From f75c9cdeab681db5b576cb972bb14be9150dab5b Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 14 Aug 2026 18:54:56 -0400 Subject: [PATCH] Traverse `BaseExceptionGroup` peer-close errors `_peer_closed_errno()` followed cause and context links but did not descend through grouped exceptions. A reset below a group could therefore escape as `trio.BrokenResourceError` instead of the normalized `TransportClosed` boundary. Walk the exception tree with cycle protection, requiring every group branch to represent peer closure before normalization. Extend the `MsgpackTransport.send()` regression to prove all-transport and mixed-failure behavior. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_server.py | 68 ++++++++++++++++++++++++++++----------- tractor/ipc/_transport.py | 61 +++++++++++++++++++++++++++-------- 2 files changed, 98 insertions(+), 31 deletions(-) diff --git a/tests/ipc/test_server.py b/tests/ipc/test_server.py index 6186454e..7689aeff 100644 --- a/tests/ipc/test_server.py +++ b/tests/ipc/test_server.py @@ -34,31 +34,49 @@ from tractor.msg.types import Aid # from ._tcp import TCPAddress -def test_send_normalizes_peer_reset(): +def test_send_normalizes_only_grouped_peer_resets(): ''' - Normalize Darwin's pre-handshake peer reset as transport closure. + Normalize only all-peer-close grouped transport failures. A UDS peer may disconnect before completing the actor handshake. Darwin can report the server's first handshake write as - `ECONNRESET`, wrapped by `trio.BrokenResourceError`; allowing - that raw error to escape cancels the daemon's shared IPC nursery. - This fake stream reproduces the exact exception chain and proves - `.send()` raises the expected `TransportClosed` boundary instead. + `ECONNRESET`, wrapped by `trio.BrokenResourceError` and potentially + nested in an `ExceptionGroup`. This fake stream first groups reset + and broken-pipe branches, proving `.send()` normalizes a complete + peer-close tree to `TransportClosed`. It then groups a reset with + an unrelated `ValueError`, proving the mixed failure remains a + `trio.BrokenResourceError` instead of hiding the application error. ''' - class ResetStream: - async def send_all(self, data: bytes) -> None: + def broken_resource(err_no: int) -> trio.BrokenResourceError: + try: + raise OSError( + err_no, + 'Peer closed', + ) + except OSError as peer_err: try: - raise OSError( - errno.ECONNRESET, - 'Connection reset by peer', - ) - except OSError as reset_err: - raise trio.BrokenResourceError from reset_err + raise trio.BrokenResourceError from peer_err + except trio.BrokenResourceError as broken_err: + return broken_err + + class GroupedFailureStream: + def __init__(self, exceptions: list[Exception]) -> None: + self.exceptions = exceptions + + async def send_all(self, data: bytes) -> None: + grouped_err = ExceptionGroup( + 'concurrent send failures', + self.exceptions, + ) + raise trio.BrokenResourceError from grouped_err async def main(): transport = object.__new__(MsgpackTransport) - transport.stream = ResetStream() + transport.stream = GroupedFailureStream([ + broken_resource(errno.ECONNRESET), + broken_resource(errno.EPIPE), + ]) transport._send_lock = trio.StrictFIFOLock() transport._laddr = 'local' transport._raddr = 'remote' @@ -70,9 +88,23 @@ def test_send_normalizes_peer_reset(): strict_types=False, ) - assert exc_info.value.src_exc.__cause__.errno == ( - errno.ECONNRESET - ) + grouped_err = exc_info.value.src_exc.__cause__ + assert isinstance(grouped_err, ExceptionGroup) + assert len(grouped_err.exceptions) == 2 + + transport.stream = GroupedFailureStream([ + ValueError('unrelated failure'), + broken_resource(errno.ECONNRESET), + ]) + with pytest.raises(trio.BrokenResourceError) as exc_info: + await transport.send( + {'probe': True}, + strict_types=False, + ) + + grouped_err = exc_info.value.__cause__ + assert isinstance(grouped_err, ExceptionGroup) + assert isinstance(grouped_err.exceptions[0], ValueError) trio.run(main) diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py index 48692f79..dfa36696 100644 --- a/tractor/ipc/_transport.py +++ b/tractor/ipc/_transport.py @@ -64,29 +64,64 @@ log = get_logger() def _peer_closed_errno(exc: BaseException) -> int|None: ''' - Find a peer-close errno in a transport exception chain. + Classify a complete transport exception tree as peer closure. + + Follow explicit cause/context links. For a `BaseExceptionGroup`, + require every child branch to resolve to a peer-close errno so an + unrelated concurrent failure is never hidden as `TransportClosed`. ''' - seen: set[int] = set() - while ( - exc - and - id(exc) not in seen - ): - seen.add(id(exc)) + def find_peer_errno( + current_exc: BaseException, + ancestors: set[int], + ) -> int|None: + exc_id: int = id(current_exc) + if exc_id in ancestors: + return None + + ancestors = ancestors | {exc_id} if ( - isinstance(exc, OSError) + isinstance(current_exc, OSError) and - exc.errno in { + current_exc.errno in { errno.ECONNRESET, errno.EPIPE, } ): - return exc.errno + return current_exc.errno - exc = exc.__cause__ or exc.__context__ + if isinstance(current_exc, BaseExceptionGroup): + child_errnos: list[int|None] = [ + find_peer_errno( + child_exc, + ancestors, + ) + for child_exc in current_exc.exceptions + ] + if all( + child_errno is not None + for child_errno in child_errnos + ): + return child_errnos[0] + return None - return None + chained_exc: BaseException|None = ( + current_exc.__cause__ + or + current_exc.__context__ + ) + if chained_exc is not None: + return find_peer_errno( + chained_exc, + ancestors, + ) + + return None + + return find_peer_errno( + exc, + set(), + ) # (codec, transport)