From 25d92bb8ddb35f3f897e8b6a5c8cc2fd374e78cc Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 23:20:12 +0000 Subject: [PATCH 1/3] fix(coordinator): make multi-daemon stop best-effort instead of aborting on first failure stop_dataflow iterated over dataflow.daemons and used `?` on each send/receive, so the first daemon that errored (e.g. a disconnected daemon whose connection lookup returns None) returned early from the whole function. Daemons later in iteration never received StopDataflow, leaving their nodes running as orphans while the CLI only saw the first error. Attempt the stop on every daemon, aggregate per-daemon failures, and return a combined error only after all have been attempted. This mirrors the established best-effort pattern in run::rollback_spawned_daemons. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CCyEn8YqaYQYwh3RLjEZ7B --- binaries/coordinator/src/handlers.rs | 55 +++++++++++++++++++++------- 1 file changed, 41 insertions(+), 14 deletions(-) diff --git a/binaries/coordinator/src/handlers.rs b/binaries/coordinator/src/handlers.rs index 8d6c359c30..e953439cb7 100644 --- a/binaries/coordinator/src/handlers.rs +++ b/binaries/coordinator/src/handlers.rs @@ -243,25 +243,52 @@ pub(crate) async fn stop_dataflow<'a>( timestamp, })?; + // Best-effort: attempt the stop on every daemon even if some fail, so one + // unreachable/erroring daemon does not orphan the nodes running on the + // remaining healthy daemons. Errors are aggregated and reported after all + // daemons have been attempted (mirrors `run::rollback_spawned_daemons`). + let mut errors: Vec<(DaemonId, eyre::Report)> = Vec::new(); for daemon_id in &dataflow.daemons { - let daemon_connection = daemon_connections - .get_mut(daemon_id) - .wrap_err("no daemon connection")?; + let result: eyre::Result<()> = async { + let daemon_connection = daemon_connections + .get_mut(daemon_id) + .wrap_err("no daemon connection")?; + + let reply_raw = daemon_connection + .send_and_receive(&message) + .await + .wrap_err("failed to send/receive stop message")?; + match serde_json::from_slice(&reply_raw) + .wrap_err("failed to deserialize stop reply from daemon")? + { + DaemonCoordinatorReply::StopResult(result) => result + .map_err(|e| eyre!(e)) + .wrap_err("failed to stop dataflow")?, + other => bail!("unexpected reply after sending stop: {other:?}"), + } + Ok(()) + } + .await; - let reply_raw = daemon_connection - .send_and_receive(&message) - .await - .wrap_err("failed to send/receive stop message")?; - match serde_json::from_slice(&reply_raw) - .wrap_err("failed to deserialize stop reply from daemon")? - { - DaemonCoordinatorReply::StopResult(result) => result - .map_err(|e| eyre!(e)) - .wrap_err("failed to stop dataflow")?, - other => bail!("unexpected reply after sending stop: {other:?}"), + if let Err(err) = result { + errors.push((daemon_id.clone(), err)); } } + if let Some((_, first)) = errors.first() { + let daemon_list = errors + .iter() + .map(|(id, err)| format!("{id}: {err:#}")) + .collect::>() + .join("; "); + return Err(eyre!( + "failed to stop dataflow `{dataflow_uuid}` on {} of {} daemon(s): {daemon_list}", + errors.len(), + dataflow.daemons.len(), + ) + .wrap_err(format!("first failure: {first:#}"))); + } + tracing::info!("successfully send stop dataflow `{dataflow_uuid}` to all daemons"); Ok(dataflow) From 4bf6b443149e4725419cf234c121ca1d618d3bff Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 23:41:44 +0000 Subject: [PATCH 2/3] refactor(coordinator): drop redundant first-failure wrap in stop error Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CCyEn8YqaYQYwh3RLjEZ7B --- binaries/coordinator/src/handlers.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/binaries/coordinator/src/handlers.rs b/binaries/coordinator/src/handlers.rs index e953439cb7..7659d057d1 100644 --- a/binaries/coordinator/src/handlers.rs +++ b/binaries/coordinator/src/handlers.rs @@ -275,7 +275,7 @@ pub(crate) async fn stop_dataflow<'a>( } } - if let Some((_, first)) = errors.first() { + if !errors.is_empty() { let daemon_list = errors .iter() .map(|(id, err)| format!("{id}: {err:#}")) @@ -285,8 +285,7 @@ pub(crate) async fn stop_dataflow<'a>( "failed to stop dataflow `{dataflow_uuid}` on {} of {} daemon(s): {daemon_list}", errors.len(), dataflow.daemons.len(), - ) - .wrap_err(format!("first failure: {first:#}"))); + )); } tracing::info!("successfully send stop dataflow `{dataflow_uuid}` to all daemons"); From dd5fb6cf8caf086574c27f079f52c9264b369f3c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 13:06:46 +0000 Subject: [PATCH 3/3] =?UTF-8?q?test(coordinator):=20pin=20best-effort=20st?= =?UTF-8?q?op=20=E2=80=94=20a=20failed=20daemon=20still=20stops=20the=20re?= =?UTF-8?q?st?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a RED-capable regression test for stop_dataflow: two daemons where the one iterated first (a BTreeSet ordered by machine id) is unreachable. Under the old first-failure `?` code the healthy daemon was never reached and its nodes were orphaned; the test asserts the healthy daemon still receives a StopDataflow and that the error aggregates the per-daemon outcome. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CCyEn8YqaYQYwh3RLjEZ7B --- binaries/coordinator/src/handlers.rs | 156 +++++++++++++++++++++++++++ 1 file changed, 156 insertions(+) diff --git a/binaries/coordinator/src/handlers.rs b/binaries/coordinator/src/handlers.rs index 7659d057d1..8a2fcd60c5 100644 --- a/binaries/coordinator/src/handlers.rs +++ b/binaries/coordinator/src/handlers.rs @@ -900,4 +900,160 @@ mod tests { "subscriber should be evicted after 100 consecutive timeouts" ); } + + /// Build a minimal `RunningDataflow` running on exactly `daemons`. + fn dataflow_on(uuid: Uuid, daemons: impl IntoIterator) -> RunningDataflow { + let descriptor: Descriptor = serde_json::from_value(serde_json::json!({ + "nodes": [{ "id": "sender", "outputs": ["message"] }] + })) + .expect("valid test descriptor"); + + RunningDataflow { + name: None, + uuid, + descriptor, + daemons: daemons.into_iter().collect(), + pending_daemons: BTreeSet::new(), + exited_before_subscribe: vec![], + nodes: BTreeMap::new(), + node_to_daemon: BTreeMap::new(), + node_metrics: BTreeMap::new(), + node_finalized: BTreeSet::new(), + node_stopped_at: BTreeMap::new(), + network_metrics: None, + ready_barrier_released: false, + spawn_result: CachedResult::default(), + stop_reply_senders: vec![], + buffered_log_messages: vec![], + log_subscribers: vec![], + topic_subscribers: BTreeMap::new(), + pending_spawn_results: BTreeSet::new(), + spawn_started_at: std::time::Instant::now(), + created_at: 0, + store_generation: 0, + last_recovery_attempt: BTreeMap::new(), + last_replay_attempt: BTreeMap::new(), + uv: false, + state_log_sequence: 0, + state_log: Vec::new(), + daemon_ack_sequence: BTreeMap::new(), + } + } + + /// A connection emulating the ws_daemon handler task for a *healthy* daemon: + /// it records each request it receives and completes the matching + /// `pending_replies` oneshot with a successful `StopResult`. Returns the + /// connection plus the shared log of requests the daemon actually received. + fn healthy_daemon() -> ( + DaemonConnection, + std::sync::Arc>>, + ) { + let received = std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new())); + let pending: std::sync::Arc< + tokio::sync::Mutex>>, + > = std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())); + let (tx, mut rx) = tokio::sync::mpsc::channel::(8); + let conn = DaemonConnection::new(tx, pending.clone(), BTreeMap::new()); + + let received_task = received.clone(); + tokio::spawn(async move { + while let Some(msg) = rx.recv().await { + let value: serde_json::Value = + serde_json::from_str(&msg).expect("outgoing request is JSON"); + let id: Uuid = value["id"] + .as_str() + .expect("request carries an id") + .parse() + .expect("id is a uuid"); + // Record receipt *before* replying, so a returned `Ok` from + // `send_and_receive` guarantees the request is already logged. + received_task.lock().await.push(msg.clone()); + let reply = serde_json::to_string(&DaemonCoordinatorReply::StopResult(Ok(()))) + .expect("serialize reply"); + if let Some(sender) = pending.lock().await.remove(&id) { + let _ = sender.send(reply); + } + } + }); + + (conn, received) + } + + /// A connection whose receiver is dropped, so the very first `send` fails — + /// modelling an unreachable/disconnected daemon. + fn broken_daemon() -> DaemonConnection { + let (tx, rx) = tokio::sync::mpsc::channel::(1); + drop(rx); + DaemonConnection::new( + tx, + std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())), + BTreeMap::new(), + ) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stop_dataflow_is_best_effort_when_a_daemon_fails() { + // A dataflow on two daemons where the one iterated *first* is + // unreachable. `DaemonId` orders by machine id, so `broken` (b) sorts + // before `healthy` (h) in `dataflow.daemons` (a `BTreeSet`): under the + // old first-failure `?` code the healthy daemon would never be reached + // and its nodes would be orphaned. Best-effort must still stop it. + let dataflow_uuid = Uuid::new_v4(); + let broken = DaemonId::new(Some("broken".to_string())); + let healthy = DaemonId::new(Some("healthy".to_string())); + + let mut daemon_connections = DaemonConnections::default(); + daemon_connections.add(broken.clone(), broken_daemon()); + let (healthy_conn, healthy_received) = healthy_daemon(); + daemon_connections.add(healthy.clone(), healthy_conn); + + let mut running_dataflows = HashMap::new(); + running_dataflows.insert( + dataflow_uuid, + dataflow_on(dataflow_uuid, [broken.clone(), healthy.clone()]), + ); + + let timestamp = HLC::default().new_timestamp(); + // `stop_dataflow`'s `Ok` is `&mut RunningDataflow` (not `Debug`), so + // destructure by hand rather than via `expect_err`. + let err = match stop_dataflow( + &mut running_dataflows, + dataflow_uuid, + &mut daemon_connections, + timestamp, + None, + false, + ) + .await + { + Ok(_) => panic!("a failing daemon must surface an aggregated error"), + Err(err) => err, + }; + + // The core property: the healthy daemon was still told to stop even + // though the daemon iterated before it failed. + let received = healthy_received.lock().await; + assert_eq!( + received.len(), + 1, + "healthy daemon must receive exactly one stop request" + ); + assert!( + received[0].contains("StopDataflow"), + "healthy daemon must receive a StopDataflow, got: {}", + received[0] + ); + + // The error is aggregated: it reports one of two daemons failed and + // names the failing one (so the CLI still learns what broke). + let msg = format!("{err:#}"); + assert!( + msg.contains("1 of 2 daemon(s)"), + "error must aggregate the per-daemon outcome: {msg}" + ); + assert!( + msg.contains("broken"), + "error must name the failed daemon: {msg}" + ); + } }