diff --git a/backend/tests/retry.rs b/backend/tests/retry.rs index abd7c38d8d..e804b96f0e 100644 --- a/backend/tests/retry.rs +++ b/backend/tests/retry.rs @@ -262,6 +262,65 @@ def main(last, port): Ok(()) } + /* `skip_if` evaluation goes through windmill-jseval and requires `quickjs` */ + #[cfg(all(feature = "deno_core", feature = "quickjs"))] + #[sqlx::test(fixtures("base"))] + async fn test_skip_if_sees_previous_step_result_on_retry( + db: Pool, + ) -> anyhow::Result<()> { + initialize_tracing().await; + + /* step `b` fails on its first attempt and succeeds on retry. Its `skip_if` + * references the previous step's result: when re-evaluated for the retry, + * `previous_result` must still be step `a`'s result, not the error of step + * `b`'s failed attempt (which would wrongly flip `skip_if` to true and skip + * the retry) */ + let value = serde_json::from_value(json!({ + "modules": [{ + "id": "a", + "value": { + "input_transforms": {}, + "type": "rawscript", + "language": "deno", + "content": "export function main() { return \"ok\" }", + }, + }, { + "id": "b", + "skip_if": { "expr": "results.a?.error !== undefined" }, + "value": { + "input_transforms": { + "index": { "type": "static", "value": 1 }, + "port": { "type": "javascript", "expr": "flow_input.port" }, + }, + "type": "rawscript", + "language": "deno", + "content": inner_step(), + }, + "retry": { "constant": { "attempts": 1, "seconds": 0 } }, + }], + })) + .unwrap(); + + let (attempts, responses) = [ + /* fail step `b` once, then pass on retry */ + (1, None), + (1, Some(42)), + ] + .into_iter() + .unzip::<_, _, Vec<_>, Vec<_>>(); + let server = Server::start(responses).await; + let result = RunJob::from(JobPayload::RawFlow { value, path: None, restarted_from: None }) + .arg("port", json!(server.addr.port())) + .run_until_complete(&db, false, server.addr.port()) + .await + .json_result() + .unwrap(); + + assert_eq!(server.close().await, attempts); + assert_eq!(json!(42), result); + Ok(()) + } + #[cfg(feature = "python")] #[sqlx::test(fixtures("base"))] async fn test_with_failure_module(db: Pool) -> anyhow::Result<()> { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 699b8683ec..86263350c3 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -3268,7 +3268,7 @@ async fn push_next_flow_job( } // Compute and initialize last_job_result - let arc_last_job_result = if status_module.is_failure() { + let mut arc_last_job_result = if status_module.is_failure() { // if job is being retried, pass the result of its previous failure last_job_result.unwrap_or_else(|| Arc::new(to_raw_value(&json!("{}")))) } else if matches!(step, Step::Step { idx: 0, .. }) || step.is_preprocessor_step() { @@ -3699,6 +3699,30 @@ async fn push_next_flow_job( .context("update flow retry")?; status_module = FlowStatusModule::WaitingForPriorSteps { id: status_module.id() }; + + // The failed attempt's error has already been consumed by `evaluate_retry` + // above. Restore `previous_result` to the preceding step's result (or the + // flow args for the first step) so that predicates re-evaluated for the + // retry (skip_if, loop iterator expressions, ...) don't see the failed + // attempt's error instead. Like the suspend/restart path above, this + // falls back to `"{}"` when the preceding step has no Success status + // (e.g. it failed with continue_on_error). + if !matches!(step, Step::FailureStep) { + arc_last_job_result = if matches!(step, Step::Step { idx: 0, .. }) + || step.is_preprocessor_step() + { + Arc::new(to_raw_value(&flow_job.args)) + } else { + match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status) + .warn_after_seconds(3) + .await? + { + None => Arc::new(to_raw_value(&json!("{}"))), + Some(previous_job_result) => Arc::new(previous_job_result), + } + }; + } + // we get the args from the last failed job status.retry.failed_jobs.last() /* Start the failure module ... */