starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-21 00:02:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: resolve $flow_expr[...] dynamic tags on flow steps

Browse Source 
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-09-16 13:38:35 +02:00
co-authored by Claude Opus 5
parent 54553b2add
commit 5e97ec22b4
4 changed files with 175 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/flow_engine_parity.rs
+57
View File
@@ -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<Postgres>) -> 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(())
}
backend/windmill-queue/src/jobs.rs
+8
View File
@@ -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<RawValue> {
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")]
backend/windmill-worker/src/worker_flow.rs
+94 -8
View File
@@ -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<Box<RawValue>>,
flow_args: Marc<HashMap<String, Box<RawValue>>>,
flow_env: Option<&HashMap<String, Box<RawValue>>>,
client: &AuthedClient,
by_id: &IdContext,
) -> error::Result<String> {
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::<Value>(
&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: &regex::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);
frontend/src/lib/components/AssignableTagsInner.svelte
+16 -6
View File
@@ -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 @@
</div>
</div>
{: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)}
<div class="text-2xs text-primary p-2 bg-surface-secondary rounded border">
<div class="font-medium mb-1 text-red-500">Invalid tag</div>
<div>
@@ -194,7 +195,7 @@
{:else}
<div class="text-2xs text-primary p-2 bg-surface-secondary rounded border">
<div class="font-medium mb-1">
{#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}
<div>Interpolated tag based on workspace id the job was created in </div>
{/if}
{#if dynamicTag}
<div>Interpolated tag based on args input of <b>{dynamicTag}</b></div>
{#if dynamicTag?.kind == 'flow_expr'}
<div>
Interpolated tag based on flow expression <b>{dynamicTag.path}</b>, resolved when the
flow step starts
</div>
{:else if dynamicTag}
<div>Interpolated tag based on args input of <b>{dynamicTag.path}</b></div>
{/if}
</div>
{/if}
@@ -252,6 +258,10 @@
>
based on args input, use <pre class="inline text-emphasis">$args[a.b.c]</pre> where
<pre class="inline">a.b.c</pre> is the path to the value in the args object.
<br />{#if variant !== 'drawer'}<br />{/if}
On flow steps, use a flow expression path to base the tag on earlier results, flow inputs or flow
env, e.g. <pre class="inline text-emphasis">$flow_expr[results.a.region]</pre> or
<pre class="inline text-emphasis">$flow_expr[flow_input.region]</pre>.
</span>
{/if}
</div>