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

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`))
wkt/big_boi_docs_472_follow_ups
Gud Boi 2026-08-29 20:05:28 -04:00
parent 97aff52050
commit 38883dca03
3 changed files with 450 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
docs/guide/discovery.rst
View File

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

122
examples/dedicated_registrar.py
View File

@ -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()

394
examples/discovery/dedicated_registrar.py 100644
View File

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