Compare commits

...

12 Commits

Author SHA1 Message Date
Gud Boi 51d7133f47 Add the interim `/tipc/` maddr grammar
`mk_maddr()`/`parse_maddr()` learn,

    /tipc/<stype>/<instance>/<scope>

mirroring how `uds` maps onto the spec-legal `/unix`.

XXX `str`-ONLY for now: there is no registered `/tipc` proto
in the multiaddr table (upstream track gh #483 +
multiformats/py-multiaddr#107) and `Multiaddr()` rejects an
unregistered name outright. `MsgTransport.maddr`s return type
is already `Multiaddr|str` (and `MsgpackUDSStream` already
exercises the `str` branch), so this fits — but it IS why gh

`parse_maddr()` therefore special-cases the `/tipc/` prefix
BEFORE handing anything to `Multiaddr()`.

Also drive the maddr mapping-table tests off `_address_types`
instead of a hardcoded len/dict so the next backend can't
fail them for the wrong reason.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 10:02:44 -04:00
Gud Boi e0f66616cd Register `tipc` in the tpt tables + test harness
Wire the backend through every registration site (contract §2)
so `--tpt-proto tipc` is a first-class suite mode,
- `_state.TransportProtocolKey` gains the key
- `_addr._address_types` + `._default_lo_addrs`
- `_addr.wrap_address()` gets a `case ('tipc', *_)`; being a
  4-elem seq it can't collide w/ `tcp`s or `uds`s 2-tuple
  cases, so NO ordering hazard (and a bare seq-pattern matches
  the `list` form `msgpack` decodes to).
- `_types`: the `Address` union, `_msg_transports`,
  `_key_to_transport`, `_addr_to_transport` and the
  `transport_from_stream()` family match. That last one keys
  off `._tipc.AF_TIPC` (which carries the uapi fallback) NOT
  `socket.AF_TIPC` which is linux-only.

Test-harness side,
- `get_rando_addr()` gains a `tipc` branch; `.get_random()`
  already salts w/ `uuid4`+pid so both within- and cross-proc
  isolation come for free.
- the `tpt_protos` fixture calls an addr-type's optional
  `.is_available()` and `pytest.fail()`s w/ its reason. Keeps
  a module-less box from turning `--tpt-proto tipc` into a few
  hundred confusing connect-timeouts. Generic on purpose —
  plans 02/03 need the same hook.
- the discovery `daemon` fixture's readiness probe learns to
  dial a TIPC service name (it previously assumed tcp-or-uds
  and blew up on the 4-tuple).

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 10:02:44 -04:00
Gud Boi 2204979492 Widen `UnwrappedAddress` to admit proto-keyed addrs
`SpawnSpec.reg_addrs`/`.bind_addrs` pinned the wire shape to
a 2-tuple, so a `tipc` addr (`('tipc', stype, inst, scope)`)
died at the child w/ `msgspec.ValidationError: Expected array
of length 2, got 4` -> `invalid SpawnSpec IPC msg`.

Point those fields at `UnwrappedAddress` (which `SpawnSpec`s
own TODO already asked for) and widen the alias.

XXX VARIADIC (`tuple[str|int, ...]`) rather than a union of
the two concrete shapes, bc `msgspec` refuses a union holding
more than one array-like type.

?TODO, the real fix is the full proto-key migration (contract
§1.1) after which this becomes a tagged union keyed off elem
0 and per-proto validation comes back.

Note the alias is declared TWICE — `.msg.types` re-declares it
to dodge a circular import (`._addr` -> `.ipc._tcp` -> `.msg`)
and *that* copy is what actually validates the wire msg.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 10:02:44 -04:00
Gud Boi 8c0ae140cd 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`))
2026-08-14 10:02:44 -04:00
Gud Boi e3089ba356 Add `TIPCAddress` + `start_listener()`, gh #378
First slice of the `AF_TIPC` tpt backend: the addr type, the
`is_tipc_available()` capability predicate and the
name-publishing listener. No `MsgTransport` yet.

An actor's TIPC addr is a *service name* `(stype, instance)`:
`.bind()`ing the singleton `TIPC_ADDR_NAMESEQ` range IS the
service registration (it shows up in `tipc nametable show`)
and a peer's `.connect()`-by-name IS the lookup — so the
kernel does discovery for us, no registrar hop.

Deats,
- `.unwrap()` is proto-keyed as `('tipc', stype, inst, scope)`
  using the `multiaddr` proto spelling so `wrap_address()`
  can't confuse it with `tcp`s or `uds`s 2-tuples.
- `.rebind_from_sockname = False` bc `getsockname()` answers
  a port-id; `.from_addr()` raises on a bare `TIPC_ADDR_ID`
  rather than fabricate an un-dialable addr.
- `.bindspace` is the TIPC *scope*, i.e. literally the set of
  hosts a published name is reachable from. `ZONE` scope is
  deprecated/aliased so fold it to `CLUSTER` on input.
- mod stays importable on non-linux (uapi-value fallbacks,
  the `_uds.SO_PASSCRED` precedent) bc `._addr` builds its
  registration tables at import time.

XXX a `.get_random()` clash does NOT raise `EADDRINUSE` —
TIPC accepts multiple publishers of one name and round-robins
connects between them (verified against a live kernel), so a
collision is *silent crosstalk*. Hence the `blake2b` digest
and its (birthday-bounded) collision test.

Also,
- a generic `.is_available() -> (ok, why_not)` classmethod;
  deliberately spelled generically (NOT `is_tipc_*`) so the
  sibling env-dependent backends — `quic`/`iroh` (gh #353)
  and the `wg` netns bindspace (gh #482) — get the same gate
  for free. Its consumer lands w/ the reg tables.
- register a `tipc` pytest mark; the kernel-touching cases
  self-skip unless `sudo modprobe tipc` has been run.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 10:02:44 -04:00
Gud Boi cca3a70de4 Add `Address.rebind_from_sockname` opt-out
Gate `Endpoint.start_listener()`s `getsockname()`-vs-`.addr`
reconciliation on a new per-addr-type `ClassVar[bool]`, set
`True` on both `TCPAddress` and `UDSAddress` so existing
behaviour is bit-for-bit unchanged.

That reconciliation exists ONLY to learn a kernel-assigned
port from a `port=0` tcp bind (its own comment says so). The
incoming `tipc` backend (gh #378) has no late-binding
analogue AND its `getsockname()` answers a `TIPC_ADDR_ID`
port-id rather than the name-seq it published — rebinding
from that would swap a dialable service name for an
un-dialable, un-reconstructable port id.

So opting out is semantically right rather than a hack.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 10:02:44 -04:00
Gud Boi d4737e957f Pin `Endpoint` addr-reconciliation for tcp/uds
Guard test for `.start_listener()`s post-bind
`getsockname()`-vs-`.addr` round-trip, landed *before* that
reconciliation gets gated on an opt-out `ClassVar`.

- tcp: a `port=0` bind MUST still learn the kernel-picked
  port, since the reconciliation is the only path that ever
  does.
- uds: the sock-file path must survive the `.from_addr()`
  round-trip unchanged.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 10:02:44 -04:00
Gud Boi f9f98eeb06 Fix `pformat_caller_frame()`s bogus `indent` kwarg
Just drop it — `pformat_boxed_tb()` spells its knobs
`tb_box_indent`/`tb_body_indent`, and that fn's default
(1-space box indent) is what the caller wanted anyway.

Regressed-by: 888af602 (`pformat_cs()` mv into `.devx.pformat`)
Found-via: `/run-tests` test_pformat_caller_frame_renders

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 10:02:44 -04:00
Gud Boi 22ef362d9f Add a `pformat_caller_frame()` render guard test
`pformat_boxed_tb()` has never accepted an `indent` kwarg but
`pformat_caller_frame(box_tb=True)` has been passing one since
`888af602`. Nothing in the suite covered the branch, so the
`TypeError` only ever surfaced from `_mk_send_mte()` — i.e.
EVERY send-side `MsgTypeError` blew up while formatting itself
and masked the real msg-spec violation behind a bogus
`TypeError`.

Red on purpose per the test-first convention; the 1-line fix
lands next.

Also pin `pformat_boxed_tb()`s signature so a future typo'd
kwarg fails loudly at the call site instead of only when some
rare error path runs.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 10:02:44 -04:00
Gud Boi ee17ed9f6e Update `wg` docs for the merged py-multiaddr#108
it lands" framing in plan-03 and the example README was stale in
both directions: the branch pin is obsolete, yet you still can't
just `pip install multiaddr`.

Deats,
- §3.2's grammar table is now re-verified against the upstream
  merge (`f86519da`) rather than only `baudco@wg_support` in a
  throwaway venv. Also notes the codec enforces a 32-byte key,
  so a truncated one is a `StringParseError` and not a silently
  mangled parse.
- §1 says merged-but-unreleased; the still-open work is spec
  registration (py-multiaddr#107 + gh #483).
- §3.4 swaps "pin the branch" for the `[tool.uv.sources]` `rev`
  pin, and fixes the `_have_wg_maddr_proto()` recipe it
  suggested — probing w/ `Multiaddr('/wg/uAAAA')` now ALWAYS
  raises bc the codec wants 32B, i.e. that feature-detect would
  report `False` even w/ the proto perfectly well known.
- risk table row goes "#108 not merged" -> "merged but
  unreleased".
- example README: `uv sync` alone now suffices bc of the pin;
  documents the 32B check and points at
  `_have_wg_maddr_proto()` as the gate.

The one surviving `baudco` mention is deliberate, it records
where the grammar was *first* verified.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 09:51:36 -04:00
Gud Boi 7d6e79551e Fix silently-corrupt keys in `parse_wg_maddr()`
`_segments()` called `Multiaddr(maddr)` purely to validate, then
swallowed every failure under `except Exception: pass`. That was
harmless pre-#108 — w/o a `wg` codec there was nothing to
validate — but now that the codec is pinned in, the swallow is
load-bearing and disabled: a malformed key sails past validation
into `wg8_pubkey()`, which happily emits a corrupt b64 str, and
the returned struct then fails its own `.maddr` round-trip. No
raise, just quietly wrong output.

Deats,
- add `_have_wg_maddr_proto()`, the gate plan-03 already
  referenced but which never actually existed. Impl'd as
  `protocols.protocol_with_name('wg')` under
  `except ProtocolNotFoundError` and cached in a mod global,
  same shape as the TIPC plan's `is_tipc_available()`.
- only validate when that gate is `True`, and let
  `StringParseError` propagate — a maddr which doesn't parse
  must NOT reach `wg8_pubkey()`.
- keep the degraded split for a pre-#108 install, now w/ an
  explicit `XXX` naming the validation you give up.

So parsing stays pure but becomes total-or-raises. Our own
`ValueError`s (missing `/wg/` seg, bare tunnel w/o an overlay
ep) are unaffected, as is the `wg(8)` b64 round-trip.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 09:50:48 -04:00
Gud Boi cc85f17f5f Pin `multiaddr` to the merged `wg` codec rev
py-multiaddr#108 (the `/wg/u<key>` maddr proto) merged upstream
on 2026-07-28 as `f86519da`, but ships in no release yet — the
latest `0.2.0` predates it by ~4 months and carries no `wg`
codec at all. So `examples/multihost/wg_lan/` can't parse its
own maddrs off PyPI.

Pinned by `rev` and not `branch` so CI stays reproducible. Note
the lock now records the git source *instead of* the `>=0.2.0`
specifier, i.e. the dep floor above is fully overridden for as
long as this pin lives.

TODO, drop the pin (and bump that floor) the moment a release
carries the codec; the only consumer is the `wg_lan` example
set.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))
2026-08-14 09:38:58 -04:00
22 changed files with 1859 additions and 58 deletions

View File

@ -26,10 +26,11 @@ onto `trio` as the library's sans-io layer allows.
`/ip4|ip6/<h>/tcp/<p>` and `/unix/<p>`; a `.../wg/u<key>` `/ip4|ip6/<h>/tcp/<p>` and `/unix/<p>`; a `.../wg/u<key>`
maddr raises `ValueError('Unsupported multiaddr protocol maddr raises `ValueError('Unsupported multiaddr protocol
combo')`. combo')`.
- there is no `wg` proto in the multiaddr spec; the first-draft - there is no `wg` proto in the multiaddr *spec* yet, but
upstream PR is multiformats/py-multiaddr#108 with key form multiformats/py-multiaddr#108 (key form `u<base64url>`) is
`u<base64url>` (commit `8be3a8b`), tracked by **merged** as of 2026-07-28 (`f86519da`) — and unreleased, the
multiformats/py-multiaddr#107 and gh #483. latest `0.2.0` predating it. Spec registration is still tracked
by multiformats/py-multiaddr#107 and gh #483.
- so **today's deployable story is declarative**: run `wg-quick` - so **today's deployable story is declarative**: run `wg-quick`
out-of-band, parse the maddr, strip to the inner out-of-band, parse the maddr, strip to the inner
`(host, port)`, verify the pubkey against the live tunnel, `(host, port)`, verify the pubkey against the live tunnel,
@ -112,9 +113,12 @@ class WGTunnelSpec(
### 3.2 `parse_maddr()`/`mk_maddr()` ### 3.2 `parse_maddr()`/`mk_maddr()`
Grammar — **verified** against py-multiaddr#108 Grammar — **verified** against py-multiaddr#108, first on the
(`baudco/py-multiaddr@wg_support`, installed in a throwaway venv; `baudco/py-multiaddr@wg_support` branch and re-verified after it
all three forms below parse *and* round-trip): merged upstream (`multiformats/py-multiaddr@f86519da`); all three
forms below parse *and* round-trip. Note the codec also validates
that the key decodes to exactly 32 bytes, so a truncated key is a
`StringParseError`, not a silently-mangled parse:
``` ```
/ip4/192.168.1.50/udp/51820/wg/u<A_pub>/ip4/10.0.11.1/tcp/1616 /ip4/192.168.1.50/udp/51820/wg/u<A_pub>/ip4/10.0.11.1/tcp/1616
@ -169,13 +173,16 @@ Observed protocol-name lists, for writing the `match`:
(bearer_names, tunnel_specs, inner_names)`. This is also what (bearer_names, tunnel_specs, inner_names)`. This is also what
makes a wg-inside-wg stack fall out for free. makes a wg-inside-wg stack fall out for free.
- `mk_maddr()` inverse for `TunnelledAddress`. - `mk_maddr()` inverse for `TunnelledAddress`.
- **blocked on upstream**: `Multiaddr('/…/wg/u…')` only parses - **pending an upstream release**: py-multiaddr#108 is merged, so
once py-multiaddr#108 lands. Until then: pin the branch in the `Multiaddr('/…/wg/u…')` parses — but off a `[tool.uv.sources]`
`wg` extra / dev-group and gate the tests on `rev` pin, since no release carries the codec. Gate the tests
`_have_wg_maddr_proto()` (a cheap try/except around on `_have_wg_maddr_proto()`, implemented as
`Multiaddr('/wg/uAAAA')`). Do **not** hand-roll a `wg` parser `protocols.protocol_with_name('wg')` under
in `tractor` — the whole point of #429 was dropping the NIH `except ProtocolNotFoundError`. Do **not** probe by parsing a
parser. dummy like `Multiaddr('/wg/uAAAA')` — the codec enforces a
32-byte key, so that raises even when the proto *is* known. Do
**not** hand-roll a `wg` parser in `tractor` — the whole point
of #429 was dropping the NIH parser.
### 3.3 verification helper (pure, composable) ### 3.3 verification helper (pure, composable)
@ -437,7 +444,7 @@ consider doing it *first* for exactly that reason.
| risk | mitigation | | risk | mitigation |
| --- | --- | | --- | --- |
| `to_thread` worker runs in the wrong netns | §5.3; pass `netns=` to pyroute2 or pin a worker; test-first | | `to_thread` worker runs in the wrong netns | §5.3; pass `netns=` to pyroute2 or pin a worker; test-first |
| py-multiaddr#108 not merged | branch pin + `_have_wg_maddr_proto()` gate; layer A's inner-addr path works regardless | | py-multiaddr#108 merged but unreleased | `[tool.uv.sources]` `rev` pin + `_have_wg_maddr_proto()` gate; layer A's inner-addr path works regardless |
| `TunnelledAddress` leaks into `Endpoint` and breaks `inspect.getmodule()` | unwrap at parse/bindspace boundary; assert `not isinstance(ep.addr, TunnelledAddress)` in `Endpoint.__post_init__` | | `TunnelledAddress` leaks into `Endpoint` and breaks `inspect.getmodule()` | unwrap at parse/bindspace boundary; assert `not isinstance(ep.addr, TunnelledAddress)` in `Endpoint.__post_init__` |
| privileged ops in a library | never `sudo`; explicit cap probe + actionable error; pre-provisioned is the default | | privileged ops in a library | never `sudo`; explicit cap probe + actionable error; pre-provisioned is the default |
| pyroute2 0.9 asyncio core drags a loop into the actor | option (1) is a *thread*, not a loop; forbid `trio-asyncio` here (§4.1) | | pyroute2 0.9 asyncio core drags a loop into the actor | option (1) is a *thread*, not a loop; forbid `trio-asyncio` here (§4.1) |

View File

@ -39,16 +39,27 @@ this composed form parses and round-trips
## requirements ## requirements
The `wg` proto isn't in released `py-multiaddr` yet (`0.2.0` has py-multiaddr #108 is **merged** (2026-07-28) but ships in no
no `wg` codec), so until #108 lands: release yet — the latest `0.2.0` (2026-03-17) predates it and has
no `wg` codec. So `pyproject.toml` carries a temporary
`[tool.uv.sources]` `rev` pin at the merge commit, and a plain
```bash ```bash
uv pip install 'git+https://github.com/baudco/py-multiaddr.git@wg_support' multibase uv sync
``` ```
`wg_maddr.py` degrades to a plain segment split when the codec is gets you a `wg`-aware `multiaddr`. That pin goes away once a
absent, so the examples still run — but you lose per-segment release carries the codec. You also need `multibase`:
validation. It deliberately does **not** hand-roll a `wg` codec
```bash
uv pip install multibase
```
Without the codec `wg_maddr.py` degrades to a plain segment split
— the examples still run, but you lose per-segment validation
(incl. the 32-byte key-length check), so a malformed key reaches
the returned struct instead of raising. `_have_wg_maddr_proto()`
is the gate. It deliberately does **not** hand-roll a `wg` codec
(gh #429 was about *dropping* our NIH parser). (gh #429 was about *dropping* our NIH parser).
## 0. tunnel setup (out-of-band, both hosts) ## 0. tunnel setup (out-of-band, both hosts)

View File

@ -2,9 +2,9 @@
r''' r'''
Parse `wg`-tunnelled multiaddrs into `tractor`-ready addrs. Parse `wg`-tunnelled multiaddrs into `tractor`-ready addrs.
The canonical form (per py-multiaddr PR #108, verified to parse + The canonical form (per py-multiaddr #108, verified to parse +
round-trip on that branch) nests the *overlay* endpoint **after** round-trip against its upstream merge) nests the *overlay*
the `/wg/` segment: endpoint **after** the `/wg/` segment:
/ip4/10.0.0.1/udp/51820/wg/u<key>/ip4/10.0.11.1/tcp/1616 /ip4/10.0.0.1/udp/51820/wg/u<key>/ip4/10.0.11.1/tcp/1616
\_______ wg bearer ______/\_ key _/\____ tractor ep _____/ \_______ wg bearer ______/\_ key _/\____ tractor ep _____/
@ -111,8 +111,10 @@ def parse_wg_maddr(
Split a `wg`-tunnelled maddr into its bearer/key/overlay Split a `wg`-tunnelled maddr into its bearer/key/overlay
parts. Pure no I/O. parts. Pure no I/O.
Uses `py-multiaddr` when it knows the `wg` proto (PR #108), Total-or-raises: with a `wg`-aware `py-multiaddr` (#108) an
else falls back to a minimal segment split. unparseable maddr raises instead of yielding a struct built
from garbage segments. See `_segments()` for the degraded
pre-#108 path.
''' '''
segs: list[str] = _segments(maddr) segs: list[str] = _segments(maddr)
@ -164,22 +166,53 @@ def parse_wg_maddr(
) )
def _segments(maddr: str) -> list[str]: _wg_proto_known: bool|None = None
def _have_wg_maddr_proto() -> bool:
''' '''
Deliver a maddr's `/`-split segments, preferring the real True iff the installed `py-multiaddr` knows the `/wg/` proto,
parser when it supports `wg`. i.e. carries py-multiaddr#108.
Merged upstream 2026-07-28 but in no release as of `0.2.0`,
hence the `[tool.uv.sources]` `rev` pin.
Pure predicate; result cached since it can't change without a
reinstall.
''' '''
from multiaddr import Multiaddr global _wg_proto_known
try: if _wg_proto_known is None:
# the real thing: validates every proto + value from multiaddr.protocols import protocol_with_name
from multiaddr.exceptions import ProtocolNotFoundError
try:
protocol_with_name('wg')
_wg_proto_known = True
except ProtocolNotFoundError:
_wg_proto_known = False
return _wg_proto_known
def _segments(maddr: str) -> list[str]:
'''
Deliver a maddr's `/`-split segments, validating via the real
parser whenever it knows `wg`.
'''
if _have_wg_maddr_proto():
from multiaddr import Multiaddr
# the real thing: validates every proto + value, incl.
# that the `wg` key decodes to exactly 32 bytes. Let it
# raise — a maddr that doesn't parse must NOT reach
# `wg8_pubkey()`, which would happily emit a corrupt key.
Multiaddr(maddr) Multiaddr(maddr)
except Exception:
# XXX STOPGAP, only until py-multiaddr#108 lands; then # XXX, degraded path for a pre-#108 `py-multiaddr` ONLY: no
# this branch is dead and `Multiaddr` is authoritative. # per-segment validation, so a malformed key survives to the
# We deliberately DON'T hand-roll a `wg` codec (the whole # returned struct. We deliberately DON'T hand-roll a `wg`
# point of gh #429 was dropping the NIH parser). # codec (the whole point of gh #429 was dropping the NIH
pass # parser) — install the pinned rev to get validation back.
return [s for s in maddr.split('/') if s] return [s for s in maddr.split('/') if s]

View File

@ -166,6 +166,17 @@ sync_pause = {requires-python = ">=3.13, <3.14"}
# linux kernel networking # linux kernel networking
# 'pyroute2 # 'pyroute2
# XXX TEMP, the `/wg/u<key>` maddr proto is MERGED upstream (in
# py-multiaddr#108, 2026-07-28) but is in NO release yet; the
# latest `0.2.0` (2026-03-17) predates the merge by ~4 months.
# Pinned by `rev` (not `branch`) so CI stays reproducible.
#
# Drop this pin (and bump the `multiaddr` dep floor above) the
# moment a release carries the `wg` codec; the only consumer is
# `examples/multihost/wg_lan/`.
# |_https://github.com/multiformats/py-multiaddr/pull/108
multiaddr = { git = 'https://github.com/multiformats/py-multiaddr.git', rev = 'f86519daaa21699023d0037c58cdff600313dd09' }
# ------ tool.uv.sources ------ # ------ tool.uv.sources ------
[tool.uv] [tool.uv]

View File

@ -0,0 +1,52 @@
'''
Unit tests for the `tractor.devx.pformat` render helpers.
'''
from __future__ import annotations
import pytest
from tractor.devx.pformat import (
pformat_boxed_tb,
pformat_caller_frame,
)
@pytest.mark.parametrize(
'box_tb',
[True, False],
ids=['boxed', 'bare'],
)
def test_pformat_caller_frame_renders(box_tb: bool):
'''
`pformat_caller_frame()` must render, not raise.
XXX the `box_tb=True` branch was passing an `indent=''` kwarg
that `pformat_boxed_tb()` never accepted, so it blew up with
a `TypeError`. Nothing in the test suite covered it, and the
only caller is `_mk_send_mte()` i.e. EVERY send-side
`MsgTypeError` died while formatting itself, masking the real
msg-spec violation behind a bogus `TypeError`.
'''
report: str = pformat_caller_frame(
stack_limit=3,
box_tb=box_tb,
)
assert isinstance(report, str)
assert 'test_pformat_caller_frame_renders' in report
def test_pformat_boxed_tb_rejects_unknown_kwargs():
'''
Pin the signature so a future typo'd kwarg fails loudly at the
call site rather than only when some rare error path runs.
'''
assert pformat_boxed_tb(tb_str='doggy\n')
with pytest.raises(TypeError):
pformat_boxed_tb(
tb_str='doggy\n',
indent='',
)

View File

@ -84,6 +84,35 @@ def _wait_for_daemon_ready(
timeout=poll_interval, timeout=poll_interval,
): ):
return return
elif tpt_proto == 'tipc':
# TIPC — `reg_addr` is the proto-keyed
# `('tipc', stype, instance, scope)` per
# `tractor.ipc._tipc.TIPCAddress.unwrap()`.
#
# NOTE, connecting *by name* IS the readiness
# probe: until the daemon `.bind()`s (i.e.
# publishes) the name, the kernel answers
# `EHOSTUNREACH` immediately — no timeout wait.
from tractor.ipc._tipc import (
AF_TIPC,
TIPC_ADDR_NAME,
)
_, stype, instance, scope = reg_addr
sock = socket.socket(AF_TIPC, socket.SOCK_STREAM)
try:
sock.settimeout(poll_interval)
sock.connect((
TIPC_ADDR_NAME,
stype,
instance,
0, # domain: 0 == "anywhere in scope"
scope,
))
return
finally:
sock.close()
else: else:
# UDS — `reg_addr` is a `(filedir, sockname)` # UDS — `reg_addr` is a `(filedir, sockname)`
# tuple per `tractor.ipc._uds.UDSAddress.unwrap`. # tuple per `tractor.ipc._uds.UDSAddress.unwrap`.

View File

@ -19,7 +19,10 @@ from tractor.discovery._multiaddr import (
_tpt_proto_to_maddr, _tpt_proto_to_maddr,
_maddr_to_tpt_proto, _maddr_to_tpt_proto,
) )
from tractor.discovery._addr import wrap_address from tractor.discovery._addr import (
wrap_address,
_address_types,
)
def test_tpt_proto_to_maddr_mapping(): def test_tpt_proto_to_maddr_mapping():
@ -30,7 +33,12 @@ def test_tpt_proto_to_maddr_mapping():
''' '''
assert _tpt_proto_to_maddr['tcp'] == 'tcp' assert _tpt_proto_to_maddr['tcp'] == 'tcp'
assert _tpt_proto_to_maddr['uds'] == 'unix' assert _tpt_proto_to_maddr['uds'] == 'unix'
assert len(_tpt_proto_to_maddr) == 2 assert _tpt_proto_to_maddr['tipc'] == 'tipc'
# NOTE, drive the expected set off the registration table
# (per the "drive-the-set-from-the-`Literal`" pattern) so
# adding a backend can't fail this for the wrong reason.
assert set(_tpt_proto_to_maddr) == set(_address_types)
def test_mk_maddr_tcp_ipv4(): def test_mk_maddr_tcp_ipv4():
@ -153,9 +161,12 @@ def test_maddr_to_tpt_proto_mapping():
''' '''
assert _maddr_to_tpt_proto == { assert _maddr_to_tpt_proto == {
'tcp': 'tcp', maddr_proto: proto_key
'unix': 'uds', for proto_key, maddr_proto in _tpt_proto_to_maddr.items()
} }
assert _maddr_to_tpt_proto['tcp'] == 'tcp'
assert _maddr_to_tpt_proto['unix'] == 'uds'
assert _maddr_to_tpt_proto['tipc'] == 'tipc'
def test_parse_maddr_tcp_ipv4(): def test_parse_maddr_tcp_ipv4():

View File

@ -14,6 +14,7 @@ from tractor import (
from tractor._testing.addr import ( from tractor._testing.addr import (
get_rando_addr, get_rando_addr,
) )
from tractor.ipc._tcp import TCPAddress
# TODO, use/check-roundtripping with some of these wrapper types? # TODO, use/check-roundtripping with some of these wrapper types?
# #
# from .._addr import Address # from .._addr import Address
@ -70,3 +71,71 @@ def test_basic_ipc_server(
pdb=debug_mode, pdb=debug_mode,
): ):
trio.run(main) trio.run(main)
@pytest.mark.parametrize(
'_tpt_proto',
['uds', 'tcp']
)
def test_ep_addr_reconciled_from_sockname(
_tpt_proto: str,
debug_mode: bool,
):
'''
Guard `Endpoint.start_listener()`'s post-bind reconciliation of
`.addr` against the listener's `socket.getsockname()`.
For `tcp` that reconciliation is the ONLY way a kernel-assigned
port (from a `port=0` bind) is ever learned, so it must keep
firing; for `uds` the sock-file path must survive the
round-trip through `.from_addr()` unchanged.
Both are pinned here *before* the reconciliation gets gated on
an `Address.rebind_from_sockname` opt-out (for backends whose
`getsockname()` reports something other than what was bound).
'''
async def main():
async with ipc._server.open_ipc_server() as server:
accept_addr: tuple[str, int|str]
match _tpt_proto:
# XXX the whole point: ask the kernel to pick.
case 'tcp':
accept_addr = (
TCPAddress.def_bindspace,
0,
)
case 'uds':
accept_addr = get_rando_addr(
tpt_proto=_tpt_proto,
)
eps: list[ipc._server.Endpoint] = await server.listen_on(
accept_addrs=[accept_addr],
stream_handler_nursery=None,
)
assert len(eps) == 1
ep: ipc._server.Endpoint = eps[0]
sockname = ep._listener.socket.getsockname()
match _tpt_proto:
case 'tcp':
# the bind req was for "any port"..
assert accept_addr[1] == 0
# ..and the ep learned the real one.
assert ep.addr._port != 0
assert ep.addr.unwrap() == tuple(sockname[:2])
case 'uds':
# sock-file path is stable across the
# `.from_addr()` round-trip.
assert ep.addr.unwrap() == accept_addr
assert str(ep.addr.sockpath) == sockname
server._parent_tn.cancel_scope.cancel()
with devx.maybe_open_crash_handler(
pdb=debug_mode,
):
trio.run(main)

View File

@ -0,0 +1,629 @@
'''
Unit tests for the `AF_TIPC` transport backend, `tractor.ipc._tipc`.
The kernel-touching cases are gated on `is_tipc_available()` since
the `tipc` module is NOT loaded by default (`sudo modprobe tipc`);
the pure address-algebra cases run everywhere.
'''
from __future__ import annotations
import errno
from socket import (
SOCK_STREAM,
SOL_SOCKET,
SO_ACCEPTCONN,
SOL_TIPC,
)
import pytest
import trio
from tractor.msg.types import Aid
from tractor.discovery import _addr
from tractor.discovery._addr import wrap_address
from tractor.discovery._multiaddr import (
mk_maddr,
parse_maddr,
)
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,
start_listener,
)
pytestmark = pytest.mark.tipc
requires_tipc = pytest.mark.skipif(
not is_tipc_available(),
reason=(
'`tipc` kernel module not loaded (`sudo modprobe tipc`)'
),
)
# ------------------------------------------------------------------
# address algebra (no kernel needed)
# ------------------------------------------------------------------
@pytest.mark.parametrize(
'addr',
[
TIPCAddress.get_root(),
TIPCAddress(
_stype=TRACTOR_STYPE,
_instance=42,
_scope=TIPC_NODE_SCOPE,
),
],
ids=['root', 'node-scoped'],
)
def test_addr_unwrap_roundtrip(addr: TIPCAddress):
'''
`.unwrap()` is proto-keyed and `.from_addr()` inverts it for
both the `tuple` form and the `list` form msgpack decodes to.
'''
unwrapped: tuple = addr.unwrap()
assert unwrapped[0] == 'tipc' == TIPCAddress.proto_key
assert len(unwrapped) == 4
assert TIPCAddress.from_addr(unwrapped) == addr
assert TIPCAddress.from_addr(list(unwrapped)) == addr
def test_addr_scope_defaults_when_omitted():
'''
A 3-elem `('tipc', stype, inst)` form defaults to the
cluster-scope bindspace.
'''
addr: TIPCAddress = TIPCAddress.from_addr(
('tipc', TRACTOR_STYPE, 99),
)
assert addr._scope == TIPC_CLUSTER_SCOPE
assert addr.bindspace == TIPCAddress.def_bindspace
def test_zone_scope_normalized_to_cluster():
'''
`TIPC_ZONE_SCOPE` is deprecated/aliased in modern kernels;
accept it on input, fold it to cluster.
'''
addr: TIPCAddress = TIPCAddress.from_addr(
('tipc', TRACTOR_STYPE, 7, TIPC_ZONE_SCOPE),
)
assert addr._scope == TIPC_CLUSTER_SCOPE
assert addr.is_valid
def test_addr_from_bare_port_id_raises():
'''
A `TIPC_ADDR_ID` 5-tuple carries no service-name so it can
NEVER be wrapped; it must fail loudly rather than silently
fabricate an un-dialable addr.
This is the invariant that lets
`TIPCAddress.rebind_from_sockname` be `False`.
'''
with pytest.raises(ValueError) as excinfo:
TIPCAddress.from_addr((TIPC_ADDR_ID, 0, 12345, 0, 0))
assert 'port-id' in str(excinfo.value)
def test_addr_is_valid_predicate():
assert TIPCAddress.get_root().is_valid
# instance 0 is not a bindable name
assert not TIPCAddress(
_stype=TRACTOR_STYPE,
_instance=0,
).is_valid
# service-types 0..63 are TIPC-internal (`TIPC_CFG_SRV`,
# `TIPC_TOP_SRV`, ..)
assert not TIPCAddress(
_stype=1,
_instance=1616,
).is_valid
def test_port_id_is_annotation_only():
'''
`.maybe_node`/`.maybe_ref` are *observed* metadata, excluded
from `.unwrap()` exactly like `UDSAddress.maybe_pid`.
'''
addr: TIPCAddress = TIPCAddress.get_root()
annotated: TIPCAddress = addr.with_port_id(
node=0xdead,
ref=1234,
)
assert annotated.unwrap() == addr.unwrap()
assert annotated.maybe_ref == 1234
assert '1234' in repr(annotated)
def test_instance_from_seed_is_pure():
'''
Same seed -> same instance (what the follow-up registrar-less
discovery fast-path will lean on), and always clear of the
reserved low range.
'''
for seed in ('doggy@123', 'kitty@456', ''):
inst: int = instance_from_seed(seed)
assert inst == instance_from_seed(seed)
assert 64 <= inst < 2**32
def test_get_random_collision_resistance():
'''
A `.get_random()` clash does NOT raise `EADDRINUSE` TIPC
accepts multiple publishers of one name and round-robins
connects between them, so a collision is *silent crosstalk*.
Assert the 4-byte digest spreads well enough for that to stay
improbable.
NOTE the bound is birthday-statistical, not absolute:
P(collision) ~= 1 - exp(-n**2 / 2**33) ~= 1.2e-2 for n=10k, so
a strict `== n` assert would be ~1-in-86 flaky. P(>2
collisions) is ~1e-7, hence the slack. See plan 01 §9 for the
escalation path if this ever trips.
'''
n: int = 10_000
addrs: list[TIPCAddress] = [
TIPCAddress.get_random()
for _ in range(n)
]
instances: set[int] = {
addr._instance
for addr in addrs
}
assert len(instances) >= n - 2
# every one is a legal, bindable name
assert all(addr.is_valid for addr in addrs)
def test_get_random_honors_bindspace():
addr: TIPCAddress = TIPCAddress.get_random(
bindspace=TIPC_NODE_SCOPE,
)
assert addr.bindspace == TIPC_NODE_SCOPE == addr._scope
def test_wrap_address_dispatches_on_the_proto_key():
'''
The proto-keyed unwrapped form must round-trip through the
*global* `wrap_address()` and NOT get stolen by `tcp`s
`(str(), int())` case nor `uds`s `(_, str())` one.
'''
addr: TIPCAddress = TIPCAddress.get_random()
assert wrap_address(addr.unwrap()) == addr
# ..and via the `list` form `msgpack` decodes to
assert wrap_address(list(addr.unwrap())) == addr
assert _addr._address_types['tipc'] is TIPCAddress
assert _addr.get_address_cls('tipc') is TIPCAddress
# the host-singleton registrar default is import-time cheap
# (no kernel module, no I/O) and mirrors the `1616` idiom
assert _addr._default_lo_addrs['tipc'] == (
'tipc', TRACTOR_STYPE, 1616, TIPC_CLUSTER_SCOPE,
)
def test_maddr_roundtrip():
'''
Interim `str`-only `/tipc/` maddr grammar (there's no
registered `/tipc` multiaddr proto yet, gh #483), which
`parse_maddr()` special-cases before `Multiaddr()` ever sees
the string.
'''
addr: TIPCAddress = TIPCAddress.get_random()
maddr: str = mk_maddr(addr)
assert isinstance(maddr, str)
assert maddr == (
f'/tipc/{addr._stype}/{addr._instance}/{addr._scope}'
)
assert parse_maddr(maddr) == addr
# ..and through the generic entrypoint
assert wrap_address(maddr) == addr
def test_eafnosupport_is_actionable_connerr(
monkeypatch: pytest.MonkeyPatch,
):
'''
With no `tipc` module the kernel answers `EAFNOSUPPORT`; that
MUST surface as a `ConnectionError` naming the fix rather than
a bare `OSError`.
'''
class _NoTIPCKernel:
@staticmethod
def socket(*args, **kwargs):
raise OSError(
errno.EAFNOSUPPORT,
'Address family not supported by protocol',
)
monkeypatch.setattr(_tipc, 'trio_socket', _NoTIPCKernel)
async def main():
await start_listener(addr=TIPCAddress.get_root())
with pytest.raises(ConnectionError) as excinfo:
trio.run(main)
report: str = str(excinfo.value)
assert 'modprobe tipc' in report
assert type(excinfo.value.__cause__) is OSError
# ------------------------------------------------------------------
# kernel-touching
# ------------------------------------------------------------------
@requires_tipc
def test_listener_tolerates_so_acceptconn():
'''
`trio.SocketListener.__init__` asserts
`getsockopt(SOL_SOCKET, SO_ACCEPTCONN)` is truthy, suppressing
`OSError` for exotic families.
Pin which of the two branches `AF_TIPC` actually takes (plan 01
§3.1 left it as an assumption) so a kernel-side regression is
caught here rather than as a mystery bind failure.
'''
async def main():
addr: TIPCAddress = TIPCAddress.get_random()
lstnr = await start_listener(addr=addr)
try:
assert lstnr.socket.getsockopt(
SOL_SOCKET,
SO_ACCEPTCONN,
)
finally:
lstnr.socket.close()
trio.run(main)
@requires_tipc
def test_bind_publishes_a_dialable_service_name():
'''
"Publishing a bind IS registration": `.bind()` a singleton
name-seq and a second task resolves it by *name* with NO
`tractor` registrar in the loop.
This is the core #378 property.
'''
async def main():
addr: TIPCAddress = TIPCAddress.get_random()
lstnr = await start_listener(addr=addr)
accepted: list = []
async def _accept():
stream = await lstnr.accept()
accepted.append(stream)
await stream.send_all(b'woof')
await stream.aclose()
async with trio.open_nursery() as tn:
tn.start_soon(_accept)
await trio.sleep(0.05)
sock = _tipc.trio_socket.socket(
AF_TIPC,
SOCK_STREAM,
)
# NOTE, connect by *name* -> the kernel does the
# lookup, i.e. this call IS the discovery query.
await sock.connect((
TIPC_ADDR_NAME,
addr._stype,
addr._instance,
0, # domain: 0 == "anywhere in scope"
addr._scope,
))
stream = trio.SocketStream(sock)
assert await stream.receive_some(16) == b'woof'
await stream.aclose()
assert len(accepted) == 1
lstnr.socket.close()
trio.run(main)
@requires_tipc
def test_getsockname_is_a_port_id_not_the_bound_name():
'''
The reason `TIPCAddress.rebind_from_sockname` is `False`.
A NAMESEQ-bound listener's `getsockname()` answers a
`TIPC_ADDR_ID` port-id, which never equals `.unwrap()` and
cannot be wrapped back into a service name.
'''
async def main():
addr: TIPCAddress = TIPCAddress.get_random()
lstnr = await start_listener(addr=addr)
try:
sockname: tuple = lstnr.socket.getsockname()
assert sockname[0] == TIPC_ADDR_ID
assert sockname != addr.unwrap()
with pytest.raises(ValueError):
TIPCAddress.from_addr(sockname)
finally:
lstnr.socket.close()
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():
'''
Unlike every other backend, TIPC permits *two* publishers of
one service name and round-robins connects between them.
Pin that observed behaviour it's the whole reason
`.get_random()` bothers with a well-spread digest, and a
future kernel that starts raising `EADDRINUSE` here would be
very good news worth noticing.
'''
async def main():
addr: TIPCAddress = TIPCAddress.get_random()
first = await start_listener(addr=addr)
second = await start_listener(addr=addr)
try:
assert first.socket.getsockname() != second.socket.getsockname()
finally:
first.socket.close()
second.socket.close()
trio.run(main)

View File

@ -99,6 +99,20 @@ def get_rando_addr(
assert addr.sockpath.resolve() assert addr.sockpath.resolve()
testrun_reg_addr = addr.unwrap() testrun_reg_addr = addr.unwrap()
# NOTE, `.get_random()` already derives the service
# *instance* from a `uuid4`+pid-salted seed, so both the
# within- and cross-proc isolation the other 2 protos
# hand-roll above comes for free.
#
# XXX matters MORE here than for tcp/uds: a TIPC name
# clash doesn't raise `EADDRINUSE`, it silently
# round-robins connects between both publishers.
case 'tipc':
from tractor.ipc._tipc import TIPCAddress
addr: TIPCAddress = addr_type.get_random()
assert addr.is_valid
testrun_reg_addr = addr.unwrap()
# XXX, as sanity it should never the same as the default for the # XXX, as sanity it should never the same as the default for the
# host-singleton registry actor. # host-singleton registry actor.
assert def_reg_addr != testrun_reg_addr assert def_reg_addr != testrun_reg_addr

View File

@ -501,6 +501,12 @@ def pytest_configure(
'trio: legacy mark for tests meant to run under the `trio` ' 'trio: legacy mark for tests meant to run under the `trio` '
'spawn backend (e.g. `test_local.py`).' 'spawn backend (e.g. `test_local.py`).'
) )
config.addinivalue_line(
'markers',
'tipc: test targets the `AF_TIPC` tpt backend; the kernel- '
'touching cases self-skip unless the `tipc` module is loaded '
'(`sudo modprobe tipc`).'
)
# `--enable-stackscope`: install SIGUSR1 → trio task-tree # `--enable-stackscope`: install SIGUSR1 → trio task-tree
# dump in pytest itself + propagate to every subactor via # dump in pytest itself + propagate to every subactor via
@ -797,6 +803,27 @@ def tpt_protos(
addr_type = _addr._address_types[proto_key] addr_type = _addr._address_types[proto_key]
assert addr_type.proto_key == proto_key assert addr_type.proto_key == proto_key
# XXX, generic capability gate: an env-dependent tpt
# whose backing kernel-mod/lib/netns isn't present here
# must fail LOUDLY and EARLY rather than as a few hundred
# confusing connect-timeouts downstream.
#
# Any `Address` type MAY expose `.is_available()`
# returning `(ok, why_not)`; absence means "always
# available" (i.e. tcp/uds).
is_avail = getattr(
addr_type,
'is_available',
None,
)
if is_avail:
avail, why_not = is_avail()
if not avail:
pytest.fail(
f'--tpt-proto={proto_key!r} is NOT usable here!\n'
f'{why_not}\n'
)
yield proto_keys yield proto_keys

View File

@ -215,7 +215,6 @@ def pformat_caller_frame(
tb_str: str = pformat_boxed_tb( tb_str: str = pformat_boxed_tb(
tb_str=tb_str, tb_str=tb_str,
field_prefix=' ', field_prefix=' ',
indent='',
) )
return tb_str return tb_str

View File

@ -33,6 +33,7 @@ from ..runtime._state import (
) )
from ..ipc._tcp import TCPAddress from ..ipc._tcp import TCPAddress
from ..ipc._uds import UDSAddress from ..ipc._uds import UDSAddress
from ..ipc._tipc import TIPCAddress
if TYPE_CHECKING: if TYPE_CHECKING:
from ..runtime._runtime import Actor from ..runtime._runtime import Actor
@ -65,9 +66,22 @@ log = get_logger()
# #
UnwrappedAddress = ( UnwrappedAddress = (
# tcp/udp/uds # tcp/udp/uds
# ('127.0.0.1', 1616)
# ('/run/user/1000/tractor', 'registry@1616.sock')
#
# ..and the explicitly proto-keyed (`multiaddr`-spelled)
# form, which is where ALL backends should eventually land
# per the note below,
# ('tipc', 1953628160, 1616, 2)
#
# XXX VARIADIC bc `msgspec` refuses a union of >1 array-like
# type, so the two shapes can't be spelled as a union. Keep
# in sync with `.msg.types.UnwrappedAddress` which
# re-declares this to dodge a circular import AND is what
# actually validates the `SpawnSpec` wire msg!
tuple[ tuple[
str, # host/domain(tcp), filesys-dir(uds) str|int,
int|str, # port/path(uds) ...,
] ]
# ?TODO? should we also include another 2 fields from # ?TODO? should we also include another 2 fields from
# our `Aid` msg such that we include the runtime `Actor.uid` # our `Aid` msg such that we include the runtime `Actor.uid`
@ -83,6 +97,17 @@ class Address(Protocol):
proto_key: ClassVar[str] proto_key: ClassVar[str]
unwrapped_type: ClassVar[UnwrappedAddress] unwrapped_type: ClassVar[UnwrappedAddress]
# whether `.ipc._server.Endpoint.start_listener()` should
# reconcile a bound `.addr` against its listener's
# `socket.getsockname()`.
#
# XXX NOTE, that reconciliation exists ONLY to learn the
# kernel-*assigned* port from a `port=0` tcp bind; a backend
# whose `getsockname()` reports a categorically different thing
# than what was `.bind()`ed must opt out with `False`, else the
# ep's addr gets clobbered by an un-dialable one.
rebind_from_sockname: ClassVar[bool]
# TODO, i feel like an `.is_bound()` is a better thing to # TODO, i feel like an `.is_bound()` is a better thing to
# support? # support?
# Lke, what use does this have besides a noop and if it's not # Lke, what use does this have besides a noop and if it's not
@ -172,7 +197,8 @@ class Address(Protocol):
_address_types: bidict[str, Type[Address]] = { _address_types: bidict[str, Type[Address]] = {
'tcp': TCPAddress, 'tcp': TCPAddress,
'uds': UDSAddress 'uds': UDSAddress,
'tipc': TIPCAddress,
} }
@ -184,6 +210,9 @@ _default_lo_addrs: dict[
] = { ] = {
'tcp': TCPAddress.get_root().unwrap(), 'tcp': TCPAddress.get_root().unwrap(),
'uds': UDSAddress.get_root().unwrap(), 'uds': UDSAddress.get_root().unwrap(),
# NOTE, pure/cheap: a service-name pair, no kernel module
# nor I/O required at import time.
'tipc': TIPCAddress.get_root().unwrap(),
} }
@ -229,6 +258,16 @@ def wrap_address(
# import pdbp; pdbp.set_trace() # import pdbp; pdbp.set_trace()
match addr: match addr:
# XXX, the explicitly proto-keyed form (spelled with the
# `multiaddr` proto name) which is where ALL backends
# should eventually land per the `UnwrappedAddress`
# migration note above.
#
# NOTE, a bare seq-pattern matches `list` too, which is
# what `msgpack` decodes our tuples back to.
case ('tipc', *_):
cls = TIPCAddress
# classic network socket-address as tuple/list # classic network socket-address as tuple/list
case ( case (
(str(), int()) (str(), int())

View File

@ -38,8 +38,23 @@ if TYPE_CHECKING:
_tpt_proto_to_maddr: dict[str, str] = { _tpt_proto_to_maddr: dict[str, str] = {
'tcp': 'tcp', 'tcp': 'tcp',
'uds': 'unix', 'uds': 'unix',
'tipc': 'tipc',
} }
# XXX, there is NO `/tipc` in the multiaddr protocol table yet
# (upstream track: gh #483 + multiformats/py-multiaddr#107), and
# `Multiaddr()` rejects an unregistered proto name outright.
#
# So until that lands `tipc` maddrs stay **`str`**-only — which
# `MsgTransport.maddr`s `Multiaddr|str` return type already
# allows and `MsgpackUDSStream.maddr` already exercises — and
# `parse_maddr()` special-cases the prefix BEFORE handing
# anything to `Multiaddr()`.
#
# This is also why gh #443's "always return `Multiaddr`" item
# stays blocked.
_tipc_maddr_prefix: str = '/tipc/'
# reverse mapping: multiaddr protocol name -> tractor proto_key # reverse mapping: multiaddr protocol name -> tractor proto_key
_maddr_to_tpt_proto: dict[str, str] = { _maddr_to_tpt_proto: dict[str, str] = {
v: k for k, v in _tpt_proto_to_maddr.items() v: k for k, v in _tpt_proto_to_maddr.items()
@ -49,7 +64,7 @@ _maddr_to_tpt_proto: dict[str, str] = {
def mk_maddr( def mk_maddr(
addr: 'Address', addr: 'Address',
) -> Multiaddr: ) -> Multiaddr|str:
''' '''
Construct a `Multiaddr` from a tractor `Address` instance, Construct a `Multiaddr` from a tractor `Address` instance,
dispatching on the `.proto_key` to build the correct dispatching on the `.proto_key` to build the correct
@ -75,6 +90,18 @@ def mk_maddr(
f'/{net_proto}/{host}/{maddr_proto}/{port}' f'/{net_proto}/{host}/{maddr_proto}/{port}'
) )
# NOTE, interim `str`-only grammar (see the
# `_tipc_maddr_prefix` note above),
#
# /tipc/<stype>/<instance>/<scope>
#
# mirroring how `uds` maps onto the spec-legal `/unix`.
case 'tipc':
_, stype, instance, scope = addr.unwrap()
return (
f'/{maddr_proto}/{stype}/{instance}/{scope}'
)
case 'uds': case 'uds':
filedir, filename = addr.unwrap() filedir, filename = addr.unwrap()
filepath = Path(filedir) / filename filepath = Path(filedir) / filename
@ -100,6 +127,17 @@ def parse_maddr(
# lazy imports to avoid circular deps # lazy imports to avoid circular deps
from tractor.ipc._tcp import TCPAddress from tractor.ipc._tcp import TCPAddress
from tractor.ipc._uds import UDSAddress from tractor.ipc._uds import UDSAddress
from tractor.ipc._tipc import TIPCAddress
# XXX MUST come before `Multiaddr()` which rejects the
# not-yet-registered `/tipc` proto name outright.
if maddr_str.startswith(_tipc_maddr_prefix):
_, _, stype, instance, scope = maddr_str.split('/')
return TIPCAddress(
_stype=int(stype),
_instance=int(instance),
_scope=int(scope),
)
maddr = Multiaddr(maddr_str) maddr = Multiaddr(maddr_str)
proto_names: list[str] = [ proto_names: list[str] = [

View File

@ -661,7 +661,16 @@ class Endpoint(Struct):
# NOTE, for handling the resolved non-0 port for # NOTE, for handling the resolved non-0 port for
# TCP/UDP network sockets. # TCP/UDP network sockets.
#
# XXX, gated on the addr-type's opt-in since for some
# backends `getsockname()` does NOT answer "the addr you
# bound"; `tipc` reports a `TIPC_ADDR_ID` port-id instead
# of the published name-seq, so rebinding from it would
# replace a dialable service-name with an un-dialable
# (and un-reconstructable) port-id.
if ( if (
self.addr.rebind_from_sockname
and
(unwrapped := lstnr.socket.getsockname()) (unwrapped := lstnr.socket.getsockname())
!= !=
self.addr.unwrap() self.addr.unwrap()

View File

@ -65,6 +65,10 @@ class TCPAddress(
unwrapped_type: ClassVar[type] = tuple[str, int] unwrapped_type: ClassVar[type] = tuple[str, int]
def_bindspace: ClassVar[str] = '127.0.0.1' def_bindspace: ClassVar[str] = '127.0.0.1'
# XXX, REQUIRED here since a `port=0` bind means the kernel
# picks and `getsockname()` is the only way we learn it.
rebind_from_sockname: ClassVar[bool] = True
# ?TODO, actually validate ipv4/6 with stdlib's `ipaddress` # ?TODO, actually validate ipv4/6 with stdlib's `ipaddress`
@property @property
def is_valid(self) -> bool: def is_valid(self) -> bool:

View File

@ -0,0 +1,777 @@
# tractor: distributed structured concurrency.
# Copyright 2018-eternity Tyler Goodlet.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
`AF_TIPC` (Transparent Inter-Process Communication) implementation of
the `tractor.ipc._transport.MsgTransport` protocol.
TIPC is a linux-kernel cluster IPC protocol whose *service names* are
published in a cluster-wide name-table by the kernel itself. That
makes a `.bind()` literally a service **registration** and
a `.connect()` literally a service **lookup**, i.e. the discovery
machinery `tractor.discovery` normally implements with a registrar
actor comes for free, in-kernel.
An actor's TIPC address is therefore a *service name* pair,
`(stype, instance)`,
- a listener `.bind()`s the singleton published range
`(stype, instance, instance)` as a `TIPC_ADDR_NAMESEQ`,
- a peer `.connect()`s that name as a `TIPC_ADDR_NAME` and the kernel
resolves it,
- `TIPC_ADDR_ID` (a `(node, ref)` port-id) is only ever an *observed*
address, never a user-facing one.
NOTE, the `tipc` kernel module is NOT loaded by default; see
`is_tipc_available()` and the `sudo modprobe tipc` hint carried in
this module's `ConnectionError` messages.
Normative refs are the kernel sources (the tipc.io docs are stale),
- `include/uapi/linux/tipc.h`
- `net/tipc/socket.c`
'''
from __future__ import annotations
from contextlib import (
contextmanager as cm,
)
import errno
from hashlib import blake2b
import os
import socket
from socket import SOCK_STREAM
from typing import (
Callable,
ClassVar,
Type,
TYPE_CHECKING,
)
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,
)
if TYPE_CHECKING:
from tractor.runtime._runtime import Actor
log = get_logger()
# XXX, `AF_TIPC` and every `TIPC_*` constant are linux-ONLY in
# CPython's `socketmodule.c`. Mirror the `_uds.py` `SO_PASSCRED`
# precedent and fall back to the uapi values so this module stays
# **importable everywhere** — `.discovery._addr` builds its
# registration tables at import time (contract §2.3) — while
# `is_tipc_available()` remains the single *runtime* gate.
#
# values verified against `include/uapi/linux/tipc.h`
try:
from socket import (
AF_TIPC,
SOL_TIPC,
TIPC_ADDR_ID,
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,
)
except ImportError:
AF_TIPC: int = 30
SOL_TIPC: int = 271
TIPC_ADDR_NAMESEQ: int = 1
TIPC_ADDR_NAME: int = 2
TIPC_ADDR_ID: int = 3
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
# ascii 'tr' in the high half and leaving the low 16b free for
# app-side partitioning via an explicit `TIPCAddress._stype`.
#
# NOTE, two `tractor` trees sharing BOTH a cluster and an `_stype`
# share a service-name space; see `.get_random()` on why that's
# only a *probabilistic* hazard.
TRACTOR_STYPE: int = 0x74_72_00_00
# TIPC reserves service-*types* 0..63 for its own internal services
# (`TIPC_CFG_SRV == 0`, `TIPC_TOP_SRV == 1`); see
# `include/uapi/linux/tipc.h`.
_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 `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',
TIPC_NODE_SCOPE: 'node',
}
# see `is_tipc_available()`
_tipc_avail: bool|None = None
def is_tipc_available() -> bool:
'''
`True` iff this kernel can create an `AF_TIPC` socket, i.e. the
`tipc` module is loaded (`sudo modprobe tipc`).
Pure predicate; no side effects, no logging. The answer can't
change without a `modprobe` so it's memoized after the first
(one syscall) probe.
'''
global _tipc_avail
if _tipc_avail is None:
try:
socket.socket(
AF_TIPC,
SOCK_STREAM,
).close()
_tipc_avail = True
except OSError:
_tipc_avail = False
return _tipc_avail
class TIPCAddress(
msgspec.Struct,
frozen=True,
):
'''
A TIPC *service name* as an address, i.e. the
`(type, instance)` pair a listener publishes and a peer
resolves, plus the optionally-*observed* `TIPC_ADDR_ID`
port-id of a live connection.
'''
_stype: int
_instance: int
_scope: int = TIPC_CLUSTER_SCOPE
# observed-only, from a `TIPC_ADDR_ID` `getsockname()`/
# `getpeername()`; excluded from `.unwrap()` exactly like
# `UDSAddress.maybe_pid`.
maybe_node: int|None = None
maybe_ref: int|None = None
proto_key: ClassVar[str] = 'tipc'
unwrapped_type: ClassVar[type] = tuple[str, int, int, int]
def_bindspace: ClassVar[int] = TIPC_CLUSTER_SCOPE
# XXX, TIPC's `getsockname()` answers a `TIPC_ADDR_ID` port-id
# and NEVER the name-seq we bound, so the `Endpoint`-level
# reconciliation would clobber a dialable service-name with an
# un-dialable port-id. There's also nothing to learn: unlike
# tcp's `port=0` there is no kernel-assigned-name analogue.
rebind_from_sockname: ClassVar[bool] = False
@property
def bindspace(self) -> int:
'''
The TIPC *scope*, i.e. literally "the set of hosts from
which this published name is reachable": `TIPC_NODE_SCOPE`
for same-host-only (the UDS analogue),
`TIPC_CLUSTER_SCOPE` for cluster-visible.
'''
return self._scope
@property
def is_valid(self) -> bool:
'''
Is this a *publishable/dialable* service name?
NOTE the `> 0` (rather than `!= 0`) guards double-duty as
the `TIPC_NAME_UNKNOWN` reject, i.e. an addr merely
*observed* off a peer's port-id is never dialable.
'''
return (
self._instance > 0
and
self._stype > 0
and
self._stype not in _tipc_reserved_stypes
and
self._scope in (
TIPC_NODE_SCOPE,
TIPC_CLUSTER_SCOPE,
)
)
@classmethod
def from_addr(
cls,
addr: tuple[str, int, int, int],
) -> TIPCAddress:
match addr:
# our proto-keyed unwrapped form, w/ scope optional
case (
('tipc', int() as stype, int() as inst, int() as scope)
|
['tipc', int() as stype, int() as inst, int() as scope]
):
return TIPCAddress(
_stype=stype,
_instance=inst,
_scope=_norm_scope(scope),
)
case (
('tipc', int() as stype, int() as inst)
|
['tipc', int() as stype, int() as inst]
):
return TIPCAddress(
_stype=stype,
_instance=inst,
)
# a kernel-observed `TIPC_ADDR_ID` 5-tuple.
#
# XXX, a port-id carries NO service-name info, so we
# cannot reconstruct `(stype, instance)` from it. This
# is exactly why `.rebind_from_sockname` is `False`;
# if you land here something re-enabled that path.
case (int() as atype, *_) if atype == TIPC_ADDR_ID:
raise ValueError(
f'Can not wrap a bare TIPC_ADDR_ID port-id !\n'
f'addr: {addr!r}\n'
f'\n'
f'A port-id carries no service-name, so the\n'
f'`(stype, instance)` identity is unrecoverable.\n'
f'Use `.with_port_id()` to *annotate* a known\n'
f'{cls.__name__} instead.\n'
)
case _:
raise TypeError(
f'Bad unwrapped-address for {cls} !\n'
f'{addr!r}\n'
)
def unwrap(self) -> tuple[str, int, int, int]:
# NOTE, proto-keyed (w/ the `multiaddr` proto spelling) so
# `wrap_address()` can dispatch unambiguously against the
# other backends' 2-tuple forms; see contract §1.1.
return (
'tipc',
self._stype,
self._instance,
self._scope,
)
def with_port_id(
self,
node: int,
ref: int,
) -> TIPCAddress:
'''
A copy annotated with an *observed* `TIPC_ADDR_ID`
port-id, purely for logging/`__repr__`.
'''
return msgspec.structs.replace(
self,
maybe_node=node,
maybe_ref=ref,
)
@classmethod
def get_random(
cls,
bindspace: int|None = None,
) -> TIPCAddress:
'''
A per-subactor ephemeral service-name.
XXX, TIPC has NO kernel-assigned-instance analogue of tcp's
`port=0`, so we must choose the instance ourselves and a
clash does **not** raise `EADDRINUSE`: TIPC happily accepts
multiple publishers of one name and round-robins connects
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
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
escalation (post-bind verification) if that ever bites.
'''
pid: int = os.getpid()
actor: Actor|None = current_actor(
err_on_no_runtime=False,
)
if actor:
seed: str = f'{actor.aid.name}@{pid}'
else:
if is_root_process():
prefix: str = 'no_runtime_root'
else:
prefix: str = 'no_runtime_actor'
# XXX, no live actor -> no `Aid` to key off, so mix
# a per-CALL token in; w/o it the seed degenerates to
# a pure fn of `(prefix, pid)` and two calls in one
# proc alias to the SAME service name — the `_uds.py`
# `.get_random()` hazard, but silent here.
seed: str = f'{prefix}.{uuid4().hex[:8]}@{pid}'
return TIPCAddress(
_stype=TRACTOR_STYPE,
_instance=instance_from_seed(seed),
_scope=(
bindspace
if bindspace is not None
else cls.def_bindspace
),
)
@classmethod
def is_available(cls) -> tuple[bool, str]:
'''
Generic tpt-capability hook: `(ok, why_not)`.
Consumed by the `tpt_protos` test fixture so a
`--tpt-proto tipc` run on a box with no `tipc` module
fails loudly and early rather than as a few hundred
confusing connect timeouts. Apps can use it too.
NOTE, deliberately spelled generically (NOT `is_tipc_*`)
so the sibling env-dependent backends `quic`/`iroh`
(gh #353) and the `wg` netns bindspace (gh #482) — get
the same gate for free.
'''
if is_tipc_available():
return (True, '')
return (
False,
'the `tipc` kernel module is not loaded'
' |_try: `sudo modprobe tipc`',
)
@classmethod
def get_root(cls) -> TIPCAddress:
# NOTE, `1616` mirrors `TCPAddress.get_root()`s port and
# the UDS `registry@1616.sock` filename so the "1616 is
# tractor's registrar" idiom holds across all backends.
return TIPCAddress(
_stype=TRACTOR_STYPE,
_instance=1616,
_scope=TIPC_CLUSTER_SCOPE,
)
def __repr__(self) -> str:
if self._instance == TIPC_NAME_UNKNOWN:
name: str = '<unknown-service>'
else:
name: str = f'0x{self._stype:08x}:{self._instance}'
body: str = (
f'{name}, {_scope_names.get(self._scope, self._scope)}'
)
if (node := self.maybe_node) is not None:
body += f', @0x{node:08x}:{self.maybe_ref}'
return (
f'{type(self).__name__}'
f'['
f'{body}'
f']'
)
def instance_from_seed(seed: str) -> int:
'''
Derive a TIPC service *instance* from an actor-identity `seed`.
A `blake2b` digest folded into `[64, 2**32)`; the low values are
skipped to stay clear of TIPC's own reserved numbering
conventions.
'''
inst: int = int.from_bytes(
blake2b(
seed.encode(),
digest_size=4,
).digest(),
'big',
)
return 64 + (inst % (2**32 - 64))
def _norm_scope(scope: int) -> int:
'''
Normalize a `TIPC_*_SCOPE` value.
`TIPC_ZONE_SCOPE` is deprecated and aliased to cluster-scope by
modern kernels; accept it on input and fold it.
'''
if scope == TIPC_ZONE_SCOPE:
log.transport(
f'Normalizing deprecated TIPC_ZONE_SCOPE -> cluster\n'
f'scope: {scope!r}\n'
)
return TIPC_CLUSTER_SCOPE
return scope
@cm
def _reraise_as_connerr(
src_excs: tuple[Type[Exception]],
addr: TIPCAddress,
):
'''
Normalize TIPC's `OSError`s into `ConnectionError`s.
XXX REQUIRED, not polish: TIPC answers a lookup for an
unpublished name with `EHOSTUNREACH` which python maps to a
**bare** `OSError`, NOT a `ConnectionError` subtype (unlike
`ECONNREFUSED` -> `ConnectionRefusedError`). Contract §4's
discovery-ping path requires the `ConnectionError` shape.
'''
try:
yield
except src_excs as src_exc:
match src_exc.errno:
case errno.EAFNOSUPPORT:
why: str = (
'TIPC unavailable — is the kernel module loaded?\n'
' |_try: `sudo modprobe tipc`\n'
)
case errno.EHOSTUNREACH:
why: str = (
'No TIPC publisher for this service name\n'
' |_nothing has `.bind()`ed it in-scope\n'
)
case _:
why: str = 'Bad TIPC service-name-as-address ??\n'
raise ConnectionError(
f'{why}'
f'{addr}\n'
f'\n'
f'from src: {src_exc!r}\n'
) from src_exc
async def start_listener(
addr: TIPCAddress,
backlog: int = 128,
**kwargs,
) -> SocketListener:
'''
Publish `addr` as a TIPC service name and listen on it.
The `.bind()` of a singleton `TIPC_ADDR_NAMESEQ` range
`(stype, instance, instance)` **is** the service registration
it's what shows up in `tipc nametable show` and what a peer's
`.connect()`-by-name resolves against.
NOTE, unlike every other backend a duplicate bind does NOT
raise: TIPC permits multiple publishers of one name and
round-robins connects between them. See
`TIPCAddress.get_random()`.
'''
log.info(
f'Attempting to publish TIPC service name\n'
f'>[\n'
f'|_{addr}\n'
)
with _reraise_as_connerr(
src_excs=(OSError,),
addr=addr,
):
sock = trio_socket.socket(
AF_TIPC,
SOCK_STREAM,
)
await sock.bind((
TIPC_ADDR_NAMESEQ,
addr._stype,
addr._instance, # lower
addr._instance, # upper
addr._scope,
))
# NOTE, backlog matches `_uds.start_listener()`'s hard-won
# value; a backlog of 1 overflows during concurrent
# deregistration storms at actor-tree teardown.
sock.listen(backlog)
log.info(
f'Published TIPC service name\n'
f'[>\n'
f' |_{addr}\n'
)
return SocketListener(sock)
# NOTE, deliberately NO `close_listener()`: there's no filesys
# 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,
)

View File

@ -38,17 +38,23 @@ from tractor.ipc._uds import (
UDSAddress, UDSAddress,
MsgpackUDSStream, MsgpackUDSStream,
) )
from tractor.ipc._tipc import (
AF_TIPC,
TIPCAddress,
MsgpackTIPCStream,
)
# if TYPE_CHECKING: # if TYPE_CHECKING:
# from tractor._addr import Address # from tractor._addr import Address
Address = TCPAddress|UDSAddress Address = TCPAddress|UDSAddress|TIPCAddress
# manually updated list of all supported msg transport types # manually updated list of all supported msg transport types
_msg_transports = [ _msg_transports = [
MsgpackTCPStream, MsgpackTCPStream,
MsgpackUDSStream MsgpackUDSStream,
MsgpackTIPCStream,
] ]
@ -59,15 +65,17 @@ _key_to_transport: dict[
] = { ] = {
('msgpack', 'tcp'): MsgpackTCPStream, ('msgpack', 'tcp'): MsgpackTCPStream,
('msgpack', 'uds'): MsgpackUDSStream, ('msgpack', 'uds'): MsgpackUDSStream,
('msgpack', 'tipc'): MsgpackTIPCStream,
} }
# convert an Address wrapper to its corresponding transport type # convert an Address wrapper to its corresponding transport type
_addr_to_transport: dict[ _addr_to_transport: dict[
Type[TCPAddress|UDSAddress], Type[TCPAddress|UDSAddress|TIPCAddress],
Type[MsgTransport] Type[MsgTransport]
] = { ] = {
TCPAddress: MsgpackTCPStream, TCPAddress: MsgpackTCPStream,
UDSAddress: MsgpackUDSStream, UDSAddress: MsgpackUDSStream,
TIPCAddress: MsgpackTIPCStream,
} }
@ -108,6 +116,12 @@ def transport_from_stream(
case socket.AF_UNIX: case socket.AF_UNIX:
transport = 'uds' transport = 'uds'
# NOTE, `AF_TIPC` is linux-only in CPython so we
# match the `._tipc` constant (which carries a uapi
# fallback) rather than `socket.AF_TIPC`.
case _ if sock.family == AF_TIPC:
transport = 'tipc'
case _: case _:
raise NotImplementedError( raise NotImplementedError(
f'Unsupported socket family: {sock.family}' f'Unsupported socket family: {sock.family}'

View File

@ -117,6 +117,12 @@ class UDSAddress(
unwrapped_type: ClassVar[type] = tuple[str, int] unwrapped_type: ClassVar[type] = tuple[str, int]
def_bindspace: ClassVar[Path] = get_rt_dir() def_bindspace: ClassVar[Path] = get_rt_dir()
# NOTE, `getsockname()` answers the sock-file path as a `str`
# which never `==` our 2-tuple `.unwrap()`, so the round-trip
# always fires; it's a no-op modulo `.maybe_pid` and is kept
# `True` to preserve pre-existing behaviour exactly.
rebind_from_sockname: ClassVar[bool] = True
@property @property
def bindspace(self) -> Path: def bindspace(self) -> Path:
''' '''

View File

@ -188,6 +188,31 @@ class Aid(
__repr__ = pretty_struct.Struct.__repr__ __repr__ = pretty_struct.Struct.__repr__
# NOTE, mirrors `.discovery._addr.UnwrappedAddress` but is
# re-declared here to dodge the circular import
# (`._addr` -> `.ipc._tcp` -> `.msg`).
#
# XXX this is the **wire** shape, so widening it is a wire-format
# change; keep the two decls in sync.
# XXX VARIADIC on purpose! `msgspec` rejects a union holding
# more than one array-like type ("Type unions may not contain
# more than one array-like (list, set, tuple) type"), so the two
# concrete shapes,
#
# ('127.0.0.1', 1616) # tcp
# ('/run/user/1000/tractor', 'x.sock') # uds
# ('tipc', 1953628160, 1616, 2) # proto-keyed (tipc)
#
# can't be spelled as `tuple[str, str|int]|tuple[str, int, int,
# int]`. Widen to one homogeneous variadic tuple instead.
#
# ?TODO, the real fix is the `UnwrappedAddress` proto-key
# migration (see `.discovery._addr`) after which this becomes a
# tagged union keyed off elem 0 and full per-proto validation
# comes back.
UnwrappedAddress = tuple[str|int, ...]
class SpawnSpec( class SpawnSpec(
pretty_struct.Struct, pretty_struct.Struct,
tag=True, tag=True,
@ -213,8 +238,8 @@ class SpawnSpec(
# TODO: not just sockaddr pairs? # TODO: not just sockaddr pairs?
# -[ ] abstract into a `TransportAddr` type? # -[ ] abstract into a `TransportAddr` type?
reg_addrs: list[tuple[str, str|int]] reg_addrs: list[UnwrappedAddress]
bind_addrs: list[tuple[str, str|int]]|None bind_addrs: list[UnwrappedAddress]|None
# TODO: caps based RPC support in the payload? # TODO: caps based RPC support in the payload?

View File

@ -47,6 +47,7 @@ if TYPE_CHECKING:
TransportProtocolKey = Literal[ TransportProtocolKey = Literal[
'tcp', 'tcp',
'uds', 'uds',
'tipc',
] ]
_def_tpt_proto: TransportProtocolKey = 'tcp' _def_tpt_proto: TransportProtocolKey = 'tcp'

View File

@ -518,7 +518,7 @@ wheels = [
[[package]] [[package]]
name = "multiaddr" name = "multiaddr"
version = "0.2.0" version = "0.2.0"
source = { registry = "https://pypi.org/simple" } source = { git = "https://github.com/multiformats/py-multiaddr.git?rev=f86519daaa21699023d0037c58cdff600313dd09#f86519daaa21699023d0037c58cdff600313dd09" }
dependencies = [ dependencies = [
{ name = "base58" }, { name = "base58" },
{ name = "dnspython" }, { name = "dnspython" },
@ -533,10 +533,6 @@ dependencies = [
{ name = "trio-typing" }, { name = "trio-typing" },
{ name = "varint" }, { name = "varint" },
] ]
sdist = { url = "https://files.pythonhosted.org/packages/c7/10/4e26a8577cfce1c0febc8d83087e1373e93c695c6e73ad010546fb67e229/multiaddr-0.2.0.tar.gz", hash = "sha256:acb6b25c332ec1b2f1f8fef8d03a8c63385d34a87d690df0f4bba43cdf6efe8d", size = 58356, upload-time = "2026-03-17T21:51:00.274Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b5/13/56e503d01218d1ca27ea9fda862045a4b400cae5e756f47315f5aaba0eee/multiaddr-0.2.0-py3-none-any.whl", hash = "sha256:bcff7bf3d7de3d6da0b865b25423bcb411de1d20d70cc6abfacf75170d17866c", size = 40424, upload-time = "2026-03-17T21:50:58.833Z" },
]
[[package]] [[package]]
name = "mypy-extensions" name = "mypy-extensions"
@ -1178,7 +1174,7 @@ requires-dist = [
{ name = "bidict", specifier = ">=0.23.1" }, { name = "bidict", specifier = ">=0.23.1" },
{ name = "colorlog", specifier = ">=6.8.2,<7" }, { name = "colorlog", specifier = ">=6.8.2,<7" },
{ name = "msgspec", specifier = ">=0.20.0" }, { name = "msgspec", specifier = ">=0.20.0" },
{ name = "multiaddr", specifier = ">=0.2.0" }, { name = "multiaddr", git = "https://github.com/multiformats/py-multiaddr.git?rev=f86519daaa21699023d0037c58cdff600313dd09" },
{ name = "pdbp", specifier = ">=1.8.2,<2" }, { name = "pdbp", specifier = ">=1.8.2,<2" },
{ name = "platformdirs", specifier = ">=4.4.0" }, { name = "platformdirs", specifier = ">=4.4.0" },
{ name = "setproctitle", specifier = ">=1.3,<2" }, { name = "setproctitle", specifier = ">=1.3,<2" },