From 0b63af020e82569054ab5a421f5d84e047c662f3 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 23:40:12 -0400 Subject: [PATCH] Retry transient registrar handshakes A loaded runner can accept a registry transport while delaying its actor handshake beyond the first one-second attempt. Treating that timeout as final classifies a healthy daemon as occupied and cascades into unrelated discovery failures. Retry connected handshake failures on fresh channels under a shared three-second budget, while returning immediately for truly absent listeners. Bound every connect-plus-handshake attempt, use incremental backoff, and retain the one-second unauthenticated server limit. Also bound shielded probe-channel cleanup to 200ms and cover the real timeout, reconnection, backoff, fresh-channel, and stalled-close paths. 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/discovery/test_tpt_bind_addrs.py | 104 +++++++++++++++++++++++++ tractor/_root.py | 69 ++++++++++------ tractor/ipc/_chan.py | 15 +++- tractor/ipc/_server.py | 4 +- 4 files changed, 166 insertions(+), 26 deletions(-) diff --git a/tests/discovery/test_tpt_bind_addrs.py b/tests/discovery/test_tpt_bind_addrs.py index 423e041a..df696888 100644 --- a/tests/discovery/test_tpt_bind_addrs.py +++ b/tests/discovery/test_tpt_bind_addrs.py @@ -9,6 +9,13 @@ bind-address selection in `_root.py`: 3. Explicit bind given -> wraps via `wrap_address()` and uses them ''' +from contextlib import asynccontextmanager as acm +from unittest.mock import ( + AsyncMock, + call, + Mock, +) + import pytest import trio import tractor @@ -20,6 +27,103 @@ from tractor.discovery._multiaddr import mk_maddr from tractor._testing.addr import get_rando_addr +def test_registry_probe_retries_transient_handshake( + monkeypatch: pytest.MonkeyPatch, +): + ''' + Retry a connected registrar after transient handshake timeout. + + Loaded macOS runners can accept the transport while delaying the + actor handshake beyond one second. Treating that first timeout as + final makes a healthy remote daemon look occupied and cascades into + discovery failures. This deterministic fake fails once, succeeds + on the second complete handshake, and proves one bounded backoff. + + ''' + async def stall_handshake(**kwargs): + await trio.sleep_forever() + + first_handshake = AsyncMock(side_effect=stall_handshake) + second_handshake = AsyncMock( + return_value=tractor.msg.Aid( + name='registrar', + uuid='registrar-uuid', + pid=1234, + is_registrar=True, + ), + ) + chans = [ + Mock(_do_handshake=first_handshake), + Mock(_do_handshake=second_handshake), + ] + closed: list[object] = [] + + @acm + async def connect_chan(addr, close_timeout): + assert close_timeout == .2 + chan = chans[len(closed)] + try: + yield chan + finally: + closed.append(chan) + + sleep = AsyncMock() + monkeypatch.setattr(_root, '_connect_chan', connect_chan) + monkeypatch.setattr(_root.trio, 'sleep', sleep) + + async def main(): + status = await _root._probe_registry( + addr=wrap_address(('127.0.0.1', 1616)), + timeout=.3, + attempt_timeout=.1, + max_attempts=3, + retry_delay=.01, + ) + assert status == 'registrar' + + trio.run(main) + + first_handshake.assert_awaited_once() + second_handshake.assert_awaited_once() + assert first_handshake.await_args.kwargs['timeout'] == .1 + assert second_handshake.await_args.kwargs['timeout'] == .1 + assert closed == chans + sleep.assert_has_awaits([call(.01)]) + + +def test_probe_channel_close_is_bounded( + monkeypatch: pytest.MonkeyPatch, +): + ''' + Bound shielded channel cleanup after a registry probe. + + `_connect_chan()` shields `.aclose()` so cancellation cannot leak + ordinary channels. A stalled close previously let registry probing + exceed every connect and handshake deadline. This fake close never + completes; the explicit cleanup allowance must still return control + to the caller without cancelling its surrounding task. + + ''' + chan = Mock() + chan.aclose = AsyncMock(side_effect=trio.sleep_forever) + monkeypatch.setattr( + tractor.Channel, + 'from_addr', + AsyncMock(return_value=chan), + ) + + async def main(): + with trio.fail_after(.5): + async with _root._connect_chan( + ('127.0.0.1', 1616), + close_timeout=.01, + ): + pass + + trio.run(main) + chan.aclose.assert_awaited_once() + + def test_transport_only_listener_is_not_registrar(): ''' Require a Tractor handshake before accepting a registry address. diff --git a/tractor/_root.py b/tractor/_root.py index 3813b31d..a8316f49 100644 --- a/tractor/_root.py +++ b/tractor/_root.py @@ -88,7 +88,11 @@ _DEBUG_COMPATIBLE_BACKENDS: tuple[str, ...] = ( async def _probe_registry( addr: Address, - timeout: float = 1, + timeout: float = 3, + attempt_timeout: float = 1, + max_attempts: int = 3, + retry_delay: float = .05, + close_timeout: float = .2, ) -> Literal[ 'absent', 'occupied', @@ -97,30 +101,49 @@ async def _probe_registry( ''' Confirm an address serves the Tractor actor handshake. + Connection and handshake work share `timeout`; each attempt gets + `attempt_timeout`. Shielded channel cleanup may consume at most one + additional `close_timeout` after either deadline fires. + ''' - try: - with trio.move_on_after(timeout) as cs: - async with _connect_chan(addr.unwrap()) as chan: - peer_aid: Aid = await chan._do_handshake( - aid=Aid( - name='registry-probe', - uuid=mk_uuid(), - pid=os.getpid(), - is_probe=True, - ), - timeout=timeout, + connected_once: bool = False + with trio.move_on_after(timeout): + for attempt in range(max_attempts): + try: + with trio.move_on_after(attempt_timeout) as attempt_cs: + async with _connect_chan( + addr.unwrap(), + close_timeout=close_timeout, + ) as chan: + connected_once = True + peer_aid: Aid = await chan._do_handshake( + aid=Aid( + name='registry-probe', + uuid=mk_uuid(), + pid=os.getpid(), + is_probe=True, + ), + timeout=attempt_timeout, + ) + if peer_aid.is_registrar is not False: + return 'registrar' + return 'occupied' + + if attempt_cs.cancelled_caught: + if not connected_once: + return 'absent' + + except OSError: + return ( + 'occupied' + if connected_once + else 'absent' ) - if peer_aid.is_registrar is not False: - return 'registrar' - return 'occupied' + except TransportClosed: + pass - if cs.cancelled_caught: - return 'occupied' - - except OSError: - return 'absent' - except TransportClosed: - return 'occupied' + if attempt + 1 < max_attempts: + await trio.sleep(retry_delay * (attempt + 1)) return 'occupied' @@ -499,7 +522,7 @@ async def open_root_actor( async def ping_tpt_socket( addr: Address, - timeout: float = 1, + timeout: float = 3, ) -> None: ''' Attempt temporary connection to see if a registry is diff --git a/tractor/ipc/_chan.py b/tractor/ipc/_chan.py index a4a4f941..bf9a05f0 100644 --- a/tractor/ipc/_chan.py +++ b/tractor/ipc/_chan.py @@ -538,7 +538,8 @@ class Channel: @acm async def _connect_chan( - addr: UnwrappedAddress + addr: UnwrappedAddress, + close_timeout: float|None = None, ) -> typing.AsyncGenerator[Channel, None]: ''' Create and connect a `Channel` to the provided `addr`, disconnect @@ -553,4 +554,14 @@ async def _connect_chan( yield chan finally: with trio.CancelScope(shield=True): - await chan.aclose() + if close_timeout is None: + await chan.aclose() + else: + with trio.move_on_after(close_timeout) as close_cs: + await chan.aclose() + if close_cs.cancelled_caught: + log.warning( + f'Timed out closing channel after ' + f'{close_timeout}s\n' + f'|_{chan}\n' + ) diff --git a/tractor/ipc/_server.py b/tractor/ipc/_server.py index b430dfeb..b0ada242 100644 --- a/tractor/ipc/_server.py +++ b/tractor/ipc/_server.py @@ -72,6 +72,8 @@ if TYPE_CHECKING: log = log.get_logger() +_PRE_REG_HANDSHAKE_TIMEOUT: float = 1 + async def maybe_wait_on_canced_subs( uid: tuple[str, str], @@ -333,7 +335,7 @@ async def handle_stream_from_peer( if actor := _state.current_actor(): peer_aid: msgtypes.Aid = await chan._do_handshake( aid=actor.aid, - timeout=1, + timeout=_PRE_REG_HANDSHAKE_TIMEOUT, ) except ( TransportClosed,