diff --git a/tests/ipc/test_tipc.py b/tests/ipc/test_tipc.py index 141aa4e7..9d022b20 100644 --- a/tests/ipc/test_tipc.py +++ b/tests/ipc/test_tipc.py @@ -354,6 +354,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 # ------------------------------------------------------------------ diff --git a/tractor/ipc/_tipc.py b/tractor/ipc/_tipc.py index da22f499..4f982653 100644 --- a/tractor/ipc/_tipc.py +++ b/tractor/ipc/_tipc.py @@ -57,6 +57,7 @@ from socket import ( SOCK_STREAM, ) import struct +import sys from typing import ( AsyncGenerator, Callable, @@ -197,6 +198,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 +345,8 @@ class TIPCAddress( def with_port_id( self, - node: int, - ref: int, + node: int|None, + ref: int|None, ) -> TIPCAddress: ''' A copy annotated with an *observed* `TIPC_ADDR_ID` @@ -683,22 +688,24 @@ 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` - # 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 + 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()`. + 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 @classmethod def get_stream_addrs(