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

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

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
+        # XXX the event carries NO scope — do not fabricate one
-        # report it — so we echo back the subscription's own. Fine
+        # from caller context and present it as observed data.
-        # for our use (we subscribe per-scope) but don't mistake
-        # it for 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:
+            await tx.send(ev)
-                tx.send_nowait(ev)
+            if ev.kind == 'timeout':
-            except trio.WouldBlock:
+                return
-                # 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,
@@ -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: