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

@@ -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.

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,10 +101,21 @@ 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.
+
     '''
+    connected_once: bool = False
+    with trio.move_on_after(timeout):
+        for attempt in range(max_attempts):
             try:
-        with trio.move_on_after(timeout) as cs:
+                with trio.move_on_after(attempt_timeout) as attempt_cs:
-            async with _connect_chan(addr.unwrap()) as chan:
+                    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',
@@ -108,19 +123,27 @@ async def _probe_registry(
                                 pid=os.getpid(),
                                 is_probe=True,
                             ),
-                    timeout=timeout,
+                            timeout=attempt_timeout,
                         )
                         if peer_aid.is_registrar is not False:
                             return 'registrar'
                         return 'occupied'
 
-        if cs.cancelled_caught:
+                if attempt_cs.cancelled_caught:
-            return 'occupied'
+                    if not connected_once:
+                        return 'absent'
 
             except OSError:
-        return 'absent'
+                return (
+                    'occupied'
+                    if connected_once
+                    else 'absent'
+                )
             except TransportClosed:
-        return 'occupied'
+                pass
+
+            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

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):
+            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'
+                    )

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,