goodboy
/
tractor
1
0
Fork 2
Code Issues 5 Pull Requests 6 Packages Projects Releases Wiki Activity

Add `MsgpackTIPCStream`, the `AF_TIPC` `MsgTransport` 

Wire `.connect_to()` (dial by service name), `.connected()`
and `.get_stream_addrs()` on top of `MsgpackTransport` so
`trio.SocketStream` + the existing `<I`-prefix framing carry
`msgpack` msgs over TIPC unchanged.

XXX both ends of a connected TIPC sock answer `TIPC_ADDR_ID`
port-ids and a port-id carries NO service name, so,
- the *dialling* side re-asserts the name it actually dialled
  over `._raddr` (same move as `MsgpackUDSStream`s peer-pid
  re-assign),
- the *accepting* side keeps a `TIPC_NAME_UNKNOWN` sentinel
  plus the observed `(node, ref)`. It doesn't need more — the
  `Aid` from `._do_handshake()` already carries the peer's
  logical identity.

Also normalize dial failures: TIPC answers an unpublished-name
lookup with `EHOSTUNREACH`, which python maps to a **bare**
`OSError` and NOT a `ConnectionError` subtype the way
`ECONNREFUSED` maps to `ConnectionRefusedError`. The
discovery-ping path needs the `ConnectionError` shape, so the
`_reraise_as_connerr()` wrap is load-bearing, not polish.

XXX ALSO tolerate a dead peer in `.get_stream_addrs()`!
Unlike tcp/uds — where the kernel keeps answering the peer
addr until *we* close — TIPC answers `ENOTCONN` once the peer
is gone. Since `MsgpackTransport.__init__()` calls
`.get_stream_addrs()` (via `Channel.from_stream()`) BEFORE the
handshake, an unguarded `OSError` there escapes
`handle_stream_from_peer()`s handshake tolerance (contract §4)
and tears down the WHOLE actor. Any connect-then-drop peer — a
port scan, a liveness probe, a cancelled dial — was a remote
actor-kill. A dead peer must cost us an addr, not the runtime.

Deats,
- `TIPC_IMPORTANCE` exposed as a `.connect_to()` kwarg — TIPC
  can rank a conn's traffic under congestion, which no other
  backend can do. Defaulted to the kernel default for now;
  wiring the parent<->child chan to `HIGH` is a follow-up.
- `TIPC_DEST_DROPPABLE = 0` so undeliverable msgs surface as
  errors instead of being silently dropped.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
wkt/pr493_review
Gud Boi 2026-08-14 09:52:25 -04:00
parent e3089ba356
commit 8c0ae140cd
2 changed files with 451 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

223
tests/ipc/test_tipc.py
View File

@ -12,20 +12,27 @@ from socket import (
SOCK_STREAM,
SOL_SOCKET,
SO_ACCEPTCONN,
SOL_TIPC,
)
import pytest
import trio
from tractor.msg.types import Aid
from tractor.ipc import _tipc
from tractor.ipc._tipc import (
AF_TIPC,
TIPC_ADDR_ID,
TIPC_ADDR_NAME,
TIPC_CLUSTER_SCOPE,
TIPC_DEST_DROPPABLE,
TIPC_HIGH_IMPORTANCE,
TIPC_IMPORTANCE,
TIPC_NAME_UNKNOWN,
TIPC_NODE_SCOPE,
TIPC_ZONE_SCOPE,
TRACTOR_STYPE,
MsgpackTIPCStream,
TIPCAddress,
instance_from_seed,
is_tipc_available,
@ -333,6 +340,222 @@ def test_getsockname_is_a_port_id_not_the_bound_name():
trio.run(main)
@requires_tipc
def test_msgpack_roundtrip_over_service_name():
'''
Two `trio` tasks in ONE proc exchange `msgpack`-framed msgs
over a TIPC service name — no `tractor` runtime involved.
Also pins the `(laddr, raddr)` story of plan 01 §3.4a: the
dialling side knows the name it dialled, the accepting side
only ever learns a port-id.
'''
async def main():
addr: TIPCAddress = TIPCAddress.get_random()
lstnr = await start_listener(addr=addr)
ping = Aid(name='doggy', uuid='abc123', pid=1)
pong = Aid(name='kitty', uuid='def456', pid=2)
srv_got: list = []
async def _serve():
stream = await lstnr.accept()
tpt = MsgpackTIPCStream(stream)
# accepting side can NOT know the peer's service name
assert not tpt.raddr.is_valid
assert tpt.raddr._instance == TIPC_NAME_UNKNOWN
assert tpt.raddr.maybe_ref is not None
srv_got.append(await tpt.recv())
await tpt.send(pong)
await stream.aclose()
async with trio.open_nursery() as tn:
tn.start_soon(_serve)
await trio.sleep(0.05)
cli: MsgpackTIPCStream = await MsgpackTIPCStream.connect_to(
destaddr=addr,
)
assert cli.connected()
# dialling side DOES know the name, and it round-trips
assert cli.raddr.is_valid
assert cli.raddr.unwrap() == addr.unwrap()
assert cli.raddr.maybe_ref is not None
await cli.send(ping)
assert await cli.recv() == pong
await cli.stream.aclose()
assert srv_got == [ping]
lstnr.socket.close()
trio.run(main)
@requires_tipc
def test_dial_unpublished_name_is_connerr():
'''
Contract §4: a dead/absent peer must normalize to
`ConnectionError` — the discovery-ping path depends on it.
XXX TIPC answers an unpublished-name lookup with
`EHOSTUNREACH`, which python maps to a **bare** `OSError` (NOT
a `ConnectionError` subtype the way `ECONNREFUSED` maps to
`ConnectionRefusedError`), so the normalization is load-bearing
rather than cosmetic.
'''
async def main():
# nothing has ever `.bind()`ed this one
nowhere = TIPCAddress(
_stype=TRACTOR_STYPE,
_instance=0xDEADBEEF,
)
with trio.fail_after(5):
await MsgpackTIPCStream.connect_to(destaddr=nowhere)
with pytest.raises(ConnectionError) as excinfo:
trio.run(main)
src_exc = excinfo.value.__cause__
assert src_exc.errno == errno.EHOSTUNREACH
assert 'No TIPC publisher' in str(excinfo.value)
@requires_tipc
def test_importance_sockopt_roundtrips():
'''
The `TIPC_IMPORTANCE` QoS knob (plan 01 §3.3) is settable and
readable back — TIPC can rank a conn's traffic under
congestion, which no other backend can do.
'''
async def main():
addr: TIPCAddress = TIPCAddress.get_random()
lstnr = await start_listener(addr=addr)
# XXX hold the accepted conn open for the duration; a
# peer that closes first makes `getpeername()` (called
# from `MsgpackTransport.__init__`) raise `ENOTCONN`.
done = trio.Event()
async def _accept():
stream = await lstnr.accept()
await done.wait()
await stream.aclose()
async with trio.open_nursery() as tn:
tn.start_soon(_accept)
await trio.sleep(0.05)
cli = await MsgpackTIPCStream.connect_to(
destaddr=addr,
importance=TIPC_HIGH_IMPORTANCE,
)
sock = cli.stream.socket
assert sock.getsockopt(
SOL_TIPC,
TIPC_IMPORTANCE,
) == TIPC_HIGH_IMPORTANCE
assert sock.getsockopt(
SOL_TIPC,
TIPC_DEST_DROPPABLE,
) == 0
await cli.stream.aclose()
done.set()
lstnr.socket.close()
trio.run(main)
@requires_tipc
def test_dropped_peer_does_not_kill_the_listener():
'''
A peer that connects then drops BEFORE we read must cost us
an addr, not the runtime.
XXX unlike tcp/uds — where the kernel keeps answering the
peer addr until *we* close — TIPC answers `ENOTCONN` on
`getpeername()` once the peer is gone. Since
`MsgpackTransport.__init__()` calls `.get_stream_addrs()`
(via `Channel.from_stream()`) BEFORE the handshake, an
unguarded `OSError` there escapes
`handle_stream_from_peer()`s handshake tolerance (contract
§4) and tears down the whole actor.
Real-world triggers: a port scan, a liveness probe (our own
`tests/discovery/conftest.py::daemon` readiness poll does
exactly this!), or a cancelled dial.
'''
async def main():
addr: TIPCAddress = TIPCAddress.get_random()
lstnr = await start_listener(addr=addr)
tpts: list = []
async def _accept():
stream = await lstnr.accept()
# MUST NOT raise even though the peer is already gone
tpts.append(MsgpackTIPCStream(stream))
async with trio.open_nursery() as tn:
tn.start_soon(_accept)
await trio.sleep(0.05)
# connect-then-immediately-drop
sock = _tipc.trio_socket.socket(AF_TIPC, SOCK_STREAM)
await sock.connect((
TIPC_ADDR_NAME,
addr._stype,
addr._instance,
0,
addr._scope,
))
sock.close()
await trio.sleep(0.2)
# NOTE the assertion that matters is simply that
# `MsgpackTIPCStream()` above did NOT raise; whether
# `getpeername()` still answers is a kernel-side race on
# the disconnect indication, so don't pin `.maybe_ref`.
assert len(tpts) == 1
raddr: TIPCAddress = tpts[0].raddr
assert not raddr.is_valid
# ..and it still reprs cleanly for the con-status logs
assert 'unknown-service' in repr(raddr)
lstnr.socket.close()
trio.run(main)
def test_observed_addr_tolerates_a_dead_peer():
'''
The deterministic half of the above: `_maybe_sockaddr()`
swallows the `ENOTCONN` and `_observed_addr()` still yields
a usable (name-less, port-id-less) addr.
'''
def _enotconn():
raise OSError(
errno.ENOTCONN,
'Transport endpoint is not connected',
)
assert _tipc._maybe_sockaddr(_enotconn) is None
addr: TIPCAddress = _tipc._observed_addr(None)
assert not addr.is_valid
assert addr.maybe_node is None
assert addr.maybe_ref is None
assert 'unknown-service' in repr(addr)
@requires_tipc
def test_duplicate_name_bind_does_not_raise():
'''

229
tractor/ipc/_tipc.py
View File

@ -53,6 +53,7 @@ import os
import socket
from socket import SOCK_STREAM
from typing import (
Callable,
ClassVar,
Type,
TYPE_CHECKING,
@ -60,12 +61,19 @@ from typing import (
from uuid import uuid4
import msgspec
import trio
from trio import (
socket as trio_socket,
SocketListener,
)
from multiaddr import Multiaddr
from tractor.msg import MsgCodec
from tractor.log import get_logger
from tractor.discovery._multiaddr import mk_maddr
from tractor.ipc._transport import (
MsgpackTransport,
)
from tractor.runtime._state import (
current_actor,
is_root_process,
@ -94,6 +102,10 @@ try:
TIPC_ADDR_NAME,
TIPC_ADDR_NAMESEQ,
TIPC_CLUSTER_SCOPE,
TIPC_DEST_DROPPABLE,
TIPC_HIGH_IMPORTANCE,
TIPC_IMPORTANCE,
TIPC_LOW_IMPORTANCE,
TIPC_NODE_SCOPE,
TIPC_ZONE_SCOPE,
)
@ -106,6 +118,10 @@ except ImportError:
TIPC_ZONE_SCOPE: int = 1
TIPC_CLUSTER_SCOPE: int = 2
TIPC_NODE_SCOPE: int = 3
TIPC_LOW_IMPORTANCE: int = 0
TIPC_HIGH_IMPORTANCE: int = 2
TIPC_IMPORTANCE: int = 127
TIPC_DEST_DROPPABLE: int = 129
# `tractor`'s reserved TIPC service-class ("type"), spelling out
@ -124,9 +140,19 @@ _tipc_reserved_stypes: range = range(0, 64)
# sentinel for "this addr was *observed* off a `TIPC_ADDR_ID`, so
# the peer's service-name is unknowable from the socket alone".
# See plan 01 §3.4.
# See `MsgpackTIPCStream.get_stream_addrs()` and plan 01 §3.4.
TIPC_NAME_UNKNOWN: int = -1
# XXX, the kernel default (`TIPC_LOW_IMPORTANCE`), i.e. today this
# is a no-op knob preserving stock behaviour.
#
# ?TODO, TIPC can rank a connection's traffic under congestion —
# something no other backend can do — so the parent<->child
# *supervision* chan deserves `TIPC_HIGH_IMPORTANCE` while bulk app
# streams stay low. Wiring `_runtime.py`'s parent-chan path to pass
# it is deliberately a follow-up; see plan 01 §3.3 + §10.
TRACTOR_DEF_IMPORTANCE: int = TIPC_LOW_IMPORTANCE
_scope_names: dict[int, str] = {
TIPC_ZONE_SCOPE: 'zone',
TIPC_CLUSTER_SCOPE: 'cluster',
@ -548,3 +574,204 @@ async def start_listener(
# entry to unlink and the kernel withdraws the published name on
# socket close. Per contract §1.2 absence means "closing is
# implicit".
@cm
def _close_on_error(sock):
'''
Close `sock` if the wrapped block raises.
Equivalent to `trio._highlevel_open_unix_stream.close_on_error`
but inlined so this (linux-cluster) backend doesn't import a
*unix-domain* private module.
'''
try:
yield sock
except BaseException:
sock.close()
raise
class MsgpackTIPCStream(MsgpackTransport):
'''
A `trio.SocketStream` around an `AF_TIPC` service-name
connection delivering `msgpack` encoded msgs via the `msgspec`
codec lib.
'''
address_type = TIPCAddress
layer_key: int = 4
@property
def maddr(self) -> Multiaddr|str:
if not self.raddr:
return '<unknown-peer>'
return mk_maddr(self.raddr)
def connected(self) -> bool:
return self.stream.socket.fileno() != -1
@classmethod
async def connect_to(
cls,
destaddr: TIPCAddress,
prefix_size: int = 4,
codec: MsgCodec|None = None,
importance: int = TRACTOR_DEF_IMPORTANCE,
**kwargs,
) -> MsgpackTIPCStream:
'''
Dial `destaddr` **by service name**.
NOTE, the `.connect()` here *is* the discovery lookup — the
kernel resolves the published name-table entry for us, so
there's no registrar hop on this path.
'''
sock = trio_socket.socket(
AF_TIPC,
SOCK_STREAM,
)
with _close_on_error(sock):
sock.setsockopt(
SOL_TIPC,
TIPC_IMPORTANCE,
importance,
)
# NOTE, surface undeliverable msgs as errors rather
# than let the kernel silently drop them.
sock.setsockopt(
SOL_TIPC,
TIPC_DEST_DROPPABLE,
0,
)
with _reraise_as_connerr(
src_excs=(OSError,),
addr=destaddr,
):
await sock.connect((
TIPC_ADDR_NAME,
destaddr._stype,
destaddr._instance,
0, # domain: 0 == "anywhere in scope"
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
@classmethod
def get_stream_addrs(
cls,
stream: trio.SocketStream,
) -> tuple[
TIPCAddress,
TIPCAddress,
]:
'''
Derive `(laddr, raddr)` from a connected TIPC socket.
XXX, BOTH ends answer `TIPC_ADDR_ID` port-ids and a port-id
carries NO service-name, so neither addr is dialable here;
they're name-`TIPC_NAME_UNKNOWN` and carry only the
observed `(node, ref)`.
That's fine and deliberate (plan 01 §3.4a):
- the *dialling* side overrides `._raddr` with the name it
actually dialled (see `.connect_to()`),
- the *accepting* side genuinely cannot know the peer's
name from the socket — but it doesn't need to, since the
`Aid` from `Channel._do_handshake()` already carries the
peer's logical identity.
'''
sock = stream.socket
return (
_observed_addr(_maybe_sockaddr(sock.getsockname)),
_observed_addr(_maybe_sockaddr(sock.getpeername)),
)
def _maybe_sockaddr(
getter: Callable[[], tuple],
) -> tuple|None:
'''
Call a `sock.getsockname`/`.getpeername` tolerantly.
XXX REQUIRED for TIPC: unlike tcp/uds — where the kernel keeps
answering the peer addr until *we* close — a TIPC socket whose
peer has already gone answers `ENOTCONN`. That happens for any
connect-then-immediately-drop peer: a port scan, a liveness
probe, a cancelled dial.
Since `MsgpackTransport.__init__()` calls `.get_stream_addrs()`
(via `Channel.from_stream()`) BEFORE the handshake, letting the
`OSError` fly would escape `handle_stream_from_peer()`s
handshake tolerance (contract §4) and tear down the whole
actor. A dead peer must cost us an addr, not the runtime.
'''
try:
return getter()
except OSError as oserr:
log.transport(
f'TIPC peer already gone, no port-id available\n'
f'from src: {oserr!r}\n'
)
return None
def _port_id(
sockaddr: tuple[int, int, int, int, int],
) -> tuple[int, int]:
'''
Unpack the `(node, ref)` of a `TIPC_ADDR_ID` 5-tuple as
delivered by `getsockname()`/`getpeername()`.
Layout is `(addrtype, node, ref, 0, scope)`; see
`makesockaddr()`s `AF_TIPC` case in CPython's `socketmodule.c`.
'''
_, node, ref, *_ = sockaddr
return (node, ref)
def _observed_addr(
sockaddr: tuple[int, int, int, int, int]|None,
) -> TIPCAddress:
'''
Wrap a `TIPC_ADDR_ID` port-id as a name-less `TIPCAddress`
usable for logging/`repr` only.
A `None` `sockaddr` (peer already gone, see
`_maybe_sockaddr()`) yields the same addr sans port-id.
'''
node: int|None = None
ref: int|None = None
if sockaddr is not None:
node, ref = _port_id(sockaddr)
return TIPCAddress(
_stype=TIPC_NAME_UNKNOWN,
_instance=TIPC_NAME_UNKNOWN,
maybe_node=node,
maybe_ref=ref,
)