diff --git a/tests/ipc/test_server.py b/tests/ipc/test_server.py index 1d63bd1b..01ddb222 100644 --- a/tests/ipc/test_server.py +++ b/tests/ipc/test_server.py @@ -3,6 +3,7 @@ High-level `.ipc._server` unit tests. ''' from __future__ import annotations +import errno import pytest import trio @@ -14,6 +15,8 @@ from tractor import ( from tractor._testing.addr import ( get_rando_addr, ) +from tractor._exceptions import TransportClosed +from tractor.ipc._transport import MsgpackTransport # TODO, use/check-roundtripping with some of these wrapper types? # # from .._addr import Address @@ -23,6 +26,49 @@ from tractor._testing.addr import ( # from ._tcp import TCPAddress +def test_send_normalizes_peer_reset(): + ''' + Normalize Darwin's pre-handshake peer reset as transport closure. + + A raw UDS readiness client connects and immediately disconnects. + Darwin reports 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. + + ''' + class ResetStream: + async def send_all(self, data: bytes) -> None: + try: + raise OSError( + errno.ECONNRESET, + 'Connection reset by peer', + ) + except OSError as reset_err: + raise trio.BrokenResourceError from reset_err + + async def main(): + transport = object.__new__(MsgpackTransport) + transport.stream = ResetStream() + transport._send_lock = trio.StrictFIFOLock() + transport._laddr = 'local' + transport._raddr = 'remote' + transport._task = trio.lowlevel.current_task() + + with pytest.raises(TransportClosed) as exc_info: + await transport.send( + {'probe': True}, + strict_types=False, + ) + + assert exc_info.value.src_exc.__cause__.errno == ( + errno.ECONNRESET + ) + + trio.run(main) + + @pytest.mark.parametrize( '_tpt_proto', ['uds', 'tcp'] diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py index dd3be179..48692f79 100644 --- a/tractor/ipc/_transport.py +++ b/tractor/ipc/_transport.py @@ -33,6 +33,7 @@ from collections.abc import ( AsyncGenerator, AsyncIterator, ) +import errno import struct import trio @@ -61,6 +62,33 @@ if TYPE_CHECKING: log = get_logger() +def _peer_closed_errno(exc: BaseException) -> int|None: + ''' + Find a peer-close errno in a transport exception chain. + + ''' + seen: set[int] = set() + while ( + exc + and + id(exc) not in seen + ): + seen.add(id(exc)) + if ( + isinstance(exc, OSError) + and + exc.errno in { + errno.ECONNRESET, + errno.EPIPE, + } + ): + return exc.errno + + exc = exc.__cause__ or exc.__context__ + + return None + + # (codec, transport) MsgTransportKey = tuple[str, str] @@ -443,23 +471,23 @@ class MsgpackTransport(MsgTransport): trans_err = _re tpt_name: str = f'{type(self).__name__!r}' - trans_err_msg: str = trans_err.args[0] + trans_err_msg: str = ( + str(trans_err.args[0]) + if trans_err.args + else '' + ) by_whom: str = { 'another task closed this fd': 'locally', 'this socket was already closed': 'by peer', }.get(trans_err_msg) match trans_err: - # XXX, specifc to UDS transport and its, - # well, "speediness".. XD - # |_ likely todo with races related to how fast - # the socket is setup/torn-down on linux - # as it pertains to rando pings from the - # `.discovery` subsys and protos. + # UDS peers can disconnect before handshake. + # Linux normally reports `EPIPE`; Darwin reports + # `ECONNRESET` for the same expected closure. case trio.BrokenResourceError() if ( - '[Errno 32] Broken pipe' - in - trans_err_msg + _peer_closed_errno(trans_err) + is not None ): tpt_closed = TransportClosed.from_src_exc( message=(