diff --git a/binaries/coordinator/src/handlers.rs b/binaries/coordinator/src/handlers.rs index 8d6c359c30..8a2fcd60c5 100644 --- a/binaries/coordinator/src/handlers.rs +++ b/binaries/coordinator/src/handlers.rs @@ -243,25 +243,51 @@ 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 !errors.is_empty() { + 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(), + )); + } + tracing::info!("successfully send stop dataflow `{dataflow_uuid}` to all daemons"); Ok(dataflow) @@ -874,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}" + ); + } }