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 + )); + } } }