From 5b31dec048856b160997ee86397d3bfbc8aa96a6 Mon Sep 17 00:00:00 2001 From: hugocasa Date: Mon, 9 Mar 2026 10:45:35 +0100 Subject: [PATCH] fix: parallel branchall hang on bad stop_after_all_iters_if + results.x.length null (#8276) Two fixes: 1. When a parallel branchall/forloop has a `stop_after_all_iters_if` expression that fails (e.g. bad JS syntax), the error was propagated with `?`, causing the transaction to roll back the parallel index increment. Since all parallel jobs were already completed, nothing could ever increment the index again and the flow hung forever. Now the error is caught and converted to a stop-early failure so the transaction commits and the flow fails gracefully. 2. Expressions like `results.a.length` in step input transforms resolved to null because the `handle_full_regex` fast path intercepted them and used PostgreSQL's `#>` JSON path operator, which can't resolve JS runtime properties like `.length` on arrays. Now the fast path skips expressions ending with JS-only properties (like `length`), falling through to full QuickJS evaluation where they work correctly. Co-authored-by: Claude Opus 4.6 --- backend/tests/worker.rs | 167 +++++++++++++++++++++ backend/windmill-jseval/src/lib.rs | 22 +++ backend/windmill-worker/src/worker_flow.rs | 25 ++- 3 files changed, 210 insertions(+), 4 deletions(-) diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 9548986a5a..32fada32fb 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -3548,3 +3548,170 @@ async fn test_flow_substep_tag_availability_check(db: Pool) -> anyhow: Ok(()) } + +#[cfg(all(feature = "quickjs", feature = "python"))] +#[sqlx::test(fixtures("base"))] +async fn test_stop_after_all_iters_if_bad_expr_parallel_branchall( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + + let port = 123; + let flow: FlowValue = serde_json::from_value(serde_json::json!({ + "modules": [ + { + "id": "a", + "value": { + "branches": [ + {"modules": [{ + "id": "b", + "value": { + "input_transforms": { "n": { "type": "javascript", "expr": "flow_input.n" } }, + "type": "rawscript", + "language": "python3", + "content": "def main(n): return n", + }, + }]} + ], + "type": "branchall", + "parallel": true, + }, + "stop_after_all_iters_if": { + "expr": "invalid!!!syntax", + "skip_if_stopped": false, + }, + }, + ], + })) + .unwrap(); + let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; + + let cjob = RunJob::from(job) + .arg("n", json!(42)) + .run_until_complete(&db, false, port) + .await; + + assert!( + !cjob.success, + "flow should fail when stop_after_all_iters_if has bad expression" + ); + + let result = cjob.json_result().unwrap(); + let error_msg = result["error"]["message"].as_str().unwrap_or(""); + assert!( + error_msg.contains("stop_after_all_iters_if"), + "error should mention stop_after_all_iters_if, got: {error_msg}" + ); + + Ok(()) +} + +#[cfg(all(feature = "quickjs", feature = "python"))] +#[sqlx::test(fixtures("base"))] +async fn test_stop_after_all_iters_if_bad_expr_parallel_forloop( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + + let port = 123; + let flow: FlowValue = serde_json::from_value(serde_json::json!({ + "modules": [ + { + "id": "a", + "value": { + "type": "forloopflow", + "iterator": { "type": "javascript", "expr": "result.items" }, + "skip_failures": false, + "parallel": true, + "modules": [{ + "value": { + "input_transforms": { + "n": { "type": "javascript", "expr": "flow_input.iter.value" }, + }, + "type": "rawscript", + "language": "python3", + "content": "def main(n): return n", + }, + }], + }, + "stop_after_all_iters_if": { + "expr": "invalid!!!syntax", + "skip_if_stopped": false, + }, + }, + ], + })) + .unwrap(); + let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; + + let cjob = RunJob::from(job) + .arg("items", json!([1, 2, 3])) + .run_until_complete(&db, false, port) + .await; + + assert!( + !cjob.success, + "flow should fail when stop_after_all_iters_if has bad expression" + ); + + let result = cjob.json_result().unwrap(); + let error_msg = result["error"]["message"].as_str().unwrap_or(""); + assert!( + error_msg.contains("stop_after_all_iters_if"), + "error should mention stop_after_all_iters_if, got: {error_msg}" + ); + + Ok(()) +} + +#[cfg(all(feature = "quickjs", feature = "python"))] +#[sqlx::test(fixtures("base"))] +async fn test_results_length_in_input_transform(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // Step a returns a list, step b accesses results.a.length via input transform. + // This tests that the handle_full_regex fast path falls through to QuickJS + // when the SQL JSON path operator can't resolve JS properties like .length. + let flow: FlowValue = serde_json::from_value(json!({ + "modules": [ + { + "id": "a", + "value": { + "type": "rawscript", + "language": "python3", + "content": "def main(): return [10, 20, 30]", + }, + }, + { + "id": "b", + "value": { + "input_transforms": { + "v": { "type": "javascript", "expr": "results.a.length" }, + }, + "type": "rawscript", + "language": "python3", + "content": "def main(v): return v", + }, + }, + ], + })) + .unwrap(); + + let result = + RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None }) + .run_until_complete(&db, false, port) + .await + .json_result() + .unwrap(); + + assert_eq!( + result, + json!(3), + "results.a.length should resolve to 3, not null" + ); + + Ok(()) +} diff --git a/backend/windmill-jseval/src/lib.rs b/backend/windmill-jseval/src/lib.rs index 89feaf7b3e..7172e8cfc8 100644 --- a/backend/windmill-jseval/src/lib.rs +++ b/backend/windmill-jseval/src/lib.rs @@ -152,6 +152,21 @@ pub fn try_exact_property_access( None } +/// JS runtime properties (not methods) that cannot be resolved by PostgreSQL's +/// #> JSON path operator. Function calls like .map(...) already don't match the +/// RE_FULL regex due to parentheses, so only property accesses need listing here. +const JS_ONLY_PROPERTIES: &[&str] = &["length"]; + +fn ends_with_js_only_property(rest: Option<&str>) -> bool { + match rest { + None => false, + Some(rest) => { + let last_segment = rest.rsplit('.').next().unwrap_or(""); + JS_ONLY_PROPERTIES.contains(&last_segment) + } + } +} + pub async fn handle_full_regex( expr: &str, authed_client: &AuthedClient, @@ -162,6 +177,13 @@ pub async fn handle_full_regex( let obj_key = captures.get(2).unwrap().as_str(); let idx_o = captures.get(3).map(|y| y.as_str()); let rest = captures.get(4).map(|y| y.as_str()); + + // Skip the SQL fast path when the expression accesses a JS runtime + // property (e.g. .length) that the PostgreSQL #> operator can't resolve. + if ends_with_js_only_property(rest) { + return None; + } + let query = if let Some(idx) = idx_o { match rest { Some(rest) => Some(format!("{}{}", idx, rest)), diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 33dafbb482..8a45f85a90 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -865,7 +865,7 @@ pub async fn update_flow_status_after_job_completion_internal( .and_then(|x| x.stop_after_all_iters_if.as_ref()) { let args = from_result_to_args(args.as_ref().await.get_ref())?; - evaluate_stop_after_all_iters_if( + if let Err(e) = evaluate_stop_after_all_iters_if( db, stop_after_all_iters_if, module_status, @@ -879,7 +879,16 @@ pub async fn update_flow_status_after_job_completion_internal( flow, &old_status, ) - .await?; + .await + { + tracing::error!("error evaluating stop_after_all_iters_if: {e:#}"); + stop_early = true; + skip_if_stop_early = false; + stop_early_err_msg = Some(format!( + "Error evaluating stop_after_all_iters_if expression `{}`: {e:#}", + stop_after_all_iters_if.expr + )); + } } let new_status = if @@ -1074,7 +1083,7 @@ pub async fn update_flow_status_after_job_completion_internal( { let args = from_result_to_args(args.as_ref().await.get_ref())?; - evaluate_stop_after_all_iters_if( + if let Err(e) = evaluate_stop_after_all_iters_if( db, stop_after_all_iters_if, module_status, @@ -1088,7 +1097,15 @@ pub async fn update_flow_status_after_job_completion_internal( flow, &old_status, ) - .await?; + .await + { + stop_early = true; + skip_if_stop_early = false; + stop_early_err_msg = Some(format!( + "Error evaluating stop_after_all_iters_if expression `{}`: {e:#}", + stop_after_all_iters_if.expr + )); + } } }