fix(daemon): increment pending counter when propagating NodeFailed downstream

[phil-opp]

Summary

Fixes #2827.

When an upstream node exits with a non-zero code, the daemon propagates a NodeEvent::NodeFailed to each downstream (subscriber) node. It enqueued that event into the subscriber's channel without the matching inc_pending() that every other delivery site performs. The per-node pending_messages counter (an Arc<AtomicU64>) is still decremented unconditionally by the Listener when it drains the event, so the counter ended up one lower than the true in-flight count — and if the subscriber's channel was already empty (counter == 0), the fetch_sub(1) wrapped to u64::MAX. The bogus value was then reported to the coordinator (metrics / dora node list) until the node re-subscribed.

Change

• Extract the NodeFailed propagation loop out of the SpawnedNodeResult handler in lib.rs into a new RunningDataflow::propagate_node_failed(failed_node, error_msg, clock) method.
• The method pairs each successful enqueue (send_with_timestamp(...).ok() == Some(true)) with inc_pending(&recv_id), mirroring every other delivery site in the daemon.
• Add a regression test (propagate_node_failed_keeps_idle_receiver_counter_consistent) that drives a source → sink mapping with an idle receiver, propagates a NodeFailed, then drains the event the way the Listener does (one unconditional fetch_sub) and asserts the counter returns to 0 rather than underflowing to u64::MAX.

Testing

• cargo test -p dora-daemon — all 113 tests pass, including the new one.
• cargo fmt -p dora-daemon -- --check and cargo clippy -p dora-daemon -- -D warnings are clean.

---

Generated by Claude Code

[trunk-io]

😎 Merged successfully - details.

[phil-opp]

I reviewed this change and found no issues.

I confirmed the bug and the fix against the surrounding code:

• send_with_timestamp returns Ok(true) when the event is actually delivered, Ok(false) when dropped (channel full), and Err when closed (event_types.rs). The new .ok() == Some(true) gate therefore increments the counter only when an event truly enters the channel.
• The node-facing listener decrements pending_messages unconditionally for every drained event — including NodeFailed (node_communication/mod.rs, all three drain sites). So an enqueue without the matching inc_pending does leave the count one low, and fetch_sub(1) on a 0 counter wraps to u64::MAX, exactly as described.
• The extracted propagate_node_failed is behavior-preserving relative to the old inline loop and mirrors the existing inc_pending pattern used at the NodeEvent::Stop delivery sites (running_dataflow.rs).

The regression test is meaningful: it drives the real propagation, asserts the counter reaches 1, then drains exactly as the listener does and asserts it returns to 0 rather than underflowing. Reads safe to merge.

---

🤖 This is a fully automated review by Claude. No human has verified these findings — please treat them as suggestions to check rather than confirmed defects.

---

Generated by Claude Code