From be6f9e86d155837d1fab5ded06c431bb0029e063 Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 18 Aug 2026 13:44:17 -0400 Subject: [PATCH] Make `TIPCNameEvent` delivery lossless Stop labeling topology events with caller-supplied scope that the kernel never reports. Event addresses now carry an explicit unknown scope instead of fabricated reachability. Apply memory-channel backpressure rather than silently dropping publish/withdraw transitions, and close the stream after delivering the terminal event from a finite subscription. Review: PR #493 (copilot-pull-request-reviewer[bot],goodboy) https://github.com/goodboy/tractor/pull/493 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- tests/ipc/test_tipc.py | 117 ++++++++++++++++++++++++++++++++++++++++- tractor/ipc/_tipc.py | 34 +++++------- 2 files changed, 127 insertions(+), 24 deletions(-) diff --git a/tests/ipc/test_tipc.py b/tests/ipc/test_tipc.py index 9d022b20..e8cd677c 100644 --- a/tests/ipc/test_tipc.py +++ b/tests/ipc/test_tipc.py @@ -39,6 +39,9 @@ from tractor.ipc._tipc import ( TIPC_IMPORTANCE, TIPC_NAME_UNKNOWN, TIPC_NODE_SCOPE, + TIPC_PUBLISHED, + TIPC_SCOPE_UNKNOWN, + TIPC_SUBSCR_TIMEOUT, TIPC_ZONE_SCOPE, TRACTOR_STYPE, MsgpackTIPCStream, @@ -817,6 +820,22 @@ def test_topology_struct_layouts(): assert len(sub) == 28 +def _pack_topology_event( + event: int, + instance: int = 71, +) -> bytes: + return struct.pack( + _tipc._EVENT_FMT, + event, + instance, + instance, + 123, + 456, + 0, 0, 0, 0, 0, + b'\0' * 8, + ) + + def test_wait_forever_is_masked_for_packing(): ''' Python exposes `TIPC_WAIT_FOREVER` as `-1`, which `struct` @@ -848,7 +867,6 @@ def test_decode_name_event_rejects_junk(): assert _tipc._decode_name_event( b'\x00' * 12, stype=TRACTOR_STYPE, - scope=TIPC_CLUSTER_SCOPE, ) is None bogus: bytes = struct.pack( @@ -861,10 +879,104 @@ def test_decode_name_event_rejects_junk(): assert _tipc._decode_name_event( bogus, stype=TRACTOR_STYPE, - scope=TIPC_CLUSTER_SCOPE, ) is None +def test_topology_event_scope_is_unknown(): + ''' + Neither `struct tipc_subscr` nor `struct tipc_event` carries a + publication scope. The old decoder copied caller context into + each `TIPCAddress`, falsely presenting cluster scope as kernel- + observed data even for a node-scoped publisher. + + Decode a valid publication and assert its address uses the + explicit unknown sentinel rather than inventing reachability. + + ''' + event: _tipc.TIPCNameEvent|None = _tipc._decode_name_event( + _pack_topology_event(TIPC_PUBLISHED), + stype=TRACTOR_STYPE, + ) + assert event is not None + assert event.addr._scope == TIPC_SCOPE_UNKNOWN + assert not event.addr.is_valid + + +def test_topology_stream_applies_backpressure(): + ''' + A full memory channel used to drop topology transitions, letting + a future push registry continue with permanently incomplete + state. A zero-capacity channel deterministically exercises that + pressure: `send_nowait()` loses the event, while `await send()` + synchronizes the reader and consumer without loss. + + The receive completing with the exact publication proves the + stream waits for capacity instead of discarding the transition. + + ''' + frame_ready = trio.Event() + + class _OneEventSocket: + def __init__(self): + self.sent: bool = False + + async def recv(self, size: int) -> bytes: + if not self.sent: + self.sent = True + # `Event.set()` does not checkpoint, so the old + # `send_nowait()` runs before the consumer below. + frame_ready.set() + return _pack_topology_event(TIPC_PUBLISHED) + await trio.sleep_forever() + + async def main() -> None: + tx, rx = trio.open_memory_channel(0) + async with trio.open_nursery() as tn: + tn.start_soon( + _tipc._stream_name_events, + _OneEventSocket(), + TRACTOR_STYPE, + tx, + ) + await frame_ready.wait() + with trio.fail_after(1): + event: _tipc.TIPCNameEvent = await rx.receive() + assert event.kind == 'published' + tn.cancel_scope.cancel() + + trio.run(main) + + +def test_topology_timeout_closes_stream(): + ''' + A finite kernel subscription expires after its timeout event. + The old reader forwarded that event and waited forever for a + second frame that could never arrive, so consumers blocked on a + channel that looked live despite having no subscription. + + Feed one timeout frame directly to the reader; receiving that + event followed by `EndOfChannel` proves the stream terminates at + the kernel subscription boundary. + + ''' + class _TimeoutSocket: + async def recv(self, size: int) -> bytes: + return _pack_topology_event(TIPC_SUBSCR_TIMEOUT) + + async def main() -> None: + tx, rx = trio.open_memory_channel(1) + await _tipc._stream_name_events( + _TimeoutSocket(), + TRACTOR_STYPE, + tx, + ) + assert (await rx.receive()).kind == 'timeout' + with pytest.raises(trio.EndOfChannel): + await rx.receive() + + trio.run(main) + + @requires_tipc def test_topology_reports_publish_and_withdraw(): ''' @@ -898,6 +1010,7 @@ def test_topology_reports_publish_and_withdraw(): for ev in got: assert ev.addr._instance == addr._instance assert ev.addr._stype == addr._stype + assert ev.addr._scope == TIPC_SCOPE_UNKNOWN assert isinstance(ev.ref, int) # both transitions name the SAME publisher port assert got[0].ref == got[1].ref diff --git a/tractor/ipc/_tipc.py b/tractor/ipc/_tipc.py index 4f982653..e286bec6 100644 --- a/tractor/ipc/_tipc.py +++ b/tractor/ipc/_tipc.py @@ -167,6 +167,10 @@ _tipc_reserved_stypes: range = range(0, 64) # See `MsgpackTIPCStream.get_stream_addrs()` and plan 01 ยง3.4. TIPC_NAME_UNKNOWN: int = -1 +# The topology event wire format carries no publication scope. +# Never promote caller context into observed address data. +TIPC_SCOPE_UNKNOWN: int = 0 + # XXX, the kernel default (`TIPC_LOW_IMPORTANCE`), i.e. today this # is a no-op knob preserving stock behaviour. # @@ -178,6 +182,7 @@ TIPC_NAME_UNKNOWN: int = -1 TRACTOR_DEF_IMPORTANCE: int = TIPC_LOW_IMPORTANCE _scope_names: dict[int, str] = { + TIPC_SCOPE_UNKNOWN: 'unknown', TIPC_ZONE_SCOPE: 'zone', TIPC_CLUSTER_SCOPE: 'cluster', TIPC_NODE_SCOPE: 'node', @@ -911,7 +916,6 @@ def _mk_subscr( def _decode_name_event( raw: bytes, stype: int, - scope: int, ) -> TIPCNameEvent|None: ''' Decode one `struct tipc_event`, or `None` if it's a runt/ @@ -946,14 +950,12 @@ def _decode_name_event( # NOTE, `tractor` only ever publishes *singleton* ranges # (`lower == upper`) so the lower bound IS the instance. # - # XXX the event carries NO scope โ€” the name-table doesn't - # report it โ€” so we echo back the subscription's own. Fine - # for our use (we subscribe per-scope) but don't mistake - # it for observed data. + # XXX the event carries NO scope โ€” do not fabricate one + # from caller context and present it as observed data. addr=TIPCAddress( _stype=stype, _instance=found_lower, - _scope=scope, + _scope=TIPC_SCOPE_UNKNOWN, ), node=node, ref=ref, @@ -963,7 +965,6 @@ def _decode_name_event( async def _stream_name_events( sock, stype: int, - scope: int, tx: trio.MemorySendChannel, ) -> None: ''' @@ -980,21 +981,12 @@ async def _stream_name_events( 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' - ) + await tx.send(ev) + if ev.kind == 'timeout': + return except ( trio.ClosedResourceError, @@ -1013,7 +1005,6 @@ async def open_topology_events( 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[ @@ -1042,7 +1033,7 @@ async def open_topology_events( topsrv_addr = TIPCAddress( _stype=TIPC_TOP_SRV, _instance=TIPC_TOP_SRV, - _scope=scope, + _scope=TIPC_CLUSTER_SCOPE, ) sock = trio_socket.socket( AF_TIPC, @@ -1085,7 +1076,6 @@ async def open_topology_events( _stream_name_events, sock, stype, - scope, tx, ) try: