goodboy
/
tractor
1
0
Fork 2
Code Issues 5 Pull Requests 6 Packages Projects Releases Wiki Activity

Move registrar probes to `discovery._api` 

Registrar election and multi-address probing are discovery-protocol
concerns, but their implementation lived in root-runtime ignition.
Move the bounded handshake probe and its concurrent address
classifier into `discovery._api`, leaving `open_root_actor()` to
consume the classified results.

Update probe tests for the canonical module and clarify that the
daemon-fixture regressions directly exercise their sibling
`conftest` plugin.

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-14 18:43:36 -04:00
parent f454cefe56
commit 75cda1933c
4 changed files with 135 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tests/discovery/test_daemon_fixture.py
View File

@ -1,6 +1,10 @@
'''
Discovery daemon fixture regressions.
This module imports private helpers from the sibling
`tests.discovery.conftest` plugin to exercise that fixture machinery
directly, rather than testing a production `tractor` API.
'''
from unittest.mock import (
call,

16
tests/discovery/test_tpt_bind_addrs.py
View File

@ -2,7 +2,8 @@
`open_root_actor(tpt_bind_addrs=...)` test suite.
Verify all three runtime code paths for explicit IPC-server
bind-address selection in `_root.py`:
bind-address selection in `_root.py` and registry probing in
`discovery._api`:
1. Non-registrar, no explicit bind -> random addrs from registry proto
2. Registrar, no explicit bind -> binds to registry_addrs
@ -19,11 +20,12 @@ from unittest.mock import (
import pytest
import trio
import tractor
from tractor import _root
from tractor.discovery import _api
from tractor.discovery._addr import (
wrap_address,
)
from tractor.discovery._multiaddr import mk_maddr
from tractor.ipc import _connect_chan
from tractor._testing.addr import get_rando_addr
@ -68,11 +70,11 @@ def test_registry_probe_retries_transient_handshake(
closed.append(chan)
sleep = AsyncMock()
monkeypatch.setattr(_root, '_connect_chan', connect_chan)
monkeypatch.setattr(_root.trio, 'sleep', sleep)
monkeypatch.setattr(_api, '_connect_chan', connect_chan)
monkeypatch.setattr(_api.trio, 'sleep', sleep)
async def main():
status = await _root._probe_registry(
status = await _api._probe_registry(
addr=wrap_address(('127.0.0.1', 1616)),
timeout=.3,
attempt_timeout=.1,
@ -114,7 +116,7 @@ def test_probe_channel_close_is_bounded(
async def main():
with trio.fail_after(.5):
async with _root._connect_chan(
async with _connect_chan(
('127.0.0.1', 1616),
close_timeout=.01,
):
@ -193,7 +195,7 @@ def test_registry_probe_preserves_no_peers_state(
actor = tractor.current_actor()
server = actor.ipc_server
probe_status = await _root._probe_registry(
probe_status = await _api._probe_registry(
addr=wrap_address(reg_addr),
)
assert probe_status == 'registrar'

115
tractor/_root.py
View File

@ -31,7 +31,6 @@ import sys
from typing import (
Any,
Callable,
Literal,
)
import warnings
@ -48,9 +47,7 @@ from .devx import (
from .spawn import _spawn
from .runtime import _state
from . import log
from .ipc import (
_connect_chan,
)
from .discovery._api import _probe_registry_addrs
from .discovery._addr import (
Address,
UnwrappedAddress,
@ -64,9 +61,7 @@ from .trionics import (
)
from ._exceptions import (
RuntimeFailure,
TransportClosed,
)
from .msg.types import Aid
logger = log.get_logger('tractor')
@ -86,68 +81,6 @@ _DEBUG_COMPATIBLE_BACKENDS: tuple[str, ...] = (
)
async def _probe_registry(
addr: Address,
timeout: float = 3,
attempt_timeout: float = 1,
max_attempts: int = 3,
retry_delay: float = .05,
close_timeout: float = .2,
) -> Literal[
'absent',
'occupied',
'registrar',
]:
'''
Confirm an address serves the Tractor actor handshake.
Connection and handshake work share `timeout`; each attempt gets
`attempt_timeout`. Shielded cleanup may add up to `close_timeout`
per attempted channel.
'''
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'
)
except TransportClosed:
pass
if attempt + 1 < max_attempts:
await trio.sleep(retry_delay * (attempt + 1))
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?
@ -516,46 +449,12 @@ async def open_root_actor(
from .devx._stackscope import enable_stack_on_sig
enable_stack_on_sig()
# closed into below ping task-func
ponged_addrs: list[Address] = []
occupied_addrs: list[Address] = []
async def ping_tpt_socket(
addr: Address,
timeout: float = 3,
) -> None:
'''
Probe with a bounded Tractor actor handshake.
Classify the address as a registrar, occupied by a
non-registrar, or absent.
'''
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'
)
# !TODO, this is basically just another (abstract)
# happy-eyeballs, so we should try for formalize it somewhere
# in a `.[_]discovery` ya?
#
async with trio.open_nursery() as tn:
for uw_addr in uw_reg_addrs:
addr: Address = wrap_address(uw_addr)
tn.start_soon(
ping_tpt_socket,
addr,
)
ponged_addrs: list[Address]
occupied_addrs: list[Address]
(
ponged_addrs,
occupied_addrs,
) = await _probe_registry_addrs(uw_reg_addrs)
if (
not ponged_addrs

115
tractor/discovery/_api.py
View File

@ -21,14 +21,18 @@ management of (service) actors.
"""
from __future__ import annotations
import ipaddress
import os
import socket
from typing import (
AsyncGenerator,
AsyncContextManager,
Literal,
TYPE_CHECKING,
)
from contextlib import asynccontextmanager as acm
import trio
from tractor.log import get_logger
from ..trionics import (
gather_contexts,
@ -40,6 +44,7 @@ from ..ipc._uds import UDSAddress
from ._addr import (
UnwrappedAddress,
Address,
mk_uuid,
wrap_address,
)
from ..runtime._portal import (
@ -52,6 +57,7 @@ from ..runtime._state import (
_runtime_vars,
_def_tpt_proto,
)
from ..msg.types import Aid
if TYPE_CHECKING:
from ..runtime._runtime import Actor
@ -60,6 +66,115 @@ if TYPE_CHECKING:
log = get_logger()
async def _probe_registry(
addr: Address,
timeout: float = 3,
attempt_timeout: float = 1,
max_attempts: int = 3,
retry_delay: float = .05,
close_timeout: float = .2,
) -> Literal[
'absent',
'occupied',
'registrar',
]:
'''
Confirm an address serves the Tractor actor handshake.
Connection and handshake work share `timeout`; each attempt gets
`attempt_timeout`. Shielded cleanup may add up to `close_timeout`
per attempted channel.
'''
from .._exceptions import TransportClosed
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'
)
except TransportClosed:
pass
if attempt + 1 < max_attempts:
await trio.sleep(retry_delay * (attempt + 1))
return 'occupied'
async def _probe_registry_addrs(
addrs: list[UnwrappedAddress],
timeout: float = 3,
) -> tuple[
list[Address],
list[Address],
]:
'''
Concurrently classify candidate registrar addresses.
Return confirmed registrar addresses followed by addresses occupied
by non-registrar or unresponsive Tractor peers.
'''
registrar_addrs: list[Address] = []
occupied_addrs: list[Address] = []
async def probe_addr(addr: Address) -> None:
probe_status = await _probe_registry(
addr=addr,
timeout=timeout,
)
if probe_status == 'registrar':
registrar_addrs.append(addr)
elif probe_status == 'occupied':
occupied_addrs.append(addr)
else:
# ?TODO, make this a "discovery" log level?
log.info(
f'No root-actor registry found @ {addr!r}\n'
)
async with trio.open_nursery() as nursery:
for unwrapped_addr in addrs:
nursery.start_soon(
probe_addr,
wrap_address(unwrapped_addr),
)
return (
registrar_addrs,
occupied_addrs,
)
def _is_local_addr(addr: Address) -> bool:
'''
Determine whether `addr` is reachable on the