From 33a040b312f7f482384917b1b0e882d1846d1f32 Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 15 Aug 2026 01:39:37 -0400 Subject: [PATCH] Add `open_topology_events()`, push-based discovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second half of layer B: an `@acm` yielding a `trio` receive-chan of `TIPCNameEvent` fed by a nursery-spawned reader on a `SOCK_SEQPACKET` conn to `TIPC_TOP_SRV`. This is the bit that makes #378's "end game cluster proto" claim real — the kernel *tells* us when any actor anywhere in the cluster publishes or withdraws a service name, so a registrar never has to poll `find_actor()`. Groundwork for the push registry in `discovery/_registry.py` (gh #184, #216). Deats, - `filt` selects granularity; `TIPC_SUB_SERVICE` is one event per *name*, `TIPC_SUB_PORTS` one per *publisher* — the latter makes the §2.3 duplicate-name/round-robin crosstalk case externally observable, which is how a push-registry could ever detect it. - a full event buf **drops** w/ a loud warning rather than blocking the reader; stalling it just backs up the kernel's own queue and loses the event less visibly. - `SOCK_SEQPACKET` is fine here bc this sock never goes through `MsgpackTransport` — the contract's "`SOCK_STREAM` only" rule is about `MsgTransport` streams, not this. XXX teardown order is load-bearing: cancel the nursery BEFORE closing the fd. `.close()`ing out from under a pending `.recv()` races — trio's retry can land on an already-freed fd and raise a bare `OSError(EBADF)` instead of the `ClosedResourceError` the reader guards for, which then escapes the nursery as an eg. (this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`)) --- tests/ipc/test_tipc.py | 106 ++++++++++++++++++++++++++++ tractor/ipc/_tipc.py | 153 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 258 insertions(+), 1 deletion(-) diff --git a/tests/ipc/test_tipc.py b/tests/ipc/test_tipc.py index a51e28b9..88ab4971 100644 --- a/tests/ipc/test_tipc.py +++ b/tests/ipc/test_tipc.py @@ -43,6 +43,7 @@ from tractor.ipc._tipc import ( TIPCAddress, instance_from_seed, is_tipc_available, + open_topology_events, start_listener, ) @@ -703,3 +704,108 @@ def test_decode_name_event_rejects_junk(): stype=TRACTOR_STYPE, scope=TIPC_CLUSTER_SCOPE, ) is None + + +@requires_tipc +def test_topology_reports_publish_and_withdraw(): + ''' + "Publishing a bind IS registration" — but *observed from the + outside*, by the kernel pushing us the name-table transition. + + This is the whole point of layer B: cluster-wide service + (de)registration with NO registrar actor and NO polling. + + ''' + async def main(): + addr: TIPCAddress = TIPCAddress.get_random() + got: list = [] + + async with open_topology_events( + stype=addr._stype, + ) as events: + # publish.. + lstnr = await start_listener(addr=addr) + with trio.fail_after(5): + got.append(await events.receive()) + + # ..then withdraw + lstnr.socket.close() + with trio.fail_after(5): + got.append(await events.receive()) + + kinds: list[str] = [ev.kind for ev in got] + assert kinds == ['published', 'withdrawn'] + + for ev in got: + assert ev.addr._instance == addr._instance + assert ev.addr._stype == addr._stype + assert isinstance(ev.ref, int) + # both transitions name the SAME publisher port + assert got[0].ref == got[1].ref + assert 'published' in repr(got[0]) + + trio.run(main) + + +@requires_tipc +def test_topology_acm_closes_cleanly(): + ''' + The `@acm` must tear down w/o leaking its reader task — exit + closes the sock, the reader's `.recv()` raises + `ClosedResourceError`, the nursery collapses. + + Also assert an *unconsumed* subscription doesn't wedge exit. + + ''' + async def main(): + addr: TIPCAddress = TIPCAddress.get_random() + with trio.fail_after(10): + async with open_topology_events(stype=addr._stype): + lstnr = await start_listener(addr=addr) + # deliberately DON'T read the event + await trio.sleep(0.2) + lstnr.socket.close() + + # ..and a second open/close round still works + async with open_topology_events( + stype=addr._stype, + ) as events: + assert events is not None + + trio.run(main) + + +@requires_tipc +def test_topology_sub_ports_reports_each_publisher(): + ''' + `TIPC_SUB_PORTS` gives one event per *publisher*, so the + duplicate-name/round-robin case (§2.3) is observable from the + topology feed — which is how a push-registry would ever + detect silent crosstalk. + + ''' + async def main(): + addr: TIPCAddress = TIPCAddress.get_random() + + async with open_topology_events( + stype=addr._stype, + filt=_tipc.TIPC_SUB_PORTS, + ) as events: + first = await start_listener(addr=addr) + second = await start_listener(addr=addr) + + refs: set[int] = set() + with trio.fail_after(5): + for _ in range(2): + ev = await events.receive() + assert ev.kind == 'published' + assert ev.addr._instance == addr._instance + refs.add(ev.ref) + + # two DISTINCT publisher ports on one service name + assert len(refs) == 2 + + first.socket.close() + second.socket.close() + + trio.run(main) diff --git a/tractor/ipc/_tipc.py b/tractor/ipc/_tipc.py index a66bea5a..89d5eca6 100644 --- a/tractor/ipc/_tipc.py +++ b/tractor/ipc/_tipc.py @@ -45,15 +45,20 @@ Normative refs are the kernel sources (the tipc.io docs are stale), ''' from __future__ import annotations from contextlib import ( + asynccontextmanager as acm, contextmanager as cm, ) import errno from hashlib import blake2b import os import socket -from socket import SOCK_STREAM +from socket import ( + SOCK_SEQPACKET, + SOCK_STREAM, +) import struct from typing import ( + AsyncGenerator, Callable, ClassVar, Literal, @@ -945,3 +950,149 @@ def _decode_name_event( node=node, ref=ref, ) + + +async def _stream_name_events( + sock, + stype: int, + scope: int, + tx: trio.MemorySendChannel, +) -> None: + ''' + Read `struct tipc_event`s off a topology-server socket until + it's closed, forwarding decoded ones to `tx`. + + ''' + try: + while True: + raw: bytes = await sock.recv(_EVENT_SIZE) + if not raw: + return + + if (ev := _decode_name_event( + raw, + stype=stype, + scope=scope, + )) is None: + continue + + try: + tx.send_nowait(ev) + except trio.WouldBlock: + # XXX drop rather than block: stalling this reader + # backs up the kernel's own queue and we'd lose the + # event anyway, just less visibly. + log.warning( + f'TIPC topology event buffer full, dropping!\n' + f'{ev}\n' + f' |_raise `buf_size` or consume faster\n' + ) + + except ( + trio.ClosedResourceError, + trio.BrokenResourceError, + ): + # normal `@acm` teardown: the socket was closed under us + return + + finally: + tx.close() + + +@acm +async def open_topology_events( + stype: int = TRACTOR_STYPE, + lower: int = 0, + upper: int = 0xFFFF_FFFF, + filt: int = TIPC_SUB_SERVICE, + scope: int = TIPC_CLUSTER_SCOPE, + timeout: int = TIPC_WAIT_FOREVER, + buf_size: int = 64, +) -> AsyncGenerator[ + trio.MemoryReceiveChannel[TIPCNameEvent], + None, +]: + ''' + Subscribe to kernel name-table events for `stype` and yield a + `trio` receive-channel of `TIPCNameEvent`. + + This is *push-based* service discovery: the kernel tells us + when any actor in the cluster publishes or withdraws a name, + so a registrar never has to poll `find_actor()`. + + `filt` selects the granularity, + - `TIPC_SUB_SERVICE`: one event per *name* becoming + (un)available — "does anyone serve this?" + - `TIPC_SUB_PORTS`: one event per *publisher*, so N binders on + one name give N events. Verified. + + NOTE this socket is `SOCK_SEQPACKET` and never goes through + `MsgpackTransport` — the contract's "`SOCK_STREAM` only" + constraint is about `MsgTransport` streams, not this. + + ''' + topsrv_addr = TIPCAddress( + _stype=TIPC_TOP_SRV, + _instance=TIPC_TOP_SRV, + _scope=scope, + ) + sock = trio_socket.socket( + AF_TIPC, + SOCK_SEQPACKET, + ) + with _close_on_error(sock): + with _reraise_as_connerr( + src_excs=(OSError,), + addr=topsrv_addr, + ): + await sock.connect(( + TIPC_ADDR_NAME, + TIPC_TOP_SRV, + TIPC_TOP_SRV, + 0, # domain: 0 == "anywhere in scope" + )) + await sock.send( + _mk_subscr( + stype=stype, + lower=lower, + upper=upper, + filt=filt, + timeout=timeout, + ) + ) + + log.info( + f'Subscribed to TIPC name-table events\n' + f'[>\n' + f' |_stype: 0x{stype:08x}\n' + f' |_range: [{lower}, {upper}]\n' + f' |_filter: {filt}\n' + ) + tx: trio.MemorySendChannel + rx: trio.MemoryReceiveChannel + tx, rx = trio.open_memory_channel(buf_size) + try: + async with trio.open_nursery() as tn: + tn.start_soon( + _stream_name_events, + sock, + stype, + scope, + tx, + ) + try: + yield rx + finally: + # XXX cancel BEFORE closing the fd! + # + # `.close()`ing out from under a pending + # `.recv()` races: trio's retry can land on an + # already-freed fd and raise a bare + # `OSError(EBADF)` instead of the + # `ClosedResourceError` the reader guards for — + # which then escapes as an eg from the nursery. + # Cancelling first makes teardown deterministic. + tn.cancel_scope.cancel() + finally: + sock.close() + await rx.aclose()