examples/multihost/tipc_cluster/host_a_srv.py

@@ -47,20 +47,14 @@ async def main() -> None:
     reg: TIPCAddress = TIPCAddress.get_root()
     print(f'host A publishing {reg}')
 
-    # A registrar root is not an entry in its own
-    # `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(
+    async with tractor.open_root_actor(
+        name='host_a',
         enable_transports=['tipc'],
         registry_addrs=[reg.unwrap()],
-    ) as an:
-        await an.start_actor(
-            'host_a',
-            enable_modules=[__name__],
-        )
+        enable_modules=[__name__],
+    ):
         print(
-            'host_a up — `tipc nametable show` on EITHER host\n'
+            'registrar 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,7 +16,6 @@ from __future__ import annotations
 
 import trio
 import tractor
-from host_a_srv import echo
 from tractor.ipc._tipc import (
     TIPCAddress,
     is_tipc_available,
@@ -40,13 +39,9 @@ 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(
-                    echo,
+                    'host_a_srv: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,10 +27,8 @@ 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,
@@ -39,9 +37,6 @@ 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,
@@ -119,29 +114,6 @@ 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
@@ -235,43 +207,6 @@ 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,
@@ -300,12 +235,6 @@ 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():
     '''
@@ -357,97 +286,6 @@ 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
 # ------------------------------------------------------------------
@@ -820,22 +658,6 @@ 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`
@@ -867,6 +689,7 @@ 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(
@@ -879,113 +702,10 @@ 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():
     '''
@@ -1019,7 +739,6 @@ 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,11 +265,7 @@ def wrap_address(
         #
         # NOTE, a bare seq-pattern matches `list` too, which is
         # what `msgpack` decodes our tuples back to.
-        case (
-            ('tipc', int(), int())
-            |
-            ('tipc', int(), int(), int())
-        ):
+        case ('tipc', *_):
             cls = TIPCAddress
 
         # classic network socket-address as tuple/list

tractor/discovery/_multiaddr.py

@@ -70,9 +70,6 @@ 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)
@@ -135,18 +132,12 @@ 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.from_addr((
-                'tipc',
-                int(stype),
-                int(instance),
-                int(scope),
-            ))
-        except (TypeError, ValueError) as src_err:
-            raise ValueError(
-                f'Invalid TIPC multiaddr: {maddr_str!r}'
-            ) from src_err
+        _, _, stype, instance, scope = maddr_str.split('/')
+        return TIPCAddress(
+            _stype=int(stype),
+            _instance=int(instance),
+            _scope=int(scope),
+        )
 
     maddr = Multiaddr(maddr_str)
     proto_names: list[str] = [

tractor/ipc/_tipc.py

@@ -57,7 +57,6 @@ from socket import (
     SOCK_STREAM,
 )
 import struct
-import sys
 from typing import (
     AsyncGenerator,
     Callable,
@@ -167,10 +166,6 @@ _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.
 #
@@ -182,7 +177,6 @@ TIPC_SCOPE_UNKNOWN: int = 0
 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',
@@ -203,10 +197,6 @@ 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(
@@ -350,8 +340,8 @@ class TIPCAddress(
 
     def with_port_id(
         self,
-        node: int|None,
-        ref: int|None,
+        node: int,
+        ref: int,
     ) -> TIPCAddress:
         '''
         A copy annotated with an *observed* `TIPC_ADDR_ID`
@@ -379,11 +369,10 @@ class TIPCAddress(
         between them (verified). I.e. a collision manifests as
         *silent crosstalk*, not an error.
 
-        So the instance is a `blake2b` digest of the actor UUID, or
-        a per-call token outside a live runtime, giving a well-spread
-        32b value. Being a pure fn of the seed it is also
-        *reproducible*, which the (follow-up) registrar-less
-        discovery fast-path wants.
+        So the instance is a `blake2b` digest of a per-call-unique
+        seed, giving a well-spread 32b value. Being a pure fn of the
+        seed it is also *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
@@ -395,7 +384,7 @@ class TIPCAddress(
             err_on_no_runtime=False,
         )
         if actor:
-            seed: str = '.'.join(actor.aid.uid)
+            seed: str = f'{actor.aid.name}@{pid}'
         else:
             if is_root_process():
                 prefix: str = 'no_runtime_root'
@@ -693,39 +682,22 @@ class MsgpackTIPCStream(MsgpackTransport):
                     destaddr._scope,
                 ))
 
-            tpt_stream = MsgpackTIPCStream(
-                trio.SocketStream(sock),
-                prefix_size=prefix_size,
-                codec=codec,
-            )
-            # XXX, the dialling side is the ONLY side that knows
-            # the peer's *service name* (a port-id can't be
-            # reversed into one), so re-assert it over the
-            # observed-only `._raddr` derived above.
-            #
-            # Reuse that tolerant observation: a peer can withdraw
-            # 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(
-                node=observed_raddr.maybe_node,
-                ref=observed_raddr.maybe_ref,
-            )
-            return tpt_stream
+        tpt_stream = MsgpackTIPCStream(
+            trio.SocketStream(sock),
+            prefix_size=prefix_size,
+            codec=codec,
+        )
+        # XXX, the dialling side is the ONLY side that knows the
+        # peer's *service name* (a port-id can't be reversed into
+        # one), so re-assert it over the observed-only `._raddr`
+        # that `.get_stream_addrs()` just derived.
+        #
+        # Same move as `MsgpackUDSStream.connect_to()`s peer-pid
+        # re-assign.
+        tpt_stream._raddr = destaddr.with_port_id(
+            *_port_id(sock.getpeername()),
+        )
+        return tpt_stream
 
     @classmethod
     def get_stream_addrs(
@@ -875,17 +847,8 @@ class TIPCNameEvent(
     frozen=True,
 ):
     '''
-    A kernel name-table transition, not an application msg.
-
-    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
+    A kernel name-table transition: some service name was
+    published or withdrawn somewhere in the cluster.
 
     '''
     kind: Literal[
@@ -906,28 +869,6 @@ 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,
@@ -962,6 +903,7 @@ 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/
@@ -996,12 +938,14 @@ 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 — do not fabricate one
-        # from caller context and present it as observed data.
+        # 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.
         addr=TIPCAddress(
             _stype=stype,
             _instance=found_lower,
-            _scope=TIPC_SCOPE_UNKNOWN,
+            _scope=scope,
         ),
         node=node,
         ref=ref,
@@ -1011,6 +955,7 @@ def _decode_name_event(
 async def _stream_name_events(
     sock,
     stype: int,
+    scope: int,
     tx: trio.MemorySendChannel,
 ) -> None:
     '''
@@ -1027,15 +972,21 @@ 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 as src_err:
-                raise TIPCNameEventOverflow(ev) from src_err
-            if ev.kind == 'timeout':
-                return
+            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,
@@ -1054,6 +1005,7 @@ 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[
@@ -1082,7 +1034,7 @@ async def open_topology_events(
     topsrv_addr = TIPCAddress(
         _stype=TIPC_TOP_SRV,
         _instance=TIPC_TOP_SRV,
-        _scope=TIPC_CLUSTER_SCOPE,
+        _scope=scope,
     )
     sock = trio_socket.socket(
         AF_TIPC,
@@ -1125,6 +1077,7 @@ 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, str]
+    unwrapped_type: ClassVar[type] = tuple[str, int]
     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, str]:
+    def unwrap(self) -> tuple[str, int]:
         # XXX NOTE, since this gets passed DIRECTLY to
         # `.ipc._uds.open_unix_socket_w_passcred()`
         return (