Abort `TIPC` topology streams on overflow

Keep `_stream_name_events()` non-blocking so a slow memory-channel
consumer cannot back up the kernel topology queue. Raise
`TIPCNameEventOverflow` and end the subscription rather than drop a
transition or let the socket reader stall. Discovery consumers must
then resubscribe and rebuild their name-table view.

Also,
- document topology semantics and scope with Linux references
- diagram the `.connect()`/`.getpeername()` withdrawal schedules
- explain the child-service and callable requirements in the
  two-host example

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

examples/multihost/tipc_cluster/host_a_srv.py

@@ -47,14 +47,20 @@ async def main() -> None:
     reg: TIPCAddress = TIPCAddress.get_root()
     print(f'host A publishing {reg}')
 
-    async with tractor.open_root_actor(
+    # A registrar root is not an entry in its own
-        name='host_a',
+    # `Registrar._registry`, so host B cannot discover that root by
+    # its actor name. `open_nursery()` keeps the root as registrar
+    # while the named child below registers as the dialable service.
+    async with tractor.open_nursery(
         enable_transports=['tipc'],
         registry_addrs=[reg.unwrap()],
+    ) as an:
+        await an.start_actor(
+            'host_a',
             enable_modules=[__name__],
-    ):
+        )
         print(
-            'registrar up — `tipc nametable show` on EITHER host\n'
+            'host_a up — `tipc nametable show` on EITHER host\n'
             'should now list this service. ctrl-c to stop.'
         )
         await trio.sleep_forever()

examples/multihost/tipc_cluster/host_b_client.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 
 import trio
 import tractor
+from host_a_srv import echo
 from tractor.ipc._tipc import (
     TIPCAddress,
     is_tipc_available,
@@ -39,9 +40,13 @@ async def main() -> None:
                     ' |_does `tipc link list` show the peer?\n'
                 )
 
+            # `.open_context()` derives a `NamespacePath` from a
+            # callable. Importing `echo` also loads its module path
+            # locally, while host A's `enable_modules` authorizes the
+            # corresponding remote callable.
             async with (
                 ptl.open_context(
-                    'host_a_srv:echo',
+                    echo,
                 ) as (ctx, _),
                 ctx.open_stream() as stream,
             ):

tests/ipc/test_tipc.py

@@ -9,11 +9,11 @@ the pure address-algebra cases run everywhere.
 from __future__ import annotations
 import errno
 import struct
+from types import SimpleNamespace
 from socket import (
     SOCK_STREAM,
     SOL_SOCKET,
     SO_ACCEPTCONN,
-    SOL_TIPC,
 )
 
 import pytest
@@ -27,8 +27,10 @@ from tractor.discovery._multiaddr import (
     parse_maddr,
 )
 from tractor.ipc import _tipc
+from tractor.ipc._uds import UDSAddress
 from tractor.ipc._tipc import (
     AF_TIPC,
+    SOL_TIPC,
     TIPC_ADDR_ID,
     TIPC_ADDR_NAME,
     TIPC_CLUSTER_SCOPE,
@@ -37,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,
@@ -114,6 +119,29 @@ def test_zone_scope_normalized_to_cluster():
     assert addr.is_valid
 
 
+def test_maddr_parse_normalizes_and_reports_bad_input():
+    '''
+    The interim `/tipc` parser must route through
+    `TIPCAddress.from_addr()` so deprecated zone scope is normalized
+    exactly like every other unwrapped-address entrypoint.
+
+    A malformed segment count previously leaked the tuple-unpacking
+    `ValueError`, which gave callers no indication that the TIPC
+    multiaddr grammar itself was invalid.
+
+    '''
+    addr: TIPCAddress = parse_maddr(
+        f'/tipc/{TRACTOR_STYPE}/7/{TIPC_ZONE_SCOPE}'
+    )
+    assert addr._scope == TIPC_CLUSTER_SCOPE
+
+    with pytest.raises(
+        ValueError,
+        match='Invalid TIPC multiaddr',
+    ):
+        parse_maddr('/tipc/not-enough-segments')
+
+
 def test_addr_from_bare_port_id_raises():
     '''
     A `TIPC_ADDR_ID` 5-tuple carries no service-name so it can
@@ -207,6 +235,43 @@ def test_get_random_collision_resistance():
     assert all(addr.is_valid for addr in addrs)
 
 
+def test_get_random_keys_live_actors_by_uuid(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    '''
+    TIPC names are cluster-wide while PIDs are host-local. Hashing
+    only `(actor name, pid)` therefore made same-named actors with
+    equal PIDs on different hosts publish one service name, where
+    TIPC silently round-robins connects between them.
+
+    Hold the actor name and PID fixed while changing only its UUID;
+    distinct instances prove the globally unique identity field is
+    now part of the derivation.
+
+    '''
+    monkeypatch.setattr(_tipc.os, 'getpid', lambda: 1616)
+
+    def get_addr(uuid: str) -> TIPCAddress:
+        actor = SimpleNamespace(
+            aid=Aid(
+                name='worker',
+                uuid=uuid,
+                pid=1616,
+            )
+        )
+        monkeypatch.setattr(
+            _tipc,
+            'current_actor',
+            lambda **kwargs: actor,
+        )
+        return TIPCAddress.get_random()
+
+    first: TIPCAddress = get_addr('actor-uuid-a')
+    second: TIPCAddress = get_addr('actor-uuid-b')
+    assert first._instance != second._instance
+    assert get_addr('actor-uuid-a')._instance == first._instance
+
+
 def test_get_random_honors_bindspace():
     addr: TIPCAddress = TIPCAddress.get_random(
         bindspace=TIPC_NODE_SCOPE,
@@ -235,6 +300,12 @@ def test_wrap_address_dispatches_on_the_proto_key():
         'tipc', TRACTOR_STYPE, 1616, TIPC_CLUSTER_SCOPE,
     )
 
+    # A UDS directory named `tipc` is still a valid classic
+    # 2-element address, not a malformed proto-keyed TIPC one.
+    uds: UDSAddress = wrap_address(('tipc', 'actor.sock'))
+    assert isinstance(uds, UDSAddress)
+    assert UDSAddress.unwrapped_type == tuple[str, str]
+
 
 def test_maddr_roundtrip():
     '''
@@ -286,6 +357,97 @@ def test_eafnosupport_is_actionable_connerr(
     assert type(excinfo.value.__cause__) is OSError
 
 
+def test_availability_probe_is_linux_only(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    '''
+    The fallback `AF_TIPC` integer keeps imports portable but can
+    alias an unrelated socket family on another OS. Socket creation
+    alone must never turn that numeric collision into a false TIPC
+    capability result.
+
+    Simulate Darwin with a socket constructor that would otherwise
+    succeed; proving it is never called pins the platform gate ahead
+    of the syscall probe.
+
+    '''
+    socket_called: bool = False
+
+    def fake_socket(*args, **kwargs):
+        nonlocal socket_called
+        socket_called = True
+
+    monkeypatch.setattr(_tipc, '_tipc_avail', None)
+    monkeypatch.setattr(_tipc.sys, 'platform', 'darwin')
+    monkeypatch.setattr(_tipc.socket, 'socket', fake_socket)
+
+    assert not is_tipc_available()
+    assert not socket_called
+
+
+def test_connect_reuses_tolerant_peer_observation(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    '''
+    A TIPC peer can withdraw immediately after `.connect()`. The
+    transport constructor already tolerates that race, but the dial
+    path used to call `getpeername()` a second time and leak a raw
+    `ENOTCONN` after an otherwise successful connection.
+
+    The fake socket permits exactly the constructor's first peer
+    lookup. A second lookup raises deterministically, so successful
+    construction proves `connect_to()` reuses the tolerant observed
+    address and transfers socket ownership only after setup.
+
+    '''
+    class _DropAfterObservation:
+        def __init__(self):
+            self.peer_lookups: int = 0
+            self.closed: bool = False
+
+        def setsockopt(self, *args) -> None:
+            pass
+
+        async def connect(self, addr) -> None:
+            pass
+
+        def getsockname(self) -> tuple[int, int, int, int, int]:
+            return (TIPC_ADDR_ID, 1, 2, 0, TIPC_CLUSTER_SCOPE)
+
+        def getpeername(self) -> tuple[int, int, int, int, int]:
+            self.peer_lookups += 1
+            if self.peer_lookups > 1:
+                raise OSError(errno.ENOTCONN, 'peer withdrew')
+            return (TIPC_ADDR_ID, 3, 4, 0, TIPC_CLUSTER_SCOPE)
+
+        def close(self) -> None:
+            self.closed = True
+
+    async def main() -> None:
+        sock = _DropAfterObservation()
+        monkeypatch.setattr(
+            _tipc.trio_socket,
+            'socket',
+            lambda *args: sock,
+        )
+        monkeypatch.setattr(
+            _tipc.trio,
+            'SocketStream',
+            lambda raw_sock: SimpleNamespace(socket=raw_sock),
+        )
+
+        addr: TIPCAddress = TIPCAddress.get_root()
+        stream: MsgpackTIPCStream = (
+            await MsgpackTIPCStream.connect_to(addr)
+        )
+        assert sock.peer_lookups == 1
+        assert stream.raddr.unwrap() == addr.unwrap()
+        assert stream.raddr.maybe_ref == 4
+        assert not sock.closed
+
+    trio.run(main)
+
+
 # ------------------------------------------------------------------
 # kernel-touching
 # ------------------------------------------------------------------
@@ -658,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`
@@ -689,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(
@@ -702,10 +879,113 @@ 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.
+
+    TIPC address types and publication scopes:
+    https://docs.kernel.org/networking/tipc.html
+    Event wire format (which has no scope field):
+    https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h
+
+    '''
+    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_surfaces_overflow():
+    '''
+    Topology transitions are `TIPC_PUBLISHED`/`TIPC_WITHDRAWN`
+    changes to the set of ports matching a subscribed name sequence,
+    as emitted by the kernel topology server:
+    https://github.com/torvalds/linux/blob/master/net/tipc/topsrv.c
+
+    Blocking this reader on `tx.send()` stops draining the topology
+    socket and merely pushes pressure into the kernel queue. Silently
+    dropping with `send_nowait()` is worse: a push registry keeps a
+    stale view without knowing it missed a transition.
+
+    A zero-capacity channel deterministically fills the user-space
+    boundary. The dedicated overflow error proves the reader remains
+    non-blocking while forcing consumers to resubscribe and rebuild
+    instead of trusting incomplete state.
+
+    '''
+    class _OneEventSocket:
+        def __init__(self):
+            self.sent: bool = False
+
+        async def recv(self, size: int) -> bytes:
+            if not self.sent:
+                self.sent = True
+                return _pack_topology_event(TIPC_PUBLISHED)
+            raise AssertionError('reader continued after overflow')
+
+    async def main() -> None:
+        tx, rx = trio.open_memory_channel(0)
+        with pytest.raises(
+            _tipc.TIPCNameEventOverflow,
+            match='resubscribe and rebuild',
+        ) as excinfo:
+            await _tipc._stream_name_events(
+                _OneEventSocket(),
+                TRACTOR_STYPE,
+                tx,
+            )
+        assert excinfo.value.event.kind == 'published'
+        with pytest.raises(trio.EndOfChannel):
+            await rx.receive()
+
+    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.
+
+    Subscription timeout semantics:
+    https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h
+
+    '''
+    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():
     '''
@@ -739,6 +1019,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/discovery/_addr.py

@@ -265,7 +265,11 @@ def wrap_address(
         #
         # NOTE, a bare seq-pattern matches `list` too, which is
         # what `msgpack` decodes our tuples back to.
-        case ('tipc', *_):
+        case (
+            ('tipc', int(), int())
+            |
+            ('tipc', int(), int(), int())
+        ):
             cls = TIPCAddress
 
         # classic network socket-address as tuple/list

tractor/discovery/_multiaddr.py

@@ -70,6 +70,9 @@ def mk_maddr(
     dispatching on the `.proto_key` to build the correct
     multiaddr-spec-compliant protocol path.
 
+    Return a `Multiaddr` for registered protocols. TIPC remains
+    an interim `str` until its upstream multiaddr protocol lands.
+
     '''
     proto_key: str = addr.proto_key
     maddr_proto: str|None = _tpt_proto_to_maddr.get(proto_key)
@@ -132,12 +135,18 @@ def parse_maddr(
     # XXX MUST come before `Multiaddr()` which rejects the
     # not-yet-registered `/tipc` proto name outright.
     if maddr_str.startswith(_tipc_maddr_prefix):
+        try:
             _, _, stype, instance, scope = maddr_str.split('/')
-        return TIPCAddress(
+            return TIPCAddress.from_addr((
-            _stype=int(stype),
+                'tipc',
-            _instance=int(instance),
+                int(stype),
-            _scope=int(scope),
+                int(instance),
-        )
+                int(scope),
+            ))
+        except (TypeError, ValueError) as src_err:
+            raise ValueError(
+                f'Invalid TIPC multiaddr: {maddr_str!r}'
+            ) from src_err
 
     maddr = Multiaddr(maddr_str)
     proto_names: list[str] = [

tractor/ipc/_tipc.py

@@ -57,6 +57,7 @@ from socket import (
     SOCK_STREAM,
 )
 import struct
+import sys
 from typing import (
     AsyncGenerator,
     Callable,
@@ -166,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.
 #
@@ -177,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',
@@ -197,6 +203,10 @@ def is_tipc_available() -> bool:
 
     '''
     global _tipc_avail
+    if sys.platform != 'linux':
+        _tipc_avail = False
+        return _tipc_avail
+
     if _tipc_avail is None:
         try:
             socket.socket(
@@ -340,8 +350,8 @@ class TIPCAddress(
 
     def with_port_id(
         self,
-        node: int,
+        node: int|None,
-        ref: int,
+        ref: int|None,
     ) -> TIPCAddress:
         '''
         A copy annotated with an *observed* `TIPC_ADDR_ID`
@@ -369,10 +379,11 @@ class TIPCAddress(
         between them (verified). I.e. a collision manifests as
         *silent crosstalk*, not an error.
 
-        So the instance is a `blake2b` digest of a per-call-unique
+        So the instance is a `blake2b` digest of the actor UUID, or
-        seed, giving a well-spread 32b value. Being a pure fn of the
+        a per-call token outside a live runtime, giving a well-spread
-        seed it is also *reproducible*, which the (follow-up)
+        32b value. Being a pure fn of the seed it is also
-        registrar-less discovery fast-path wants.
+        *reproducible*, which the (follow-up) registrar-less
+        discovery fast-path wants.
 
         NOTE the residual risk is birthday-bounded: ~1.2e-2 for 10k
         names sharing one `_stype`. See plan 01 §9 for the
@@ -384,7 +395,7 @@ class TIPCAddress(
             err_on_no_runtime=False,
         )
         if actor:
-            seed: str = f'{actor.aid.name}@{pid}'
+            seed: str = '.'.join(actor.aid.uid)
         else:
             if is_root_process():
                 prefix: str = 'no_runtime_root'
@@ -687,15 +698,32 @@ class MsgpackTIPCStream(MsgpackTransport):
                 prefix_size=prefix_size,
                 codec=codec,
             )
-        # XXX, the dialling side is the ONLY side that knows the
+            # XXX, the dialling side is the ONLY side that knows
-        # peer's *service name* (a port-id can't be reversed into
+            # the peer's *service name* (a port-id can't be
-        # one), so re-assert it over the observed-only `._raddr`
+            # reversed into one), so re-assert it over the
-        # that `.get_stream_addrs()` just derived.
+            # observed-only `._raddr` derived above.
             #
-        # Same move as `MsgpackUDSStream.connect_to()`s peer-pid
+            # Reuse that tolerant observation: a peer can withdraw
-        # re-assign.
+            # between `.connect()` and a second `.getpeername()`.
+            #
+            # Peer / kernel                 dial task
+            #      |                            |
+            #      |<----- connect(name) -------|
+            #      |------ connected ---------->|  (A)
+            #      |<----- getpeername() -------|  (B, ctor)
+            #      |-- port-id or `ENOTCONN` -->|
+            #      X withdraws                  |
+            #      |<----- getpeername() -------|  (C, old code)
+            #      |------ `ENOTCONN` --------->|
+            #
+            # If withdrawal precedes B, `_maybe_sockaddr()` records
+            # no port-id and the handshake observes the close. If it
+            # follows B, that first observation remains useful. The
+            # removed C lookup created the only raw-`ENOTCONN` race.
+            observed_raddr: TIPCAddress = tpt_stream._raddr
             tpt_stream._raddr = destaddr.with_port_id(
-            *_port_id(sock.getpeername()),
+                node=observed_raddr.maybe_node,
+                ref=observed_raddr.maybe_ref,
             )
             return tpt_stream
 
@@ -847,8 +875,17 @@ class TIPCNameEvent(
     frozen=True,
 ):
     '''
-    A kernel name-table transition: some service name was
+    A kernel name-table transition, not an application msg.
-    published or withdrawn somewhere in the cluster.
+
+    A transition reports that the set of TIPC ports matching a
+    subscribed service-name sequence changed: `TIPC_PUBLISHED`
+    adds a matching `(node, ref)`, `TIPC_WITHDRAWN` removes one,
+    and `TIPC_SUBSCR_TIMEOUT` ends a finite subscription.
+
+    Wire definitions:
+    https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h
+    Topology server:
+    https://github.com/torvalds/linux/blob/master/net/tipc/topsrv.c
 
     '''
     kind: Literal[
@@ -869,6 +906,28 @@ class TIPCNameEvent(
         )
 
 
+class TIPCNameEventOverflow(RuntimeError):
+    '''
+    The user-space event buffer lost authoritative continuity.
+
+    The socket reader must not remain blocked while its kernel
+    topology subscription is live. Raising aborts that subscription
+    and forces the consumer to resubscribe and rebuild its name-table
+    view instead of using silently stale state.
+
+    '''
+    def __init__(
+        self,
+        event: TIPCNameEvent,
+    ) -> None:
+        self.event = event
+        super().__init__(
+            'TIPC topology event buffer overflowed; '
+            'resubscribe and rebuild the name-table view.\n'
+            f'lost event: {event!r}'
+        )
+
+
 def _mk_subscr(
     stype: int,
     lower: int,
@@ -903,7 +962,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/
@@ -938,14 +996,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,
@@ -955,7 +1011,6 @@ def _decode_name_event(
 async def _stream_name_events(
     sock,
     stype: int,
-    scope: int,
     tx: trio.MemorySendChannel,
 ) -> None:
     '''
@@ -972,21 +1027,15 @@ 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:
+            except trio.WouldBlock as src_err:
-                # XXX drop rather than block: stalling this reader
+                raise TIPCNameEventOverflow(ev) from src_err
-                # backs up the kernel's own queue and we'd lose the
+            if ev.kind == 'timeout':
-                # event anyway, just less visibly.
+                return
-                log.warning(
-                    f'TIPC topology event buffer full, dropping!\n'
-                    f'{ev}\n'
-                    f' |_raise `buf_size` or consume faster\n'
-                )
 
     except (
         trio.ClosedResourceError,
@@ -1005,7 +1054,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[
@@ -1034,7 +1082,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,
@@ -1077,7 +1125,6 @@ async def open_topology_events(
                 _stream_name_events,
                 sock,
                 stype,
-                scope,
                 tx,
             )
             try:

tractor/ipc/_uds.py

@@ -114,7 +114,7 @@ class UDSAddress(
     # -[ ] need to check what other mult-transport frameworks do
     #     like zmq, nng, uri-spec et al!
     proto_key: ClassVar[str] = 'uds'
-    unwrapped_type: ClassVar[type] = tuple[str, int]
+    unwrapped_type: ClassVar[type] = tuple[str, str]
     def_bindspace: ClassVar[Path] = get_rt_dir()
 
     # NOTE, `getsockname()` answers the sock-file path as a `str`
@@ -176,7 +176,7 @@ class UDSAddress(
                     f'{addr!r}\n'
                 )
 
-    def unwrap(self) -> tuple[str, int]:
+    def unwrap(self) -> tuple[str, str]:
         # XXX NOTE, since this gets passed DIRECTLY to
         # `.ipc._uds.open_unix_socket_w_passcred()`
         return (