Compare commits

..

10 Commits

Author SHA1 Message Date
Gud Boi 1e9a110194 Fix TIPC integration after the `main` rebase
Carry the WG-aware py-multiaddr rev into the refreshed lock while
retaining `main`'s current dependency set.

Keep TIPC's `Multiaddr` annotation off the eager import path, and
extend lazy annotation checks for TIPC's interim `Multiaddr|str`
shape.

Prompt-IO: ai/prompt-io/opencode/20260830T045303Z_69a0e504_prompt_io.md

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:08:10 -04:00
Gud Boi d9d7cc491a Draft the `/tipc` multiaddr protocol
Propose a fixed-width service endpoint carrying the TIPC type,
instance and publication scope, with one canonical structured value
that generic multiaddr parsers can compose normally.

Retain the kernel-standard `tipc` name while using “Cluster Domain
Sockets” as explanatory terminology. Document the binary and text
encodings, WireGuard composition, deployment-management boundary,
upstream sequence, test vector and open maintainer questions.

Prompt-IO: ai/prompt-io/opencode/20260819T003328Z_53516b09_prompt_io.md

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:46 -04:00
Gud Boi 65dae3ae51 Harden the `tipc` two-host smoke test
Turn the physical-host sketch into an operator runbook covering
cluster identity, interface and bearer setup, link validation,
failure/rejoin testing, diagnostic capture and cleanup.

Explain the cluster-domain-socket analogy and identify a future
`pyroute2` TIPC codec as the path from manual `tipc(8)` commands to
the same netlink management stack planned for WireGuard.

Authorize `host_a_srv` by its stable import name so direct script
execution does not expose only `__main__` while host B requests the
callable's actual `NamespacePath`.

Prompt-IO: ai/prompt-io/opencode/20260819T003327Z_53516b09_prompt_io.md

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:46 -04:00
Gud Boi c0826e00bf Make the `tipc` CI leg blocking
The refreshed PR matrix passes on Ubuntu with the TIPC kernel
module loaded, along with the TCP, UDS and macOS legs. Remove the
temporary `continue-on-error` expression so future TIPC
regressions block CI.

Prompt-IO: ai/prompt-io/opencode/20260819T003326Z_53516b09_prompt_io.md

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:46 -04:00
Gud Boi 44f5c46274 Abort `TIPC` topology streams on overflow
Keep `_stream_name_events()` non-blocking so a slow memory-channel
consumer cannot back up the kernel topology queue. Raise
`TIPCNameEventOverflow` and end the subscription rather than drop a
transition or let the socket reader stall. Discovery consumers must
then resubscribe and rebuild their name-table view.

Also,
- document topology semantics and scope with Linux references
- diagram the `.connect()`/`.getpeername()` withdrawal schedules
- explain the child-service and callable requirements in the
  two-host example

Review: PR #493 (copilot-pull-request-reviewer[bot],goodboy)
https://github.com/goodboy/tractor/pull/493

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:46 -04:00
Gud Boi 55ab72e7dc Fix `TIPC` two-host service example
A registrar root does not register itself in its own actor-name
registry, so host B could never discover the advertised `host_a`.
Boot that service as a child actor under the `TIPC` registrar
instead.

Import and pass the enabled `echo` callable to `.open_context()`;
the prior module-path string could not produce a `NamespacePath`.

Review: PR #493 (copilot-pull-request-reviewer[bot],goodboy)
https://github.com/goodboy/tractor/pull/493

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:46 -04:00
Gud Boi 0b439b850f Make `TIPCNameEvent` delivery lossless
Stop labeling topology events with caller-supplied scope that the
kernel never reports. Event addresses now carry an explicit unknown
scope instead of fabricated reachability.

Apply memory-channel backpressure rather than silently dropping
publish/withdraw transitions, and close the stream after delivering
the terminal event from a finite subscription.

Review: PR #493 (copilot-pull-request-reviewer[bot],goodboy)
https://github.com/goodboy/tractor/pull/493

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:46 -04:00
Gud Boi 5563baaf17 Harden `TIPC` socket setup races
Reject TIPC availability outside Linux before probing the fallback
socket-family integer, which can alias an unrelated family on
another OS.

Keep dialled sockets under setup ownership through transport
construction, then reuse the constructor's tolerant peer
observation. A peer withdrawing after `.connect()` can no longer
trigger a second raw `getpeername()` or leak setup resources.

Review: PR #493 (copilot-pull-request-reviewer[bot],goodboy)
https://github.com/goodboy/tractor/pull/493

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:46 -04:00
Gud Boi c798c94514 Key `TIPCAddress` instances by actor UUID
TIPC service names span the cluster while PIDs remain host-local.
Hashing only `(name, pid)` could therefore make same-named actors
on different hosts silently share one round-robin service name.

Derive the live-runtime seed from `Aid.uid` so the actor UUID
separates those names while keeping each identity reproducible.
Pin both properties with a deterministic regression test.

Review: PR #493 (copilot-pull-request-reviewer[bot],goodboy)
https://github.com/goodboy/tractor/pull/493

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:46 -04:00
Gud Boi 91e06910b6 Tighten `TIPC` address-shape dispatch
Restrict proto-key matching to numeric 3- or 4-element
descriptors so a UDS directory named `tipc` stays UDS.

Route `/tipc` parsing through `TIPCAddress.from_addr()` to
normalize zone scope and report malformed input clearly. Also
align UDS unwrapped metadata with its actual `(str, str)` shape.

Keep the TIPC test module portable by importing `SOL_TIPC` from
the backend's UAPI fallback instead of the host `socket` module.

Review: PR #493 (copilot-pull-request-reviewer[bot],goodboy)
https://github.com/goodboy/tractor/pull/493

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-09-01 20:02:44 -04:00
19 changed files with 977 additions and 81 deletions

View File

@ -96,12 +96,6 @@ jobs:
# are hard signals. Promote the test step to required once the
# suite is green.
# XXX the `tipc` leg stays NON-blocking until the `modprobe`
# step below is proven stable on GH's runners; flip this off
# once it's had a few green runs. See plan 01 §7.3 in
# `ai/tpt-backends/01_tipc_backend.md`.
continue-on-error: ${{ matrix.tpt_proto == 'tipc' }}
strategy:
fail-fast: false
matrix:

View File

@ -0,0 +1,29 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: a998acff-af66-4969-a65e-f77cbf5ef8dd
timestamp: 2026-08-19T00:33:26Z
git_ref: wkt/tipc_backend_378
scope: config
substantive: true
raw_file: 20260819T003326Z_53516b09_prompt_io.raw.md
---
## Prompt
Confirm whether the successful TIPC CI leg can become blocking.
## Response summary
Removed the temporary TIPC-only `continue-on-error` expression after
every refreshed PR #493 check passed.
## Files changed
- `.github/workflows/ci.yml` - make TIPC failures block CI.
## Human edits
The human explicitly requested reconsidering the gate after the CI
run and chose to leave the generated change local for review before
requesting this commit plan.

View File

@ -0,0 +1,23 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-19T00:33:26Z
git_ref: wkt/tipc_backend_378
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
The user asked whether the now-green TIPC CI leg could stop using
`continue-on-error`.
## Response
All checks on PR #493 passed, including the Ubuntu TIPC matrix job.
The generated config change removes the temporary non-blocking gate
and its obsolete explanatory comment:
> `git diff HEAD~1..HEAD -- .github/workflows/ci.yml`
The remaining matrix, Linux-only exclusion and `modprobe tipc` setup
stay unchanged.

View File

@ -0,0 +1,37 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: a998acff-af66-4969-a65e-f77cbf5ef8dd
timestamp: 2026-08-19T00:33:27Z
git_ref: wkt/tipc_backend_378
scope: code
substantive: true
raw_file: 20260819T003327Z_53516b09_prompt_io.raw.md
---
## Prompt
Develop a practical bootstrap plan for testing TIPC across two
physical motherboards.
## Response summary
Expanded the two-host example into a reproducible smoke-test runbook
and made host A authorize the stable module name imported by host B.
Documented the cluster-domain-socket analogy and the prospective
`pyroute2` management path.
## Files changed
- `examples/multihost/tipc_cluster/README.md` - document setup,
validation, resilience testing, capture and cleanup.
- `examples/multihost/tipc_cluster/host_a_srv.py` - authorize the
`host_a_srv` RPC module under direct script execution.
## Human edits
The human selected two separate physical motherboards as the target
environment and requested that the resulting local work be packaged
as its own commit boundary. The human also proposed “Cluster Domain
Sockets” as clearer terminology and reusing the planned `pyroute2`
dependency for TIPC management.

View File

@ -0,0 +1,27 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-19T00:33:27Z
git_ref: wkt/tipc_backend_378
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
The user asked for a bootstrap plan to try TIPC with two real,
physically separate Linux motherboards.
## Response
The generated changes turn the existing sketch into an operator
runbook and correct the RPC module authorization used by direct
script execution:
> `git diff HEAD~1..HEAD -- examples/multihost/tipc_cluster/README.md examples/multihost/tipc_cluster/host_a_srv.py`
The runbook covers matching revisions, cluster identity, interface
selection, Ethernet bearer setup, link and name-table validation,
RPC, failure/rejoin testing, diagnostics, cleanup and a future
network-namespace fixture. It also records “Cluster Domain Sockets”
as explanatory terminology and identifies a future `pyroute2` TIPC
generic-netlink codec as the path away from manual `tipc(8)` calls.

View File

@ -0,0 +1,37 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: a998acff-af66-4969-a65e-f77cbf5ef8dd
timestamp: 2026-08-19T00:33:28Z
git_ref: wkt/tipc_backend_378
scope: docs
substantive: true
raw_file: 20260819T003328Z_53516b09_prompt_io.raw.md
---
## Prompt
Draft an upstream `/tipc` multiaddr issue modeled on the prior `wg`
proposal.
## Response summary
Added a reviewable upstream issue draft with a concrete binary and
text encoding, composition examples, rollout plan and open design
questions. Retained the ecosystem-standard `tipc` name while using
“Cluster Domain Sockets” as explanatory terminology, and separated
the future `pyroute2` management track from address encoding.
## Files changed
- `ai/tpt-backends/04_tipc_multiaddr_upstream.md` - candidate
`multiformats/multiaddr` issue body.
- `ai/tpt-backends/README.md` - roster the upstream proposal.
## Human edits
The human chose to track the draft under `ai/tpt-backends/` instead
of leaving it in the ignored gish cache, and explicitly requested a
Prompt-IO record for this docs-only commit. The human proposed the
`cds` alternative and unifying TIPC/WireGuard management through
`pyroute2`.

View File

@ -0,0 +1,26 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-19T00:33:28Z
git_ref: wkt/tipc_backend_378
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
The user asked to begin drafting an upstream `/tipc` multiaddr issue
similar to the existing WireGuard proposal, then chose to track the
draft as an AI planning document.
## Response
The generated proposal is intended for `multiformats/multiaddr`:
> `git diff HEAD~1..HEAD -- ai/tpt-backends/04_tipc_multiaddr_upstream.md ai/tpt-backends/README.md`
It proposes a code allocated upstream, a fixed 72-bit value holding
the service type, instance and scope, and the canonical text form
`/tipc/<type>:<instance>:<scope>`. It includes WireGuard composition,
security boundaries, the `tipc` versus `cds` naming decision,
deployment-management separation, implementation phases, test data
and open questions for maintainers.

View File

@ -0,0 +1,40 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: a998acff-af66-4969-a65e-f77cbf5ef8dd
timestamp: 2026-08-30T04:53:03Z
git_ref: wkt/tipc_backend_378
scope: code
substantive: true
raw_file: 20260830T045303Z_69a0e504_prompt_io.raw.md
---
## Prompt
Inspect the TIPC backend rebase onto current `main`, repair and verify any
integration problems before force-push, and prepare the result through the
proper `/commit-plan` workflow.
## Response summary
Repaired the stale lock produced by replaying the historical py-multiaddr
source-pin commit over current `main`. Adapted TIPC to the newer lazy-import
contract and extended runtime annotation tests for the interim `/tipc`
multiaddr representation. Verified both the full default and TIPC transport
test matrices.
## Files changed
- `uv.lock` - select the pinned WG-capable py-multiaddr revision while
retaining current `main` dependencies.
- `tractor/ipc/_tipc.py` - make the `Multiaddr` annotation import lazy.
- `tests/test_lazy_imports.py` - cover TIPC annotation resolution and the
interim `Multiaddr|str` return shape.
## Human edits
The human restarted and advanced the rebase one conflict at a time, required
the agent to provide exact continuation commands after each resolution,
authorized lock repair and verification, and retained ownership of the
force-push. The human then rejected the ad hoc staging suggestion and required
the complete `/commit-plan` workflow before committing.

View File

@ -0,0 +1,54 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-30T04:53:03Z
git_ref: wkt/tipc_backend_378
diff_cmd: git diff HEAD~1..HEAD
---
## Prompt
After rebasing the TIPC backend branch onto current `main`, inspect the
result before the human force-pushes it. Repair and verify any integration
problems, then provide a proper `/commit-plan` rather than an ad hoc staging
and commit suggestion.
## Response
The range-diff showed that the two omitted pformat commits were already
present on `main`, while the conflict resolutions correctly retained current
platform-aware transport behavior. It also exposed a lost `uv.lock` update:
`pyproject.toml` retained the unreleased WG-capable `multiaddr` source pin,
but the lock still selected the PyPI release.
> `git diff HEAD~1..HEAD -- uv.lock`
Regenerated the lock against current `main`, preserving its newly landed
dependencies while selecting py-multiaddr revision `f86519da`.
The full test suite then found that eagerly importing `Multiaddr` from the
new TIPC module violated `main`'s cold-import contract.
> `git diff HEAD~1..HEAD -- tractor/ipc/_tipc.py`
Moved `Multiaddr` under `TYPE_CHECKING` and supplied the runtime `Any` alias
used by the sibling TCP and UDS backends, retaining lazy multiaddr loading
and runtime annotation resolution.
> `git diff HEAD~1..HEAD -- tests/test_lazy_imports.py`
Extended lazy annotation checks to the TIPC stream and updated
`mk_maddr()`'s expected runtime return type for the branch's interim
`Multiaddr|str` TIPC grammar.
Validation completed:
- `uv lock --check` passed.
- Focused lazy-import, TIPC, and multiaddr checks: 60 passed.
- Full default transport suite: 548 passed, 9 skipped, 7 xfailed,
3 xpassed.
- Full `--tpt-proto=tipc` suite: 548 passed, 9 skipped, 8 xfailed,
2 xpassed.
The human corrected the initial ad hoc commit suggestion and required the
repository's complete `/commit-plan` workflow.

View File

@ -0,0 +1,191 @@
# `/tipc` multiaddr protocol: upstream issue draft
Candidate issue for `multiformats/multiaddr`, to be submitted after
the encoding questions below have been reviewed locally.
## Context
Linux TIPC (Transparent Inter-Process Communication) addresses a
service by a location-independent `(service type, instance)` name.
A server publishes that name into the kernel-maintained cluster name
table and a client connects by the same name; no host or transport
port forms part of the service identity.
TIPC is also known as **Cluster Domain Sockets**, a useful description
of its relationship to Unix-domain sockets. The registered protocol
name should nevertheless remain `tipc`: it matches Linux's
`AF_TIPC`, socket constants, kernel module and iproute2 frontend.
Registering `cds` would create an ecosystem-specific alias that is
harder to map back to the normative kernel API.
We would like to register a `tipc` multiaddr component so these
service endpoints can be represented directly and composed with a
bearer or tunnel description:
```text
/tipc/1953628160:1616:2
/ip4/192.168.1.50/udp/51820/wg/u<key>/tipc/1953628160:1616:2
\____________ WireGuard bearer ____________/\____ TIPC service ____/
```
In the composed form, the components through `/wg/<key>` identify
the routed bearer and tunnel peer. The `/tipc/...` component is pure
service identity, resolved to a current publisher by the TIPC kernel
name table. Unlike a TCP endpoint, it deliberately has no inner IP
address or port.
This proposal does not imply that parsing the multiaddr configures a
TIPC bearer. In particular, TIPC over a WireGuard interface requires
a separately configured TIPC UDP bearer; WireGuard interfaces are L3
devices and cannot carry TIPC Ethernet media directly.
Today that bearer is configured through iproute2's `tipc` frontend,
which speaks the kernel's `TIPCv2` generic-netlink family. `pyroute2`
already provides WireGuard support and generic-netlink primitives but
has no TIPC codec/module; adding one is a complementary deployment
automation track, not part of this address-format proposal.
## Proposed protocol
- Name: `tipc`
- Code: TBD, allocated in `multiformats/multicodec` under the
`multiaddr` tag before implementations stabilize one
- Size: 72 bits
- Value: service type, service instance and publication scope
### Binary form
Exactly nine bytes with no value-length prefix:
| Offset | Size | Field | Encoding |
| ---: | ---: | --- | --- |
| 0 | 4 bytes | service type | unsigned 32-bit big-endian |
| 4 | 4 bytes | service instance | unsigned 32-bit big-endian |
| 8 | 1 byte | publication scope | unsigned enum byte |
```text
tipc-value = uint32be(type) || uint32be(instance) || uint8(scope)
```
For type `1953628160` (`0x74720000`), instance `1616` and
cluster scope `2`, the payload is:
```text
74 72 00 00 00 00 06 50 02
```
### String form
Use one multiaddr value segment containing three canonical decimal
integers:
```text
/tipc/<type>:<instance>:<scope>
```
Canonical values have no sign, whitespace, alternate radix or
leading zeroes, except that zero itself is `0`. `type` and `instance`
must fit unsigned 32-bit fields. Scope is one of:
- `2`: `TIPC_CLUSTER_SCOPE`
- `3`: `TIPC_NODE_SCOPE`
The existing experimental spelling
`/tipc/<type>/<instance>/<scope>` cannot be registered as one normal
multiaddr protocol: generic parsing treats each slash-delimited name
as another protocol component. A single structured value preserves
TIPC's atomic service-address semantics without registering three
artificial protocols.
## Why scope is included
TIPC scope controls where a bound service publication is visible.
The same address representation is used for listener configuration
and dialing, so retaining scope lets a multiaddr round-trip the full
socket address rather than silently turning a node-local bind into a
cluster publication.
Modern Linux UAPI defines cluster and node scopes. The deprecated
zone spelling should not receive a new wire value; implementations
may normalize legacy input to cluster scope before encoding.
## Composition
Standalone service:
```text
/tipc/1953628160:1616:2
```
TIPC service reached through a WireGuard bearer:
```text
/ip4/192.168.1.50/udp/51820/wg/u<key>/tipc/1953628160:1616:2
```
This differs intentionally from TCP over WireGuard:
```text
/ip4/192.168.1.50/udp/51820/wg/u<key>/ip4/10.0.11.1/tcp/1616
```
TCP repeats an inner locative address. TIPC does not: its service
name is resolved and load-balanced in-kernel across current
publishers.
## Semantics and security
- A TIPC service name identifies a service, not a unique process.
Multiple publishers may bind the same name and connections can be
distributed among them.
- Publication scope is reachability metadata, not authentication.
- A composed `/wg` key authenticates the tunnel peer, not the TIPC
service publisher.
- TIPC's optional native AES-GCM link encryption is independent of
this address codec and of WireGuard.
- Codec implementations should validate field widths and canonical
text only; cluster membership and publisher authorization remain
deployment concerns.
## Implementation plan
1. Reserve a `multiaddr`-tagged code in
`multiformats/multicodec`.
2. Add the fixed-size protocol row and normative encoding text to
`multiformats/multiaddr`.
3. Add codecs and cross-language test vectors, beginning with
`multiformats/py-multiaddr`.
4. Verify standalone and composed `wg` + `tipc` string/binary
round-trips.
## Open questions
1. Is a fixed 72-bit value preferred over a self-describing or
variable-width tuple for this kernel-defined address?
2. Should node scope be representable in a generally shareable
multiaddr, or should the registered form be cluster-only?
3. Does multiaddr have an existing convention for structured numeric
values that should replace the colon-separated text form?
4. Should the specification describe TIPC service *ranges*, or keep
this protocol limited to singleton service names used for
connection endpoints?
## References
- Linux TIPC documentation:
https://docs.kernel.org/networking/tipc.html
- Cluster Domain Sockets terminology:
https://en.wikipedia.org/wiki/Transparent_Inter-process_Communication
- Linux socket UAPI:
https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h
- Linux TIPC generic-netlink UAPI:
https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc_netlink.h
- pyroute2 WireGuard and generic-netlink APIs:
https://docs.pyroute2.org/wireguard.html
- WireGuard multiaddr implementation discussion:
https://github.com/multiformats/py-multiaddr/issues/107
- WireGuard codec implementation:
https://github.com/multiformats/py-multiaddr/pull/108
- Downstream tracking and prototype:
https://github.com/goodboy/tractor/issues/498

View File

@ -16,6 +16,7 @@ assume it and document only their own deltas.
| [01 — TIPC](./01_tipc_backend.md) | [#378] | **none** (stdlib) | small | **landed**, PR [#493] — see the [handoff](./01_tipc_HANDOFF.md) |
| [02 — QUIC/`iroh`](./02_quic_iroh_backend.md) | [#353] | `iroh` (uniffi FFI) | large | needs a prep PR |
| [03 — `wg` bindspace](./03_wg_tunnel_bindspace.md) | [#482], [#443] | `pyroute2` | medium, 3 layers | layer A now |
| [04 — `/tipc` multiaddr](./04_tipc_multiaddr_upstream.md) | [#498] | upstream `multiformats` | issue draft | local review |
Headline conclusions:
@ -54,5 +55,6 @@ lands first ships it.
[#353]: https://github.com/goodboy/tractor/issues/353
[#482]: https://github.com/goodboy/tractor/issues/482
[#443]: https://github.com/goodboy/tractor/issues/443
[#498]: https://github.com/goodboy/tractor/issues/498
[#493]: https://github.com/goodboy/tractor/pull/493

View File

@ -2,6 +2,10 @@
TIPC is a linux-kernel cluster IPC protocol whose service names
live in a **cluster-wide name table maintained by the kernel**.
It is also described as **Cluster Domain Sockets**: the Unix-domain
socket model extended from one kernel to a cluster. That name is a
useful explanation for new users, while the code keeps `tipc` as its
protocol key to match Linux's `AF_TIPC`, kernel module and tooling.
For `tractor` that means:
- an actor's IPC address is a service name `(stype, instance)`,
@ -85,30 +89,55 @@ Everything above is single-node (`modprobe` is enough). To span
hosts you need a **bearer** on both, which is the one thing that
can't be CI'd.
For the first physical test, use two wired Linux hosts on the same
L2 segment. Prefer a direct cable or uncomplicated switch; avoid
Wi-Fi, guest VLANs and port isolation until the basic link works.
Use the same checkout and Python environment on both hosts:
```bash
# on BOTH hosts
git rev-parse HEAD # must match on A and B
uv sync --all-extras --dev
sudo modprobe tipc
# over ethernet (L2) — simplest when the hosts share a segment
sudo tipc bearer enable media eth device eth0
# choose the real wired iface; do not assume `eth0`
ip -br link
IFACE=enp3s0
# inspect existing cluster identity before changing anything
tipc node get address # must differ between hosts
tipc node get netid # must match between hosts
# use one private test netid on BOTH hosts, before enabling bearers
sudo tipc node set netid 37801
# ethernet is simplest when the hosts share an L2 segment
sudo tipc bearer enable media eth device "$IFACE"
# ..or over UDP when L2 isn't available — and MANDATORY over a
# `wg` mesh, see below
sudo tipc bearer enable media udp name uc localip 10.0.11.1
# verify BEFORE running anything: this must list the peer
tipc bearer list
tipc link list
tipc node list
```
Then:
If the link does not appear, first verify carrier, a common TIPC
network ID, distinct node addresses, a common VLAN and compatible
MTUs. Ethernet TIPC uses EtherType traffic rather than IP routing,
so a successful `ping` alone does not prove the bearer can work.
Run each command from `examples/multihost/tipc_cluster/`:
```bash
# host A
python host_a_srv.py
uv run python host_a_srv.py
# host B
python host_b_client.py
watch -n 0.5 tipc nametable show # optional second terminal
uv run python host_b_client.py
```
Note what's absent from both scripts: any IP, hostname or port.
@ -116,6 +145,51 @@ Both sides name the *same service*, and the kernel routes it.
Move `host_a_srv.py` to a third node and host B's dial keeps
working, unchanged.
For a first resilience pass, use a local console or separate
management link so the test does not cut off your own SSH session:
```bash
# host B: record the healthy baseline
tipc link list
tipc link statistics show
# either host: withdraw and recreate the bearer
sudo tipc bearer disable media eth device "$IFACE"
tipc link list
sudo tipc bearer enable media eth device "$IFACE"
tipc link list
# prove name withdrawal/republication and RPC recovery
tipc nametable show
uv run python host_b_client.py
```
Capture `uname -a`, both node addresses, `tipc bearer list`,
`tipc link list`, `tipc link statistics show`, the name table and
both Python transcripts. Those artifacts distinguish an actor bug
from bearer discovery, cluster identity or switch configuration.
Clean up a disposable Ethernet test on both hosts with:
```bash
sudo tipc bearer disable media eth device "$IFACE"
tipc link list
```
Restore any pre-existing network ID only after all bearers are
disabled. The first useful automation target is a two-node network
namespace fixture that asserts link-up, remote publication, RPC,
withdrawal and republication in that order; physical hardware then
remains the validation layer for real NIC and switch behaviour.
The commands above use iproute2's `tipc` frontend, which speaks the
kernel's `TIPCv2` generic-netlink family. The planned `pyroute2`
dependency already manages WireGuard, interfaces and namespaces and
provides generic-netlink primitives, but it does not currently ship a
TIPC message codec. Adding one upstream would let `tractor` replace
these manual commands with one Python netlink stack instead of
shelling out; until then, `tipc(8)` remains the canonical frontend.
### over a `wg` mesh
TIPC over WireGuard is the intended reference multihost
@ -231,12 +305,10 @@ self-skip when the module is absent.
## CI
Single-host TIPC *is* CI-able — the module ships with the
standard Ubuntu kernel package, so a `sudo modprobe tipc` step
plus a `--tpt-proto tipc` matrix entry should work. That's not
wired up yet; verify in a throwaway workflow first, and fall
back to a container job with `--cap-add NET_ADMIN` if the
runners refuse. Cross-node (bearer) testing stays manual — this
README is that smoke test.
standard Ubuntu kernel package. CI loads it with `sudo modprobe
tipc` and runs the suite with `--tpt-proto tipc` as a blocking
matrix leg. Cross-node bearer testing stays manual — this README
is that smoke test.
## normative refs

View File

@ -47,14 +47,23 @@ async def main() -> None:
reg: TIPCAddress = TIPCAddress.get_root()
print(f'host A publishing {reg}')
async with tractor.open_root_actor(
name='host_a',
# A registrar root is not an entry in its own
# `Registrar._registry`, so host B cannot discover that root by
# its actor name. `open_nursery()` keeps the root as registrar
# while the named child below registers as the dialable service.
async with tractor.open_nursery(
enable_transports=['tipc'],
registry_addrs=[reg.unwrap()],
enable_modules=[__name__],
):
) as an:
await an.start_actor(
'host_a',
# Host B imports this same module name to construct the
# RPC `NamespacePath`; direct script execution would
# otherwise expose it as `__main__` on host A.
enable_modules=['host_a_srv'],
)
print(
'registrar up — `tipc nametable show` on EITHER host\n'
'host_a up — `tipc nametable show` on EITHER host\n'
'should now list this service. ctrl-c to stop.'
)
await trio.sleep_forever()

View File

@ -16,6 +16,7 @@ from __future__ import annotations
import trio
import tractor
from host_a_srv import echo
from tractor.ipc._tipc import (
TIPCAddress,
is_tipc_available,
@ -39,9 +40,13 @@ async def main() -> None:
' |_does `tipc link list` show the peer?\n'
)
# `.open_context()` derives a `NamespacePath` from a
# callable. Importing `echo` also loads its module path
# locally, while host A's `enable_modules` authorizes the
# corresponding remote callable.
async with (
ptl.open_context(
'host_a_srv:echo',
echo,
) as (ctx, _),
ctx.open_stream() as stream,
):

View File

@ -13,7 +13,6 @@ from socket import (
SOCK_STREAM,
SOL_SOCKET,
SO_ACCEPTCONN,
SOL_TIPC,
)
import pytest
@ -27,8 +26,10 @@ from tractor.discovery._multiaddr import (
parse_maddr,
)
from tractor.ipc import _tipc
from tractor.ipc._uds import UDSAddress
from tractor.ipc._tipc import (
AF_TIPC,
SOL_TIPC,
TIPC_ADDR_ID,
TIPC_ADDR_NAME,
TIPC_CLUSTER_SCOPE,
@ -37,6 +38,9 @@ from tractor.ipc._tipc import (
TIPC_IMPORTANCE,
TIPC_NAME_UNKNOWN,
TIPC_NODE_SCOPE,
TIPC_PUBLISHED,
TIPC_SCOPE_UNKNOWN,
TIPC_SUBSCR_TIMEOUT,
TIPC_ZONE_SCOPE,
TRACTOR_STYPE,
MsgpackTIPCStream,
@ -114,6 +118,29 @@ def test_zone_scope_normalized_to_cluster():
assert addr.is_valid
def test_maddr_parse_normalizes_and_reports_bad_input():
'''
The interim `/tipc` parser must route through
`TIPCAddress.from_addr()` so deprecated zone scope is normalized
exactly like every other unwrapped-address entrypoint.
A malformed segment count previously leaked the tuple-unpacking
`ValueError`, which gave callers no indication that the TIPC
multiaddr grammar itself was invalid.
'''
addr: TIPCAddress = parse_maddr(
f'/tipc/{TRACTOR_STYPE}/7/{TIPC_ZONE_SCOPE}'
)
assert addr._scope == TIPC_CLUSTER_SCOPE
with pytest.raises(
ValueError,
match='Invalid TIPC multiaddr',
):
parse_maddr('/tipc/not-enough-segments')
def test_addr_from_bare_port_id_raises():
'''
A `TIPC_ADDR_ID` 5-tuple carries no service-name so it can
@ -207,6 +234,43 @@ def test_get_random_collision_resistance():
assert all(addr.is_valid for addr in addrs)
def test_get_random_keys_live_actors_by_uuid(
monkeypatch: pytest.MonkeyPatch,
):
'''
TIPC names are cluster-wide while PIDs are host-local. Hashing
only `(actor name, pid)` therefore made same-named actors with
equal PIDs on different hosts publish one service name, where
TIPC silently round-robins connects between them.
Hold the actor name and PID fixed while changing only its UUID;
distinct instances prove the globally unique identity field is
now part of the derivation.
'''
monkeypatch.setattr(_tipc.os, 'getpid', lambda: 1616)
def get_addr(uuid: str) -> TIPCAddress:
actor = SimpleNamespace(
aid=Aid(
name='worker',
uuid=uuid,
pid=1616,
)
)
monkeypatch.setattr(
_tipc,
'current_actor',
lambda **kwargs: actor,
)
return TIPCAddress.get_random()
first: TIPCAddress = get_addr('actor-uuid-a')
second: TIPCAddress = get_addr('actor-uuid-b')
assert first._instance != second._instance
assert get_addr('actor-uuid-a')._instance == first._instance
def test_get_random_honors_bindspace():
addr: TIPCAddress = TIPCAddress.get_random(
bindspace=TIPC_NODE_SCOPE,
@ -235,6 +299,12 @@ def test_wrap_address_dispatches_on_the_proto_key():
'tipc', TRACTOR_STYPE, 1616, TIPC_CLUSTER_SCOPE,
)
# A UDS directory named `tipc` is still a valid classic
# 2-element address, not a malformed proto-keyed TIPC one.
uds: UDSAddress = wrap_address(('tipc', 'actor.sock'))
assert isinstance(uds, UDSAddress)
assert UDSAddress.unwrapped_type == tuple[str, str]
def test_maddr_roundtrip():
'''
@ -286,6 +356,97 @@ def test_eafnosupport_is_actionable_connerr(
assert type(excinfo.value.__cause__) is OSError
def test_availability_probe_is_linux_only(
monkeypatch: pytest.MonkeyPatch,
):
'''
The fallback `AF_TIPC` integer keeps imports portable but can
alias an unrelated socket family on another OS. Socket creation
alone must never turn that numeric collision into a false TIPC
capability result.
Simulate Darwin with a socket constructor that would otherwise
succeed; proving it is never called pins the platform gate ahead
of the syscall probe.
'''
socket_called: bool = False
def fake_socket(*args, **kwargs):
nonlocal socket_called
socket_called = True
monkeypatch.setattr(_tipc, '_tipc_avail', None)
monkeypatch.setattr(_tipc.sys, 'platform', 'darwin')
monkeypatch.setattr(_tipc.socket, 'socket', fake_socket)
assert not is_tipc_available()
assert not socket_called
def test_connect_reuses_tolerant_peer_observation(
monkeypatch: pytest.MonkeyPatch,
):
'''
A TIPC peer can withdraw immediately after `.connect()`. The
transport constructor already tolerates that race, but the dial
path used to call `getpeername()` a second time and leak a raw
`ENOTCONN` after an otherwise successful connection.
The fake socket permits exactly the constructor's first peer
lookup. A second lookup raises deterministically, so successful
construction proves `connect_to()` reuses the tolerant observed
address and transfers socket ownership only after setup.
'''
class _DropAfterObservation:
def __init__(self):
self.peer_lookups: int = 0
self.closed: bool = False
def setsockopt(self, *args) -> None:
pass
async def connect(self, addr) -> None:
pass
def getsockname(self) -> tuple[int, int, int, int, int]:
return (TIPC_ADDR_ID, 1, 2, 0, TIPC_CLUSTER_SCOPE)
def getpeername(self) -> tuple[int, int, int, int, int]:
self.peer_lookups += 1
if self.peer_lookups > 1:
raise OSError(errno.ENOTCONN, 'peer withdrew')
return (TIPC_ADDR_ID, 3, 4, 0, TIPC_CLUSTER_SCOPE)
def close(self) -> None:
self.closed = True
async def main() -> None:
sock = _DropAfterObservation()
monkeypatch.setattr(
_tipc.trio_socket,
'socket',
lambda *args: sock,
)
monkeypatch.setattr(
_tipc.trio,
'SocketStream',
lambda raw_sock: SimpleNamespace(socket=raw_sock),
)
addr: TIPCAddress = TIPCAddress.get_root()
stream: MsgpackTIPCStream = (
await MsgpackTIPCStream.connect_to(addr)
)
assert sock.peer_lookups == 1
assert stream.raddr.unwrap() == addr.unwrap()
assert stream.raddr.maybe_ref == 4
assert not sock.closed
trio.run(main)
# ------------------------------------------------------------------
# kernel-touching
# ------------------------------------------------------------------
@ -658,6 +819,22 @@ def test_topology_struct_layouts():
assert len(sub) == 28
def _pack_topology_event(
event: int,
instance: int = 71,
) -> bytes:
return struct.pack(
_tipc._EVENT_FMT,
event,
instance,
instance,
123,
456,
0, 0, 0, 0, 0,
b'\0' * 8,
)
def test_wait_forever_is_masked_for_packing():
'''
Python exposes `TIPC_WAIT_FOREVER` as `-1`, which `struct`
@ -689,7 +866,6 @@ def test_decode_name_event_rejects_junk():
assert _tipc._decode_name_event(
b'\x00' * 12,
stype=TRACTOR_STYPE,
scope=TIPC_CLUSTER_SCOPE,
) is None
bogus: bytes = struct.pack(
@ -702,10 +878,113 @@ def test_decode_name_event_rejects_junk():
assert _tipc._decode_name_event(
bogus,
stype=TRACTOR_STYPE,
scope=TIPC_CLUSTER_SCOPE,
) is None
def test_topology_event_scope_is_unknown():
'''
Neither `struct tipc_subscr` nor `struct tipc_event` carries a
publication scope. The old decoder copied caller context into
each `TIPCAddress`, falsely presenting cluster scope as kernel-
observed data even for a node-scoped publisher.
Decode a valid publication and assert its address uses the
explicit unknown sentinel rather than inventing reachability.
TIPC address types and publication scopes:
https://docs.kernel.org/networking/tipc.html
Event wire format (which has no scope field):
https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h
'''
event: _tipc.TIPCNameEvent|None = _tipc._decode_name_event(
_pack_topology_event(TIPC_PUBLISHED),
stype=TRACTOR_STYPE,
)
assert event is not None
assert event.addr._scope == TIPC_SCOPE_UNKNOWN
assert not event.addr.is_valid
def test_topology_stream_surfaces_overflow():
'''
Topology transitions are `TIPC_PUBLISHED`/`TIPC_WITHDRAWN`
changes to the set of ports matching a subscribed name sequence,
as emitted by the kernel topology server:
https://github.com/torvalds/linux/blob/master/net/tipc/topsrv.c
Blocking this reader on `tx.send()` stops draining the topology
socket and merely pushes pressure into the kernel queue. Silently
dropping with `send_nowait()` is worse: a push registry keeps a
stale view without knowing it missed a transition.
A zero-capacity channel deterministically fills the user-space
boundary. The dedicated overflow error proves the reader remains
non-blocking while forcing consumers to resubscribe and rebuild
instead of trusting incomplete state.
'''
class _OneEventSocket:
def __init__(self):
self.sent: bool = False
async def recv(self, size: int) -> bytes:
if not self.sent:
self.sent = True
return _pack_topology_event(TIPC_PUBLISHED)
raise AssertionError('reader continued after overflow')
async def main() -> None:
tx, rx = trio.open_memory_channel(0)
with pytest.raises(
_tipc.TIPCNameEventOverflow,
match='resubscribe and rebuild',
) as excinfo:
await _tipc._stream_name_events(
_OneEventSocket(),
TRACTOR_STYPE,
tx,
)
assert excinfo.value.event.kind == 'published'
with pytest.raises(trio.EndOfChannel):
await rx.receive()
trio.run(main)
def test_topology_timeout_closes_stream():
'''
A finite kernel subscription expires after its timeout event.
The old reader forwarded that event and waited forever for a
second frame that could never arrive, so consumers blocked on a
channel that looked live despite having no subscription.
Feed one timeout frame directly to the reader; receiving that
event followed by `EndOfChannel` proves the stream terminates at
the kernel subscription boundary.
Subscription timeout semantics:
https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h
'''
class _TimeoutSocket:
async def recv(self, size: int) -> bytes:
return _pack_topology_event(TIPC_SUBSCR_TIMEOUT)
async def main() -> None:
tx, rx = trio.open_memory_channel(1)
await _tipc._stream_name_events(
_TimeoutSocket(),
TRACTOR_STYPE,
tx,
)
assert (await rx.receive()).kind == 'timeout'
with pytest.raises(trio.EndOfChannel):
await rx.receive()
trio.run(main)
@requires_tipc
def test_topology_reports_publish_and_withdraw():
'''
@ -739,6 +1018,7 @@ def test_topology_reports_publish_and_withdraw():
for ev in got:
assert ev.addr._instance == addr._instance
assert ev.addr._stype == addr._stype
assert ev.addr._scope == TIPC_SCOPE_UNKNOWN
assert isinstance(ev.ref, int)
# both transitions name the SAME publisher port
assert got[0].ref == got[1].ref

View File

@ -19,6 +19,7 @@ from tractor.discovery import (
)
from tractor.ipc import (
_tcp,
_tipc,
_uds,
)
@ -271,13 +272,16 @@ def test_lazy_annotation_names_resolve():
introspection.
'''
assert get_type_hints(_multiaddr.mk_maddr)['return'] is Any
assert get_type_hints(_multiaddr.mk_maddr)['return'] == Any|str
assert get_type_hints(_tcp.MsgpackTCPStream.maddr.fget)[
'return'
] is Any
assert get_type_hints(_uds.MsgpackUDSStream.maddr.fget)[
'return'
] == Any|str
assert get_type_hints(_tipc.MsgpackTIPCStream.maddr.fget)[
'return'
] == Any|str
assert get_type_hints(_addr.Address.get_random)[
'current_actor'
] is Any

View File

@ -330,7 +330,11 @@ def wrap_address(
#
# NOTE, a bare seq-pattern matches `list` too, which is
# what `msgpack` decodes our tuples back to.
case ('tipc', *_):
case (
('tipc', int(), int())
|
('tipc', int(), int(), int())
):
cls = TIPCAddress
# classic network socket-address as tuple/list

View File

@ -83,6 +83,9 @@ def mk_maddr(
dispatching on the `.proto_key` to build the correct
multiaddr-spec-compliant protocol path.
Return a `Multiaddr` for registered protocols. TIPC remains
an interim `str` until its upstream multiaddr protocol lands.
'''
from multiaddr import Multiaddr
@ -154,12 +157,18 @@ def parse_maddr(
# XXX MUST come before `Multiaddr()` which rejects the
# not-yet-registered `/tipc` proto name outright.
if maddr_str.startswith(_tipc_maddr_prefix):
_, _, stype, instance, scope = maddr_str.split('/')
return TIPCAddress(
_stype=int(stype),
_instance=int(instance),
_scope=int(scope),
)
try:
_, _, stype, instance, scope = maddr_str.split('/')
return TIPCAddress.from_addr((
'tipc',
int(stype),
int(instance),
int(scope),
))
except (TypeError, ValueError) as src_err:
raise ValueError(
f'Invalid TIPC multiaddr: {maddr_str!r}'
) from src_err
try:
maddr = Multiaddr(maddr_str)

View File

@ -57,7 +57,9 @@ from socket import (
SOCK_STREAM,
)
import struct
import sys
from typing import (
Any,
AsyncGenerator,
Callable,
ClassVar,
@ -74,10 +76,8 @@ from trio import (
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,
)
@ -87,8 +87,13 @@ from tractor.runtime._state import (
)
if TYPE_CHECKING:
# ONLY type-annots, the eager import costs
# `import tractor` wall-time (gh #470).
from multiaddr import Multiaddr
from tractor.discovery._addr import TaggedTIPCAddress
from tractor.runtime._runtime import Actor
else:
Multiaddr = Any
log = get_logger()
@ -167,6 +172,10 @@ _tipc_reserved_stypes: range = range(0, 64)
# See `MsgpackTIPCStream.get_stream_addrs()` and plan 01 §3.4.
TIPC_NAME_UNKNOWN: int = -1
# The topology event wire format carries no publication scope.
# Never promote caller context into observed address data.
TIPC_SCOPE_UNKNOWN: int = 0
# XXX, the kernel default (`TIPC_LOW_IMPORTANCE`), i.e. today this
# is a no-op knob preserving stock behaviour.
#
@ -178,6 +187,7 @@ TIPC_NAME_UNKNOWN: int = -1
TRACTOR_DEF_IMPORTANCE: int = TIPC_LOW_IMPORTANCE
_scope_names: dict[int, str] = {
TIPC_SCOPE_UNKNOWN: 'unknown',
TIPC_ZONE_SCOPE: 'zone',
TIPC_CLUSTER_SCOPE: 'cluster',
TIPC_NODE_SCOPE: 'node',
@ -198,6 +208,10 @@ def is_tipc_available() -> bool:
'''
global _tipc_avail
if sys.platform != 'linux':
_tipc_avail = False
return _tipc_avail
if _tipc_avail is None:
try:
socket.socket(
@ -341,8 +355,8 @@ class TIPCAddress(
def with_port_id(
self,
node: int,
ref: int,
node: int|None,
ref: int|None,
) -> TIPCAddress:
'''
A copy annotated with an *observed* `TIPC_ADDR_ID`
@ -370,10 +384,11 @@ class TIPCAddress(
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.
So the instance is a `blake2b` digest of the actor UUID, or
a per-call token outside a live runtime, giving a well-spread
32b value. Being a pure fn of the seed it is also
*reproducible*, which the (follow-up) registrar-less
discovery fast-path wants.
NOTE the residual risk is birthday-bounded: ~1.2e-2 for 10k
names sharing one `_stype`. See plan 01 §9 for the
@ -385,7 +400,7 @@ class TIPCAddress(
err_on_no_runtime=False,
)
if actor:
seed: str = f'{actor.aid.name}@{pid}'
seed: str = '.'.join(actor.aid.uid)
else:
if is_root_process():
prefix: str = 'no_runtime_root'
@ -629,6 +644,8 @@ class MsgpackTIPCStream(MsgpackTransport):
@property
def maddr(self) -> Multiaddr|str:
from tractor.net import mk_maddr
if not self.raddr:
return '<unknown-peer>'
@ -683,22 +700,39 @@ class MsgpackTIPCStream(MsgpackTransport):
destaddr._scope,
))
tpt_stream = MsgpackTIPCStream(
trio.SocketStream(sock),
prefix_size=prefix_size,
codec=codec,
)
# XXX, the dialling side is the ONLY side that knows the
# peer's *service name* (a port-id can't be reversed into
# one), so re-assert it over the observed-only `._raddr`
# that `.get_stream_addrs()` just derived.
#
# Same move as `MsgpackUDSStream.connect_to()`s peer-pid
# re-assign.
tpt_stream._raddr = destaddr.with_port_id(
*_port_id(sock.getpeername()),
)
return tpt_stream
tpt_stream = MsgpackTIPCStream(
trio.SocketStream(sock),
prefix_size=prefix_size,
codec=codec,
)
# XXX, the dialling side is the ONLY side that knows
# the peer's *service name* (a port-id can't be
# reversed into one), so re-assert it over the
# observed-only `._raddr` derived above.
#
# Reuse that tolerant observation: a peer can withdraw
# between `.connect()` and a second `.getpeername()`.
#
# Peer / kernel dial task
# | |
# |<----- connect(name) -------|
# |------ connected ---------->| (A)
# |<----- getpeername() -------| (B, ctor)
# |-- port-id or `ENOTCONN` -->|
# X withdraws |
# |<----- getpeername() -------| (C, old code)
# |------ `ENOTCONN` --------->|
#
# If withdrawal precedes B, `_maybe_sockaddr()` records
# no port-id and the handshake observes the close. If it
# follows B, that first observation remains useful. The
# removed C lookup created the only raw-`ENOTCONN` race.
observed_raddr: TIPCAddress = tpt_stream._raddr
tpt_stream._raddr = destaddr.with_port_id(
node=observed_raddr.maybe_node,
ref=observed_raddr.maybe_ref,
)
return tpt_stream
@classmethod
def get_stream_addrs(
@ -848,8 +882,17 @@ class TIPCNameEvent(
frozen=True,
):
'''
A kernel name-table transition: some service name was
published or withdrawn somewhere in the cluster.
A kernel name-table transition, not an application msg.
A transition reports that the set of TIPC ports matching a
subscribed service-name sequence changed: `TIPC_PUBLISHED`
adds a matching `(node, ref)`, `TIPC_WITHDRAWN` removes one,
and `TIPC_SUBSCR_TIMEOUT` ends a finite subscription.
Wire definitions:
https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h
Topology server:
https://github.com/torvalds/linux/blob/master/net/tipc/topsrv.c
'''
kind: Literal[
@ -870,6 +913,28 @@ class TIPCNameEvent(
)
class TIPCNameEventOverflow(RuntimeError):
'''
The user-space event buffer lost authoritative continuity.
The socket reader must not remain blocked while its kernel
topology subscription is live. Raising aborts that subscription
and forces the consumer to resubscribe and rebuild its name-table
view instead of using silently stale state.
'''
def __init__(
self,
event: TIPCNameEvent,
) -> None:
self.event = event
super().__init__(
'TIPC topology event buffer overflowed; '
'resubscribe and rebuild the name-table view.\n'
f'lost event: {event!r}'
)
def _mk_subscr(
stype: int,
lower: int,
@ -904,7 +969,6 @@ def _mk_subscr(
def _decode_name_event(
raw: bytes,
stype: int,
scope: int,
) -> TIPCNameEvent|None:
'''
Decode one `struct tipc_event`, or `None` if it's a runt/
@ -939,14 +1003,12 @@ def _decode_name_event(
# NOTE, `tractor` only ever publishes *singleton* ranges
# (`lower == upper`) so the lower bound IS the instance.
#
# XXX the event carries NO scope — the name-table doesn't
# report it — so we echo back the subscription's own. Fine
# for our use (we subscribe per-scope) but don't mistake
# it for observed data.
# XXX the event carries NO scope — do not fabricate one
# from caller context and present it as observed data.
addr=TIPCAddress(
_stype=stype,
_instance=found_lower,
_scope=scope,
_scope=TIPC_SCOPE_UNKNOWN,
),
node=node,
ref=ref,
@ -956,7 +1018,6 @@ def _decode_name_event(
async def _stream_name_events(
sock,
stype: int,
scope: int,
tx: trio.MemorySendChannel,
) -> None:
'''
@ -973,21 +1034,15 @@ async def _stream_name_events(
if (ev := _decode_name_event(
raw,
stype=stype,
scope=scope,
)) is None:
continue
try:
tx.send_nowait(ev)
except trio.WouldBlock:
# XXX drop rather than block: stalling this reader
# backs up the kernel's own queue and we'd lose the
# event anyway, just less visibly.
log.warning(
f'TIPC topology event buffer full, dropping!\n'
f'{ev}\n'
f' |_raise `buf_size` or consume faster\n'
)
except trio.WouldBlock as src_err:
raise TIPCNameEventOverflow(ev) from src_err
if ev.kind == 'timeout':
return
except (
trio.ClosedResourceError,
@ -1006,7 +1061,6 @@ async def open_topology_events(
lower: int = 0,
upper: int = 0xFFFF_FFFF,
filt: int = TIPC_SUB_SERVICE,
scope: int = TIPC_CLUSTER_SCOPE,
timeout: int = TIPC_WAIT_FOREVER,
buf_size: int = 64,
) -> AsyncGenerator[
@ -1035,7 +1089,7 @@ async def open_topology_events(
topsrv_addr = TIPCAddress(
_stype=TIPC_TOP_SRV,
_instance=TIPC_TOP_SRV,
_scope=scope,
_scope=TIPC_CLUSTER_SCOPE,
)
sock = trio_socket.socket(
AF_TIPC,
@ -1078,7 +1132,6 @@ async def open_topology_events(
_stream_name_events,
sock,
stype,
scope,
tx,
)
try: