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

fix: run an unresolvable $flow_expr tag on the default tag and give it the step input scope

Browse Source 
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-09-16 13:54:29 +02:00
co-authored by Claude Opus 5
parent 5e97ec22b4
commit 1731e518be
2 changed files with 27 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-queue/src/jobs.rs
+6 -3
View File
@@ -4650,8 +4650,8 @@ 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.
/// Whether the tag reads the flow's state (`$flow_expr[results.a.foo]`), which only the flow
/// runtime can resolve, right before pushing the step.
pub fn tag_reads_flow_expr(tag: &str) -> bool {
RE_FLOW_EXPR_TAG.is_match(tag)
}
@@ -6545,7 +6545,10 @@ async fn push_inner<'c, 'd>(
);
windmill_common::worker::dedicated_worker_tag(workspace_id, &full_path)
} else {
if tag == Some("".to_string()) {
// The flow runtime resolves a step's `$flow_expr[...]` before pushing it, so one still here
// was pushed with no flow state to read (a step test, a dependency job) and would name a
// queue no worker serves: the job runs on its default tag instead.
if tag == Some("".to_string()) || tag.as_deref().is_some_and(tag_reads_flow_expr) {
tag = None;
}
backend/windmill-worker/src/worker_flow.rs
+21 -8
View File
@@ -3115,11 +3115,12 @@ 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.
/// (`results.a.foo`, `flow_input.region`, `flow_env.pool`, `resume.region`). 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>>,
env: HashMap<String, Arc<Box<RawValue>>>,
flow_args: Marc<HashMap<String, Box<RawValue>>>,
flow_env: Option<&HashMap<String, Box<RawValue>>>,
client: &AuthedClient,
@@ -3131,16 +3132,22 @@ async fn interpolate_flow_expr_tag(
if rendered.contains_key(expr) {
continue;
}
let value = evaluate_input_transform::<Value>(
&InputTransform::new_javascript_expr(expr),
last_result.clone(),
let value = eval_timeout(
expr.to_string(),
env.clone(),
Some(flow_args.clone()),
flow_env,
Some(client),
Some(by_id),
None,
)
.await
.map_err(|e| Error::ExecutionErr(format!("Could not resolve the step tag `{tag}`: {e}")))?;
.map_err(|e| {
Error::ExecutionErr(format!(
"Could not resolve the step tag `{tag}`: error during evaluation of `{expr}`:\n{e:#}"
))
})?;
let value = serde_json::from_str::<Value>(value.get()).map_err(to_anyhow)?;
rendered.insert(expr, render_flow_expr_tag_value(value));
}
Ok(RE_FLOW_EXPR_TAG
@@ -4508,9 +4515,15 @@ async fn push_next_flow_job(
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);
let env = HashMap::from([
("previous_result".to_string(), arc_last_job_result.clone()),
("resume".to_string(), resume.clone()),
("resumes".to_string(), resumes.clone()),
("approvers".to_string(), approvers.clone()),
]);
match interpolate_flow_expr_tag(
&t,
arc_last_job_result.clone(),
env,
expr_flow_args.clone(),
flow_env,
client,