Normalize peer resets in `.send()`

A raw UDS readiness client can disconnect before the actor handshake.
Darwin reports the first server write as `ECONNRESET`, wrapped in
`trio.BrokenResourceError`; letting it escape cancels the daemon's
shared IPC nursery and makes later roots elect themselves registrar.

Walk the exception chain for `EPIPE` or `ECONNRESET` and translate
either into the existing `TransportClosed` boundary. Also handle
argument-less resource errors without raising `IndexError`.

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`))
wkt/uds_macos_473
Gud Boi 2026-08-13 18:30:44 -04:00
parent 0a3b0efcc6
commit 64e820e18e
2 changed files with 84 additions and 10 deletions

View File

@ -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']

View File

@ -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=(