From dedd77587f3234f3e480388b00340394794ac5bb Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 10 Oct 2026 12:40:00 +0200 Subject: [PATCH] fix: keep preserve_step_tags on saved branch and loop flow nodes (#11660) * fix: keep preserve_step_tags on saved branch and loop flow nodes Co-Authored-By: Claude Opus 5.5 * style: format reduce_flow call Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- backend/tests/worker.rs | 95 +++++++++++++++++++ .../windmill-worker/src/worker_lockfiles.rs | 9 ++ 2 files changed, 104 insertions(+) diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index eebb14e07e..751752eb00 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -5000,6 +5000,101 @@ async fn test_flow_lock_all(db: Pool) -> anyhow::Result<()> { Ok(()) } +/// A deployed flow runs a multi-step loop body as the flow node its deploy stored, which +/// carries its own copy of the flow's `preserve_step_tags`. A preview never loads that node. +#[sqlx::test(fixtures("base"))] +async fn test_saved_flow_preserves_nested_step_tags(db: Pool) -> anyhow::Result<()> { + use futures::StreamExt; + + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let client = reqwest::Client::new(); + let flow = json!({ + "path": "g/all/saved_step_tags", "summary": "", "description": "", "tag": "python3", + "schema": { "type": "object", "properties": {}, "required": [] }, + "value": { + "preserve_step_tags": true, + "modules": [{ + "id": "loop", + "value": { + "type": "forloopflow", "skip_failures": false, + "iterator": { "type": "static", "value": [2, 3] }, + "modules": [{ + "id": "own_tag", + "value": { + "type": "rawscript", "language": "bun", "lock": "", "tag": "bun", + "content": "export function main(n: number) { return n * n; }", + "input_transforms": { + "n": { "type": "javascript", "expr": "flow_input.iter.value" } + } + } + }, { + "id": "untagged", + "value": { + "type": "rawscript", "language": "bun", "lock": "", + "content": "export function main(n: number) { return n + 1; }", + "input_transforms": { + "n": { "type": "javascript", "expr": "results.own_tag" } + } + } + }] + } + }] + } + }); + let mut completed = listen_for_completed_jobs(&db).await; + client + .post(format!( + "http://localhost:{port}/api/w/test-workspace/flows/create" + )) + .bearer_auth("SECRET_TOKEN") + .json(&flow) + .send() + .await? + .error_for_status()?; + let deployment = in_test_worker(&db, completed.next(), port).await.unwrap(); + let deployed = completed_job(deployment, &db).await; + assert!(deployed.success, "{:?}", deployed.result); + + let job_id = client + .post(format!( + "http://localhost:{port}/api/w/test-workspace/jobs/run/f/g/all/saved_step_tags" + )) + .bearer_auth("SECRET_TOKEN") + .json(&json!({})) + .send() + .await? + .error_for_status()? + .text() + .await? + .parse::()?; + in_test_worker(&db, completed.find(&job_id), port).await; + let job = completed_job(job_id, &db).await; + assert!(job.success, "{:?}", job.result); + let tags = sqlx::query_as::<_, (String, String)>( + "WITH RECURSIVE descendants AS ( + SELECT id FROM v2_job WHERE id = $1 + UNION ALL + SELECT child.id FROM v2_job child JOIN descendants ON child.parent_job = descendants.id + ) + SELECT DISTINCT flow_step_id, tag FROM v2_job + WHERE id IN (SELECT id FROM descendants) AND flow_step_id IN ('own_tag', 'untagged') + ORDER BY flow_step_id, tag", + ) + .bind(job_id) + .fetch_all(&db) + .await?; + assert_eq!( + tags, + vec![ + ("own_tag".to_owned(), "bun".to_owned()), + ("untagged".to_owned(), "python3".to_owned()) + ] + ); + Ok(()) +} + #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index b608601991..6189d39b8d 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -1244,6 +1244,7 @@ pub async fn handle_flow_dependency_job( &job.workspace_id, flow.failure_module.as_ref(), flow.same_worker, + flow.preserve_step_tags, ) .await?; @@ -2134,6 +2135,7 @@ async fn insert_flow_modules<'c>( workspace_id: &str, failure_module: Option<&Box>, same_worker: bool, + preserve_step_tags: bool, modules: &mut Vec, modules_node: &mut Option, ) -> Result> { @@ -2144,6 +2146,7 @@ async fn insert_flow_modules<'c>( workspace_id, failure_module, same_worker, + preserve_step_tags, )) .await?; if modules.is_empty() || crate::worker_flow::is_simple_modules(modules, failure_module) { @@ -2160,6 +2163,7 @@ async fn insert_flow_modules<'c>( modules: std::mem::take(modules), failure_module: failure_module.cloned(), same_worker, + preserve_step_tags, ..Default::default() }))), None, @@ -2176,6 +2180,7 @@ async fn reduce_flow<'c>( workspace_id: &str, failure_module: Option<&Box>, same_worker: bool, + preserve_step_tags: bool, ) -> Result> { use FlowModuleValue::*; for module in &mut *modules { @@ -2234,6 +2239,7 @@ async fn reduce_flow<'c>( workspace_id, failure_module, same_worker, + preserve_step_tags, modules, modules_node, ) @@ -2247,6 +2253,7 @@ async fn reduce_flow<'c>( workspace_id, failure_module, same_worker, + preserve_step_tags, &mut branch.modules, &mut branch.modules_node, ) @@ -2258,6 +2265,7 @@ async fn reduce_flow<'c>( workspace_id, failure_module, same_worker, + preserve_step_tags, default, default_node, ) @@ -2271,6 +2279,7 @@ async fn reduce_flow<'c>( workspace_id, failure_module, same_worker, + preserve_step_tags, &mut branch.modules, &mut branch.modules_node, )