feat(memory-pool): cross-machine zero-copy tensor transport for end-edge-cloud collaboration

[tang-canran]

Memory Pool: Zero-Copy Tensor Transport across Daemons and Machines for End-Edge-Cloud Robotics

(1) Design — new architecture for multi-daemon and cross-machine deployment

This PR extends the memory-pool transport to the deployment topologies real edge-cloud robotics actually run: multiple daemons on one host (sensor, perception, and inference pipelines as separate daemons on a robot or edge box) and daemon clusters across machines (edge ↔ cloud). The additions:

Same-host multi-daemon — direct read across daemons

• The receiver locates the sender's pool segment directly by machine-qualified name (pool_{node_id}_{machine}_{counter}) and reads it in place, bypassing the daemon relay and even the mirror push entirely (same-host direct detection). Native dora has no such path: outputs to a consumer on another daemon are pinned to the daemon relay (Readiness barrier (#2666) counts unreachable remote subscribers: fixed 5 s per-node startup stall + lost direct-zenoh fast path for local subscribers in multi-machine dataflows #2738), so the same topology costs 89 MB/s instead of ~5.8 GB/s (65×).
• For GPU pools, the sender exports the CUDA IPC handle into the pool header at registration; a same-host cross-daemon receiver imports it once and reads device memory zero-copy (32 GB/s measured).
• Control-plane notifications (RegisterPool/RegisterPoolAck/FreePool) between daemons ride a zenoh SHM payload for same-host delivery.

Cross-machine — reliable large-frame data plane with GPU staging

• The data plane uses the daemon's zenoh channel with blocking congestion control: large frames queue and drain over slow WAN links instead of being dropped. Native dora's relay uses Drop + express: fragments are silently dropped when the TX queue (16-batch cap) backs up on a slow link, and the daemon↔daemon data plane hangs at any frame size once the RTT crosses ~200 ms — reproducible on loopback with tc qdisc add dev lo root netem delay 100ms (a 0.5 MiB frame hangs identically to 40 MiB). There is no working native path for WAN large frames.
• GPU endpoints get CPU staging pools: a pageable transit pool on the send side (GPU_A → DtoH → CPU_A) and a pinned pool on the receive side (CPU_B → HtoD → GPU_B), both registered and reused per frame; the full path GPU→CPU→zenoh TCP→CPU→GPU runs at link bandwidth (~38 MB/s on the 1-Gbps LAN link; GPU staging adds no per-frame allocation).
• Machine-qualified pool names and the daemon-side memory manager (touched_by / targeted cleanup) prevent cross-machine ID aliasing and clean up pools on every daemon involved — orphan sweeps are scoped to the daemon's own machine so a sibling daemon's live segments are never touched.

Existing single-daemon behavior is unchanged; the event stream, zenoh, and daemon relay code are untouched.

(2) Performance: memory pool vs native dora vs ROS 2 (3 scenarios × 4 device pairs + cross-machine links)

Benchmark: 100 frames of 10000×512 int64 (40.96 MB), turn-based per-frame handshake, Average transfer throughput (data_bytes / (t_received − t_send)), same payload and cadence on both sides. Native dora runs with the 256 MiB zenoh SHM pool configured (its best zero-copy configuration). GPU columns (cpu2cuda/cuda2cpu/cuda2cuda) are measured with the same benchmark; the WAN row is the cross-machine example's 61.44 MB frames (15000×512) on the same data path.

Memory pool (MB/s, this PR):

Scenario	cpu2cpu	cpu2cuda	cuda2cpu	cuda2cuda
Single daemon	6157.9	3197.1	6792.7	44145.8 (same-GPU IPC)
Same-host cross-daemon	5787.5	2616.6	6977.6	32691.6 (IPC direct read)
Cross-machine 1-Gbps LAN (5090↔A100, RTT ~0.2 ms)	38.0	37.5	38.3	37.5 (GPU_A→CPU→TCP→CPU→GPU_B)
Cross-machine true WAN (workstation↔A100, RTT 4.7 ms, link ~7 MB/s)	~4	—	—	—

Native dora (MB/s, same benchmark):

Scenario	cpu2cpu	note
Single daemon	2409.6	zenoh SHM zero-copy active
Same-host cross-daemon	89.3	pinned to the daemon relay (#2738); node-to-node zenoh mesh is same-machine-only
Cross-machine 1-Gbps LAN	~100	relay-bound, ≈82–85% of line rate (independently reproduced: 819 Mbps on the same link)
Cross-machine true WAN	no working path	Drop + express silently drops fragments on the slow link; the producer blocks on its ack forever (observed with 20–61 MB frames); loopback tc netem delay 100ms reproduces the data-plane hang at any size

Speedup: 2.6× single-daemon, 65× same-host cross-daemon. On a 1-Gbps LAN the native relay already runs at ≈85% of line rate, so that link leaves little headroom (the pool is at 32% of line rate there — docker bridge/NAT and the per-frame handshake are the current limiters, not the link; we report it as-is rather than label it WAN). The claims that matter are the same-host 65× (reproducible on one machine without a network) and the WAN rows below.

Cross-machine landscape — ROS 2 network DDS on a real WAN (RTT 38.3 ms, n=12/size, byte-contract all green):

Payload	Raw TCP per-frame (control arm)	ROS 2 network DDS	ROS 2 vs raw TCP
1 MiB	21.80 MB/s	1.29 MB/s	17.0×
4 MiB	43.74 MB/s	1.19 MB/s	36.9×
16 MiB	68.08 MB/s	1.07 MB/s	63.4×

• ROS 2 is not bandwidth-limited on the WAN — it is locked to a fixed advance per RTT. Raw TCP scales with frame size (21.8 → 68.1 MB/s, the fixed 38.3 ms RTT amortizing over larger frames), while ROS 2 stays flat at 1.07–1.29 MB/s. 1.1 MB/s × 38.3 ms ≈ 42 KB/RTT: the reliable-transport pacing advances only a fixed chunk per round trip.
• The memory pool has no such signature. Across the two links measured, pool throughput tracks link bandwidth (38–44.6 MB/s on the 1-Gbps LAN, ~4 MB/s on the ~7 MB/s WAN with 61.44 MB frames at 13–16 s/frame, ≈60% of the link), and blocking congestion control queues and drains instead of dropping fragments.
• Same-host control (16 MiB, loopback): ROS 2 network DDS 801 MB/s vs dora's cross-daemon relay 107 MB/s — ROS 2's two-hop publisher→subscriber path is 7.5× faster than dora's four-hop node→daemon→daemon→node relay. The relay path is replaceable; the pool removes it for same-host and replaces it with a reliable queueing data plane cross-machine.

(3) Purpose and significance — enabling end-edge-cloud tensor pipelines

VLA models and world models are driving explosive demand for on-device inference [1–4]; end-edge co-computing is becoming the mainstream deployment paradigm, keeping data transport inside the LAN at 1–6 ms round trips versus tens-to-hundreds of milliseconds over the WAN to the cloud [10,11]. A robot pipeline in this paradigm moves tens-of-MB GPU tensors (camera frames, point clouds, VLA features) through several processes — sensor → preprocessing → inference — on the same host, and then to the edge cluster or the cloud.

This PR makes that pipeline zero-copy at every hop:

• On the end device: tensors stay on the GPU across processes (same-GPU CUDA IPC at 42–44 GB/s; CPU↔GPU at 3–7 GB/s), and the zero-copy read frees the device's limited CPU for inference instead of serialization and relay copies.
• At the edge (multi-daemon on one host): separate daemons per pipeline stage no longer pay the daemon-relay tax — direct cross-daemon reads deliver 65× over native dora (89 MB/s → 5.8 GB/s), so multi-process edge pipelines scale without the message-passing overhead. (On a 1-Gbps LAN the relay is already ≈85% of line rate, so the cross-machine LAN gain is small by physics; the same-host win is where the edge multi-process tax is.)
• Edge ↔ cloud (WAN): native dora cannot carry large frames across machines at all (Drop + express drops fragments; the producer blocks forever), and ROS 2's network DDS collapses to ~1.1 MB/s on a 38 ms-RTT WAN — RTT-paced, not bandwidth-bound. The pool's blocking-control relay delivers large frames at link bandwidth (~60% of a ~7 MB/s WAN link, scaling with frame size) — closing the gap where distributed deployments currently have no viable data path for training-data uploads, fleet telemetry, or cloud-assisted inference.

One API (write_memory_pool / read_memory_pool) covers end, edge, and cloud with automatic device- and topology-aware path selection, turning message-passing deployments into shared-state pipelines without changing the dataflow description.

[tang-canran]

Thanks — both notes addressed in 170278f2:

1. Retry window gate refined (the residual of point 1). You're right that name-equality only separated single-daemon from everything-else: a same-host cross-daemon read resolves to the sender's machine-A name, which differs from the receiver's local derivation, so it inherited the 1-hour window. The gate now checks whether the resolved sender-side segment actually exists on this host (/dev/shm/<resolved-name>): same-host pools (single-daemon or cross-daemon) fail fast in 500 ms when a producer crashes, and only genuinely remote pools (whose segment is not openable locally) keep the 3600 s window. Verified: same-host cross-daemon smoke still passes end-to-end (the strengthened condition below), single-daemon regression at 6660 MB/s.
2. Smoke test success condition strengthened. The marker alone is no longer sufficient — it must be accompanied by a clean finish (dataflow finished / finished successfully, no Failed). Combined with the receiver's per-frame tensor[0] == i assertions (which run before the marker prints), a degraded or short transfer now fails the test.

CI re-running on 170278f2.

[phil-opp]

🤖 Automated review by Claude — fully automated, no human in the loop; please verify before acting.

Follow-up on 170278f2, which landed after the last review. It addresses the two non-blocking notes from that pass, and I found no new issues in the delta:

1. Retry-window gate — now keyed on whether the resolved sender-side segment actually exists at /dev/shm/<name> on this host, so same-host reads (single- and cross-daemon) fail fast in ~500 ms when a producer crashes, and only genuinely remote pools keep the 3600 s window. This closes the residual of the original point 1 (the gate was previously single-daemon vs. everything-else).
2. Smoke-test success condition — no longer marker-only; it now also requires a clean finish (no Failed) and is backed by the receiver's per-frame tensor[0] == i assertions, so a degraded/short transfer fails the test.

I re-checked the safety-critical paths alongside the delta — seqlock write bounds in write_cross_pool_data, appended-last enum discriminants, the restored dora_pool_ free-path namespace guard, and coordinator reply-routing cleanup — and see nothing new to flag. Standing non-blocker unchanged: the true two-host *_cross.yml path still has no CI coverage (it needs two machines), so that path remains manually validated only.

---

Generated by Claude Code

[phil-opp]

🤖 Automated review by Claude Code — fully automated review, not vetted by a human.

The diff is unchanged since the last pass and the earlier rounds look genuinely addressed. Looking more closely at orphan cleanup, though, I think there's a latent correctness issue the earlier passes and I both missed — worth a look before merge.

cleanup_orphans is not machine-scoped, so a daemon can unlink a sibling daemon's live same-dataflow segments. In libraries/extensions/memory-pool/src/lib.rs the sweep matches:

let is_this_dataflow = name.starts_with("dora_pool_")
    && (name.starts_with(&unqualified_prefix)      // dora_pool_{df}_
        || name.contains(&qualified_segment));      // _{df}_

The contains("_{df}_") arm matches machine-qualified names dora_pool_{machine}_{df}_{node}_{counter} for any machine, not just this one. And with DORA_MACHINE_ID set (the multi-daemon case), the Python auto-name for a node's own local pool is also machine-qualified (dora_pool_{machine}_{df}_{node}_{counter}), so the match hits live local sender pools, not only mirrors.

cleanup_orphans runs per-daemon at the top of spawn_dataflow (binaries/daemon/src/lib.rs) with the coordinator-assigned dataflow UUID that every participating daemon shares. So the doc-comment rationale — "safe … because dataflow IDs are UUIDs and no two daemons run the same one concurrently" — doesn't hold for a same-host multi-daemon dataflow. The PR's own create_cross_pool_shmem comment even notes both daemons share one /dev/shm namespace on a single host.

Severity is race-gated rather than every-run, which is why it slipped past:

• remove_file is unlink(2); on /dev/shm it drops the name only, so consumers that already hold the mapping are unaffected. The concrete harm is a consumer that opens the segment by name after the sweep getting ENOENT (the Python read path does reopen by resolved name), or the sender later recreating a same-named segment with a divergent inode.
• The sweep runs once, before nodes are built, and pools are created lazily — so a simultaneous single launch is fine (every sweep precedes any pool). The harmful window needs a peer daemon's segments to already be live when another daemon enters spawn_dataflow: staggered spawn, a late-joining/reconnecting daemon, or a re-spawn of an already-running dataflow.

For contrast, cleanup_all is deliberately own-machine-scoped (and cleanup_all_removes_only_own_machine_mirrors asserts a foreign machine's mirror survives), so cleanup_orphans is the one path that crosses the machine boundary. Scoping its sweep to the unqualified dora_pool_{df}_ plus the own-machine dora_pool_{own_machine}_{df}_ forms would close the gap, and a regression test in the spirit of cleanup_all_removes_only_own_machine_mirrors would pin it.

Everything else I re-checked held up: the seqlock write bounds in write_cross_pool_data, the machine-id / Remote-only-locality gates on the pool messages, the appended enum variant, the GPU HtoD staging bounds, and the concurrent-writer interleave test all look correct.

---

Generated by Claude Code

[tang-canran]

Good catch — the contains("_{df}_") arm indeed crosses the machine boundary, and you are right that the "UUIDs can never collide" rationale fails for a same-host multi-daemon dataflow where both daemons share one /dev/shm namespace. Fixed in 49c848f2:

The sweep is now machine-scoped. cleanup_orphans(dataflow_id, own_machine_id) only matches:

• the unqualified local form dora_pool_{df}_* (no DORA_MACHINE_ID in play), and
• this daemon's own qualified prefix dora_pool_{own_machine}_{df}_* (starts_with, with the trailing _ guarding against prefix aliasing) — the same form the Python side auto-generates for this machine's local pools and mirrors;

a sibling daemon's dora_pool_{other_machine}_{df}_* segments are never touched. The daemon passes self.machine_id.as_deref() at the spawn_dataflow call site (the existing cleanup_orphan_mirrors at daemon startup was already own-machine-scoped by prefix).

Regression test (cleanup_orphans_removes_local_and_own_machine_qualified_segments): sweeps with Some("machine-1") and asserts the unqualified segment and the own-machine segment are removed while dora_pool_machine-2_{df}_* (sibling daemon, same dataflow) and a foreign dataflow segment survive — the sibling-holds-live-segment case you described, in the spirit of cleanup_all_removes_only_own_machine_mirrors.

Tests: dora-memory-pool 13/13 (incl. the updated regression), dora-daemon --lib 209/209 (incl. both health_check_tests and cross_pool_write_tests); cargo clippy -p dora-memory-pool -p dora-daemon -- -D warnings and cargo fmt --check clean. CI re-running on 49c848f2.

[phil-opp]

🤖 Automated review by Claude Code — fully automated review, not vetted by a human.

The latest commit (49c848f2) addresses the earlier cleanup_orphans machine-scoping issue. The bare contains("_{df}_") arm — which matched a sibling daemon's dora_pool_{other_machine}_{df}_* segments — is replaced with a starts_with on this daemon's own dora_pool_{own_machine}_{df}_ prefix, and the only remaining match is the unqualified dora_pool_{df}_ form (which a sibling's machine-qualified name does not start with, since a UUID dataflow id won't alias a machine id). The added regression test pins that a dora_pool_machine-2_{df}_* sibling segment survives a machine-1 sweep. No new issues in this delta.

---

Generated by Claude Code

[heyong4725]

I would not approve this as-is. I found five actionable issues at head 3c88617e.

Findings

• [P1] Cross-machine writes return before the remote pool is updated. The daemon spawns the Zenoh publish and immediately replies Ok, allowing the subsequent output notification to overtake the tensor data. The receiver can then return the previous stable frame; the example works around this by polling a tensor element for up to 300 seconds. The protocol needs a remote commit acknowledgement or sequence-aware read, not application-specific polling. daemon/src/lib.rs#L5773-L5791

• [P1] Pools larger than approximately 64 MiB cannot use the new transport. Registration accepts tensors up to 1 GiB, but push_mirror_update embeds the entire tensor in a node-to-daemon TCP request, whose hard limit is 64 MiB including bincode overhead. Worse, the error is only logged, so registration/write appears successful while the receiver waits indefinitely. Either enforce the actual cross-machine limit or pass a shared-memory reference to the daemon. python/node/src/lib.rs#L4085-L4095

• [P1] Cross-pool state is not scoped by dataflow. cross_pools and CROSS_REGISTER_PENDING are keyed only by pool_id. Pool IDs repeat across concurrently running dataflows because each node process starts its counter again. A later registration can overwrite another flow's peer, an acknowledgement can satisfy the wrong registration, and freeing one pool can make the other flow's subsequent writes silently stop forwarding. Use (DataflowId, pool_id) consistently. memory-pool/src/lib.rs#L76-L85

• [P2] Every dataflow spawn leaks a Zenoh subscriber task. The task handle is discarded and the receive loop has no dataflow-shutdown branch. finish_dataflow therefore removes the dataflow while its subscriber, session clone, and event sender remain alive for the daemon's lifetime. Repeated or failed spawns accumulate resources and can create duplicate consumers. daemon/src/lib.rs#L4652-L4677

• [P2] The mandatory clippy gate fails on non-Linux platforms. machine_id is only used inside a Linux-gated block, producing an unused-variable error on macOS with -D warnings. memory-pool/src/lib.rs#L413

Validation

• cargo test -p dora-memory-pool: 11 passed.
• cargo check -p dora-message -p dora-coordinator -p dora-daemon: passed.
• cargo clippy -p dora-memory-pool -- -D warnings: failed on the non-Linux unused variable above.
• Current remote CI shows format/check/clippy passing, but Test, E2E, and semantic-contract jobs were skipped.
• CUDA and cross-machine smoke tests were not run.

[tang-canran]

Thank you for the careful human review — all five issues are addressed in c83c59bf:

P1 — Cross-machine writes now commit remotely before replying. write_memory_pool withholds its daemon reply until the mirror daemon confirms the segment write: the write is assigned a per-pool sequence, MemoryPoolWrite carries it, and after the mirror memcpy the mirror daemon publishes MemoryPoolWriteAck { seq, ok } back through the memory-pool topic. The pending reply is resolved only by the seq-matched ack (an ack for a previous write can never satisfy a newer pending reply), so the send_output notification that follows the write can never overtake the tensor data and the receiver can no longer return the previous stable frame. A publish failure resolves the reply with the error immediately, and a 120 s safety timeout fails a write whose ack never arrives (peer restart) instead of hanging the node — no application-level polling needed. (The example's 300 s tensor[0] == i retry remains as harmless belt-and-suspenders; the protocol no longer depends on it.)

P1 — Cross-machine pools > 64 MiB are now rejected at registration. Both the daemon's RegisterCrossMachinePool handler and the python register_memory_pool check size > MAX_MESSAGE_BYTES - 1024 and fail with a clear error naming the transport limit. Previously a >64 MiB pool registered fine (the register request carries only metadata), but every per-frame push — which embeds the full tensor in the node→daemon request — failed with only a log line, leaving the receiver waiting forever. The shmem-reference alternative is noted as future work in the error message's spirit; enforcing the actual limit closes the silent-failure hole now.

P1 — Cross-pool state is now keyed by (dataflow id, pool id). MemoryPoolManager::cross_pools (extension) and CROSS_REGISTER_PENDING (daemon) both carry the dataflow id in their keys; register_cross_pool / is_cross / cross_peer / unregister_cross_pool take it explicitly. A concurrent dataflow's pool_node_0 can no longer overwrite another flow's peer, satisfy its ack, or stop its forwarding on free. New regression test cross_pool_state_is_dataflow_scoped pins the aliasing case (same pool id in two dataflows: registration, peer lookup, and free stay independent).

P2 — The per-dataflow zenoh subscriber task is now tracked and terminated. The tokio::spawn handle is retained in Daemon::memory_pool_subscribers and aborted in finish_dataflow. It is also aborted on the failed-spawn path — the subscriber is spawned before the node build, so a failed spawn never reaches finish_dataflow and previously leaked the task, its session clone, and its event sender for the daemon's lifetime (with duplicate consumers on re-spawn).

P2 — Non-Linux clippy fixed. cleanup_all's machine_id parameter is now consumed on non-Linux (let _ = machine_id alongside the existing table drain), so -D warnings passes on macOS.

Validation: dora-memory-pool 14/14 (incl. the new dataflow-scoping test), dora-daemon --lib 209/209, cargo clippy -p dora-memory-pool -p dora-daemon -p dora-message -p dora-cli -- -D warnings clean, cargo fmt --check clean. The python-extension crate's clippy output shows pre-existing pyo3 lint errors unrelated to this round (that crate is excluded from CI clippy). CUDA/cross-machine smoke tests are still gated on the two-host environment; the same-host cross-daemon smoke (#[ignore] torch-gated) runs in the nightly memory-pool-smoke job. CI re-running on c83c59bf.

[phil-opp]

🤖 Automated review by Claude — this is a fully automated review with no human in the loop. Treat it as advisory.

Followed up on the commits after the last review (c83c59bf, 56dcc799). The five issues from the earlier human review are addressed correctly in the current head:

1. Write-commit ack — MemoryPoolWrite now carries a per-pool seq, and the write path withholds the node reply in CROSS_WRITE_PENDING[(df,pool,seq)] until the mirror daemon publishes the seq-matched MemoryPoolWriteAck, so the send_output notification can no longer overtake the tensor data. Publish failure and a 120s timeout fail the write loudly rather than hanging it.
2. 64 MiB cap — enforced in both the daemon RegisterCrossMachinePool handler and the Python register_memory_pool (size > MAX_MESSAGE_BYTES - 1024), so an over-limit pool is rejected at registration instead of silently failing every push.
3. Dataflow-scoped cross state — cross_pools, CROSS_REGISTER_PENDING, and CROSS_WRITE_SEQ are all keyed by (dataflow_id, pool_id); cross_pool_state_is_dataflow_scoped pins the same-pool-id-across-dataflows aliasing case.
4. Subscriber lifecycle — the per-dataflow Zenoh subscriber handle is retained and .abort()ed in both finish_dataflow and the failed-spawn arm.
5. Non-Linux clippy — cleanup_all keeps machine_id consumed on non-Linux.

No new blocking issues in this delta. Two minor, non-blocking notes:

• The same-host cross-daemon smoke test takes the direct == true read path, which bypasses the new MemoryPoolWrite/MemoryPoolWriteAck machinery — so the write-ack fix itself is still only exercised by the manual two-host runs, not CI. Given how central that ack is to correctness, a test that forces the non-direct path (even mocked) would be worth adding.
• The read fast-path bail message is hardcoded to "not ready after 3600s" even for local pools that only waited ~500ms (cosmetic).

---

Generated by Claude Code

[tang-canran]

Follow-up on P1 (64 MiB limit) — the second half of your either/or is now implemented, so the limit is gone rather than enforced:

The write path now passes a shared-memory reference instead of the tensor. write_memory_pool's cross-machine push sends only (buffer_id, size) metadata over the node→daemon channel; the daemon opens the sender's segment (its shared_memory_name was recorded at registration), reads size bytes from the DORADMA data region, and forwards them through the existing path (zenoh → mirror write → commit ack). The node→daemon request is now KB-scale, so:

• the MAX_MESSAGE_BYTES (64 MiB) transport cap no longer applies to cross-machine pools — registration accepts up to the existing 1 GiB cap, and a >64 MiB pool transfers correctly;
• the registration-time rejection added in c83c59bf is removed (it would now wrongly block pools the transport can carry);
• payload-carrying requests are still honored for compatibility (older nodes / explicit push).

The failure mode is also loud now instead of silent: a segment that cannot be opened (or a corrupt header) fails the write with a clear error rather than logging and stranding the receiver.

Validation: dora-memory-pool 14/14, dora-daemon --lib 209/209, clippy/fmt clean; the same-host cross-daemon smoke (smoke_local_memory_pool_cpu2cpu_cross_local, torch-gated) reruns end-to-end on the new path (result to follow). Pushed as c83c59bf + 56dcc799 + this commit.

Update after implementation: the same-host cross-daemon smoke (smoke_local_memory_pool_cpu2cpu_cross_local) passes end-to-end on the new path (68.8 s, 100-frame transfer + clean finish). One subtlety surfaced during verification and is fixed in the final commit: the register-time initial push arrives at the daemon before the python's local registration completes, so the segment name is resolved deterministically (the machine-qualified auto-name) first, with the daemon table as fallback for explicit name= pools; and write errors now reply to the node instead of propagating (a handler error was tearing down the node connection and cascading into a daemon disconnect). dora-memory-pool 14/14, dora-daemon --lib 209/209, clippy/fmt clean. Pushed as f420a542.

[tang-canran]

Thanks — both notes addressed in c7d2a8e2:

1. The write-ack path now has CI coverage without a two-host environment. You're right that the same-host cross-daemon smoke reads via direct == true and bypasses the MemoryPoolWrite/MemoryPoolWriteAck machinery entirely. The ack resolution is extracted into a free function resolve_cross_write_ack() and pinned by a unit test (write_ack_resolves_only_seq_matched_pending_reply): a stale seq resolves nothing (both pending replies stay untouched), the seq-matched ack resolves exactly its own pending reply, and a failed mirror write (ok=false) surfaces as the error reply. This covers the correctness core of the commit protocol — the seq-matching — while the two-host transfer remains manually validated (as it must, given CI has no second host).

2. The read fast-path bail message now reports the actual wait window. It previously hardcoded "not ready after 3600s" even for local pools that waited only ~500 ms; it now prints the real window (wait_ms / 1000), so a local pool reports "not ready after 0s" and a remote one "after 3600s".

Validation: dora-daemon --lib 210/210 (incl. the new ack test), dora-memory-pool 14/14, clippy -D warnings and fmt clean on daemon/memory-pool/message/cli. CI re-running on c7d2a8e2.

[phil-opp]

Reviewed the delta since the last automated pass — the shared-memory-reference write path (f420a542) and the ack test (c7d2a8e2). The ack-resolution unit test and the "3600s" message fix look good. I found three new items, in priority order:

1. read_pool_segment_data runs on the daemon event loop. In the DaemonNodeEvent::WriteMemoryPool handler, the shmem-reference read (read_pool_segment_data(&shmem_name, size)) executes synchronously before the tokio::spawn — a vec[0u8; size] plus a full size-byte copy_nonoverlapping on the event loop for every cross-machine frame. This is the same large-payload work that the rest of this handler deliberately spawns off the loop (the comments right here note that inline multi-MB work "wedges the daemon event loop (heartbeats + node replies + output delivery)"). Each frame now blocks the loop for the duration of that read, scaling with pool size. The segment read should happen inside the spawned publish task, not on the event loop.

2. No size cap on cross-machine mirror allocation. f420a542 removed the earlier registration cap, but the "1 GiB registration cap" it references doesn't appear to be enforced anywhere — not in the Python register_memory_pool, not in the daemon RegisterCrossMachinePool handler, and not in create_cross_pool_shmem, which allocates size + data_offset in /dev/shm straight from the remote RegisterPool event's size field. A buggy or corrupted peer daemon (cross-machine zenoh input) can drive an unbounded /dev/shm allocation on the mirror host (memory-exhaustion DoS). The read/write copy paths are correctly bounds-checked against the real segment length, so this is allocation-size only, not an out-of-bounds access — but an explicit maximum should be reinstated on both the registering node and the mirror-creating daemon.

3. (minor) Unescaped dtype/device in the mirror header JSON. create_cross_pool_shmem interpolates the remote-provided dtype/device into the header JSON via format! without escaping; a value containing " or \ corrupts or injects into the JSON. Impact is limited (the reader re-validates the declared size against the segment length before any copy, so at worst the mirror is unreadable), but these untrusted cross-machine strings should be escaped or validated before being written into a parsed structure.

---

🤖 Automated review by Claude (Claude Code). This review was generated fully automatically with no human in the loop and has not been vetted by a maintainer — treat it as advisory.

---

Generated by Claude Code

[tang-canran]

All three items addressed in da17ba25:

1. The shmem-reference read is off the event loop. The WriteMemoryPool handler now does only the cheap segment-name resolution (deterministic auto-name first, daemon table for explicit name= pools); the actual read (vec![0u8; size] + copy) runs inside the spawned publish task, so a large frame no longer blocks heartbeats, node replies, or output delivery. Read failures and the no-segment case resolve the seq-matched pending reply with the error (via the same resolve_cross_write_ack path as publish failures), so the node's write still fails loudly instead of hanging.

2. The mirror side now enforces the 1 GiB cap. You're right that the local-side caps (python register_memory_pool, daemon RegisterPinnedMemory) don't protect the mirror: create_cross_pool_shmem allocates size + data_offset in /dev/shm straight from the remote RegisterPool event's size field. The mirror-creating task now rejects size > 1 GiB before creating the segment, and the error flows back to the origin through the existing RegisterPoolAck path (so the registering node gets the failure, not a silent mirror).

3. The mirror header JSON is escaped. dtype/device arrive from the remote event (untrusted cross-machine strings) and were interpolated with format!; the header is now built with serde_json::json!, so quotes/backslashes are escaped instead of corrupting or injecting into the parsed structure. The reader's existing bounds checks (declared size vs actual segment length) remain the second line of defense.

Validation: dora-daemon --lib 210/210, dora-memory-pool 14/14, clippy -D warnings and fmt clean. CI re-running on da17ba25.

[phil-opp]

Follow-up on da17ba25, which landed after the last review and addresses the three items from that pass. I checked the diff and it looks correct: the shared-memory-reference read (full-size alloc + copy) now runs inside the spawned publish task rather than on the event loop, with both failure paths (no resolvable segment, read error) routed through the seq-matched resolve_cross_write_ack so the node still gets a loud reply; the None+empty vs None+non-empty match arms correctly separate a resolution failure from a payload-carrying request. The 1 GiB mirror cap is enforced before create_cross_pool_shmem with the error returned via RegisterPoolAck, and the mirror header JSON is now built with serde_json::json! so the untrusted dtype/device strings are escaped. No new issues in this delta.

---

🤖 This is a fully automated review by Claude (Claude Code). No human has vetted this comment; please treat it as advisory.

Generated by Claude Code

---

Generated by Claude Code

[phil-opp]

🤖 Automated review by Claude — fully automated, no human in the loop; please verify before acting.

Re-reviewed the delta since the last pass — c13d9b8 (direct-TCP cross-machine data plane), which landed after the previous review. The safety-critical parts of the new serve_cross_data_frame hold up: size is validated against the actual mirror length (data_offset + size > shmem.len() → error) before the read_exact(dst), so there is no OOB write or unbounded allocation from the untrusted size field; pool_len is capped at 1024; and leaving the seqlock generation odd when a payload read fails mid-frame is the correct fail-safe (readers reject the torn frame and it self-heals on the next write, matching the documented GPU-path precedent). No blocking issues found in this delta.

Two minor, non-blocking notes:

1. start_cross_data_listener accept loop can hot-spin. let Ok((stream, peer)) = listener.accept().await else { continue; }; (binaries/daemon/src/lib.rs) retries with no backoff. Per-connection errors (ECONNABORTED) are fine, but a persistent error such as EMFILE/ENFILE (fd exhaustion) makes accept() return Err immediately every iteration → a 100% CPU busy loop. A short sleep on the error branch would bound it.

2. The new direct-TCP path has no automated coverage. The same-host cross-daemon smoke reads via the direct == true (IPC) path, which bypasses serve_cross_data_frame/send_cross_data_frame entirely — so this 437-line network data plane, like the two-host *_cross.yml transfer, is only manually validated. Given it's the new steady-state cross-machine write path, a targeted test of the frame codec (send_cross_data_frame → serve_cross_data_frame round-trip over a loopback TcpListener, asserting the mirror segment and the seq-matched ack) would be worth adding.

Minor topology note (self-healing, not a bug): the origin dials peer_addr.ip():data_port, where peer_addr is the target daemon's WS source address as seen by the coordinator. In a NAT'd/multi-homed deployment that IP may not be the one reachable from the origin daemon; it degrades to the zenoh relay on connect failure, so correctness is preserved, but the direct fast-path would silently never engage there.

---

Generated by Claude Code

[tang-canran]

New: direct-TCP cross-machine data plane (feature addition, not a review item — the user-requested "one-copy" write path).

write_memory_pool now sends only a lightweight request; the origin daemon reads the sender segment (one user-space copy) and streams the bytes over a direct TCP connection to the mirror daemon, which reads the stream straight into the mirror segment (zero user-space copies on the receive side, seqlock + per-pool async lock). The commit ack still travels over zenoh, so the pending machinery and the "no stale frame" guarantee are unchanged. Falls back to the zenoh relay when no endpoint is known or the direct send fails.

Plumbing: the mirror daemon runs a data listener (port 7410, DORA_MEMORY_POOL_DATA_PORT), advertises it in RegisterPoolAck.data_port; the coordinator tracks each daemon WS peer address and returns it in ResolveMachine; the origin keeps a persistent connection per endpoint (dead connections dropped and lazily re-established).

LAN measurement (5090↔A100, 100 × 61.44 MB frames, byte-identical previews, 0 zenoh fallbacks): 51.15 → 89.38 MB/s (+75%), ≈76% of the 117.4 MB/s line rate. The remaining gap is the send-side read copy, the receive-side recv, and the per-frame handshake — sendfile/splice kernel zero-copy is the next step. dora-daemon 210/210, dora-coordinator 122/122, dora-memory-pool 14/14, clippy/fmt clean. Pushed as c13d9b85. (WAN validation needs the rendezvous machine to publish the data port.)

[tang-canran]

Thanks — both notes addressed in a56472aa, and the topology note is acknowledged:

1. Accept-loop hot-spin is bounded. The listener now sleeps 50 ms on accept errors — transient errors (ECONNABORTED) still retry promptly, while a persistent error (EMFILE/ENFILE fd exhaustion) can no longer spin at 100% CPU.

2. The direct-TCP codec now has automated coverage. serve_cross_data_frame is split into handle_cross_data_frame (frame parse + mirror write, zenoh-free) and the ack publish; a loopback round-trip test (direct_tcp_frame_round_trip_writes_mirror) drives send_cross_data_frame → handle_cross_data_frame over a TcpListener and asserts the payload lands byte-identical in the mirror's data region under an even seqlock generation, with the returned ack info matching (dataflow, pool, seq). You're right that the same-host smoke (direct == true) bypasses this data plane entirely — this is the new steady-state write path's first automated coverage; the remaining gap (the zenoh ack publish itself) needs a real zenoh session and stays with the two-host manual validation.

3. NAT/multi-homed topology note — acknowledged, self-healing by design. The origin dials the target's coordinator-visible WS source address; where that IP is not origin-reachable (NAT'd/multi-homed), the direct path silently never engages and writes fall back to the zenoh relay. Correctness is preserved; a future enhancement could advertise a dialable address explicitly in RegisterPoolAck (e.g., from --zenoh-peer or an env override) — noted in design.md.

Validation: dora-daemon --lib 211/211 (incl. the new codec test), dora-coordinator 122/122, dora-memory-pool 14/14, clippy -D warnings and fmt clean. CI re-running on a56472aa.

[tang-canran]

Non-blocking issues ledger — all resolved (per the round-by-round reviews):

#	Item	Resolution
R3	Smoke success condition doesn't assert frame count / byte correctness	170278f2 — marker + clean-finish + per-frame tensor[0] == i assertions
R3	3600 s retry window mis-gated for same-host cross-daemon reads	170278f2 — gate keyed on /dev/shm segment existence
R7	Write-commit ack machinery had no automated coverage	c7d2a8e2 — resolve_cross_write_ack unit test (stale seq / matched seq / failed-write arms)
R7	Read fast-path bail message hardcoded "not ready after 3600s"	c7d2a8e2 — reports the actual wait window
Direct-TCP	Accept loop can hot-spin on persistent errors (EMFILE/ENFILE)	a56472aa — 50 ms sleep on the error branch
Direct-TCP	New data plane had no automated coverage	a56472aa — loopback codec round-trip test (send → handle, mirror bytes + seqlock + ack info)
Direct-TCP	NAT/multi-homed: origin dials the coordinator-visible WS source address, which may be unreachable	dbee784b + 1b9f9d3b — lazy listener (opened on first RegisterPool, not at daemon startup) + explicit dialable-address advertisement

On the topology note specifically (1b9f9d3b): the mirror daemon now advertises a dialable address via DORA_MEMORY_POOL_DATA_ADDR (full ip:port), carried in RegisterPoolAck.data_addr; the origin prefers it over the coordinator-derived address and falls back to the derived one otherwise. This matters beyond the textbook multi-homed case — with a co-located coordinator the mirror daemon's WS source address is literally 127.0.0.1 (it dials the coordinator locally), so the derived address could never be dialed by a remote origin. The listener itself is also now opened lazily (dbee784b): only daemons that actually mirror a pool bind the port, so non-participating daemons expose nothing.

Validation: dora-daemon --lib 211/211, dora-coordinator 122/122, dora-memory-pool 14/14, clippy -D warnings and fmt clean. CI re-running on 1b9f9d3b.