Probe registrar capability in actor handshakes

Transport connect alone can select a foreign, stalled, or ordinary
actor endpoint as the registry. On macOS UDS this also exercises a
fragile connect-and-bail path before every root election.

Extend `Aid` with backward-compatible probe and registrar capability
fields, require a bounded typed handshake, and classify addresses as
absent, occupied, or confirmed registrars. Reject occupied endpoints
instead of binding over them.

Also,
- close `_connect_chan()` in `finally`
- bypass normal peer tracking for election probes
- preserve legacy registrar handshakes with unknown capability
- cover foreign listeners and idle registrar peer state

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

@@ -12,6 +12,7 @@ bind-address selection in `_root.py`:
 import pytest
 import trio
 import tractor
+from tractor import _root
 from tractor.discovery._addr import (
     wrap_address,
 )
@@ -19,6 +20,87 @@ from tractor.discovery._multiaddr import mk_maddr
 from tractor._testing.addr import get_rando_addr
 
 
+def test_transport_only_listener_is_not_registrar():
+    '''
+    Require a Tractor handshake before accepting a registry address.
+
+    The old election probe marked an address live after transport
+    connect alone. A non-Tractor listener, or a registrar still
+    failing its initial handshake, was therefore selected as the
+    remote registry. This test accepts the probe and closes it without
+    replying, then proves `open_root_actor()` ignores that endpoint and
+    elects the local actor registrar instead.
+
+    '''
+    async def transport_only_handler(
+        stream: trio.SocketStream,
+    ) -> None:
+        await stream.aclose()
+
+    async def main():
+        listeners = await trio.open_tcp_listeners(0)
+        listener = listeners[0]
+        sockname = listener.socket.getsockname()
+        reg_addr: tuple[str, int] = (
+            sockname[0],
+            sockname[1],
+        )
+
+        async with trio.open_nursery() as tn:
+            tn.start_soon(
+                trio.serve_listeners,
+                transport_only_handler,
+                listeners,
+            )
+            with pytest.raises(
+                RuntimeError,
+                match='occupied but did not answer',
+            ):
+                async with tractor.open_root_actor(
+                    registry_addrs=[reg_addr],
+                    enable_transports=['tcp'],
+                ):
+                    pytest.fail('foreign listener selected as registrar')
+
+            tn.cancel_scope.cancel()
+
+    trio.run(main)
+
+
+def test_registry_probe_preserves_no_peers_state(
+    reg_addr: tuple,
+    tpt_proto: str,
+):
+    '''
+    Keep an idle registrar peer-free after an election probe.
+
+    Probe handshakes exchange registrar capability but must not enter
+    `IPCServer._peers`. Resetting `_no_more_peers` before identifying a
+    probe left an idle registrar reporting phantom peers and delayed
+    shutdown. This test probes the live local registrar and proves its
+    peer map and no-peers event remain unchanged afterward.
+
+    '''
+    async def main():
+        async with tractor.open_root_actor(
+            registry_addrs=[reg_addr],
+            enable_transports=[tpt_proto],
+        ):
+            actor = tractor.current_actor()
+            server = actor.ipc_server
+
+            probe_status = await _root._probe_registry(
+                addr=wrap_address(reg_addr),
+            )
+            assert probe_status == 'registrar'
+
+            await trio.sleep(0)
+            assert not server._peers
+            assert server._no_more_peers.is_set()
+
+    trio.run(main)
+
+
 # ------------------------------------------------------------------
 # helpers
 # ------------------------------------------------------------------

tractor/_root.py

@@ -31,6 +31,7 @@ import sys
 from typing import (
     Any,
     Callable,
+    Literal,
 )
 import warnings
 
@@ -63,7 +64,9 @@ from .trionics import (
 )
 from ._exceptions import (
     RuntimeFailure,
+    TransportClosed,
 )
+from .msg.types import Aid
 
 
 logger = log.get_logger('tractor')
@@ -83,6 +86,45 @@ _DEBUG_COMPATIBLE_BACKENDS: tuple[str, ...] = (
 )
 
 
+async def _probe_registry(
+    addr: Address,
+    timeout: float = 1,
+) -> Literal[
+    'absent',
+    'occupied',
+    'registrar',
+]:
+    '''
+    Confirm an address serves the Tractor actor handshake.
+
+    '''
+    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,
+                )
+                if peer_aid.is_registrar is not False:
+                    return 'registrar'
+                return 'occupied'
+
+        if cs.cancelled_caught:
+            return 'occupied'
+
+    except OSError:
+        return 'absent'
+    except TransportClosed:
+        return 'occupied'
+
+    return 'occupied'
+
+
 # TODO: stick this in a `@acm` defined in `devx.debug`?
 # -[ ] also maybe consider making this a `wrapt`-deco to
 #     save an indent level?
@@ -453,6 +495,7 @@ async def open_root_actor(
 
         # closed into below ping task-func
         ponged_addrs: list[Address] = []
+        occupied_addrs: list[Address] = []
 
         async def ping_tpt_socket(
             addr: Address,
@@ -467,18 +510,15 @@ async def open_root_actor(
             server is listening at that addr.
 
             '''
-            try:
+            probe_status = await _probe_registry(
-                # TODO: this connect-and-bail forces us to have to
+                addr=addr,
-                # carefully rewrap TCP 104-connection-reset errors as
+                timeout=timeout,
-                # EOF so as to avoid propagating cancel-causing errors
+            )
-                # to the channel-msg loop machinery. Likely it would
+            if probe_status == 'registrar':
-                # be better to eventually have a "discovery" protocol
-                # with basic handshake instead?
-                with trio.move_on_after(timeout):
-                    async with _connect_chan(addr.unwrap()):
                 ponged_addrs.append(addr)
-
+            elif probe_status == 'occupied':
-            except OSError:
+                occupied_addrs.append(addr)
+            else:
                 # ?TODO, make this a "discovery" log level?
                 logger.info(
                     f'No root-actor registry found @ {addr!r}\n'
@@ -496,6 +536,17 @@ async def open_root_actor(
                     addr,
                 )
 
+        if (
+            not ponged_addrs
+            and
+            occupied_addrs
+        ):
+            raise RuntimeError(
+                f'Registry address(es) are occupied but did not '
+                f'answer as Tractor registrars!\n'
+                f'occupied_addrs: {occupied_addrs!r}\n'
+            )
+
         if tpt_bind_addrs is None:
             tpt_bind_addrs: list[Address] = []
         else:

tractor/ipc/_chan.py

@@ -495,6 +495,7 @@ class Channel:
     async def _do_handshake(
         self,
         aid: Aid,
+        timeout: float|None = None,
 
     ) -> Aid:
         '''
@@ -505,8 +506,27 @@ class Channel:
         "actor model" parlance.
 
         '''
+        try:
+            with trio.fail_after(
+                timeout if timeout is not None else float('inf')
+            ):
                 await self.send(aid)
                 peer_aid: Aid = await self.recv()
+            if not isinstance(peer_aid, Aid):
+                raise TypeError(
+                    f'Expected {Aid!r}, received {peer_aid!r}'
+                )
+        except (
+            MsgTypeError,
+            TypeError,
+            UnicodeDecodeError,
+            trio.TooSlowError,
+        ) as handshake_err:
+            raise TransportClosed(
+                message='Peer sent an invalid actor handshake!\n',
+                src_exc=handshake_err,
+                loglevel='warning',
+            ) from handshake_err
         log.runtime(
             f'Received hanshake with peer\n'
             f'<= {peer_aid.reprol(sin_uuid=False)}\n'
@@ -529,6 +549,8 @@ async def _connect_chan(
 
     '''
     chan = await Channel.from_addr(addr)
+    try:
         yield chan
+    finally:
         with trio.CancelScope(shield=True):
             await chan.aclose()

tractor/ipc/_server.py

@@ -316,8 +316,6 @@ async def handle_stream_from_peer(
       )
 
     '''
-    server._no_more_peers = trio.Event()  # unset by making new
-
     # TODO, debug_mode tooling for when hackin this lower layer?
     # with debug.maybe_open_crash_handler(
     #     pdb=True,
@@ -335,6 +333,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,
             )
     except (
         TransportClosed,
@@ -364,6 +363,13 @@ async def handle_stream_from_peer(
         )
         return
 
+    # Registry election probes need only the server's `Aid` capability
+    # response; never register them as ordinary RPC peers.
+    if peer_aid.is_probe:
+        return
+
+    server._no_more_peers = trio.Event()  # unset by making new
+
     uid: tuple[str, str] = (
         peer_aid.name,
         peer_aid.uuid,

tractor/msg/types.py

@@ -144,6 +144,8 @@ class Aid(
     name: str
     uuid: str
     pid: int|None = None
+    is_registrar: bool|None = None
+    is_probe: bool = False
 
     # TODO? can/should we extend this field set?
     # -[ ] use built-in support for UUIDs? `uuid.UUID` which has

tractor/runtime/_runtime.py

@@ -259,6 +259,7 @@ class Actor:
             name=name,
             uuid=uuid,
             pid=os.getpid(),
+            is_registrar=self.is_registrar,
         )
         self._task: trio.Task|None = None