From 7e4df02bd60c4d6ee8c92d3dfd19f4e587ff9632 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 19 Jun 2026 11:57:05 +0200 Subject: [PATCH] fix: trigger flow error handler on unrecoverable (OOM/zombie) step failures (#9662) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: trigger flow error handler on unrecoverable (OOM/zombie) step failures When a worker is OOM-killed mid-step, the zombie job handler fails the step via handle_job_error with unrecoverable=true. update_flow_status_after_job_completion had `false if unrecoverable => false`, which silently completed the flow with the error and skipped the flow's failure module (error handler). It would also have pinned the failure module to the dead worker via same_worker. Unrecoverable failures now route to the failure module instead of being retried or silently dropped, and the error-handler step is pushed as a regular queued job that any live worker can pick up. Fixes WIN-2070 Co-Authored-By: Claude Opus 4.8 (1M context) * fix: skip retry on unrecoverable flow failures, add retry-skip regression test Address review: the failure-module-on-unrecoverable change must also bypass the per-step retry policy in push_next_flow_job, otherwise an OOM/zombie-killed step with a retry config would be retried instead of routing to the error handler. Gate the retry evaluation on !unrecoverable and add a regression test. Co-Authored-By: Claude Opus 4.8 (1M context) * chore: add sqlx offline cache for new flow-step zombie test query Co-Authored-By: Claude Opus 4.8 (1M context) * fix: route unrecoverable continue_on_error step failures to the error handler Addresses Codex/Pi review (P1): with continue_on_error on the failed step, the step counter was advanced before the unrecoverable decision branch, so push_next_flow_job pushed the next normal step instead of the failure module — hiding the worker death and letting the flow complete successfully. - Do not advance the step counter (inc) for an unrecoverable continue_on_error failure. - Let the Failure arm in push_next_flow_job route to the failure step even on a continue_on_error module when unrecoverable. - Add a regression test (a[continue_on_error] -> b + failure_module): asserts the failure module runs and step b does not. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- ...3db97dba77d649b637c2529e4f232dce16c88.json | 22 ++ backend/tests/worker.rs | 255 ++++++++++++++++++ backend/windmill-worker/src/worker.rs | 3 + backend/windmill-worker/src/worker_flow.rs | 51 +++- 4 files changed, 324 insertions(+), 7 deletions(-) create mode 100644 backend/.sqlx/query-fac563138c316998d4f523edd633db97dba77d649b637c2529e4f232dce16c88.json diff --git a/backend/.sqlx/query-fac563138c316998d4f523edd633db97dba77d649b637c2529e4f232dce16c88.json b/backend/.sqlx/query-fac563138c316998d4f523edd633db97dba77d649b637c2529e4f232dce16c88.json new file mode 100644 index 0000000000..9520c270dc --- /dev/null +++ b/backend/.sqlx/query-fac563138c316998d4f523edd633db97dba77d649b637c2529e4f232dce16c88.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT q.id FROM v2_job_queue q JOIN v2_job j USING (id)\n WHERE j.parent_job = $1 AND q.running = true", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false + ] + }, + "hash": "fac563138c316998d4f523edd633db97dba77d649b637c2529e4f232dce16c88" +} diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 9445e6cd6a..4ed12d3a4a 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -3990,6 +3990,261 @@ async fn test_failure_module(db: Pool) -> anyhow::Result<()> { Ok(()) } +/// Push `flow`, run it on a real worker until its first step is running, then simulate +/// `monitor::handle_zombie_jobs` reaping that step unrecoverably (its worker crashed/OOM'd) +/// by calling `handle_job_error(..., unrecoverable = true, ...)` exactly as the monitor does. +/// Returns the flow's completed result. +#[cfg(feature = "deno_core")] +async fn run_flow_until_step_running_then_fail_unrecoverably( + db: &Pool, + port: u16, + flow: FlowValue, +) -> serde_json::Value { + use std::sync::atomic::AtomicU16; + use std::sync::Arc; + use tokio::sync::mpsc; + use windmill_common::auth::create_token_for_owner; + use windmill_common::client::AuthedClient; + use windmill_common::KillpillSender; + use windmill_queue::{get_queued_job_v2, MiniCompletedJob, SameWorkerPayload}; + use windmill_worker::{JobCompletedSender, SameWorkerSender}; + + let flow_id = + RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None }) + .push(db) + .await; + + let db_ = db.clone(); + in_test_worker( + db, + async move { + let db = db_; + + // Wait for the first step to be running on the worker. + let step_job = loop { + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let running = sqlx::query_scalar!( + "SELECT q.id FROM v2_job_queue q JOIN v2_job j USING (id) + WHERE j.parent_job = $1 AND q.running = true", + flow_id + ) + .fetch_optional(&db) + .await + .unwrap(); + if let Some(step_id) = running { + if let Some(job) = get_queued_job_v2(&db, &step_id).await.unwrap() { + break job; + } + } + }; + + // The dummy `same_worker_tx` mirrors the monitor, which has no live worker channel. + let (sw_tx, _sw_rx) = mpsc::channel::(1); + let sw_tx = SameWorkerSender(sw_tx, Arc::new(AtomicU16::new(0))); + let (jc_tx, _jc_rx) = JobCompletedSender::new_never_used(); + let (_kp_tx, kp_rx) = KillpillSender::new(1); + let token = create_token_for_owner( + &db, + "test-workspace", + "u/test-user", + "", + 100, + "", + &Uuid::nil(), + None, + None, + ) + .await + .unwrap(); + let client = AuthedClient::new( + format!("http://localhost:{port}"), + "test-workspace".to_string(), + token, + None, + ); + + windmill_worker::result_processor::handle_job_error( + &db, + &client, + &MiniCompletedJob::from(step_job), + 0, + None, + windmill_common::error::Error::ExecutionErr( + "simulated worker OOM crash".to_string(), + ), + true, // unrecoverable + Some(&sw_tx), + "", + "test-monitor", + jc_tx, + &kp_rx, + ) + .await; + + // The flow should now run its failure module and complete. + loop { + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let done = sqlx::query_scalar!( + "SELECT EXISTS(SELECT 1 FROM v2_job_completed WHERE id = $1)", + flow_id + ) + .fetch_one(&db) + .await + .unwrap() + .unwrap_or(false); + if done { + break; + } + } + }, + port, + ) + .await; + + completed_job(flow_id, db).await.json_result().unwrap() +} + +/// The hanging-forever first step: only ever completed by the simulated zombie handler. +#[cfg(feature = "deno_core")] +fn hanging_step_value() -> serde_json::Value { + serde_json::json!({ + "input_transforms": {}, + "type": "rawscript", + "language": "deno", + "content": "export async function main() { await new Promise((r) => setTimeout(r, 600000)); }", + }) +} + +/// The error handler module that marks itself so tests can assert it ran. +#[cfg(feature = "deno_core")] +fn marker_failure_module() -> serde_json::Value { + serde_json::json!({ + "value": { + "input_transforms": { "error": { "type": "javascript", "expr": "previous_result", } }, + "type": "rawscript", + "language": "deno", + "content": "export function main(error) { return { handled_unrecoverable: true, error } }", + } + }) +} + +/// Regression test for WIN-2070: a flow step that fails *unrecoverably* — e.g. its worker was +/// OOM-killed and the failure is surfaced by the zombie job handler — must still trigger the +/// flow's error handler (failure module). Previously, `unrecoverable` failures silently +/// completed the flow with an error and skipped the failure module entirely. +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_failure_module_triggered_on_unrecoverable_failure( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let flow: FlowValue = serde_json::from_value(serde_json::json!({ + "modules": [{ "id": "a", "value": hanging_step_value() }], + "failure_module": marker_failure_module(), + })) + .unwrap(); + + let result = run_flow_until_step_running_then_fail_unrecoverably(&db, port, flow).await; + + server.close().await.unwrap(); + + assert_eq!( + result["handled_unrecoverable"], + json!(true), + "failure module (flow error handler) should run for an unrecoverable step failure, got: {result}" + ); + Ok(()) +} + +/// WIN-2070: an unrecoverable failure must NOT be retried even when the step has a retry +/// policy — the original worker is gone, so retrying is pointless. It should fall straight +/// through to the error handler (failure module) instead. +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_unrecoverable_failure_skips_retry_runs_failure_module( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let flow: FlowValue = serde_json::from_value(serde_json::json!({ + "modules": [{ + "id": "a", + "value": hanging_step_value(), + "retry": { "constant": { "attempts": 5, "seconds": 0 } }, + }], + "failure_module": marker_failure_module(), + })) + .unwrap(); + + let result = run_flow_until_step_running_then_fail_unrecoverably(&db, port, flow).await; + + server.close().await.unwrap(); + + assert_eq!( + result["handled_unrecoverable"], + json!(true), + "unrecoverable failure should skip retry and run the failure module, got: {result}" + ); + Ok(()) +} + +/// WIN-2070: an unrecoverable failure on a `continue_on_error` step must still route to the +/// error handler rather than silently advancing to the next step (which would hide the worker +/// death). A normal failure on a `continue_on_error` step is tolerated and the flow continues; +/// a worker crash/OOM is not. +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_unrecoverable_failure_on_continue_on_error_runs_failure_module( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // Step 'a' hangs (and tolerates failures via continue_on_error); step 'b' would run next + // if the unrecoverable failure were (wrongly) tolerated. + let flow: FlowValue = serde_json::from_value(serde_json::json!({ + "modules": [ + { + "id": "a", + "value": hanging_step_value(), + "continue_on_error": true, + }, + { + "id": "b", + "value": { + "input_transforms": {}, + "type": "rawscript", + "language": "deno", + "content": "export function main() { return { ran_b: true }; }", + }, + }, + ], + "failure_module": marker_failure_module(), + })) + .unwrap(); + + let result = run_flow_until_step_running_then_fail_unrecoverably(&db, port, flow).await; + + server.close().await.unwrap(); + + assert_eq!( + result["handled_unrecoverable"], + json!(true), + "unrecoverable failure on a continue_on_error step should run the failure module, got: {result}" + ); + assert!( + result.get("ran_b").is_none(), + "the step after a continue_on_error step must NOT run on an unrecoverable failure, got: {result}" + ); + Ok(()) +} + #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_run_wait_result_early_return_with_failure_module( diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 984acbe756..1ca77310ff 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -3808,6 +3808,9 @@ pub async fn handle_queued_job( worker_name, flow_runners, &killpill_rx, + // A freshly pulled flow job is being executed by a live worker; the prior + // step (if any) completed normally, so this is never unrecoverable here. + false, )) .warn_after_seconds(10) .await diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 86263350c3..7f90f9240e 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1279,7 +1279,11 @@ pub async fn update_flow_status_after_job_completion_internal( }), ) } else { - let inc = if continue_on_error { + // An unrecoverable failure (worker crash/OOM) must reach the error handler + // even on a continue_on_error step, so don't advance the step counter past + // the failed module — otherwise the flow would silently continue to the next + // step and hide the worker death. + let inc = if !unrecoverable && continue_on_error { let retry = current_module .as_ref() .and_then(|x| x.retry.clone()) @@ -1708,7 +1712,16 @@ pub async fn update_flow_status_after_job_completion_internal( _ if stop_early => stop_early_err_msg.is_some() && flow_value.failure_module.is_some(), // if stop_early_err_msg some, we want to trigger the error handler before stopping the flow, if any _ if flow_job.is_canceled() => false, true => !is_last_step, - false if unrecoverable => false, + // An unrecoverable failure (a step killed by a worker crash/OOM and surfaced by + // the zombie handler, or an error raised while updating the flow status itself) + // must not be retried or silently skipped, but it should still trigger the flow's + // error handler: an OOM/worker death is precisely when the error handler is expected + // to run. Continue the flow only to reach the failure module, never to retry. + false if unrecoverable => { + !is_failure_step + && !has_triggered_error_handler + && flow_value.failure_module.is_some() + } false if skip_seq_branch_failure || skip_loop_failures || continue_on_error => { !is_last_step } @@ -2059,6 +2072,7 @@ pub async fn update_flow_status_after_job_completion_internal( worker_name, flow_runners, &killpill_rx, + unrecoverable, )) .warn_after_seconds(10) .await @@ -2749,6 +2763,10 @@ pub async fn handle_flow( worker_name: &str, flow_runners: Option>, killpill_rx: &tokio::sync::broadcast::Receiver<()>, + // The previous step failed unrecoverably (e.g. a worker crash/OOM surfaced by the + // zombie handler). The next pushed step can only be the error handler (failure + // module), and it must not be pinned to the dead worker via same_worker. + unrecoverable: bool, ) -> anyhow::Result<()> { let flow = flow_data.value(); @@ -2922,6 +2940,7 @@ pub async fn handle_flow( flow_runners.clone(), job_completed_tx.clone(), &killpill_rx, + unrecoverable, )) .warn_after_seconds(10) .await?; @@ -3104,6 +3123,10 @@ async fn push_next_flow_job( flow_runners: Option>, job_completed_tx: JobCompletedSender, killpill_rx: &tokio::sync::broadcast::Receiver<()>, + // The prior step failed unrecoverably (worker crash/OOM). The only step pushed + // from here is the error handler, which must run on a live worker rather than + // being pinned to the dead one via same_worker / dedicated runners. + unrecoverable: bool, ) -> error::Result { let job_root = flow_job .flow_innermost_root_job @@ -3657,7 +3680,10 @@ async fn push_next_flow_job( } }; - let retry = if matches!(&status_module, FlowStatusModule::Failure { .. },) { + // An unrecoverable failure (worker crash/OOM) must not be retried — the original worker + // and its state are gone — so skip retry evaluation and fall straight through to the + // failure module below. + let retry = if !unrecoverable && matches!(&status_module, FlowStatusModule::Failure { .. },) { let retry = &module.retry.clone().unwrap_or_default(); evaluate_retry( retry, @@ -3672,8 +3698,13 @@ async fn push_next_flow_job( None }; let get_args_from_id = match &status_module { + // `|| unrecoverable`: a worker crash/OOM routes to the failure module even on a + // continue_on_error step (whose failures are normally tolerated), matching the + // `unrecoverable` decision in update_flow_status_after_job_completion_internal. FlowStatusModule::Failure { job, .. } - if retry.as_ref().is_some() || !module.continue_on_error.is_some_and(|x| x) => + if retry.as_ref().is_some() + || !module.continue_on_error.is_some_and(|x| x) + || unrecoverable => { if let Some((fail_count, retry_in)) = retry { tracing::debug!( @@ -4022,7 +4053,8 @@ async fn push_next_flow_job( .as_ref() .is_some_and(|fr| fr.job_id == flow_job.id); - let continue_with_runners = (start_runners || (flow_runners.is_some() && !do_not_pass_runners)) + let continue_with_runners = !unrecoverable + && (start_runners || (flow_runners.is_some() && !do_not_pass_runners)) && module.suspend.is_none() && module.sleep.is_none(); @@ -4031,8 +4063,13 @@ async fn push_next_flow_job( let job_same_worker = flow_job.same_worker && matches!(flow_job.kind, JobKind::Flow) && flow_job.runnable_id.is_some(); - let continue_on_same_worker = - (flow.same_worker || job_same_worker) && module.suspend.is_none() && module.sleep.is_none(); + // After an unrecoverable failure the original worker is gone, so the error handler + // step is pushed as a regular queued job (any live worker can pick it up) instead of + // being signaled to the dead worker via same_worker — which would strand it forever. + let continue_on_same_worker = !unrecoverable + && (flow.same_worker || job_same_worker) + && module.suspend.is_none() + && module.sleep.is_none(); /* Finally, push the job into the queue */ let mut uuids = vec![];