diff --git a/backend/.sqlx/query-1bf87fe9667f860c6089bffb803291f4ff5c979be956369fa4406d789cca92fc.json b/backend/.sqlx/query-1bf87fe9667f860c6089bffb803291f4ff5c979be956369fa4406d789cca92fc.json new file mode 100644 index 0000000000..1ee774a993 --- /dev/null +++ b/backend/.sqlx/query-1bf87fe9667f860c6089bffb803291f4ff5c979be956369fa4406d789cca92fc.json @@ -0,0 +1,112 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n j.id AS \"id!\", j.workspace_id AS \"workspace_id!\", j.parent_job, j.flow_step_id IS NOT NULL AS \"is_flow_step?\",\n COALESCE(s.flow_status, s.workflow_as_code_status)::text AS \"flow_status: Box\", r.ping AS last_ping, j.same_worker AS \"same_worker?\",\n q.worker AS \"worker?\",\n wp.ping_at AS \"worker_last_ping?\",\n wp.memory_usage AS \"worker_memory_usage?\",\n wp.wm_memory_usage AS \"worker_wm_memory_usage?\",\n wp.memory AS \"worker_memory_total?\",\n wp.worker_group AS \"worker_group?\",\n wp.wm_version AS \"worker_version?\",\n wp.current_job_id AS \"worker_current_job_id?\",\n wp.worker_instance AS \"worker_instance?\"\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n LEFT JOIN worker_ping wp ON wp.worker = q.worker\n WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()\n AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode' OR j.kind = 'singlestepflow')\n AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval\n AND q.canceled_by IS NULL\n\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "workspace_id!", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 3, + "name": "is_flow_step?", + "type_info": "Bool" + }, + { + "ordinal": 4, + "name": "flow_status: Box", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "last_ping", + "type_info": "Timestamptz" + }, + { + "ordinal": 6, + "name": "same_worker?", + "type_info": "Bool" + }, + { + "ordinal": 7, + "name": "worker?", + "type_info": "Varchar" + }, + { + "ordinal": 8, + "name": "worker_last_ping?", + "type_info": "Timestamptz" + }, + { + "ordinal": 9, + "name": "worker_memory_usage?", + "type_info": "Int8" + }, + { + "ordinal": 10, + "name": "worker_wm_memory_usage?", + "type_info": "Int8" + }, + { + "ordinal": 11, + "name": "worker_memory_total?", + "type_info": "Int8" + }, + { + "ordinal": 12, + "name": "worker_group?", + "type_info": "Varchar" + }, + { + "ordinal": 13, + "name": "worker_version?", + "type_info": "Varchar" + }, + { + "ordinal": 14, + "name": "worker_current_job_id?", + "type_info": "Uuid" + }, + { + "ordinal": 15, + "name": "worker_instance?", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + true, + null, + null, + true, + false, + true, + false, + true, + true, + true, + false, + false, + true, + false + ] + }, + "hash": "1bf87fe9667f860c6089bffb803291f4ff5c979be956369fa4406d789cca92fc" +} diff --git a/backend/.sqlx/query-387e135bf363533089493c35eb39dd4864f4e0aa0093df47df7d444ee748dbb2.json b/backend/.sqlx/query-387e135bf363533089493c35eb39dd4864f4e0aa0093df47df7d444ee748dbb2.json new file mode 100644 index 0000000000..50f0bf86e5 --- /dev/null +++ b/backend/.sqlx/query-387e135bf363533089493c35eb39dd4864f4e0aa0093df47df7d444ee748dbb2.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT count(*) FROM v2_job_completed\n WHERE workspace_id = $1 AND id = ANY($2) AND status = 'success'", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "UuidArray" + ] + }, + "nullable": [ + null + ] + }, + "hash": "387e135bf363533089493c35eb39dd4864f4e0aa0093df47df7d444ee748dbb2" +} diff --git a/backend/.sqlx/query-4371e59b85cc279dba03800aacde8a4b96c9629cdc785edc7a389c3eb7269608.json b/backend/.sqlx/query-4371e59b85cc279dba03800aacde8a4b96c9629cdc785edc7a389c3eb7269608.json new file mode 100644 index 0000000000..4bf6f3692b --- /dev/null +++ b/backend/.sqlx/query-4371e59b85cc279dba03800aacde8a4b96c9629cdc785edc7a389c3eb7269608.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT (kind = 'flow' AND parent_job IS NULL) AS \"restartable!\"\n FROM v2_job WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "restartable!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "4371e59b85cc279dba03800aacde8a4b96c9629cdc785edc7a389c3eb7269608" +} diff --git a/backend/.sqlx/query-4a35f9a0201e283d854bd77f7f79d4c24699c2c7abad4086e0bb876c5c6b2c38.json b/backend/.sqlx/query-4a35f9a0201e283d854bd77f7f79d4c24699c2c7abad4086e0bb876c5c6b2c38.json new file mode 100644 index 0000000000..6c5b364a5b --- /dev/null +++ b/backend/.sqlx/query-4a35f9a0201e283d854bd77f7f79d4c24699c2c7abad4086e0bb876c5c6b2c38.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_completed SET status = 'canceled', canceled_by = 'monitor',\n flow_status = $2 WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "4a35f9a0201e283d854bd77f7f79d4c24699c2c7abad4086e0bb876c5c6b2c38" +} diff --git a/backend/.sqlx/query-60201b41a0bfea89a201a25ae2431fe94999ca4f11934fbab4638b2653a678dc.json b/backend/.sqlx/query-60201b41a0bfea89a201a25ae2431fe94999ca4f11934fbab4638b2653a678dc.json new file mode 100644 index 0000000000..9f4c0f9bf6 --- /dev/null +++ b/backend/.sqlx/query-60201b41a0bfea89a201a25ae2431fe94999ca4f11934fbab4638b2653a678dc.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_completed SET status = 'canceled', canceled_by = 'admin',\n flow_status = $2 WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "60201b41a0bfea89a201a25ae2431fe94999ca4f11934fbab4638b2653a678dc" +} diff --git a/backend/.sqlx/query-7d2923f940cf5a8cbbdc51de91c7e00942a8f4f5251502ef8efc0f510a857551.json b/backend/.sqlx/query-7d2923f940cf5a8cbbdc51de91c7e00942a8f4f5251502ef8efc0f510a857551.json new file mode 100644 index 0000000000..7967a56868 --- /dev/null +++ b/backend/.sqlx/query-7d2923f940cf5a8cbbdc51de91c7e00942a8f4f5251502ef8efc0f510a857551.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_completed\n SET status = 'canceled', canceled_by = 'monitor', canceled_reason = 'zombie flow',\n flow_status = $2\n WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "7d2923f940cf5a8cbbdc51de91c7e00942a8f4f5251502ef8efc0f510a857551" +} diff --git a/backend/.sqlx/query-fa1b9fcdd344fa2c88a187f4a558e375d93caf76c4cec362c2b92d2a7b8704a2.json b/backend/.sqlx/query-fa1b9fcdd344fa2c88a187f4a558e375d93caf76c4cec362c2b92d2a7b8704a2.json new file mode 100644 index 0000000000..0d9e4ac3ad --- /dev/null +++ b/backend/.sqlx/query-fa1b9fcdd344fa2c88a187f4a558e375d93caf76c4cec362c2b92d2a7b8704a2.json @@ -0,0 +1,83 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n j.runnable_path as script_path, j.runnable_id AS \"script_hash: ScriptHash\",\n j.kind AS \"job_kind!: JobKind\", c.canceled_by,\n COALESCE(c.flow_status, c.workflow_as_code_status) AS \"flow_status: Json>\",\n j.raw_flow AS \"raw_flow: Json>\"\n FROM v2_job_completed c JOIN v2_job j USING (id) WHERE j.id = $1 and j.workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "script_path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "script_hash: ScriptHash", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "job_kind!: JobKind", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlestepflow", + "flowscript", + "flownode", + "appscript", + "aiagent", + "unassigned_script", + "unassigned_flow", + "unassigned_singlestepflow" + ] + } + } + } + }, + { + "ordinal": 3, + "name": "canceled_by", + "type_info": "Varchar" + }, + { + "ordinal": 4, + "name": "flow_status: Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 5, + "name": "raw_flow: Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true, + true, + false, + true, + null, + true + ] + }, + "hash": "fa1b9fcdd344fa2c88a187f4a558e375d93caf76c4cec362c2b92d2a7b8704a2" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 4947b9b211..d1e63958e2 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -4946,11 +4946,13 @@ async fn find_zombie_flow_culprit_worker( } async fn handle_zombie_flows(db: &DB) -> error::Result<()> { + // flow_status is cast ::text on purpose: decoding the jsonb column directly as Box + // yields its binary form (leading version byte) and fails serde_json parsing at column 1. let flows = sqlx::query!( r#" SELECT j.id AS "id!", j.workspace_id AS "workspace_id!", j.parent_job, j.flow_step_id IS NOT NULL AS "is_flow_step?", - COALESCE(s.flow_status, s.workflow_as_code_status) AS "flow_status: Box", r.ping AS last_ping, j.same_worker AS "same_worker?", + COALESCE(s.flow_status, s.workflow_as_code_status)::text AS "flow_status: Box", r.ping AS last_ping, j.same_worker AS "same_worker?", q.worker AS "worker?", wp.ping_at AS "worker_last_ping?", wp.memory_usage AS "worker_memory_usage?", @@ -4979,11 +4981,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> { .as_deref() .and_then(|x| serde_json::from_str::(x).ok()); if !flow.same_worker.unwrap_or(false) - && status.is_some_and(|s| { - s.modules - .get(0) - .is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. })) - }) + && status.as_ref().is_some_and(|s| s.is_not_yet_started()) { let error_message = format!( "Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.", @@ -5294,6 +5292,18 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> { format!("Flow {id} ({base_url}/run/{id}?workspace={workspace_id}) was cancelled because it") } ); + let reason = match between_steps_recovery_guidance( + db, + status.as_ref(), + id, + &workspace_id, + &base_url, + ) + .await + { + Some(guidance) => format!("{reason}\n\n{guidance}"), + None => reason, + }; report_critical_error(reason.clone(), db.clone(), Some(&flow.workspace_id), None).await; cancel_zombie_flow_job(db, flow.id, &flow.workspace_id, format!(r#"{reason} @@ -5340,6 +5350,103 @@ Please check your worker logs for more details and feel free to report it to the Ok(()) } +/// When a between-steps zombie's stuck step has every child recorded as a +/// `success` completion, the flow's state is fully derivable: only the final +/// state transition was lost to the worker failure, not any real work. In that +/// case return concrete restart-from-step recovery guidance to append to the +/// cancellation reason / critical alert. Returns `None` when the state isn't +/// derivable (some child missing or not successful), so the existing wording is +/// left untouched. Auto-recovery is deliberately not attempted (a re-driven +/// transition can OOM again on the same aggregated state; a human raises the +/// memory limit first, then restarts). +async fn between_steps_recovery_guidance( + db: &DB, + status: Option<&FlowStatus>, + flow_id: Uuid, + workspace_id: &str, + base_url: &str, +) -> Option { + // The stuck module is the current step, left InProgress because the + // transition that would have marked it Success was dropped. It is only + // derivable when its own cursor reached the end (a serial fan-out reaped + // mid-iteration has unrun work left; while-loops are never derivable). Whether + // restart reuses the children or re-runs the step (final step, or one carrying a + // stop/skip/approval/sleep) is decided by the restart path against the flow + // definition, which the reaper doesn't load; the guidance states both outcomes + // rather than promising reuse the restart might decline. + let status = status?; + let idx = usize::try_from(status.step).ok()?; + let module = status.modules.get(idx)?; + if !module.is_between_steps_complete() { + return None; + } + let step_id = module.id(); + + // Only a top-level deployed flow exposes a working restart-from-step: the run page's + // "Re-start from" button is rendered only for job_kind == 'flow' (a flowpreview, even a + // pathful editor preview, or a singlestepflow does not qualify), and a subflow child + // restarts via its root. Match that surface exactly so the guidance never points at a + // button / endpoint that isn't there; leave the existing wording otherwise. + let restartable = sqlx::query_scalar!( + r#"SELECT (kind = 'flow' AND parent_job IS NULL) AS "restartable!" + FROM v2_job WHERE id = $1"#, + flow_id, + ) + .fetch_one(db) + .await + .ok()?; + if !restartable { + return None; + } + + // Children whose completion the lost transition would have aggregated: the + // loop/branchall iterations, or the single leaf/subflow child. + let child_ids: Vec = module + .flow_jobs() + .filter(|v| !v.is_empty()) + .or_else(|| module.job().map(|j| vec![j]))?; + + // Derivable only when every child is recorded as a success completion. + let success_children = sqlx::query_scalar!( + "SELECT count(*) FROM v2_job_completed + WHERE workspace_id = $1 AND id = ANY($2) AND status = 'success'", + workspace_id, + &child_ids, + ) + .fetch_one(db) + .await + .ok()? + .unwrap_or(0); + if success_children != child_ids.len() as i64 { + return None; + } + let n = child_ids.len(); + + // For loop/branchall, name the completed iteration/branch count so the + // operator can confirm the whole fan-out is intact. + let iteration_hint = match module { + FlowStatusModule::InProgress { iterator: Some(_), .. } => { + format!(" (loop step, all {n} iterations completed)") + } + FlowStatusModule::InProgress { branchall: Some(_), .. } => { + format!(" (branchall step, all {n} branches completed)") + } + _ => String::new(), + }; + + Some(format!( + "RECOVERY: all {n} child job(s) of step `{step_id}`{iteration_hint} completed successfully; \ +only the flow's final state transition was lost to the worker failure above (not any genuine failure), so \ +the completed work is intact. To recover: first change the failure condition (raise the worker memory limit, \ +e.g. k8s `resources.limits.memory`, or move the flow to a larger worker group), then restart from step \ +`{step_id}`. Restart replays only the dropped transition and reuses the completed children where the step's \ +result is fully derivable; a step that is the flow's last, or carries a stop/skip condition, an approval, or a \ +sleep, is re-run instead (re-evaluating those on the larger worker).\n\ + UI: open {base_url}/run/{flow_id}?workspace={workspace_id} and use \"Re-start from {step_id}\".\n\ + API: POST {base_url}/api/w/{workspace_id}/jobs/restart/f/{flow_id} with body {{\"step_id\":\"{step_id}\"}}." + )) +} + async fn cancel_zombie_flow_job( db: &Pool, id: Uuid, diff --git a/backend/tests/zombie_flow_recovery.rs b/backend/tests/zombie_flow_recovery.rs new file mode 100644 index 0000000000..fb658dfb03 --- /dev/null +++ b/backend/tests/zombie_flow_recovery.rs @@ -0,0 +1,661 @@ +//! Regression test for hand-recovery of between-steps zombie flows. +//! +//! When a worker is OOM-killed mid state-transition, the zombie monitor +//! (`handle_zombie_flows` → `cancel_job` with force) reaps the flow: it lands in +//! `v2_job_completed` as `canceled`, with its `flow_status` preserved: the step +//! whose transition was lost stays `InProgress` even though all its children +//! completed successfully. This test reproduces that exact terminal state and +//! asserts that a hand-restart from the stuck step reuses every completed child +//! (no re-run) and the flow reaches success. +//! +//! The reaper itself lives in the `windmill` binary crate and is unreachable +//! from an integration test, so we reproduce the state `cancel_job(force)` +//! leaves behind directly; the fix under test is the restart-resolution path, +//! not the detection query. + +#![cfg(feature = "deno_core")] + +use serde_json::json; +use sqlx::{Pool, Postgres}; +use windmill_common::flow_status::{BranchChosen, FlowStatus, RestartedFrom}; +use windmill_common::flows::FlowValue; +use windmill_common::jobs::JobPayload; +use windmill_test_utils::*; + +/// Child job UUID for a top-level step in a completed flow's `flow_status` +/// (optionally the iteration index for a ForLoop / BranchAll container). +async fn child_job_id_for_step( + db: &Pool, + flow_job_id: uuid::Uuid, + step_id: &str, + iter: Option, +) -> uuid::Uuid { + let raw: serde_json::Value = sqlx::query_scalar!( + "SELECT flow_status FROM v2_job_completed WHERE id = $1", + flow_job_id + ) + .fetch_one(db) + .await + .unwrap() + .expect("flow_status missing"); + let status: FlowStatus = serde_json::from_value(raw).expect("parse flow_status"); + let module = status + .modules + .iter() + .find(|m| m.id() == step_id) + .expect("step in flow_status"); + match iter { + Some(i) => module.flow_jobs().expect("flow_jobs")[i], + None => module.job().expect("job"), + } +} + +/// A between-steps zombie whose fan-out completed but whose final transition was +/// lost can be hand-restarted from the stuck step, reusing every completed child +/// (including the last iteration) and reaching success. +#[sqlx::test(fixtures("base", "hello"))] +async fn test_between_steps_zombie_restart_reuses_all_children( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // A fan-out ForLoop `fanout` (2 iterations) followed by `after`, which + // consumes the loop's aggregated result. In the zombie scenario `fanout` + // finished all iterations but its final transition was lost, so `after` + // never ran. + let flow_value: FlowValue = serde_json::from_value(json!({ + "modules": [ + { + "id": "fanout", + "value": { + "type": "forloopflow", + "iterator": { "type": "javascript", "expr": "['a', 'b']" }, + "skip_failures": false, + "parallel": false, + "modules": [{ + "id": "inner", + "value": { + "type": "rawscript", + "language": "deno", + "input_transforms": { + "v": { "type": "javascript", "expr": "flow_input.iter.value" } + }, + "content": "export function main(v: string) { return v }" + } + }] + } + }, + { + "id": "after", + "value": { + "type": "rawscript", + "language": "deno", + "input_transforms": { + "loop_res": { "type": "javascript", "expr": "results.fanout" } + }, + "content": "export function main(loop_res: string[]) { return loop_res.join(',') }" + } + } + ] + })) + .unwrap(); + + // Run to completion to obtain real, successful child jobs. + let full_run = RunJob::from(JobPayload::RawFlow { + value: flow_value.clone(), + path: None, + restarted_from: None, + }) + .run_until_complete(&db, false, port) + .await; + assert!(full_run.success, "baseline run should succeed"); + assert_eq!(full_run.json_result().unwrap(), json!("a,b")); + + let orig_iter0 = child_job_id_for_step(&db, full_run.id, "fanout", Some(0)).await; + let orig_iter1 = child_job_id_for_step(&db, full_run.id, "fanout", Some(1)).await; + let orig_after = child_job_id_for_step(&db, full_run.id, "after", None).await; + + // Reproduce the zombie-reaper's terminal state: cancelled by `monitor` with + // `flow_status` frozen mid-transition: `fanout` still `InProgress` (all + // iterations done), `after` never reached. + let mut flow_status: serde_json::Value = sqlx::query_scalar!( + "SELECT flow_status FROM v2_job_completed WHERE id = $1", + full_run.id + ) + .fetch_one(&db) + .await? + .expect("flow_status"); + for m in flow_status["modules"].as_array_mut().unwrap() { + match m["id"].as_str() { + Some("fanout") => { + m["type"] = json!("InProgress"); + // A reaped loop keeps its cursor at the last iteration. + m["iterator"] = json!({ "index": 1, "itered_len": 2 }); + } + Some("after") => *m = json!({ "type": "WaitingForPriorSteps", "id": "after" }), + _ => {} + } + } + flow_status["step"] = json!(0); + sqlx::query!( + "UPDATE v2_job_completed + SET status = 'canceled', canceled_by = 'monitor', canceled_reason = 'zombie flow', + flow_status = $2 + WHERE id = $1", + full_run.id, + flow_status, + ) + .execute(&db) + .await?; + + // Hand-restart from the stuck step. `fanout` is recognised as a derivable + // between-steps zombie (all children succeeded), so it is reused verbatim and + // only the dropped transition onward is replayed. `Some(0)` is the exact value the + // run page's "Re-start from" button sends (a whole-step restart), not `None`. + let restarted = RunJob::from(JobPayload::RestartedFlow { + completed_job_id: full_run.id, + step_id: "fanout".into(), + branch_or_iteration_n: Some(0), + flow_version: None, + branch_chosen: None, + nested: None, + }) + .run_until_complete(&db, false, port) + .await; + + // Flow reaches success, reusing the loop's aggregated result. + assert!( + restarted.success, + "restarted zombie flow should succeed: {:?}", + restarted.json_result() + ); + assert_eq!(restarted.json_result().unwrap(), json!("a,b")); + + // Every completed loop iteration reuses its original child job (no re-run); + // only `after`, which never ran, executes fresh. + let new_iter0 = child_job_id_for_step(&db, restarted.id, "fanout", Some(0)).await; + let new_iter1 = child_job_id_for_step(&db, restarted.id, "fanout", Some(1)).await; + let new_after = child_job_id_for_step(&db, restarted.id, "after", None).await; + assert_eq!(new_iter0, orig_iter0, "loop iteration 0 must be reused"); + assert_eq!(new_iter1, orig_iter1, "loop iteration 1 must be reused"); + assert_ne!(new_after, orig_after, "`after` should run fresh"); + + Ok(()) +} + +/// A serial for-loop reaped *between* iterations (an all-success prefix, but the +/// cursor not yet at the last iteration) must NOT be treated as complete: reuse +/// would silently drop the remaining iterations. A downstream `after` step makes +/// the loop non-final, so the ONLY thing that can prevent reuse here is the +/// cursor-completeness guard; if it regresses, `after` would consume a truncated +/// loop result and this test fails. Restart must re-run the whole loop instead. +#[sqlx::test(fixtures("base", "hello"))] +async fn test_mid_iteration_zombie_not_reused(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let flow_value: FlowValue = serde_json::from_value(json!({ + "modules": [ + { + "id": "fanout", + "value": { + "type": "forloopflow", + "iterator": { "type": "javascript", "expr": "['a', 'b', 'c']" }, + "skip_failures": false, + "parallel": false, + "modules": [{ + "id": "inner", + "value": { + "type": "rawscript", + "language": "deno", + "input_transforms": { + "v": { "type": "javascript", "expr": "flow_input.iter.value" } + }, + "content": "export function main(v: string) { return v }" + } + }] + } + }, + { + "id": "after", + "value": { + "type": "rawscript", + "language": "deno", + "input_transforms": { + "loop_res": { "type": "javascript", "expr": "results.fanout" } + }, + "content": "export function main(loop_res: string[]) { return loop_res.join(',') }" + } + } + ] + })) + .unwrap(); + + let full_run = RunJob::from(JobPayload::RawFlow { + value: flow_value.clone(), + path: None, + restarted_from: None, + }) + .run_until_complete(&db, false, port) + .await; + assert!(full_run.success); + let orig_iter0 = child_job_id_for_step(&db, full_run.id, "fanout", Some(0)).await; + + // Reap after iteration 0: the loop is InProgress with the cursor still on + // iteration 0 (of 3), only iteration 0 recorded; `after` never reached. + let mut flow_status: serde_json::Value = sqlx::query_scalar!( + "SELECT flow_status FROM v2_job_completed WHERE id = $1", + full_run.id + ) + .fetch_one(&db) + .await? + .expect("flow_status"); + for m in flow_status["modules"].as_array_mut().unwrap() { + match m["id"].as_str() { + Some("fanout") => { + m["type"] = json!("InProgress"); + m["iterator"] = json!({ "index": 0, "itered_len": 3 }); + m["flow_jobs"] = json!([m["flow_jobs"][0]]); + m["flow_jobs_success"] = json!([true]); + } + Some("after") => *m = json!({ "type": "WaitingForPriorSteps", "id": "after" }), + _ => {} + } + } + flow_status["step"] = json!(0); + sqlx::query!( + "UPDATE v2_job_completed SET status = 'canceled', canceled_by = 'monitor', + flow_status = $2 WHERE id = $1", + full_run.id, + flow_status, + ) + .execute(&db) + .await?; + + let restarted = RunJob::from(JobPayload::RestartedFlow { + completed_job_id: full_run.id, + step_id: "fanout".into(), + branch_or_iteration_n: None, + flow_version: None, + branch_chosen: None, + nested: None, + }) + .run_until_complete(&db, false, port) + .await; + + // The loop re-runs from scratch: all three iterations execute (so `after` sees + // "a,b,c", not a truncated "a"), and iteration 0 is a fresh job. + assert!( + restarted.success, + "restart should re-run the loop and succeed: {:?}", + restarted.json_result() + ); + assert_eq!(restarted.json_result().unwrap(), json!("a,b,c")); + let new_iter0 = child_job_id_for_step(&db, restarted.id, "fanout", Some(0)).await; + assert_ne!( + new_iter0, orig_iter0, + "iteration 0 must re-run, not be reused" + ); + + Ok(()) +} + +/// A nested restart request targets an inner step of the restart-step container. +/// Even when that container is an eligible between-steps zombie, reuse must NOT +/// fire (it would skip the whole container and ignore the explicit nested target). +/// The inner step must re-run. +#[sqlx::test(fixtures("base", "hello"))] +async fn test_nested_restart_not_swallowed_by_zombie_reuse( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // `branch` is a BranchOne (single child, so branch_or_iteration_n is None on + // restart: the exact shape that would trip zombie reuse) with two inner steps, + // followed by a downstream `after`. + let flow_value: FlowValue = serde_json::from_value(json!({ + "modules": [ + { + "id": "branch", + "value": { + "type": "branchone", + "default": [], + "branches": [{ + "expr": "true", + "modules": [ + { + "id": "inner_first", + "value": { + "type": "rawscript", "language": "deno", + "input_transforms": {}, + "content": "export function main() { return 'first' }" + } + }, + { + "id": "inner_second", + "value": { + "type": "rawscript", "language": "deno", + "input_transforms": { + "first": { "type": "javascript", "expr": "results.inner_first" } + }, + "content": "export function main(first: string) { return `${first}|second` }" + } + } + ] + }] + } + }, + { + "id": "after", + "value": { + "type": "rawscript", "language": "deno", + "input_transforms": { + "b": { "type": "javascript", "expr": "results.branch" } + }, + "content": "export function main(b: string) { return `after:${b}` }" + } + } + ] + })) + .unwrap(); + + let full_run = RunJob::from(JobPayload::RawFlow { + value: flow_value.clone(), + path: None, + restarted_from: None, + }) + .run_until_complete(&db, false, port) + .await; + assert!(full_run.success); + assert_eq!(full_run.json_result().unwrap(), json!("after:first|second")); + let branch_child = child_job_id_for_step(&db, full_run.id, "branch", None).await; + let orig_inner_second = child_job_id_for_step(&db, branch_child, "inner_second", None).await; + + // Reap `branch` as a between-steps zombie (its child completed, transition lost); + // `after` never reached. + let mut flow_status: serde_json::Value = sqlx::query_scalar!( + "SELECT flow_status FROM v2_job_completed WHERE id = $1", + full_run.id + ) + .fetch_one(&db) + .await? + .expect("flow_status"); + for m in flow_status["modules"].as_array_mut().unwrap() { + match m["id"].as_str() { + Some("branch") => m["type"] = json!("InProgress"), + Some("after") => *m = json!({ "type": "WaitingForPriorSteps", "id": "after" }), + _ => {} + } + } + flow_status["step"] = json!(0); + sqlx::query!( + "UPDATE v2_job_completed SET status = 'canceled', canceled_by = 'monitor', + flow_status = $2 WHERE id = $1", + full_run.id, + flow_status, + ) + .execute(&db) + .await?; + + // Nested restart: re-run `inner_second` inside `branch`. Zombie reuse must step + // aside so the nested chain is honored. + let restarted = RunJob::from(JobPayload::RestartedFlow { + completed_job_id: full_run.id, + step_id: "branch".into(), + branch_or_iteration_n: None, + flow_version: None, + branch_chosen: Some(BranchChosen::Branch { branch: 0 }), + nested: Some(Box::new(RestartedFrom { + flow_job_id: branch_child, + step_id: "inner_second".into(), + branch_or_iteration_n: None, + flow_version: None, + branch_chosen: None, + nested: None, + })), + }) + .run_until_complete(&db, false, port) + .await; + + assert!( + restarted.success, + "nested restart of a zombie container should succeed: {:?}", + restarted.json_result() + ); + assert_eq!( + restarted.json_result().unwrap(), + json!("after:first|second") + ); + let new_branch_child = child_job_id_for_step(&db, restarted.id, "branch", None).await; + let new_inner_second = child_job_id_for_step(&db, new_branch_child, "inner_second", None).await; + assert_ne!( + new_inner_second, orig_inner_second, + "the nested target inner_second must re-run, not be skipped by zombie reuse" + ); + + Ok(()) +} + +/// A raw-flow (editor preview) restart queues the request's CURRENT definition, which the editor +/// allows to differ from the completed run. Zombie reuse must not fire there: it would validate the +/// stored step and synthesize Success from the old children, skipping the user's edit. The edited +/// step must re-run and downstream must observe its new result. +#[sqlx::test(fixtures("base", "hello"))] +async fn test_raw_flow_restart_does_not_reuse_edited_step( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let flow_of = |suffix: &str| -> FlowValue { + serde_json::from_value(json!({ + "modules": [ + { + "id": "fanout", + "value": { + "type": "forloopflow", + "iterator": { "type": "javascript", "expr": "['a', 'b']" }, + "skip_failures": false, + "parallel": false, + "modules": [{ + "id": "inner", + "value": { + "type": "rawscript", "language": "deno", + "input_transforms": { + "v": { "type": "javascript", "expr": "flow_input.iter.value" } + }, + "content": format!("export function main(v: string) {{ return v + '{suffix}' }}") + } + }] + } + }, + { + "id": "after", + "value": { + "type": "rawscript", "language": "deno", + "input_transforms": { + "loop_res": { "type": "javascript", "expr": "results.fanout" } + }, + "content": "export function main(loop_res: string[]) { return loop_res.join(',') }" + } + } + ] + })) + .unwrap() + }; + + let full_run = + RunJob::from(JobPayload::RawFlow { value: flow_of(""), path: None, restarted_from: None }) + .run_until_complete(&db, false, port) + .await; + assert!(full_run.success); + assert_eq!(full_run.json_result().unwrap(), json!("a,b")); + let orig_iter0 = child_job_id_for_step(&db, full_run.id, "fanout", Some(0)).await; + + let mut flow_status: serde_json::Value = sqlx::query_scalar!( + "SELECT flow_status FROM v2_job_completed WHERE id = $1", + full_run.id + ) + .fetch_one(&db) + .await? + .expect("flow_status"); + for m in flow_status["modules"].as_array_mut().unwrap() { + match m["id"].as_str() { + Some("fanout") => { + m["type"] = json!("InProgress"); + m["iterator"] = json!({ "index": 1, "itered_len": 2 }); + } + Some("after") => *m = json!({ "type": "WaitingForPriorSteps", "id": "after" }), + _ => {} + } + } + flow_status["step"] = json!(0); + sqlx::query!( + "UPDATE v2_job_completed SET status = 'canceled', canceled_by = 'monitor', + flow_status = $2 WHERE id = $1", + full_run.id, + flow_status, + ) + .execute(&db) + .await?; + + let restarted = RunJob::from(JobPayload::RawFlow { + value: flow_of("X"), + path: None, + restarted_from: Some(RestartedFrom { + flow_job_id: full_run.id, + step_id: "fanout".into(), + branch_or_iteration_n: None, + flow_version: None, + branch_chosen: None, + nested: None, + }), + }) + .run_until_complete(&db, false, port) + .await; + + // The edited step must run: results reflect the new definition, not the reused old children. + assert!( + restarted.success, + "edited raw-flow restart should succeed: {:?}", + restarted.json_result() + ); + assert_eq!(restarted.json_result().unwrap(), json!("aX,bX")); + let new_iter0 = child_job_id_for_step(&db, restarted.id, "fanout", Some(0)).await; + assert_ne!( + new_iter0, orig_iter0, + "the edited fanout step must re-run, not be reused" + ); + + Ok(()) +} + +/// Only a flow reaped by the zombie monitor (canceled_by = 'monitor') is eligible for reuse. A +/// plain force-cancel at the same boundary (a child succeeded, its parent transition not yet +/// landed) yields the identical InProgress/all-success shape but must keep restart-from-step +/// semantics: the selected step re-runs. +#[sqlx::test(fixtures("base", "hello"))] +async fn test_non_monitor_cancel_is_not_reused(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let flow_value: FlowValue = serde_json::from_value(json!({ + "modules": [ + { + "id": "fanout", + "value": { + "type": "forloopflow", + "iterator": { "type": "javascript", "expr": "['a', 'b']" }, + "skip_failures": false, + "parallel": false, + "modules": [{ + "id": "inner", + "value": { + "type": "rawscript", "language": "deno", + "input_transforms": { + "v": { "type": "javascript", "expr": "flow_input.iter.value" } + }, + "content": "export function main(v: string) { return v }" + } + }] + } + }, + { + "id": "after", + "value": { + "type": "rawscript", "language": "deno", + "input_transforms": { + "loop_res": { "type": "javascript", "expr": "results.fanout" } + }, + "content": "export function main(loop_res: string[]) { return loop_res.join(',') }" + } + } + ] + })) + .unwrap(); + + let full_run = RunJob::from(JobPayload::RawFlow { + value: flow_value.clone(), + path: None, + restarted_from: None, + }) + .run_until_complete(&db, false, port) + .await; + assert!(full_run.success); + let orig_iter0 = child_job_id_for_step(&db, full_run.id, "fanout", Some(0)).await; + + // Same frozen-transition shape as a zombie, but canceled by a USER, not the monitor. + let mut flow_status: serde_json::Value = sqlx::query_scalar!( + "SELECT flow_status FROM v2_job_completed WHERE id = $1", + full_run.id + ) + .fetch_one(&db) + .await? + .expect("flow_status"); + for m in flow_status["modules"].as_array_mut().unwrap() { + match m["id"].as_str() { + Some("fanout") => { + m["type"] = json!("InProgress"); + m["iterator"] = json!({ "index": 1, "itered_len": 2 }); + } + Some("after") => *m = json!({ "type": "WaitingForPriorSteps", "id": "after" }), + _ => {} + } + } + flow_status["step"] = json!(0); + sqlx::query!( + "UPDATE v2_job_completed SET status = 'canceled', canceled_by = 'admin', + flow_status = $2 WHERE id = $1", + full_run.id, + flow_status, + ) + .execute(&db) + .await?; + + let restarted = RunJob::from(JobPayload::RestartedFlow { + completed_job_id: full_run.id, + step_id: "fanout".into(), + branch_or_iteration_n: None, + flow_version: None, + branch_chosen: None, + nested: None, + }) + .run_until_complete(&db, false, port) + .await; + + assert!(restarted.success); + let new_iter0 = child_job_id_for_step(&db, restarted.id, "fanout", Some(0)).await; + assert_ne!( + new_iter0, orig_iter0, + "a non-monitor cancel must re-run the step, not reuse the child" + ); + + Ok(()) +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 433d2e4ed5..4b2196bf63 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -5666,6 +5666,10 @@ async fn push_inner<'c, 'd>( restarted_from_val.step_id.as_str(), restarted_from_val.branch_or_iteration_n, restarted_from_val.flow_version, + restarted_from_val.nested.is_some(), + // RawFlow queues the request's (possibly edited) definition, not the + // stored one, so zombie reuse of the stored step is unsafe here. + false, ) .await?; FlowStatus { @@ -6055,6 +6059,10 @@ async fn push_inner<'c, 'd>( step_id.as_str(), branch_or_iteration_n, flow_version, + nested.is_some(), + // RestartedFlow resolves and queues the completed job's stored definition, so the + // step validated for reuse is the one that will run. + true, ) .await?; @@ -7063,6 +7071,78 @@ fn create_restarted_module( } } +/// A between-steps-zombie step: an `InProgress` module (in an otherwise terminal, +/// reaped flow) whose every child is recorded as a `success` completion. Only the +/// module's final state transition was lost, so the whole step is derivable and +/// safe to reuse on restart. Children incomplete/failed/cancelled ⟹ not a zombie. +async fn is_derivable_between_steps_zombie( + db: &Pool, + workspace_id: &str, + module: &FlowStatusModule, +) -> Result { + // The module's own cursor must prove it reached the end (a serial loop/branch-all reaped + // mid-fan-out has an all-success prefix but unrun remaining iterations); while-loops are + // never derivable. Children-success is verified below. + if !module.is_between_steps_complete() { + return Ok(false); + } + let child_ids: Vec = module + .flow_jobs() + .filter(|v| !v.is_empty()) + .or_else(|| module.job().map(|j| vec![j])) + .unwrap_or_default(); + if child_ids.is_empty() { + return Ok(false); + } + let success_children = sqlx::query_scalar!( + "SELECT count(*) FROM v2_job_completed + WHERE workspace_id = $1 AND id = ANY($2) AND status = 'success'", + workspace_id, + &child_ids, + ) + .fetch_one(db) + .await? + .unwrap_or(0); + Ok(success_children == child_ids.len() as i64) +} + +/// Convert a between-steps-zombie `InProgress` module (validated by +/// [`is_derivable_between_steps_zombie`]) into the `Success` it would have become +/// had its dropped transition landed, reusing all completed children. Downstream +/// steps re-derive this step's result from `flow_jobs`/`job` on demand +/// (`get_previous_job_result`), so no aggregate needs recomputing here. +fn reuse_completed_zombie_module(module: FlowStatusModule) -> FlowStatusModule { + match module { + FlowStatusModule::InProgress { + id, + job, + flow_jobs, + flow_jobs_success, + flow_jobs_duration, + branch_chosen, + agent_actions, + agent_actions_success, + .. + } => FlowStatusModule::Success { + id, + job, + // Every child was verified successful, so normalise the success + // vector (the dropped transition may have left the last entry unset). + flow_jobs_success: flow_jobs_success + .map(|v| v.into_iter().map(|_| Some(true)).collect()), + flow_jobs, + flow_jobs_duration, + branch_chosen, + approvers: vec![], + failed_retries: vec![], + skipped: false, + agent_actions, + agent_actions_success, + }, + other => other, + } +} + async fn restarted_flows_resolution( db: &Pool, workspace_id: &str, @@ -7070,6 +7150,15 @@ async fn restarted_flows_resolution( restart_step_id: &str, branch_or_iteration_n: Option, flow_version: Option, + // A nested restart chain (RestartedFrom.nested) descends into the restart step's child to + // re-run an inner step; zombie reuse would skip the whole container and ignore it. + nested_restart: bool, + // Zombie reuse validates the restart step against the completed job's STORED definition and + // synthesizes Success from its recorded children. That is only sound when the run being queued + // uses that same definition (JobPayload::RestartedFlow). A JobPayload::RawFlow restart queues + // the editor's current, possibly EDITED, definition instead, so reuse would skip the edited + // step and reuse the old child result; disable it there. + allow_zombie_reuse: bool, ) -> Result< ( Option, @@ -7086,7 +7175,7 @@ async fn restarted_flows_resolution( let row = sqlx::query!( "SELECT j.runnable_path as script_path, j.runnable_id AS \"script_hash: ScriptHash\", - j.kind AS \"job_kind!: JobKind\", + j.kind AS \"job_kind!: JobKind\", c.canceled_by, COALESCE(c.flow_status, c.workflow_as_code_status) AS \"flow_status: Json>\", j.raw_flow AS \"raw_flow: Json>\" FROM v2_job_completed c JOIN v2_job j USING (id) WHERE j.id = $1 and j.workspace_id = $2", @@ -7102,6 +7191,12 @@ async fn restarted_flows_resolution( )) })?; + // Zombie reuse must only apply to flows the zombie monitor reaped (canceled_by = 'monitor'). + // An ordinary force-cancel copies the same live flow_status, so a user canceling after a child + // succeeds but before the parent transition lands produces the identical InProgress/all-success + // shape; those must retain restart-from-step semantics (the step re-runs). + let reaped_by_monitor = row.canceled_by.as_deref() == Some("monitor"); + let current_flow_version = row.script_hash.map(|x| x.0); let is_version_change = flow_version.is_some() && current_flow_version.is_some() @@ -7192,9 +7287,36 @@ async fn restarted_flows_resolution( continue; }; if module.id() == restart_step_id { - // if the module ID is the one we want to restart the flow at, or if it's past it in the flow, - // set the module as WaitingForPriorSteps as it needs to be re-run - if branch_or_iteration_n.is_none() || branch_or_iteration_n.unwrap() == 0 { + // Reuse is only safe when there is a NEXT step to advance into (advancing past the + // last module lands on the failure step) and the step's definition carries no + // completion/arming semantics that reuse would skip (stop predicates, skip_if, + // suspend, sleep); such a step must re-run, not be synthesized as Success. + let has_next_step = flow_value + .modules + .last() + .is_none_or(|m| m.id != restart_step_id); + // A whole-step restart is `None` (restart API with the field omitted) or `Some(0)` + // (the run page's "Re-start from" button always sends 0); both mean "redo this step", + // which for a monitor-reaped zombie means reuse it. `Some(n>=1)` is an explicit + // partial container restart and keeps its existing reuse-0..n-1 / rerun-from-n path. + if allow_zombie_reuse + && reaped_by_monitor + && branch_or_iteration_n.unwrap_or(0) == 0 + && !nested_restart + && has_next_step + && module_definition.allows_zombie_reuse() + && is_derivable_between_steps_zombie(db, workspace_id, &module).await? + { + // Between-steps-zombie recovery: this step's children all + // completed but its final state transition was dropped (the + // flow was reaped by the zombie monitor). Reuse the completed + // step verbatim and restart from the NEXT step, so no child + // re-runs and only the dropped transition is replayed onward. + step_n += 1; + truncated_modules.push(reuse_completed_zombie_module(module)); + } else if branch_or_iteration_n.is_none() || branch_or_iteration_n.unwrap() == 0 { + // if the module ID is the one we want to restart the flow at, or if it's past it in the flow, + // set the module as WaitingForPriorSteps as it needs to be re-run // The module as WaitingForPriorSteps as the entire module (i.e. all the branches) need to be re-run truncated_modules .push(FlowStatusModule::WaitingForPriorSteps { id: module.id() }); diff --git a/backend/windmill-types/src/flow_status.rs b/backend/windmill-types/src/flow_status.rs index 62da4d5517..79890e3c6e 100644 --- a/backend/windmill-types/src/flow_status.rs +++ b/backend/windmill-types/src/flow_status.rs @@ -475,6 +475,33 @@ impl FlowStatusModule { } } + /// For a still-`InProgress` module (a between-steps zombie), whether the module's own + /// iteration/branch cursor proves it actually reached the end, so the only thing left is + /// the final state transition (children-success is a separate, DB-side check). + /// + /// A serial for-loop / branch-all grows `flow_jobs` one entry at a time, so an all-success + /// prefix does NOT mean the module finished: the cursor must sit on the last element. Parallel + /// containers preallocate every child up front, so a full success set is conclusive. While-loops + /// are never derivable here (continuation depends on a condition evaluated after each iteration, + /// which a reaped zombie never persisted). Non-`InProgress` modules return false. + pub fn is_between_steps_complete(&self) -> bool { + match self { + FlowStatusModule::InProgress { while_loop: true, .. } => false, + // Parallel loop/branch-all: all children exist up front, so children-success suffices. + FlowStatusModule::InProgress { parallel: true, .. } => true, + FlowStatusModule::InProgress { iterator: Some(it), .. } => { + let total = it + .itered_len + .or_else(|| it.itered.as_ref().map(|v| v.len())); + total.is_some_and(|t| t > 0 && it.index + 1 == t) + } + FlowStatusModule::InProgress { branchall: Some(ba), .. } => ba.branch + 1 == ba.len, + // Single-child leaf / subflow / branch-one: the child ran, nothing else to advance. + FlowStatusModule::InProgress { .. } => true, + _ => false, + } + } + pub fn agent_actions(&self) -> Option> { match self { FlowStatusModule::InProgress { agent_actions, .. } => agent_actions.clone(), @@ -549,4 +576,131 @@ impl FlowStatus { let i = usize::try_from(self.step).ok()?; self.modules.get(i) } + + /// Whether no step has begun executing yet: the preprocessor (if any) and the first + /// module are both still `WaitingForPriorSteps`. A reaped flow in this state can be + /// safely re-queued because nothing ran. A preprocessor that is `InProgress` means its + /// child already ran (only the parent transition was lost), so re-queuing would + /// re-run the preprocessor and duplicate its side effects. + pub fn is_not_yet_started(&self) -> bool { + self.preprocessor_module + .as_ref() + .is_none_or(|p| matches!(p, FlowStatusModule::WaitingForPriorSteps { .. })) + && self + .modules + .first() + .is_some_and(|m| matches!(m, FlowStatusModule::WaitingForPriorSteps { .. })) + } +} + +#[cfg(test)] +mod tests { + use super::{FlowStatus, FlowStatusModule}; + + fn module(json: serde_json::Value) -> FlowStatusModule { + serde_json::from_value(json).unwrap() + } + + fn status(json: serde_json::Value) -> FlowStatus { + serde_json::from_value(json).unwrap() + } + + #[test] + fn is_not_yet_started_distinguishes_preprocessor_zombie() { + let nil = "00000000-0000-0000-0000-000000000000"; + let waiting = serde_json::json!({ "type": "WaitingForPriorSteps", "id": "a" }); + let failure = serde_json::json!({ "type": "WaitingForPriorSteps", "id": "failure" }); + // No preprocessor, first module waiting: genuinely unstarted. + assert!(status(serde_json::json!({ + "step": 0, "modules": [waiting], "failure_module": failure + })) + .is_not_yet_started()); + // First module already InProgress: started. + assert!(!status(serde_json::json!({ + "step": 0, + "modules": [{ "type": "InProgress", "id": "a", "job": nil }], + "failure_module": failure + })) + .is_not_yet_started()); + // Preprocessor still waiting, first module waiting: unstarted. + assert!(status(serde_json::json!({ + "step": -1, "modules": [waiting], "failure_module": failure, + "preprocessor_module": { "type": "WaitingForPriorSteps", "id": "pre" } + })) + .is_not_yet_started()); + // Preprocessor InProgress (its child ran) while modules[0] still waits: a + // preprocessor zombie, NOT unstarted, so it must not be auto-requeued. + assert!(!status(serde_json::json!({ + "step": -1, "modules": [waiting], "failure_module": failure, + "preprocessor_module": { "type": "InProgress", "id": "pre", "job": nil } + })) + .is_not_yet_started()); + } + + #[test] + fn between_steps_complete_serial_loop() { + // Cursor on the last iteration => complete. + assert!(module(serde_json::json!({ + "type": "InProgress", "id": "a", "job": "00000000-0000-0000-0000-000000000000", + "iterator": { "index": 1, "itered_len": 2 }, "flow_jobs": [] + })) + .is_between_steps_complete()); + // Reaped mid-iteration (iteration 1 of 2 never scheduled) => NOT complete. + assert!(!module(serde_json::json!({ + "type": "InProgress", "id": "a", "job": "00000000-0000-0000-0000-000000000000", + "iterator": { "index": 0, "itered_len": 2 }, "flow_jobs": [] + })) + .is_between_steps_complete()); + // Legacy shape: itered array present, itered_len absent. + assert!(module(serde_json::json!({ + "type": "InProgress", "id": "a", "job": "00000000-0000-0000-0000-000000000000", + "iterator": { "index": 1, "itered": ["x", "y"] } + })) + .is_between_steps_complete()); + } + + #[test] + fn between_steps_complete_while_loop_never() { + assert!(!module(serde_json::json!({ + "type": "InProgress", "id": "a", "job": "00000000-0000-0000-0000-000000000000", + "while_loop": true, "iterator": { "index": 1, "itered_len": 2 } + })) + .is_between_steps_complete()); + } + + #[test] + fn between_steps_complete_branchall_and_parallel() { + // Serial branch-all on the last branch => complete; earlier branch => not. + assert!(module(serde_json::json!({ + "type": "InProgress", "id": "a", "job": "00000000-0000-0000-0000-000000000000", + "branchall": { "branch": 1, "len": 2 } + })) + .is_between_steps_complete()); + assert!(!module(serde_json::json!({ + "type": "InProgress", "id": "a", "job": "00000000-0000-0000-0000-000000000000", + "branchall": { "branch": 0, "len": 2 } + })) + .is_between_steps_complete()); + // Parallel loop: children preallocated, so any cursor is fine (success is checked elsewhere). + assert!(module(serde_json::json!({ + "type": "InProgress", "id": "a", "job": "00000000-0000-0000-0000-000000000000", + "parallel": true, "iterator": { "index": 0, "itered_len": 3 } + })) + .is_between_steps_complete()); + } + + #[test] + fn between_steps_complete_leaf_and_non_inprogress() { + // Single-child leaf: the child ran, nothing to advance. + assert!(module(serde_json::json!({ + "type": "InProgress", "id": "a", "job": "00000000-0000-0000-0000-000000000000" + })) + .is_between_steps_complete()); + // A Success module is not a between-steps zombie. + assert!(!module(serde_json::json!({ + "type": "Success", "id": "a", "job": "00000000-0000-0000-0000-000000000000", + "skipped": false + })) + .is_between_steps_complete()); + } } diff --git a/backend/windmill-types/src/flows.rs b/backend/windmill-types/src/flows.rs index 7e70dc5acd..b06e2af05e 100644 --- a/backend/windmill-types/src/flows.rs +++ b/backend/windmill-types/src/flows.rs @@ -664,6 +664,19 @@ impl FlowModule { .is_ok_and(|x| x == "script" || x == "rawscript" || x == "flowscript") } + /// Whether a between-steps-zombie step carrying this definition can be safely reused as + /// `Success` on restart (see restart-resolution reuse). Excludes steps whose completion + /// transition or arming carries semantics that reuse would silently skip: stop predicates + /// (`stop_after_if` / `stop_after_all_iters_if`, which decide whether downstream steps run), + /// `skip_if` (skipped-state and suspend arming), a `suspend` approval boundary, and `sleep`. + pub fn allows_zombie_reuse(&self) -> bool { + self.stop_after_if.is_none() + && self.stop_after_all_iters_if.is_none() + && self.skip_if.is_none() + && self.suspend.is_none() + && self.sleep.is_none() + } + pub fn get_type(&self) -> anyhow::Result<&str> { #[derive(Deserialize)] pub struct FlowModuleValueType<'a> {