Fix/reliable inter daemon output close

[SunSunSun689]

Summary

Fix inter-daemon messages being silently dropped when the Zenoh publish drain channel is full.
send_to_remote_receivers now retries full-channel enqueue attempts briefly and returns an error if the message still cannot be queued, instead of logging a warning and returning Ok(()). This prevents control events like OutputClosed from being lost without the caller knowing.

[trunk-io]

Merging to main in this repository is managed by Trunk.

• To merge this pull request, check the box to the left or comment /trunk merge below.

After your PR is submitted to the merge queue, this comment will be automatically updated with its status. If the PR fails, failure details will also be posted here

[phil-opp]

Automated review by Claude — fully automated, may contain mistakes; please verify before acting.

I found the following issue(s):

Making send_to_remote_receivers fatal can tear down the entire daemon on the regular output path. By returning Err when the channel stays full, the failure now propagates fatally for normal Output events:

• binaries/daemon/src/lib.rs:5046 — send_out() calls send_to_remote_receivers(...).await?.
• binaries/daemon/src/lib.rs:4738 — the SendOut event arm calls send_out(...).await.context("failed to send out")? (unlike the CloseOutputs arm at ~4714, which swallows the error into a reply_sender and does not crash the daemon).
• binaries/daemon/src/lib.rs:1846 — handle_node_event(...).await? runs inside the top-level while let Some(event) = events.next().await loop (starts at line 1781), so the error exits the daemon's main event loop, taking down all local dataflows.

Net effect: sustained back-pressure to a remote daemon that keeps the 256-slot channel full for more than ~30 ms (3 × 10 ms) now converts a single dropped inter-daemon message into a fatal daemon exit. That is a strictly more severe failure mode than the previous warn-and-drop for regular outputs. Note the PR's motivating case — OutputClosed via send_output_closed_events — actually goes through the CloseOutputs reply path and is not fatal; it's the generic Output path that is.

Secondary: the retry sleep blocks the single-threaded event loop. enqueue_zenoh_outbound_reliably awaits tokio::time::sleep (up to 3 × 10 ms) inline within send_out, which is awaited directly in the main event loop (line 1846). While retrying one back-pressured remote publish, the daemon cannot process any other event (other nodes' outputs, heartbeats, coordinator events) for up to ~30 ms.

Worth deciding whether channel-full for the regular output path should be fatal at all, or whether the retry should avoid stalling the event loop (e.g. handled off the main loop / bounded without an inline sleep).

---

Generated by Claude Code

[phil-opp]

Automated review by Claude — this is a fully automated review with no human in the loop; please verify before acting on it.

One scope note in addition to the earlier review. The stated change here is the inter-daemon publish retry (enqueue_zenoh_outbound_reliably in binaries/daemon/src/lib.rs), but the diff against main also contains the entire ~385-line descriptor field classifier — libraries/core/src/descriptor/classify.rs plus the rewrite of resolve_aliases_and_set_defaults_in_topology and removal of node_kind_mut/NodeKindMut.

That refactor is the whole content of #3070 and also appears verbatim in #3072, so the same change is currently duplicated across three PRs and is hard to review independently here. If these are meant to be stacked on #3070, it would help to note the dependency (or rebase once #3070 lands); otherwise the classifier is orthogonal to the Zenoh fix and would be much easier to review on its own. It also changes descriptor acceptance — validate_against_whitelist now hard-errors on top-level fields that resolution previously ignored — which is a compatibility change worth landing separately with its own tests and a changelog note.

---

Generated by Claude Code

[phil-opp]

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

Thanks for splitting the data-plane path off — routing regular Output events through the non-fatal try_send_to_remote_receivers resolves the earlier concern about backpressure crashing the daemon on the normal send path.

I checked whether the reliable path (send_output_closed_events → send_to_remote_receivers, which now returns Err on sustained-full) is fatal on node-finish. It does propagate via ? up through handle_dora_event, but it's caught at the run_inner call site — the match result { Err(e) => warn!("daemon disconnected from coordinator… reconnect") } loop — so per the #2029 design the running nodes are preserved and the daemon reconnects rather than crashing. On dora run the path is unreachable (no remote receivers, so the closed-outputs loop never calls it). So this is not a daemon-crash path. Two smaller things do seem worth a look:

1. On that sustained-full OutputClosed case you now get a spurious "daemon disconnected from coordinator… reconnecting" cycle where the old code just logged a warning and returned Ok(()). The reconnect doesn't itself redeliver the OutputClosed, so the "reliable" intent isn't actually achieved on this path — the net effect is a misleading disconnect log plus an unnecessary reconnect. It might be cleaner to treat sustained-full here as non-fatal (log + a failure counter, like the data path) or to drive an explicit redelivery, rather than surfacing it as a coordinator-connection error.

2. (Carried over from the earlier review, still unaddressed) enqueue_zenoh_outbound_reliably awaits tokio::time::sleep (up to ~3×10 ms) inline on the single-threaded event loop, so during the retries the daemon can't process other node/heartbeat/coordinator events.

Also, as previously noted, this diff still bundles the unrelated ~385-line node-classifier refactor from #3070, which is orthogonal to the close-handling fix.

---

Generated by Claude Code

[phil-opp]

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

Reviewed the latest push: the unrelated ~385-line node-classifier refactor (flagged in the prior reviews as bundled from #3070) is now gone, so this PR is scoped to just the Zenoh publish-enqueue change — that resolves the earlier scope note. No new issues in the current diff, and the data-plane split (regular Output → non-fatal try_send_to_remote_receivers) still looks correct; the added tests are meaningful.

Two earlier points remain outstanding on the reliable path and are unchanged by this push:

1. enqueue_zenoh_outbound_reliably still awaits tokio::time::sleep (up to ~3×10 ms) inline, so while retrying one back-pressured remote publish the single-threaded daemon event loop can't process other node/heartbeat/coordinator events.
2. On the send_output_closed_events → send_to_remote_receivers path, sustained-full now returns Err and surfaces as a spurious "daemon disconnected from coordinator … reconnecting" cycle (non-fatal, but the reconnect doesn't redeliver the OutputClosed, so the "reliable" intent isn't actually achieved). A non-fatal log-plus-counter, like the data path, or an explicit redelivery would be cleaner.

---

Generated by Claude Code

[phil-opp]

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

The latest commit (d5265b7, Aug 11) replaces the earlier sleep-based retry design with a two-channel priority split, and it resolves both previously-outstanding concerns: there's no longer an inline tokio::time::sleep retry on the event loop, and a sustained-full control queue is now non-fatal in handle_control_enqueue_result (only a gone forwarder is fatal), so the spurious "daemon disconnected … reconnecting" cycle on the OutputClosed path is gone.

I found one new issue introduced by the two-channel prioritization, though:

OutputClosed-vs-Output reordering / data-loss window. Regular Output and OutputClosed for the same output are published on the same zenoh topic via the same cached publisher (zenoh_daemon_control_topic, dataflow.publishers keyed by output_id), but they now travel through separate channels — data via zenoh_publish_tx, OutputClosed via zenoh_control_tx → forwarder → control publish queue. The drain task's tokio::select! { biased; … } drains all ready control messages before data, so a later-produced OutputClosed can publisher.put() ahead of earlier-produced Output messages still sitting in the data channel. On the receiver, InterDaemonEvent::OutputClosed runs close_input(...), so those trailing Output events then arrive at an already-closed input and get dropped.

That reorders close ahead of data on a single logical stream — a data-loss window that didn't exist when both events shared one FIFO channel on main. It triggers whenever the data channel merely has a backlog at the moment the producing node finishes, not only when the queue is full. Prioritizing control traffic is reasonable, but OutputClosed needs to stay ordered behind the data on its own output — e.g. route the close through the same data channel, or drain any pending data for that publisher before publishing its close.

The added backpressure unit tests (try_enqueue_zenoh_outbound Queued/Full/Closed, data_queue_full_does_not_block_control_enqueue, control_pending_enqueue_full_is_recorded_without_error) are meaningful, but none cover cross-channel ordering, which is where the new risk lives.

---

Generated by Claude Code

[phil-opp]

🤖 Automated review by Claude — fully automated, no human reviewed this; please verify before acting.

Re-reviewed after the latest push (a20a6c2, "preserve inter-daemon close ordering"), which is new since the previous automated review.

The prior review's blocking concern — the two-channel priority split letting a later OutputClosed be published ahead of earlier Output messages on the same stream — is resolved. This commit reverts to a single FIFO (zenoh_publish_tx) shared by both paths:

• Regular Output → try_send_to_remote_receivers → try_enqueue_zenoh_outbound (binaries/daemon/src/lib.rs:5116): non-fatal, drops on Full and records net_publish_failures, returns Ok. Resolves the original daemon-teardown concern.
• OutputClosed → send_to_remote_receivers → enqueue_required_zenoh_outbound (binaries/daemon/src/lib.rs:5107, :7310): try_send, and on Full spawns a detached task that send().awaits for capacity instead of sleeping inline or dropping. It never returns Err for backpressure, so the earlier spurious "daemon disconnected … reconnecting" cycle on the close path is also gone.

Ordering within a single output holds: both events share one FIFO with one consumer (the drain task), there is no post-close data on that output, and tokio's mpsc hands the next freed permit to the queued send().await waiter, so OutputClosed cannot pass earlier data for its output. No missed/double close and no deadlock in the enqueue path. The output_closed_waits_for_publish_capacity_without_dropping and required_publish_channel_closed_is_recorded tests meaningfully exercise the reliable path.

No new issues found in the current diff.

---

Generated by Claude Code

[phil-opp]

🤖 Automated review by Claude — fully automated, no human in the loop; treat the findings as suggestions to verify rather than as authority.

First, the good news: this does not repeat the #3072 defect. The diff never touches open_external_mappings, the finish-straggler veto, or any node-liveness computation, so no healthy node can be SIGKILLed by it. There is also no event-loop deadlock — enqueue_required_zenoh_outbound hands the blocking send().await to a tokio::spawn.

1. The stated guarantee is not delivered. binaries/daemon/src/lib.rs:7484-7495 — the new try_send_to_remote_receivers looks the publisher up in the same dataflow.publishers map, so the "required" OutputClosed and the droppable data messages share one publisher declared .congestion_control(CongestionControl::Drop). Making the in-process channel enqueue reliable only moves the drop one layer down.

Failure scenario: the remote link is saturated, node A finishes, the close event now survives the 256-slot channel and reaches put() — and zenoh discards it. publish_zenoh_outbound bumps net_publish_failures and returns. Downstream node B on daemon 2 never receives InputClosed and hangs exactly as before. A real fix needs a separate CongestionControl::Block publisher (or a retry/ack) for control events.

2. None of the eight added tests exercise the fix. zenoh_publish_drain_preserves_output_closed_ordering calls only try_send/try_recv on a mpsc::channel::<&'static str> — it contains no dora code and passes identically on unmodified main. output_closed_uses_the_same_fifo_as_regular_output is the same test with a wrapper call, and regular_output_enqueue_full_is_nonfatal duplicates zenoh_publish_enqueue_reports_full_without_waiting line for line. The rest assert tokio's documented try_send semantics.

Concretely: revert the enqueue_required_zenoh_outbound call at the OutputClosed site to a plain try_send, and every test except the two that name the new helper still passes. What needs covering is close_outputs → send_to_remote_receivers with a saturated channel.

3. output_closed_waits_for_publish_capacity_without_dropping is order-fragile. It relies on a single yield_now().await being enough for the spawned task to take the permit freed by the preceding try_recv. That holds on the current-thread runtime today but breaks intermittently under flavor = "multi_thread" or if the helper gains an await before its send. publish_rx.recv().await with a timeout would be deterministic.

4. A reported metric changes meaning silently. handle_publish_enqueue_result now bumps net_publish_failures on channel-full drops, where it previously counted only zenoh put() failures. That is surfaced as NetworkMetrics.publish_failures (binaries/daemon/src/lib.rs:2074).

5. try_send_to_remote_receivers duplicates ~38 lines of publisher lookup and ZenohOutbound construction from send_to_remote_receivers. One lookup helper plus two enqueue policies would stop the two copies drifting.