From 38883dca0317330f6d7effe7a5f85a8034aaab02 Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 20:05:28 -0400 Subject: [PATCH] Harden the dedicated registrar example Run the registrar in its own process and prove that a sibling client discovers the service through registry lookup instead of an existing peer channel. Retry ephemeral bind collisions, publish readiness atomically and validate bounded cross-platform shutdown. Document actual duplicate name and multi-registrar ordering semantics alongside the example. Move the demo under the discovery examples and wrap process ownership in an `@acm`. Record the future public subsystem, Piker service and pytest isolation follow-ups. (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- docs/guide/discovery.rst | 85 +++-- examples/dedicated_registrar.py | 122 ------- examples/discovery/dedicated_registrar.py | 394 ++++++++++++++++++++++ 3 files changed, 450 insertions(+), 151 deletions(-) delete mode 100644 examples/dedicated_registrar.py create mode 100644 examples/discovery/dedicated_registrar.py diff --git a/docs/guide/discovery.rst b/docs/guide/discovery.rst index 4ce2d6ad..a14967e0 100644 --- a/docs/guide/discovery.rst +++ b/docs/guide/discovery.rst @@ -30,9 +30,9 @@ the registry tracks the live tree as it grows and shrinks. .. note:: Actor names are **not** enforced unique — the registry is keyed - by the full ``(name, uuid)`` pair. Name-based lookups simply - resolve to the *last* registered match, so if you boot five - actors all named ``'bob'``, you get the freshest ``'bob'`` B) + by the full ``(name, uuid)`` pair. A name lookup returns one + matching registration, but the API does not promise which match + wins. Use unique service names when selection matters. First boot: who's the registrar? -------------------------------- @@ -69,18 +69,33 @@ A dedicated registrar --------------------- That second rule — *"if a registrar answers, boot as a plain root"* — is all you need to run the registry as its own -**standalone process**, decoupled from any app tree's root. Boot -a bare ``tractor.run_daemon([], registry_addrs=[...])`` (a root -actor that does nothing but hold the registry), point your app -tree at the same ``registry_addrs``, and every actor discovers -through that *external* registrar instead of a tree-local one: +**standalone process**, decoupled from any app tree's root. In the +daemon process, enter ``open_root_actor()`` with an explicit +``registry_addrs`` and ``ensure_registry=True``; the latter makes +startup fail instead of silently joining a registrar that won the +address. Point each app tree at the address that daemon actually +bound: -.. literalinclude:: ../../examples/dedicated_registrar.py - :caption: examples/dedicated_registrar.py +.. literalinclude:: ../../examples/discovery/dedicated_registrar.py + :caption: examples/discovery/dedicated_registrar.py :language: python -This is the "registrar as a subsystem, not the root actor" shape. -Two caveats today (both tracked as #472 follow-ups): +The example's selector socket binds but deliberately never listens. +It owns the kernel-selected local address only long enough to read it, +then closes so Tractor's actual listener can bind the same address. +This is not a socket transfer: the close/rebind handoff is non-atomic, +so the example retries with a fresh candidate only when registrar +startup reports that another process claimed the released address. +Retries are bounded, and other startup failures remain visible. It +publishes the selected address only after the actor context enters. +It also performs the lookup inside a separate ``client`` actor. The +service is its sibling, not its child, so the client has no spawn-time +service channel to satisfy the local-peer fast path. The +``query_actor()`` assertion verifies that a registrar portal handled +the lookup before ``find_actor()`` makes the service RPC. + +This is the "registrar as a subsystem, not the app root actor" +shape. Two caveats today (both tracked as #472 follow-ups): ``enable_transports`` is single-proto per runtime, so a registrar can't yet serve multiple backends at once; and there's no way to spawn a registrar as a *sub*-actor of a shared tree (only as its @@ -114,13 +129,22 @@ Knobs worth knowing: - ``registry_addrs=[...]``: query specific (possibly multiple, possibly remote) registrars instead of your tree's default, -- ``only_first=False``: deliver a ``list[Portal]`` of *all* - matches found across the queried registrars instead of just the - first, +- ``only_first=True``: after all configured registrars are queried + concurrently, yield the result in the first ``registry_addrs`` + position. This is configured order, not first-reachable order, so + the result can be ``None`` even when a later registrar returned a + portal, -- ``raise_on_none=True``: raise a ``RuntimeError`` instead of - yielding ``None`` when no match is found — for when absence is - a hard error in your app. +- ``only_first=False``: when any query succeeds, yield an ordered + ``list[Portal | None]`` with one result per ``registry_addrs`` + position; misses remain ``None`` placeholders. When every query + misses, yield ``None`` instead of a list. This does not enumerate + every duplicate name in one registrar, + +- ``raise_on_none=True``: raise a ``RuntimeError`` when every + registrar query returns ``None``. With ``only_first=True`` it does + not raise merely because the first ordered result is ``None`` when + a later result is a portal. ``wait_for_actor()`` ******************** @@ -153,10 +177,11 @@ Yields a portal straight to the registrar actor itself — or a Fast paths and address preference --------------------------------- -Before doing any RPC to the registrar, every lookup first scans -the calling actor's *already-connected peers*: if you have a live -channel to an actor named ``name`` you get a portal over it -immediately, no registrar round-trip at all. +Before doing any RPC to the registrar, ``query_actor()``, +``wait_for_actor()``, and the default ``find_actor()`` lookup first +scan the calling actor's *already-connected peers*. If the caller +has a live channel to an actor named ``name``, it gets a portal over +that channel immediately, with no registrar round-trip. When a registry entry holds *multiple* addresses (a multihomed actor) the "best" one is chosen by locality: @@ -223,8 +248,8 @@ the existing registrar: Per the bootstrap rules above, if those addrs are absent this process becomes its own registrar root, so the same code works standalone and -as a tree-joiner. An occupied address that does not complete a Tractor -registrar handshake fails startup instead of being rebound. +as a tree-joiner. An occupied address that does not complete a +Tractor registrar handshake fails startup instead of being rebound. "Arbiter"? A legacy naming note ------------------------------- @@ -248,10 +273,11 @@ Very naive, very honest ----------------------- To be clear, this is a **very naive** discovery system: one -process-tree-local registrar holding a dict, no replication, no -re-election when it dies, no cross-host propagation. That's -intentional (for now); it covers the "wire up my services on this -host" case without dragging in a consensus protocol. +in-memory registrar holding a dict, no replication, no re-election +when it dies, and no automatic cross-host propagation. Separate +programs can use the same reachable registrar, as above, but must be +configured with its address. That's intentional (for now); it covers +the "wire up my services" case without a consensus protocol. On the roadmap (issue `#216`_ tracks a chunk of it): @@ -276,6 +302,7 @@ to hear from you. :class:`tractor.Registrar`. .. _gossip protocol: https://en.wikipedia.org/wiki/Gossip_protocol -.. _modern protocol: https://en.wikipedia.org/wiki/Rendezvous_protocol +.. _modern protocol: + https://en.wikipedia.org/wiki/Rendezvous_protocol .. _discovery: https://zguide.zeromq.org/docs/chapter8/#Discovery .. _#216: https://github.com/goodboy/tractor/issues/216 diff --git a/examples/dedicated_registrar.py b/examples/dedicated_registrar.py deleted file mode 100644 index d80af49b..00000000 --- a/examples/dedicated_registrar.py +++ /dev/null @@ -1,122 +0,0 @@ -''' -Run a *dedicated* registrar as its own standalone process — decoupled -from your app's root actor — and discover a service *through* it. - -Normally the registrar **is** the root actor of your tree. Here we -instead boot a separate `tractor.run_daemon([], registry_addrs=[...])` -process whose *sole* job is to be the registry, then point our app -tree at it via `registry_addrs`. Because a registrar is already -reachable at that addr, our app's root actor does NOT become one — it -registers with (and discovers through) the external daemon. That's the -"registrar as a subsystem, not the root actor" pattern. - -NB: `enable_transports` is single-proto per-runtime today (see -`tractor._root`), so this demos one transport; a genuinely -multi-backend registrar (and spawning one as a *sub*actor of a shared -tree) are future runtime work — see the #472 follow-ups. - -''' -from contextlib import suppress -import signal -import socket -import subprocess -import sys -import time - -import trio -import tractor - - -# the fixed addr the dedicated registrar binds and everyone points at. -REG_ADDR: tuple[str, int] = ('127.0.0.1', 1717) - - -def _wait_registrar_ready( - addr: tuple[str, int], - proc: subprocess.Popen, - deadline: float = 10.0, -) -> None: - ''' - Active-poll the registrar's bind addr until it accepts a - connection (proving it's booted + listening), bailing early if - the daemon proc dies during startup. - - ''' - end: float = time.monotonic() + deadline - while time.monotonic() < end: - if proc.poll() is not None: - raise RuntimeError( - f'registrar died on startup (rc={proc.returncode})' - ) - with suppress(OSError): - with socket.create_connection(addr, timeout=0.1): - return - time.sleep(0.05) - raise TimeoutError(f'registrar never came up @ {addr}') - - -async def greet() -> str: - '''A trivial service task any peer can RPC by name.''' - return f'hello from {tractor.current_actor().name}!' - - -async def app() -> None: - ''' - Point our app tree at the EXTERNAL registrar (not its own root) - via `registry_addrs`, register a named service, then discover + - RPC it purely by name. - - ''' - an: tractor.ActorNursery - async with tractor.open_nursery( - registry_addrs=[REG_ADDR], - enable_transports=['tcp'], - ) as an: - # this subactor registers with the DEDICATED registrar @ - # REG_ADDR (our root is a plain peer, not the registry). - await an.start_actor( - 'greeter', - enable_modules=[__name__], - ) - # discover it *through the external registrar*, by name only. - portal: tractor.Portal - async with tractor.wait_for_actor('greeter') as portal: - print(f'found `greeter` via dedicated registrar @ {REG_ADDR}') - print(await portal.run(greet)) - await an.cancel() - - -def main() -> None: - # boot the dedicated registrar as its own process/tree: an empty - # `enable_modules` `run_daemon()` is just a root actor that does - # nothing but hold + serve the registry. - code: str = ( - 'import tractor; ' - f'tractor.run_daemon([], registry_addrs={[REG_ADDR]!r}, ' - "enable_transports=['tcp'], loglevel='error')" - ) - registrar: subprocess.Popen = subprocess.Popen( - [sys.executable, '-c', code], - # the registry is a quiet background service; hush its logs + - # expected SIGINT-teardown traceback so the demo output stays - # focused on the discovery flow. - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - try: - _wait_registrar_ready(REG_ADDR, registrar) - print( - f'dedicated registrar up @ {REG_ADDR} ' - f'(pid {registrar.pid})' - ) - trio.run(app) - finally: - # graceful SIGINT teardown of the standalone registrar. - registrar.send_signal(signal.SIGINT) - with suppress(subprocess.TimeoutExpired): - registrar.wait(timeout=10) - print('dedicated registrar shut down') - - -if __name__ == '__main__': - main() diff --git a/examples/discovery/dedicated_registrar.py b/examples/discovery/dedicated_registrar.py new file mode 100644 index 00000000..f4386f13 --- /dev/null +++ b/examples/discovery/dedicated_registrar.py @@ -0,0 +1,394 @@ +''' +Run a dedicated registrar in a standalone process. + +The service and discovery client are sibling actors. The client has +no pre-existing channel to the service, so its lookup must use the +external registrar instead of the local-peer fast path. + +''' +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager as acm +import errno +from pathlib import Path +import signal +import socket +import subprocess +import sys +import tempfile +import time + +import trio +import tractor + + +MAX_BIND_ATTEMPTS: int = 5 + + +def _is_addr_collision(exc: BaseException) -> bool: + ''' + Return whether registrar startup lost the selected TCP address. + + Tractor can notice the collision while probing the address or + later when its listener binds. Exception groups are retryable + only when every contained failure reports the same collision. + + ''' + match exc: + case BaseExceptionGroup(exceptions=exceptions): + return bool(exceptions) and all( + _is_addr_collision(child) + for child in exceptions + ) + + case OSError() as os_error: + return ( + os_error.errno in {errno.EADDRINUSE, 10048} + or getattr(os_error, 'winerror', None) == 10048 + ) + + case RuntimeError() as runtime_error: + message: str = str(runtime_error) + return ( + 'Registry address(es) are occupied' in message + or 'registry socket(s) already bound' in message + ) + + case _: + return False + + +def run_registrar(ready_path: str) -> None: + ''' + Serve as the required registrar and report its selected address. + + The kernel selects ephemeral loopback candidates in this process. + If another process claims a released candidate first, retry with + a fresh candidate up to `MAX_BIND_ATTEMPTS`. Other startup errors + and the final collision remain visible. `ensure_registry=True` + prevents silently joining a registrar that won the address. + + ''' + ready_file: Path = Path(ready_path) + + async def serve() -> None: + ''' + Open the registrar, publish readiness, and serve forever. + + ''' + for attempt in range(1, MAX_BIND_ATTEMPTS + 1): + # This selector socket reserves and reports a + # kernel-selected candidate; it never listens and is + # not transferred to Tractor. Closing it lets + # `open_root_actor()` create its own listener on the + # same addr. The close/rebind handoff is non-atomic, + # hence the bounded collision retries. + sock: socket.socket + with socket.socket( + socket.AF_INET, + socket.SOCK_STREAM, + ) as sock: + sock.bind(('127.0.0.1', 0)) + selected: tuple[str, int] = sock.getsockname() + registry_addr: tuple[str, int] = ( + selected[0], + selected[1], + ) + + try: + actor: tractor.Actor + async with tractor.open_root_actor( + name='dedicated_registrar', + registry_addrs=[registry_addr], + enable_transports=['tcp'], + enable_modules=[], + ensure_registry=True, + loglevel='error', + ) as actor: + if not actor.is_registrar: + raise RuntimeError( + 'daemon did not become registrar' + ) + + tmp_file: Path = ready_file.with_suffix('.tmp') + tmp_file.write_text( + str(registry_addr[1]), + encoding='ascii', + ) + tmp_file.replace(ready_file) + await trio.sleep_forever() + + except BaseException as exc: + if ( + not _is_addr_collision(exc) + or attempt == MAX_BIND_ATTEMPTS + ): + raise + await trio.sleep(.05 * attempt) + + try: + trio.run(serve) + except KeyboardInterrupt: + pass + + +def _registrar_command(ready_path: Path) -> list[str]: + ''' + Build a child command that loads without running `main()`. + + `runpy.run_path()` also works when the docs test copies and + renames this example before executing it. + + ''' + module_path: str = repr(str(Path(__file__).resolve())) + function_name: str = repr('run_registrar') + ready_arg: str = repr(str(ready_path)) + code: str = ( + f'import runpy; module = runpy.run_path({module_path}); ' + f'module[{function_name}]({ready_arg})' + ) + return [sys.executable, '-c', code] + + +def _wait_registrar_ready( + ready_path: Path, + proc: subprocess.Popen, + deadline: float = 10.0, +) -> tuple[str, int]: + ''' + Wait until the child has entered its registrar actor context. + + The child atomically publishes its selected port only after + `open_root_actor()` completes. Fail early if startup crashes. + + ''' + end: float = time.monotonic() + deadline + while time.monotonic() < end: + if proc.poll() is not None: + returncode: int|None = proc.returncode + raise RuntimeError( + f'registrar exited during startup: {returncode=}' + ) + + try: + port: int = int( + ready_path.read_text(encoding='ascii') + ) + except ( + OSError, + ValueError, + ): + time.sleep(.05) + continue + + if not 0 < port < 2**16: + raise RuntimeError(f'invalid registrar port: {port!r}') + if proc.poll() is not None: + raise RuntimeError( + 'registrar exited after reporting ready' + ) + return ('127.0.0.1', port) + + raise TimeoutError('registrar did not report ready') + + +def _stop_registrar( + proc: subprocess.Popen, + graceful_timeout: float = 5.0, +) -> None: + ''' + Stop and reap the registrar, escalating after a bounded wait. + + Windows children receive `CTRL_C_EVENT` in their new process + group; POSIX children receive `SIGINT`. A child that ignores + graceful shutdown is killed, and every path finishes with + `wait()`. A non-zero child exit remains visible to the caller. + + ''' + if proc.poll() is None: + graceful_signal: int = ( + signal.CTRL_C_EVENT + if sys.platform == 'win32' + else signal.SIGINT + ) + try: + proc.send_signal(graceful_signal) + except OSError: + if proc.poll() is None: + proc.terminate() + + try: + proc.wait(timeout=graceful_timeout) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + + if proc.returncode: + raise RuntimeError( + 'registrar shutdown failed: ' + f'returncode={proc.returncode}' + ) + + +async def greet() -> str: + ''' + Return a greeting identifying the actor serving the RPC. + + ''' + actor_name: str = tractor.current_actor().name + return f'hello from {actor_name}!' + + +async def discover_and_greet( + registry_addr: tuple[str, int], +) -> tuple[str, str, str]: + ''' + Prove registrar lookup from a client without a service channel. + + The parent spawns this actor as `greeter`'s sibling. A non-`None` + registry portal from `query_actor()` proves that discovery did + not take the existing-peer fast path, which returns no registry + portal. + + ''' + service_addr: tuple[str, int]|None + registry_portal: tractor.Portal|None + async with tractor.query_actor( + 'greeter', + regaddr=registry_addr, + ) as (service_addr, registry_portal): + if registry_portal is None: + raise RuntimeError('lookup used a local service channel') + if service_addr is None: + raise RuntimeError('greeter was not registered') + + service_portal: tractor.Portal|None + async with tractor.find_actor( + 'greeter', + registry_addrs=[registry_addr], + ) as service_portal: + if service_portal is None: + raise RuntimeError('greeter disappeared before RPC') + greeting: str = await service_portal.run(greet) + + client_name: str = tractor.current_actor().name + return client_name, repr(service_addr), greeting + + +async def app(registry_addr: tuple[str, int]) -> None: + ''' + Use sibling service and client actors with an external registrar. + + Only the parent receives both spawn-time portals. The `client` + actor performs discovery in its own process and has no direct + `greeter` channel before the lookup. + + ''' + actor_nursery: tractor.ActorNursery + async with tractor.open_nursery( + registry_addrs=[registry_addr], + enable_transports=['tcp'], + ) as actor_nursery: + await actor_nursery.start_actor( + 'greeter', + enable_modules=[__name__], + ) + client_portal: tractor.Portal = ( + await actor_nursery.start_actor( + 'client', + enable_modules=[__name__], + ) + ) + result: tuple[str, str, str] = await client_portal.run( + discover_and_greet, + registry_addr=registry_addr, + ) + client_name: str + service_addr: str + greeting: str + ( + client_name, + service_addr, + greeting, + ) = result + print( + f'{client_name!r} found `greeter` through registrar ' + f'{registry_addr!r}; service address: {service_addr}\n' + f'{greeting}' + ) + await actor_nursery.cancel() + + +# TODO: Promote this lifecycle into an OTB `tractor.discovery` +# registrar subsystem. Reuse attach-or-create ownership from +# `piker.service.maybe_open_pikerd()` and named service supervision +# from `piker.service.Services`; replace the file readiness +# handshake, then use the API from `tractor._testing.pytest` to +# isolate remaining hard-coded `reg_addr` cases. +@acm +async def _open_registrar( +) -> AsyncIterator[tuple[str, int]]: + ''' + Start, publish, and reap one dedicated registrar process. + + The Windows child gets a distinct console process group so the + graceful control event targets it without interrupting this + process. + + ''' + temp_dir: str + with tempfile.TemporaryDirectory( + prefix='tractor-registrar-', + ) as temp_dir: + ready_path: Path = Path(temp_dir) / 'ready' + creationflags: int = ( + subprocess.CREATE_NEW_PROCESS_GROUP + if sys.platform == 'win32' + else 0 + ) + registrar: subprocess.Popen = subprocess.Popen( + _registrar_command(ready_path), + stdout=subprocess.DEVNULL, + creationflags=creationflags, + ) + primary_error: BaseException|None = None + try: + registry_addr: tuple[str, int] = _wait_registrar_ready( + ready_path, + registrar, + ) + print( + f'dedicated registrar ready at {registry_addr!r} ' + f'(pid {registrar.pid})' + ) + yield registry_addr + except BaseException as error: + primary_error = error + raise + finally: + try: + _stop_registrar(registrar) + except BaseException as cleanup_error: + if primary_error is None: + raise + cleanup_note: str = ( + 'registrar cleanup also failed: ' + f'{cleanup_error!r}' + ) + primary_error.add_note(cleanup_note) + print('dedicated registrar shut down') + + +async def main() -> None: + ''' + Run the external registrar and sibling discovery actors. + + ''' + registry_addr: tuple[str, int] + async with _open_registrar() as registry_addr: + await app(registry_addr) + + +if __name__ == '__main__': + trio.run(main)