diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 3ec21f3c84..c51f057f16 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6809,9 +6809,7 @@ async fn push_inner<'c, 'd>( 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, + Some(&value), ) .await?; FlowStatus { @@ -7207,7 +7205,7 @@ async fn push_inner<'c, 'd>( 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, + None, ) .await?; @@ -8416,6 +8414,29 @@ pub async fn fetch_restart_flow_version( cache::flow::fetch_version(db, version).await } +// A restart copies the steps before the restart step as completed, and the worker only gates the +// restart step on a copied approval when that gate opened in the original run. So a copied step +// whose gate never opened (timed out, disapproved) would let the restart skip its approval. +fn refuse_restart_past_unopened_suspend_gate( + flow_value: &FlowValue, + copied_modules: &[FlowStatusModule], +) -> Result<(), Error> { + for module in copied_modules { + let id = module.id(); + let suspend = flow_value + .modules + .iter() + .find(|m| m.id == id) + .and_then(|m| m.suspend.as_ref()); + if module.suspend_gate_never_opened(suspend) { + return Err(Error::BadRequest(format!( + "Step {id} was never approved. Restart the flow from step {id} to request the approval again." + ))); + } + } + Ok(()) +} + async fn restarted_flows_resolution( db: &Pool, workspace_id: &str, @@ -8426,12 +8447,12 @@ async fn restarted_flows_resolution( // 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, + // The definition the restart queues when it is not the completed job's stored one + // (JobPayload::RawFlow queues the editor's, possibly EDITED, definition). Approval gates are + // read from it. Zombie reuse validates the restart step against the STORED definition and + // synthesizes Success from its recorded children, so it is disabled when this is set: reuse + // would skip the edited step and reuse the old child result. + queued_flow: Option<&FlowValue>, ) -> Result< ( Option, @@ -8492,6 +8513,8 @@ async fn restarted_flows_resolution( }; let flow_value = flow_data.value(); + let allow_zombie_reuse = queued_flow.is_none(); + let gate_flow = queued_flow.unwrap_or(flow_value); let flow_status = row .flow_status .as_ref() @@ -8512,6 +8535,7 @@ async fn restarted_flows_resolution( let module_id = &module_definition.id; if module_id == restart_step_id { + refuse_restart_past_unopened_suspend_gate(gate_flow, &truncated_modules)?; // Mark this and all following modules as WaitingForPriorSteps if branch_or_iteration_n.is_none() || branch_or_iteration_n.unwrap() == 0 { truncated_modules @@ -8563,6 +8587,7 @@ async fn restarted_flows_resolution( continue; }; if module.id() == restart_step_id { + refuse_restart_past_unopened_suspend_gate(gate_flow, &truncated_modules)?; // 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, @@ -8950,3 +8975,38 @@ mod render_tag_path_tests { RawValue::from_string(json.to_string()).unwrap() } } + +#[cfg(test)] +mod restart_past_suspend_gate_tests { + use super::refuse_restart_past_unopened_suspend_gate; + use serde_json::json; + use windmill_common::{flow_status::FlowStatusModule, flows::FlowValue}; + + #[test] + fn refuses_only_a_blocking_gate_that_never_opened() { + let flow = |suspend: serde_json::Value| -> FlowValue { + serde_json::from_value(json!({"modules": [ + {"id": "a", "value": {"type": "identity"}, "suspend": suspend}, + {"id": "b", "value": {"type": "identity"}}, + ]})) + .unwrap() + }; + let copied_a = |approvers: serde_json::Value| -> Vec { + vec![serde_json::from_value(json!({ + "type": "Success", "id": "a", "job": uuid::Uuid::new_v4(), "approvers": approvers + })) + .unwrap()] + }; + let blocking = flow(json!({"required_events": 1})); + let approved = json!([{"resume_id": 0, "approver": "admin"}]); + + assert!( + refuse_restart_past_unopened_suspend_gate(&blocking, &copied_a(json!([]))).is_err() + ); + assert!(refuse_restart_past_unopened_suspend_gate(&blocking, &copied_a(approved)).is_ok()); + let continues = flow(json!({"required_events": 1, "continue_on_disapprove_timeout": true})); + assert!( + refuse_restart_past_unopened_suspend_gate(&continues, &copied_a(json!([]))).is_ok() + ); + } +} diff --git a/backend/windmill-types/src/flow_status.rs b/backend/windmill-types/src/flow_status.rs index 6dff8430b7..2fd5ef1184 100644 --- a/backend/windmill-types/src/flow_status.rs +++ b/backend/windmill-types/src/flow_status.rs @@ -4,7 +4,7 @@ use std::time::Duration; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use crate::flows::FlowValue; +use crate::flows::{FlowValue, Suspend}; const MINUTES: Duration = Duration::from_secs(60); const HOURS: Duration = MINUTES.saturating_mul(60); @@ -409,6 +409,21 @@ pub enum JobResult { } impl FlowStatusModule { + /// Whether this completed step's suspend gate could block and never opened (it timed out or + /// was disapproved): `approvers` is written when the gate opens, and is the only record of it + /// that outlives the run, since its resume_job rows are deleted with its queue row. + pub fn suspend_gate_never_opened(&self, suspend: Option<&Suspend>) -> bool { + let Some(suspend) = suspend else { + return false; + }; + if suspend.required_events.unwrap_or(0) == 0 + || suspend.continue_on_disapprove_timeout.unwrap_or(false) + { + return false; + } + matches!(self, FlowStatusModule::Success { skipped: false, approvers, .. } if approvers.is_empty()) + } + pub fn job(&self) -> Option { match self { FlowStatusModule::WaitingForPriorSteps { .. } => None,