diff --git a/examples/multihost/tipc_cluster/host_a_srv.py b/examples/multihost/tipc_cluster/host_a_srv.py index c7b9f562..a32770d2 100644 --- a/examples/multihost/tipc_cluster/host_a_srv.py +++ b/examples/multihost/tipc_cluster/host_a_srv.py @@ -47,6 +47,10 @@ 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( enable_transports=['tipc'], registry_addrs=[reg.unwrap()], diff --git a/examples/multihost/tipc_cluster/host_b_client.py b/examples/multihost/tipc_cluster/host_b_client.py index 1e5ea7b9..d2fd73ea 100644 --- a/examples/multihost/tipc_cluster/host_b_client.py +++ b/examples/multihost/tipc_cluster/host_b_client.py @@ -40,6 +40,10 @@ 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, diff --git a/tests/ipc/test_tipc.py b/tests/ipc/test_tipc.py index e8cd677c..582209fa 100644 --- a/tests/ipc/test_tipc.py +++ b/tests/ipc/test_tipc.py @@ -892,6 +892,11 @@ def test_topology_event_scope_is_unknown(): 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), @@ -902,20 +907,24 @@ def test_topology_event_scope_is_unknown(): assert not event.addr.is_valid -def test_topology_stream_applies_backpressure(): +def test_topology_stream_surfaces_overflow(): ''' - 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. + 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 - The receive completing with the exact publication proves the - stream waits for capacity instead of discarding the transition. + 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. ''' - frame_ready = trio.Event() - class _OneEventSocket: def __init__(self): self.sent: bool = False @@ -923,26 +932,23 @@ def test_topology_stream_applies_backpressure(): 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() + raise AssertionError('reader continued after overflow') 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, + with pytest.raises( + _tipc.TIPCNameEventOverflow, + match='resubscribe and rebuild', + ) as excinfo: + await _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() + assert excinfo.value.event.kind == 'published' + with pytest.raises(trio.EndOfChannel): + await rx.receive() trio.run(main) @@ -958,6 +964,9 @@ def test_topology_timeout_closes_stream(): 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: diff --git a/tractor/ipc/_tipc.py b/tractor/ipc/_tipc.py index e286bec6..ec88627c 100644 --- a/tractor/ipc/_tipc.py +++ b/tractor/ipc/_tipc.py @@ -705,6 +705,21 @@ class MsgpackTIPCStream(MsgpackTransport): # # 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, @@ -860,8 +875,17 @@ class TIPCNameEvent( frozen=True, ): ''' - A kernel name-table transition: some service name was - published or withdrawn somewhere in the cluster. + 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 ''' kind: Literal[ @@ -882,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, @@ -984,7 +1030,10 @@ async def _stream_name_events( )) is None: continue - await tx.send(ev) + try: + tx.send_nowait(ev) + except trio.WouldBlock as src_err: + raise TIPCNameEventOverflow(ev) from src_err if ev.kind == 'timeout': return