Fix reliable inter-daemon close handling

[SunSunSun689]

Summary

Fix two inter-daemon shutdown/routing cases that could leave dataflows hanging:

• retry Zenoh drain-channel enqueue instead of silently dropping inter daemon control events when the queue is temporarily full
• keep the finish-straggler watchdog armed for unrelated local nodes when another node has an open remote output

[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]

No issues found.

The straggler-watchdog change is a correct narrowing of the previous behavior. Before, finish_stragglers bailed out entirely (return Vec::new()) whenever any open_external_mappings entry existed, so a single cross-daemon output disabled the finish-escalation watchdog for every unrelated local node. The new code keeps the global guard on self.stop_sent only, and instead exempts each producing node individually via remote_output_open (select_finish_stragglers does if node.remote_output_open { continue; }). The remote_output_open flag is matched against the straggler's id (the producing node id), so only the node actually flushing to a remote consumer is skipped while other local stragglers stay covered.

The two added tests assert meaningful behavior: remote_output_open_skips_only_that_node checks that the pure selector skips just the remote producer, and unrelated_remote_output_does_not_disable_local_straggler_watchdog exercises the full finish_stragglers path to confirm an unrelated remote mapping no longer suppresses escalation of a stuck local node.

One minor maintenance note (not blocking): is_set() and the all_checkable list in validate_against_whitelist() (libraries/core/descriptor/classify.rs, shared with #3070/#3071) each duplicate the full Node field set and must be kept in sync by hand as new fields are added — a missed field would silently regress the whitelist. Consider a comment or a compile-time exhaustiveness check to guard against that.

---

🤖 This is a fully automated review by Claude (Claude Code). No human has reviewed these comments before posting.

Generated by Claude Code

---

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.

Two scope notes.

1. The PR description mentions retrying the Zenoh drain-channel enqueue instead of dropping inter-daemon control events, but that change isn't in this diff — the only functional change here is the per-node finish-straggler fix in binaries/daemon/src/running_dataflow.rs. That retry (enqueue_zenoh_outbound_reliably) is in Fix/reliable inter daemon output close #3071, so the description looks stale.

2. The diff against main also bundles the entire ~385-line descriptor field classifier — libraries/core/src/descriptor/classify.rs plus the resolve_aliases_and_set_defaults_in_topology rewrite and removal of node_kind_mut. That's the whole content of feat(core)：add node field whitelist classifier to prevent silent field drops #3070 (and also appears in Fix/reliable inter daemon output close #3071), so the same refactor is duplicated across three PRs. It's orthogonal to the straggler fix and would be easier to review/revert on its own; it also changes descriptor acceptance (hard-erroring on top-level fields that resolution previously ignored), which deserves its own tests and a changelog note.

The straggler-watchdog change itself looks correct: narrowing the global open_external_mappings bail-out to a per-node remote_output_open skip keeps unrelated local stragglers watchdog-covered, and the two added tests (remote_output_open_skips_only_that_node, unrelated_remote_output_does_not_disable_local_straggler_watchdog) exercise that meaningfully.

---

Generated by Claude Code

[SunSunSun689]

Updated this PR to narrow the scope to the per-node finish-straggler watchdog
fix only.

What changed:

• rebased/rebuilt the branch on latest origin/main
• removed the descriptor field classifier changes from this PR; that work
  remains separate in feat(core)：add node field whitelist classifier to prevent silent field drops #3070
• kept only fix(daemon): keep local straggler watchdog armed

Current diff is now limited to:

• binaries/daemon/src/running_dataflow.rs

The branch was force-pushed from d8cc0e4d0 to b79488475 using --force- with-lease.

[phil-opp]

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

Re-reviewed after the Aug 13 force-push that narrowed this PR to just the per-node finish-straggler watchdog fix in binaries/daemon/src/running_dataflow.rs (the ~385-line descriptor classifier that earlier reviews flagged as bundled from #3070 is now gone, so that scope concern is resolved).

No issues found in the current diff. The change is a correct narrowing of the previous behavior: finish_stragglers no longer bails out entirely whenever any open_external_mappings entry exists (return Vec::new() on !open_external_mappings.is_empty()), which disabled the finish-escalation watchdog for every unrelated local node. It now keeps only the self.stop_sent global guard and exempts each producing node individually via remote_output_open, computed as open_external_mappings.iter().any(|output| &output.0 == id) — i.e. matched against the producing node id (OutputId.0), so only the node actually flushing to a remote consumer is skipped (select_finish_stragglers does if node.remote_output_open { continue; }) while other local stragglers stay covered.

The two added tests are meaningful: remote_output_open_skips_only_that_node checks the pure selector skips just the remote producer, and unrelated_remote_output_does_not_disable_local_straggler_watchdog exercises the full finish_stragglers path to confirm an unrelated remote mapping no longer suppresses escalation of a stuck local node.

---

Generated by Claude Code

[phil-opp]

🤖 This is a fully automated review by Claude. No human has verified these findings.

The per-node narrowing looks right, but I think the ordering of the new skip in select_finish_stragglers (binaries/daemon/src/running_dataflow.rs) reintroduces a premature-close in the distributed case this PR targets.

The new if node.remote_output_open { continue; } runs before the never_finishes global veto (if node.never_finishes { return Vec::new(); }). never_finishes is true for sources and timer/logs-fed nodes, and it's a global veto meaning "the dataflow is still producing, escalate nobody." That veto is what protects quiet local consumers of a live source: a pure sink never refreshes last_activity (only node→daemon DaemonRequests do, in node_communication/mod.rs; delivery to the node is daemon→node), and health_check_timeout is opt-in, so a live-but-quiet sink reads as "silent past grace."

Consider a distributed dataflow where source S on this daemon feeds both a remote consumer on another daemon and a local sink B. open_external_mappings then contains an OutputId(S, …) (an output with both a local and remote consumer lands in both mappings and open_external_mappings), so remote_output_open is true for S. S is now continued before it can veto, and B — never drained (S still alive), connected, silent past grace — becomes eligible and gets escalated/SIGKILLed while S is still feeding it. On main this couldn't happen, because finish_stragglers bailed entirely whenever any external mapping existed.

Suggest evaluating never_finishes before the remote_output_open skip so a live source/timer/logs node still vetoes globally. The intended fix is unaffected: a drained intermediate producer still flushing to a remote consumer has never_finishes == false, so it's still skipped individually while unrelated local stragglers stay covered. It'd be worth adding a test for a never_finishes node that also has remote_output_open — the current tests only cover drained() producers (never_finishes == false), so this ordering is untested.

---

Generated by Claude Code

[SunSunSun689]

Updated the PR to address the latest review feedback.

The ordering in select_finish_stragglers now evaluates the never_finishes
global veto before the per-node remote_output_open skip. This preserves the
intended behavior for live sources / timer-fed / logs-fed nodes: if such a
node is still running, it still prevents finish-straggler escalation for the
whole dataflow, even when it also has an open remote output.

I also added a regression test:

• remote_output_open_never_finishing_node_still_vetoes_globally

This covers the case where a source-like node has both:

• never_finishes == true
• remote_output_open == true

and verifies that an otherwise eligible local straggler is not escalated.

[phil-opp]

The Aug 14 commit (97bd9349) fixes the ordering issue from the last review. In select_finish_stragglers (binaries/daemon/src/running_dataflow.rs) the never_finishes global veto (return Vec::new()) is now evaluated before the if node.remote_output_open { continue; } skip, so a live source/timer/log-fed node that also has an open remote output still vetoes finish-escalation for the whole dataflow — instead of being skipped and letting a quiet local consumer get SIGKILLed. The new remote_output_open_never_finishing_node_still_vetoes_globally test locks this in: it pairs a never_finishes: true, remote_output_open: true source with an otherwise-eligible drained sink and asserts nothing is selected (fails under the old ordering, passes now).

No further issues found in the latest diff.

---

🤖 This is a fully automated review by Claude (Claude Code) — no human reviewed it before posting. Treat it as advisory.

Generated by Claude Code

---

Generated by Claude Code

[phil-opp]

🤖 Automated review by Claude — fully automated, no human has vetted this. Treat it as advisory and verify before acting.

The Aug-14 fix correctly moved the never_finishes veto ahead of the new skip, but I think there's a second, distinct veto-bypass that the same continue still causes — and it hits the exact distributed case this PR targets.

if node.remote_output_open { continue; } in select_finish_stragglers (binaries/daemon/src/running_dataflow.rs) still runs before the None-branch veto, so an active or still-starting non-source producer with a cross-daemon consumer no longer vetoes.

never_finishes only covers sources / timer- / logs-fed nodes. The other half of the global veto is the drained_for: None arm:

None => {
    if node.connected && node.silent_for >= effective_grace {
        eligible.push(node.id.clone());
    } else {
        return Vec::new(); // unconnected (still starting) OR connected-but-active => veto
    }
}

Because remote_output_open continues before this match, a producer that lands in the None branch never reaches that return Vec::new(). open_external_mappings holds outputs produced by a local node and consumed by a node on another daemon, so remote_output_open is true for exactly such a local producer.

Failure scenario (the distributed case this PR targets):

• Daemon B runs producer R (fed by a remote source S on another daemon) and a local sink T that consumes R. R also feeds a remote consumer, so OutputId(R, …) is in open_external_mappings → remote_output_open == true for R.
• R is actively producing: last_activity is fresh (silent_for < grace) and its inputs are still open (S alive) → drained_for == None, so R lands in the None branch. never_finishes == false (it has inputs), so the earlier veto doesn't apply.
• T is a quiet sink: delivery to a node doesn't stamp last_activity (only node→daemon requests do), and health_check_timeout is opt-in, so T reads as connected + silent past grace → eligible.
• Iteration: R hits continue instead of the None-branch return Vec::new(), so it no longer vetoes; T is then selected and escalated/SIGKILLed while R is still actively feeding it.

On main this couldn't happen, because finish_stragglers bailed entirely whenever any open_external_mappings entry existed. The same applies to a producer that hasn't connected yet (still starting up): it now continues instead of vetoing via the !connected path.

Suggested shape: keep remote_output_open guarding only escalation eligibility (the two eligible.push(...) sites), not the veto returns — i.e. still run the None-branch veto for !connected / silent_for < grace, and only skip the push when connected && silent_for >= effective_grace && remote_output_open.

On tests: the added cases only cover drained() producers (drained_for: Some) plus the one never_finishes case, so a remote_output_open node in the None branch (active or unconnected) is untested and wouldn't catch this.

---

Generated by Claude Code

[phil-opp]

Automated review by Claude (fully automated; no human in the loop)

Re-checked after the main merge (30013f06) that updated this branch. That merge doesn't touch select_finish_stragglers, so the correctness issue from the earlier review is still open in the current diff — flagging it again since there's been activity on the PR:

if node.remote_output_open { continue; } still runs before the match node.drained_for block, so it also skips the None-branch veto (else { return Vec::new(); }), not just escalation eligibility. A non-source producer with a cross-daemon consumer that is still active (drained_for == None, silent_for < grace) or still starting up (!connected) lands in that None arm — on main it vetoes escalation for the whole dataflow; here it continues and no longer vetoes. A quiet local sink of that producer (delivery doesn't refresh last_activity, and health_check_timeout is opt-in) then reads as silent-past-grace and can be escalated/SIGKILLed while the producer is still feeding it — the same distributed premature-close this PR targets, and one that couldn't happen on main (which bailed out on any open external mapping).

Suggested shape: gate only the two eligible.push(...) sites on remote_output_open, and still run the None-branch veto for the !connected / silent_for < grace cases. The added tests only cover drained() producers (drained_for: Some) plus one never_finishes case, so the None-branch producer path is untested.

---

Generated by Claude Code

[SunSunSun689]

Addressed the latest review by keeping remote_output_open from bypassing the veto paths in select_finish_stragglers.

Behavior now is:

• never_finishes still vetoes the whole dataflow even if the node has a remote output open
• active or still-starting non-source producers with a remote consumer still hit the None-branch veto
• only the final escalation selection is skipped for the remote-output node itself

Added regressions for:

• source + remote output still vetoes globally
• active producer + remote output still vetoes globally
• still-starting producer + remote output still vetoes globally

Validation:

• cargo fmt --all -- --check
• cargo test -p dora-daemon remote_output_open_ -- --nocapture

[phil-opp]

Re-checked after b69033cd (Aug 20). This resolves the veto-bypass flagged in the previous two reviews: in select_finish_stragglers, remote_output_open now guards only the two eligible.push(...) sites, not the veto returns. Both veto paths now fire unconditionally for a producer with an open remote output:

• never_finishes → early return Vec::new() (live source / timer- or log-fed node);
• the None arm's else { return Vec::new(); } → still fires for an active (silent_for < grace) or still-starting (!connected) producer.

A drained producer still flushing to a remote consumer is skipped from escalation without vetoing, so unrelated local stragglers stay watchdog-covered — the intended fix. The two added None-branch tests (remote_output_open_does_not_bypass_active_node_veto via never_drained, and ..._starting_node_veto with connected: false) cover the producer paths that were previously untested.

No further issues found in the current diff.

---

🤖 Fully automated review by Claude (Claude Code) — no human in the loop. Advisory only; please verify before acting.

---

Generated by Claude Code

[SunSunSun689]

Resolved the 4 conflicts and rebased the branch onto current main.

I also fixed the schema drift introduced by the bad conflict resolution:

• restored the descriptor::normalize_path export for manifest injection
• removed stale custom / communication references from the descriptor code
• updated the daemon test helper for the current RunningNode shape

Verified with:

• cargo fmt --all -- --check
• cargo test -p dora-core check_module_file_accepts_operator_produced_outputs -- --nocapture
• cargo test -p dora-daemon operator_shaped_node_with_data_input_is_not_a_source -- --nocapture
• cargo test -p dora-daemon opt_in_keeps_the_straggler_watchdog_armed_for_drainable_nodes -- --nocapture

[phil-opp]

Re-review after the latest push (ce60fa5, "align rebase with current schema")

The straggler-watchdog change in binaries/daemon/src/running_dataflow.rs still looks correct — remote_output_open guards only the two eligible.push(...) sites and both veto returns fire unconditionally, matching the last review.

But the latest rebase commit has re-bundled the descriptor-classifier changes that earlier reviews had gotten removed. classify.rs, expand.rs, and mod.rs are back in the diff, and against current main they read as a revert of already-merged work (#3070 plus a merge_env fix) rather than as new changes. Two concrete problems:

1. Breaks the test suite. classify.rs drops output_metadata and pattern from ROS2_ALLOWED (reverting feat(core)：add node field whitelist classifier to prevent silent field drops #3070). libraries/core/tests/descriptor-validation.rs::valid_descriptor_field_cases_resolve asserts that cases/valid-ros2-metadata.yml — which sets both output_metadata and pattern on a ros2: node — resolves successfully. With this diff that descriptor is now rejected, so the test fails. This PR doesn't touch the test or the case file, so the failure is live.

2. Reintroduces a fixed bug and deletes its guards. mod.rs reverts merge_env to the "only normalize when both inputs empty" form and deletes the two regression tests merge_env_normalizes_empty_node_map_to_none and merge_env_normalizes_empty_global_and_node_maps_to_none. That brings back merge_env(None, Some({})) → Some({}) (an empty env: {} on the resolved node) that main currently normalizes to None.

Neither change is mentioned in the PR description, and both are unrelated to the inter-daemon close / straggler fix. I'd rebase onto current main and drop the classify.rs/expand.rs/mod.rs hunks so the diff is scoped back to running_dataflow.rs.

---

🤖 This is a fully automated review by Claude. It may contain mistakes — please verify before acting on it.

---

Generated by Claude Code

[SunSunSun689]

I removed the re-bundled descriptor-classifier changes from this branch and restored those files to current main.

The PR diff is now scoped back to binaries/daemon/src/running_dataflow.rs only, so the watchdog fix is isolated again.

Validated with cargo fmt --all -- --check and cargo test -p dora-daemon remote_output_open_ -- --nocapture.

[phil-opp]

🤖 Automated review by Claude — generated fully automatically; no human has vetted it.

The latest commit (edc1152) drops the re-bundled #3070 classifier changes, so the PR is back to running_dataflow.rs scope as the earlier review asked. I re-checked the straggler-watchdog logic on the current head: the never_finishes veto and both return Vec::new() veto paths fire unconditionally, and remote_output_open guards only the two eligible.push(...) sites — so a drained producer still flushing to a remote consumer is exempted from escalation without disabling the watchdog for unrelated local stragglers. No new issues found; the latest commit looks safe to merge.

---

Generated by Claude Code