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

fix: dispatch workflow-as-code tasks from a deployed flow's inline step

Browse Source
This commit is contained in:
Ruben Fiszel
2026-09-15 15:38:15 +02:00
parent c4e878e831
commit 5fc8ff18f8
3 changed files with 121 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/bun_jobs.rs
+47
View File
@@ -1209,6 +1209,53 @@ export function main() {
Ok(())
}
/// A deployed flow runs an inline step as the `flow_node` its deploy rewrote it into,
/// a `FlowScript` job rather than the preview job the editor runs. A workflow-as-code
/// step's `task()` children must dispatch from that kind too, as re-runs of the same
/// node, or the step passes its editor test and fails once deployed.
///
/// The step is cached: a child that shared the parent's result-cache key would hand
/// its own result (`10`) back to the parent on resume, in place of the workflow's.
#[sqlx::test(fixtures("base", "wac_flow_script"))]
async fn test_bun_wac_task_dispatch_from_flow_script(db: Pool<Postgres>) -> anyhow::Result<()> {
use windmill_common::flows::FlowNodeId;
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let node = FlowNodeId(3000000000000011);
let job = RunJob::from(JobPayload::FlowScript {
id: node,
path: "f/system/wac_flow_script/a".to_string(),
language: ScriptLang::Bun,
cache_ttl: Some(60),
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(),
})
.arg("n", serde_json::json!(5))
.run_until_complete(&db, false, port)
.await;
assert_eq!(
job.json_result().unwrap(),
serde_json::json!({"doubled": 10})
);
let children: Vec<(String, Option<i64>, Option<i32>)> = sqlx::query_as(
"SELECT kind::text, runnable_id, cache_ttl FROM v2_job WHERE parent_job = $1",
)
.bind(job.id)
.fetch_all(&db)
.await?;
assert_eq!(
children,
vec![("flowscript".to_string(), Some(node.0), None)],
"the task child re-runs the parent's flow node, outside the result cache"
);
Ok(())
}
// ============================================================================
// Environment Variable Tests
// ============================================================================
backend/tests/fixtures/wac_flow_script.sql
Vendored
+30
View File
@@ -0,0 +1,30 @@
-- A deployed flow whose inline bun step is workflow-as-code calling task(), in the
-- shape the deploy leaves behind: the RawScript module rewritten into a flow_node that
-- the step then runs as a FlowScript job. No lock, so the worker resolves
-- windmill-client at run time like the other bun fixtures.
INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES (
'test-workspace', '', '',
'f/system/wac_flow_script',
'{}',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"n":{"type":"integer","description":""}},"required":[],"type":"object"}',
'{"modules":[{"id":"a","value":{"type":"flowscript","id":3000000000000011,"language":"bun","input_transforms":{"n":{"expr":"flow_input.n","type":"javascript"}}}}]}',
'system'
);
INSERT INTO public.flow_node(id, workspace_id, path, hash_v2, lock, code) VALUES (
3000000000000011,
'test-workspace',
'f/system/wac_flow_script',
'0000000000000000000000000000000000000000000000000000000000000011',
NULL,
E'import { workflow, task } from "windmill-client";
const double = task(async (n: number) => {
return n * 2;
});
export const main = workflow(async (n: number) => {
const d = await double(n);
return { doubled: d };
});'
);
backend/windmill-worker/src/bun_executor.rs
+44 -15
View File
@@ -2874,15 +2874,17 @@ pub async fn handle_wac_v2_output(
.collect();
// Resolve job_payload once (same for all children since they re-run
// the parent script)
// the parent script). No result cache on a child: it runs with the
// parent's kind, runnable and arguments, so its cached result would be
// read back as the parent's on resume and as every sibling's.
let job_payload_template = match job.kind {
JobKind::Script => {
if let Some(hash) = job.runnable_id {
Ok(JobPayload::ScriptHash {
hash,
path: job.runnable_path.clone().unwrap_or_default(),
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
language: job.script_lang.unwrap_or(ScriptLang::Bun),
priority: job.priority,
@@ -2897,6 +2899,27 @@ pub async fn handle_wac_v2_output(
))
}
}
// A deployed flow runs an inline step as the `flow_node` its deploy
// rewrote it into; the child re-runs that node the way a `Script`
// child re-runs its hash, so `runnable_id` (the checkpoint's source
// hash) stays the same across parent and children.
JobKind::FlowScript => {
if let Some(id) = job.runnable_id {
Ok(JobPayload::FlowScript {
id: windmill_common::flows::FlowNodeId(id.0),
path: job.runnable_path.clone().unwrap_or_default(),
language: job.script_lang.unwrap_or(ScriptLang::Bun),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: ConcurrencySettings::default(),
})
} else {
Err(error::Error::internal_err(
"WAC v2 FlowScript job missing runnable_id".to_string(),
))
}
}
JobKind::Preview => {
let row: Option<(Option<String>, Option<String>)> = sqlx::query_as(
"SELECT raw_code, raw_lock FROM v2_job WHERE id = $1 AND workspace_id = $2",
@@ -2912,8 +2935,8 @@ pub async fn handle_wac_v2_output(
hash: None,
language: job.script_lang.unwrap_or(ScriptLang::Bun),
lock: lock,
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: ConcurrencySettingsWithCustom::default(),
debouncing_settings: DebouncingSettings::default(),
@@ -3106,9 +3129,11 @@ pub async fn handle_wac_v2_output(
let push_args = PushArgs { args: &child_args, extra: None };
// Apply step-level overrides to payload (cache, concurrency)
// Apply step-level overrides to payload (cache, concurrency). The
// cache one is for an external runnable only: an inline child has no
// cache key of its own (see the template above).
let mut job_payload = job_payload;
if let Some(cache_ttl) = step.cache_ttl {
if let Some(cache_ttl) = step.cache_ttl.filter(|_| is_external) {
match &mut job_payload {
JobPayload::ScriptHash { cache_ttl: ref mut ct, .. } => {
*ct = Some(cache_ttl)
@@ -3122,7 +3147,8 @@ pub async fn handle_wac_v2_output(
|| step.concurrency_time_window_s.is_some()
{
match &mut job_payload {
JobPayload::ScriptHash { concurrency_settings: ref mut cs, .. } => {
JobPayload::ScriptHash { concurrency_settings: ref mut cs, .. }
| JobPayload::FlowScript { concurrency_settings: ref mut cs, .. } => {
if let Some(limit) = step.concurrent_limit {
cs.concurrent_limit = Some(limit);
}
@@ -3188,13 +3214,16 @@ pub async fn handle_wac_v2_output(
job.visible_to_owner,
step.tag.clone().or_else(|| Some(job.tag.clone())),
step.timeout.or(job.timeout),
None, // flow_step_id
step.priority, // priority_override
None, // authed
false, // running
None, // end_user_email
None, // trigger
None, // suspended_mode
None, // flow_step_id
// An inline child queues at the parent's priority unless the task
// sets its own; an external runnable keeps its own.
step.priority
.or(if is_external { None } else { job.priority }),
None, // authed
false, // running
None, // end_user_email
None, // trigger
None, // suspended_mode
)
.await?;