From 5724c0516a87a63db51e45799906cec9edf95458 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 13 Aug 2026 20:04:48 -0400 Subject: [PATCH] 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 | 82 ++++++++++++++++++++++++++ tractor/_root.py | 75 +++++++++++++++++++---- tractor/ipc/_chan.py | 32 ++++++++-- tractor/ipc/_server.py | 10 +++- tractor/msg/types.py | 2 + tractor/runtime/_runtime.py | 1 + 6 files changed, 183 insertions(+), 19 deletions(-) diff --git a/tests/discovery/test_tpt_bind_addrs.py b/tests/discovery/test_tpt_bind_addrs.py index ae3b3437..423e041a 100644 --- a/tests/discovery/test_tpt_bind_addrs.py +++ b/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 # ------------------------------------------------------------------ diff --git a/tractor/_root.py b/tractor/_root.py index c123631b..3813b31d 100644 --- a/tractor/_root.py +++ b/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: - # TODO: this connect-and-bail forces us to have to - # carefully rewrap TCP 104-connection-reset errors as - # EOF so as to avoid propagating cancel-causing errors - # to the channel-msg loop machinery. Likely it would - # 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) - - except OSError: + probe_status = await _probe_registry( + addr=addr, + timeout=timeout, + ) + if probe_status == 'registrar': + ponged_addrs.append(addr) + elif probe_status == 'occupied': + 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: diff --git a/tractor/ipc/_chan.py b/tractor/ipc/_chan.py index a2450585..a4a4f941 100644 --- a/tractor/ipc/_chan.py +++ b/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. ''' - await self.send(aid) - peer_aid: Aid = await self.recv() + 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) - yield chan - with trio.CancelScope(shield=True): - await chan.aclose() + try: + yield chan + finally: + with trio.CancelScope(shield=True): + await chan.aclose() diff --git a/tractor/ipc/_server.py b/tractor/ipc/_server.py index 31f4c6b0..b430dfeb 100644 --- a/tractor/ipc/_server.py +++ b/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, diff --git a/tractor/msg/types.py b/tractor/msg/types.py index 4f3e33cc..334458c0 100644 --- a/tractor/msg/types.py +++ b/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 diff --git a/tractor/runtime/_runtime.py b/tractor/runtime/_runtime.py index 695090e1..67cae017 100644 --- a/tractor/runtime/_runtime.py +++ b/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