mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-23 00:03:46 +00:00
fix: refuse restarting a flow past an approval that was never granted (#11920)
Restarting a flow at the step right after an approval step copied the approval step as completed, and the worker skips the gate at the restart step, so a flow whose approval timed out or was disapproved could be restarted past it without anyone approving. Restart resolution now refuses a restart whose copied steps include an approval gate that never opened, naming the step to restart from instead. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
8dbdfa1949
commit
2c6bcc09c9
@@ -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<Postgres>,
|
||||
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<i64>,
|
||||
@@ -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<FlowStatusModule> {
|
||||
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()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Uuid> {
|
||||
match self {
|
||||
FlowStatusModule::WaitingForPriorSteps { .. } => None,
|
||||
|
||||
Reference in New Issue
Block a user