fix: trigger flow error handler on unrecoverable (OOM/zombie) step failures (#9662)

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* chore: add sqlx offline cache for new flow-step zombie test query

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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"
+}

backend/tests/worker.rs

@@ -3990,6 +3990,261 @@ async fn test_failure_module(db: Pool<Postgres>) -> 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<Postgres>,
+    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::<SameWorkerPayload>(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<Postgres>,
+) -> 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<Postgres>,
+) -> 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<Postgres>,
+) -> 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(

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

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<Arc<FlowRunners>>,
     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<Arc<FlowRunners>>,
     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<PushNextFlowJob> {
     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![];