From f02df7fc454b0c2a2afa9ae9848e26af0e379246 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 23 Jul 2026 19:09:28 +0200 Subject: [PATCH] feat(monitor): make between-steps zombie flows hand-recoverable (#10287) * feat(monitor): make between-steps zombie flows hand-recoverable When a worker is OOM-killed mid state-transition, the flow is reaped as a between-steps zombie (children all success, module still InProgress). We do not auto-recover (a re-driven transition can OOM again), so instead: - Append actionable recovery guidance to the cancellation reason when the reaped step's state is derivable (every child a success completion): which step, iterations completed, raise memory then restart-from-step (UI + API). - Restart-from-step now reuses a zombie step verbatim (InProgress with all children successful) and restarts from the next step, so no completed child re-runs; downstream steps re-derive its result from flow_jobs on demand. - Cast flow_status ::text in the reaper query: reading the jsonb column as Box included the binary version byte and silently failed FlowStatus parsing (disabling the restart-not-yet-started branch since the v2 migration). Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): only reuse a between-steps zombie step that provably finished Address review findings on the zombie-restart reuse path: - Require structural completeness (FlowStatusModule::is_between_steps_complete): a serial for-loop / branch-all reaped mid-fan-out has an all-success prefix but unrun remaining iterations, so the cursor must sit on the last element; while-loops are never derivable (continuation is a post-iteration condition). Parallel containers preallocate all children, so success alone is conclusive. Shared by the monitor guidance and the restart resolution. - Decline reuse when the step carries stop_after_if / stop_after_all_iters_if: those predicates decide whether downstream steps run, and reuse would bypass them; such a step re-runs instead. - Decline reuse when the zombie step is the last module (advancing past it lands on the failure step); it falls back to the existing re-run path. - Unit tests for is_between_steps_complete and an integration test asserting a mid-iteration serial-loop zombie is re-run, not reused. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): align zombie recovery guidance with restart eligibility Address CI review findings: - Exclude skip_if / suspend / sleep (not just stop predicates) from reuse via FlowModule::allows_zombie_reuse, so a skipped/suspend-armed step is never synthesized as Success (which would strand a restart waiting on an approval it never armed). - The reaper does not load the flow definition, so it cannot know whether restart will reuse or re-run a given step; reword the guidance to state both outcomes (reuse where derivable, re-run for the flow's last step or one carrying a stop/skip condition, approval, or sleep) instead of promising "no re-run". - Make the mid-iteration regression test exercise the cursor-completeness guard: a downstream step makes the loop non-final, so reuse is prevented only by the guard; a truncated loop result would then fail the assertion. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): never let zombie reuse swallow a nested restart request A nested restart (RestartedFrom.nested) descends into the restart step's child to re-run an inner step. For an eligible zombie BranchOne/Subflow the outer branch_or_iteration_n is None, so reuse fired, skipped the container, and the explicitly requested inner step never re-ran. Thread the presence of a nested chain into restarted_flows_resolution and decline reuse when set. Regression test added (RED without the guard: the nested target is reused instead of re-run). Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): don't auto-requeue preprocessor zombies as unstarted flows The ::text parse fix re-activated the "hasn't started yet, restart it" branch, but its `modules[0] == WaitingForPriorSteps` check also matches a flow whose preprocessor is still InProgress (step == -1, first module waiting). Requeuing such a flow re-runs the preprocessor, duplicating side effects / repeating the OOM. Gate the branch on FlowStatus::is_not_yet_started, which also requires the preprocessor (if any) to be WaitingForPriorSteps. Unit-tested. Also drop the numbered procedural narration from the happy-path test comments. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): only emit restart guidance for restartable (deployed, top-level) flows The recovery guidance points operators at the run page's "Re-start from" button and the restart API, but both require a top-level deployed flow: a preview has no flow path (the button is hidden, the API 400s) and a subflow child restarts via its root, not itself. Gate the guidance on runnable_path IS NOT NULL AND parent_job IS NULL so previews/subflows keep the existing wording instead of being told to use a button/endpoint that isn't there. Verified end-to-end: a reaped preview gets no RECOVERY block, a reaped deployed flow does. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): gate recovery guidance on kind='flow' to match the restart surface Addresses review nit: a pathful editor preview (kind='flowpreview' with a runnable_path) satisfied the previous runnable_path check but the run page only renders the "Re-start from" button for kind='flow'. Match that condition exactly so previews/singlestepflow keep the plain wording. Verified end-to-end: a reaped pathful preview now gets no RECOVERY block. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): disable zombie reuse for raw-flow (editor preview) restarts A JobPayload::RawFlow restart queues the request's current, possibly EDITED, definition, but restarted_flows_resolution validates reuse against the completed job's STORED definition. For an eligible preview zombie, editing the restart step and restarting from it would synthesize Success from the old children and skip the edit. Thread allow_zombie_reuse into the resolver (true only for JobPayload::RestartedFlow, which queues the stored definition) and decline reuse for raw-flow restarts. Regression test added (RED without the guard: the edited step is skipped and the old result is reused). Co-Authored-By: Claude Opus 4.8 (1M context) * chore(sqlx): add offline cache for zombie_flow_recovery test queries The integration test's UPDATE v2_job_completed queries had no .sqlx entry, so the CI SQLX_OFFLINE build of the test failed to compile. Regenerated with --all-targets --features deno_core,quickjs to capture the test-target queries. Co-Authored-By: Claude Opus 4.8 (1M context) * test(monitor): drop procedural narration from the raw-flow zombie test Per AGENTS.md (comments record constraints, not narration): remove the two step-describing comments the reviewer flagged; the test doc comment already carries the durable rationale. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): restrict zombie reuse to monitor-reaped flows The reuse predicate matched the InProgress/all-children-success shape without checking provenance, so an ordinary force-cancel at the same boundary (a child succeeded before its parent transition landed) would also be reused, dropping the usual restart-from-step re-run. Gate reuse on canceled_by = 'monitor' (the username the zombie reaper cancels with). Regression test added (RED without the guard: a user-cancelled flow reuses the child instead of re-running it). Co-Authored-By: Claude Opus 4.8 (1M context) * fix(monitor): reuse zombie step on Some(0) too, so the run-page button works The run page's "Re-start from" button always sends branch_or_iteration_n = 0 (never omits it), but reuse only fired for None, so the exact UI path the recovery message points to would re-run the children instead of reusing them. Treat a whole-step restart (None or Some(0)) as reuse-eligible; Some(n>=1) keeps the explicit partial-container restart. Verified against the live EE restart API with branch_or_iteration_n=0: all loop-iteration child UUIDs are reused. Happy- path test now sends Some(0) to match the button. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- ...291f4ff5c979be956369fa4406d789cca92fc.json | 112 +++ ...9dd4864f4e0aa0093df47df7d444ee748dbb2.json | 23 + ...e8a4b96c9629cdc785edc7a389c3eb7269608.json | 22 + ...9d4c24699c2c7abad4086e0bb876c5c6b2c38.json | 15 + ...31fe94999ca4f11934fbab4638b2653a678dc.json | 15 + ...7e00942a8f4f5251502ef8efc0f510a857551.json | 15 + ...8e375d93caf76c4cec362c2b92d2a7b8704a2.json | 83 +++ backend/src/monitor.rs | 119 +++- backend/tests/zombie_flow_recovery.rs | 661 ++++++++++++++++++ backend/windmill-queue/src/jobs.rs | 130 +++- backend/windmill-types/src/flow_status.rs | 154 ++++ backend/windmill-types/src/flows.rs | 13 + 12 files changed, 1352 insertions(+), 10 deletions(-) create mode 100644 backend/.sqlx/query-1bf87fe9667f860c6089bffb803291f4ff5c979be956369fa4406d789cca92fc.json create mode 100644 backend/.sqlx/query-387e135bf363533089493c35eb39dd4864f4e0aa0093df47df7d444ee748dbb2.json create mode 100644 backend/.sqlx/query-4371e59b85cc279dba03800aacde8a4b96c9629cdc785edc7a389c3eb7269608.json create mode 100644 backend/.sqlx/query-4a35f9a0201e283d854bd77f7f79d4c24699c2c7abad4086e0bb876c5c6b2c38.json create mode 100644 backend/.sqlx/query-60201b41a0bfea89a201a25ae2431fe94999ca4f11934fbab4638b2653a678dc.json create mode 100644 backend/.sqlx/query-7d2923f940cf5a8cbbdc51de91c7e00942a8f4f5251502ef8efc0f510a857551.json create mode 100644 backend/.sqlx/query-fa1b9fcdd344fa2c88a187f4a558e375d93caf76c4cec362c2b92d2a7b8704a2.json create mode 100644 backend/tests/zombie_flow_recovery.rs 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> {