diff --git a/backend/tests/flow_engine_parity.rs b/backend/tests/flow_engine_parity.rs index 78e45e404b..5f17c98a06 100644 --- a/backend/tests/flow_engine_parity.rs +++ b/backend/tests/flow_engine_parity.rs @@ -3312,3 +3312,60 @@ export function main(i: number) { Ok(()) } + +// A `$flow_expr[...]` step tag is resolved from the flow's state before the step is pushed, and +// one that cannot be evaluated fails the step instead of queueing it on a tag no worker serves. +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_flow_expr_step_tag(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + + let step = |id: &str, tag: Option<&str>| { + flow_module( + id, + FlowModuleValue::RawScript { + input_transforms: Default::default(), + language: ScriptLang::Deno, + content: "export function main() { return { lang: 'bun' } }".to_string(), + path: None, + lock: None, + tag: tag.map(str::to_string), + concurrency_settings: Default::default(), + is_trigger: None, + assets: None, + }, + ) + }; + let flow = FlowValue { + modules: vec![ + step("a", None), + step("b", Some("$flow_expr[results.a.lang]")), + step("c", Some("nobody-serves-$flow_expr[a.lang]")), + ], + same_worker: false, + ..Default::default() + }; + + let job = RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None }) + .run_until_complete(&db, false, server.addr.port()) + .await; + + let b_tag = sqlx::query_scalar::<_, String>( + "SELECT tag FROM v2_job WHERE parent_job = $1 AND flow_step_id = 'b'", + ) + .bind(job.id) + .fetch_one(&db) + .await?; + assert_eq!(b_tag, "bun"); + + assert!(!job.success); + let result = job.json_result().unwrap(); + let message = result["error"]["message"].as_str().unwrap_or_default(); + assert!( + message.contains("Could not resolve the step tag `nobody-serves-$flow_expr[a.lang]`"), + "got {result:?}" + ); + + Ok(()) +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index ab91d54287..6a99082596 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -4650,6 +4650,12 @@ pub fn tag_reads_args(tag: &str) -> bool { RE_ARG_TAG.is_match(tag) } +/// Whether the tag reads the flow's state (`$flow_expr[results.a.foo]`). Only the flow runtime +/// can resolve it, right before pushing the step; `push` leaves it verbatim. +pub fn tag_reads_flow_expr(tag: &str) -> bool { + RE_FLOW_EXPR_TAG.is_match(tag) +} + pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String { // Save this value to avoid parsing twice let workspaced = x.as_str().replace("$workspace", workspace_id).to_string(); @@ -5269,6 +5275,8 @@ pub fn empty_result() -> Box { lazy_static::lazy_static! { pub static ref RE_ARG_TAG: Regex = Regex::new(r#"\$args\[((?:\w+\.)*\w+)\]"#).unwrap(); + pub static ref RE_FLOW_EXPR_TAG: Regex = + Regex::new(r#"\$flow_expr\[((?:\w+\.)*\w+)\]"#).unwrap(); } #[cfg(feature = "cloud")] diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index ed543bd517..3b0d600607 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -71,8 +71,8 @@ use windmill_queue::{ add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job, insert_concurrency_key_capped, interpolate_args, report_error_to_workspace_handler_or_critical_side_channel, tag_reads_args, - try_schedule_next_job, CanceledBy, FlowRunners, MiniCompletedJob, MiniPulledJob, PushArgs, - PushIsolationLevel, SameWorkerPayload, WrappedError, + tag_reads_flow_expr, try_schedule_next_job, CanceledBy, FlowRunners, MiniCompletedJob, + MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload, WrappedError, RE_FLOW_EXPR_TAG, }; use windmill_audit::audit_oss::audit_log; @@ -3114,6 +3114,48 @@ fn resolve_flow_step_tag( } } +/// Resolves each `$flow_expr[path]` of a step tag by evaluating `path` as a flow expression +/// (`results.a.foo`, `flow_input.region`, `flow_env.pool`). A string renders bare, `null` (also +/// what a missing field evaluates to) renders empty, any other value as its JSON text. +async fn interpolate_flow_expr_tag( + tag: &str, + last_result: Arc>, + flow_args: Marc>>, + flow_env: Option<&HashMap>>, + client: &AuthedClient, + by_id: &IdContext, +) -> error::Result { + let mut rendered: HashMap<&str, String> = HashMap::new(); + for cap in RE_FLOW_EXPR_TAG.captures_iter(tag) { + let expr = cap.get(1).unwrap().as_str(); + if rendered.contains_key(expr) { + continue; + } + let value = evaluate_input_transform::( + &InputTransform::new_javascript_expr(expr), + last_result.clone(), + Some(flow_args.clone()), + flow_env, + Some(client), + Some(by_id), + ) + .await + .map_err(|e| Error::ExecutionErr(format!("Could not resolve the step tag `{tag}`: {e}")))?; + rendered.insert(expr, render_flow_expr_tag_value(value)); + } + Ok(RE_FLOW_EXPR_TAG + .replace_all(tag, |cap: ®ex::Captures| rendered[&cap[1]].clone()) + .into_owned()) +} + +fn render_flow_expr_tag_value(value: Value) -> String { + match value { + Value::String(s) => s, + Value::Null => String::new(), + v => v.to_string(), + } +} + #[cfg(test)] mod tag_resolution_tests { use super::resolve_flow_step_tag; @@ -4192,6 +4234,11 @@ async fn push_next_flow_job( None }; + // The scope the step's input transforms were evaluated in, which a `$flow_expr[...]` tag + // must read too. + let mut expr_flow_args = arc_flow_job_args.clone(); + let mut expr_previous_id = previous_id.as_str(); + let marc; let me; let args = match &next_status { @@ -4216,8 +4263,11 @@ async fn push_next_flow_job( if let Some(input_transforms) = simple_input_transforms { //previous id is none because we do not want to use previous id if we are in a for loop let ctx = get_transform_context(&flow_job, "", &status); + let args = Marc::new(args); + expr_flow_args = args.clone(); + expr_previous_id = ""; let ti = transform_input( - Marc::new(args), + args, flow_env, arc_last_job_result.clone(), input_transforms, @@ -4396,17 +4446,21 @@ async fn push_next_flow_job( payload_tag.tag.as_deref(), ); - // `push_args` is empty once the input transforms failed, so a tag reading `$args[...]` - // interpolates to a queue nobody serves and the step sits there instead of reporting - // the error. Send it to the flow's tag, which a worker is provably serving right now. + // A step whose inputs, or whose `$flow_expr[...]` tag, failed to evaluate is pushed only + // to report the error, and a computed tag can then name a queue nobody serves (`push_args` + // is empty, so `$args[...]` reads nothing), leaving the step stuck instead. Send it to the + // flow's tag, which a worker is provably serving right now. // // A step handed over by id, or one whose tag `push` replaces, never reaches a worker // through its tag, so rewriting theirs would be noise. let step_is_pulled_by_tag = !continue_on_same_worker && !continue_with_runners && !payload_tag.payload.is_dedicated_worker(); - let reroute_to_flow_tag = - err.is_some() && step_is_pulled_by_tag && tag.as_deref().is_some_and(tag_reads_args); + let reroute_to_flow_tag = err.is_some() + && step_is_pulled_by_tag + && tag + .as_deref() + .is_some_and(|t| tag_reads_args(t) || tag_reads_flow_expr(t)); let tag = if reroute_to_flow_tag { Some(flow_job.tag.clone()) } else { @@ -4448,6 +4502,38 @@ async fn push_next_flow_job( .await?; } + // Resolved only after the check: CUSTOM_TAGS allows the template, so its value may name + // any queue, as the value of an `$args[...]` tag does. + let mut tag_err = None; + let tag = match tag { + Some(t) if err.is_none() && tag_reads_flow_expr(&t) => { + let ctx = get_transform_context(&flow_job, expr_previous_id, &status); + match interpolate_flow_expr_tag( + &t, + arc_last_job_result.clone(), + expr_flow_args.clone(), + flow_env, + client, + &ctx, + ) + .warn_after_seconds(3) + .await + { + Ok(resolved) => Some(resolved), + Err(e) => { + tag_err = Some(e); + Some(if step_is_pulled_by_tag { + flow_job.tag.clone() + } else { + t + }) + } + } + } + t => t, + }; + let err = err.or(tag_err.as_ref()); + let evaluated_timeout = if let Some(timeout_transform) = &module.timeout { let ctx = get_transform_context(&flow_job, &previous_id, &status); diff --git a/frontend/src/lib/components/AssignableTagsInner.svelte b/frontend/src/lib/components/AssignableTagsInner.svelte index e1ad0baf8e..f014dfeed5 100644 --- a/frontend/src/lib/components/AssignableTagsInner.svelte +++ b/frontend/src/lib/components/AssignableTagsInner.svelte @@ -39,7 +39,8 @@ // Mirrors CUSTOM_TAG_REGEX in backend/windmill-common/src/worker.rs — keep both in sync. const customTagRegex = /^([\w-]+)\(((?:[\w-]+\*?\+)*[\w-]+\*?|(?:\^[\w-]+\*?)+)\)$/ - const dynamicTagRegex = /\$args\[((?:\w+\.)*\w+)\]/ + // Mirrors RE_ARG_TAG and RE_FLOW_EXPR_TAG in backend/windmill-queue/src/jobs.rs. + const dynamicTagRegex = /\$(args|flow_expr)\[((?:\w+\.)*\w+)\]/ function formatWorkspace(w: { id: string; includeForks: boolean }) { return w.includeForks ? `${w.id} (and its forks)` : w.id @@ -49,7 +50,7 @@ let r = newTag.trim() if (r == '') return undefined let matched = r.match(dynamicTagRegex) - return matched?.[1] + return matched ? { kind: matched[1], path: matched[2] } : undefined }) let extractedCustomTag = $derived.by(() => { @@ -183,7 +184,7 @@ {:else if newTag.trim()} - {#if newTag.includes('(') || newTag.includes(')') || newTag.includes('+') || newTag.includes('^') || newTag.includes('*') || ((newTag.includes('.') || newTag.includes('$args[')) && !dynamicTag)} + {#if newTag.includes('(') || newTag.includes(')') || newTag.includes('+') || newTag.includes('^') || newTag.includes('*') || ((newTag.includes('.') || newTag.includes('$args[') || newTag.includes('$flow_expr[')) && !dynamicTag)}
Invalid tag
@@ -194,7 +195,7 @@ {:else}
- {#if newTag.includes('$workspace') || newTag.includes('$args')} + {#if newTag.includes('$workspace') || dynamicTag} Dynamic tag {:else} Simple tag @@ -207,8 +208,13 @@ {#if newTag.includes('$workspace') && !dynamicTag}
Interpolated tag based on workspace id the job was created in
{/if} - {#if dynamicTag} -
Interpolated tag based on args input of {dynamicTag}
+ {#if dynamicTag?.kind == 'flow_expr'} +
+ Interpolated tag based on flow expression {dynamicTag.path}, resolved when the + flow step starts +
+ {:else if dynamicTag} +
Interpolated tag based on args input of {dynamicTag.path}
{/if}
{/if} @@ -252,6 +258,10 @@ > based on args input, use 
$args[a.b.c]
where 
a.b.c
is the path to the value in the args object. +
{#if variant !== 'drawer'}
{/if} + On flow steps, use a flow expression path to base the tag on earlier results, flow inputs or flow + env, e.g. 
$flow_expr[results.a.region]
or + 
$flow_expr[flow_input.region]
. {/if}