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

fix: keep preserve_step_tags on saved branch and loop flow nodes (#11660)

Browse Source 
* fix: keep preserve_step_tags on saved branch and loop flow nodes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* style: format reduce_flow call

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-10-10 12:40:00 +02:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 93e81ec189
commit dedd77587f
2 changed files with 104 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/worker.rs
+95
View File
@@ -5000,6 +5000,101 @@ async fn test_flow_lock_all(db: Pool<Postgres>) -> 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<Postgres>) -> 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::<Uuid>()?;
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"))]
backend/windmill-worker/src/worker_lockfiles.rs
+9
View File
@@ -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<FlowModule>>,
same_worker: bool,
preserve_step_tags: bool,
modules: &mut Vec<FlowModule>,
modules_node: &mut Option<FlowNodeId>,
) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
@@ -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<FlowModule>>,
same_worker: bool,
preserve_step_tags: bool,
) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
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,
)