diff --git a/backend/tests/bun_jobs.rs b/backend/tests/bun_jobs.rs index d121dee202..63d5341ba3 100644 --- a/backend/tests/bun_jobs.rs +++ b/backend/tests/bun_jobs.rs @@ -1256,6 +1256,53 @@ async fn test_bun_wac_task_dispatch_from_flow_script(db: Pool) -> anyh Ok(()) } +/// `task(fn, { cache_ttl })` on an inline task of a deployed flow's step: the child runs +/// the parent's code with the parent's arguments, so its result-cache key carries its +/// step key, or the parent and every sibling would read its result back as their own. +#[sqlx::test(fixtures("base", "wac_flow_script"))] +async fn test_bun_wac_inline_task_cache_is_per_task(db: Pool) -> anyhow::Result<()> { + use windmill_common::flows::FlowNodeId; + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let payload = || JobPayload::FlowScript { + id: FlowNodeId(3000000000000012), + path: "f/system/wac_flow_script/a".to_string(), + language: ScriptLang::Bun, + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(), + }; + + let mut children_from_cache = Vec::new(); + for _ in 0..2 { + let job = RunJob::from(payload()) + .arg("n", serde_json::json!(5)) + .run_until_complete(&db, false, port) + .await; + assert_eq!( + job.json_result().unwrap(), + serde_json::json!({"doubled": 10, "tripled": 15}) + ); + let from_cache: i64 = sqlx::query_scalar( + "SELECT count(*) FROM job_logs l JOIN v2_job j ON j.id = l.job_id \ + WHERE j.parent_job = $1 AND l.logs LIKE '%found in cache%'", + ) + .bind(job.id) + .fetch_one(&db) + .await?; + children_from_cache.push(from_cache); + } + assert_eq!( + children_from_cache, + vec![0, 2], + "the second run serves each task from its own cache entry" + ); + Ok(()) +} + // ============================================================================ // Environment Variable Tests // ============================================================================ diff --git a/backend/tests/fixtures/wac_flow_script.sql b/backend/tests/fixtures/wac_flow_script.sql index 019dbda586..1b780c6cf4 100644 --- a/backend/tests/fixtures/wac_flow_script.sql +++ b/backend/tests/fixtures/wac_flow_script.sql @@ -28,3 +28,26 @@ export const main = workflow(async (n: number) => { return { doubled: d }; });' ); + +-- The same flow's step with two tasks that cache their own result. +INSERT INTO public.flow_node(id, workspace_id, path, hash_v2, lock, code) VALUES ( +3000000000000012, +'test-workspace', +'f/system/wac_flow_script', +'0000000000000000000000000000000000000000000000000000000000000012', +NULL, +E'import { workflow, task } from "windmill-client"; + +const double = task(async (n: number) => { + return n * 2; +}, { cache_ttl: 60 }); +const triple = task(async (n: number) => { + return n * 3; +}, { cache_ttl: 60 }); + +export const main = workflow(async (n: number) => { + const d = await double(n); + const t = await triple(n); + return { doubled: d, tripled: t }; +});' +); diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index e1483c9448..117fbda086 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -2874,9 +2874,9 @@ pub async fn handle_wac_v2_output( .collect(); // Resolve job_payload once (same for all children since they re-run - // 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. + // the parent script). The step's cache setting is for the workflow's + // result; a task is cached only through its own `cache_ttl` option, + // under a key of its own (see `cached_result_path`). let job_payload_template = match job.kind { JobKind::Script => { if let Some(hash) = job.runnable_id { @@ -3035,6 +3035,12 @@ pub async fn handle_wac_v2_output( let mut pushed_ids: Vec = Vec::with_capacity(num_steps); let push_result: error::Result<()> = async { for (step, (_, child_uuid)) in steps.iter().zip(job_ids.iter()) { + // A task with a runnable of its own (a deployed script or flow) queues + // at that runnable's priority; any other task is the parent's code and + // queues at the parent's. + let own_runnable = matches!(step.dispatch_type.as_str(), "script" | "flow") + && !step.script.starts_with("./"); + // Resolve job payload based on dispatch_type let (job_payload, child_args, is_external, on_behalf_of) = match step.dispatch_type.as_str() { @@ -3048,8 +3054,8 @@ pub async fn handle_wac_v2_output( hash: None, language: module.language, lock: module.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(), @@ -3129,13 +3135,12 @@ pub async fn handle_wac_v2_output( let push_args = PushArgs { args: &child_args, extra: None }; - // 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). + // Apply step-level overrides to payload (cache, concurrency) let mut job_payload = job_payload; - if let Some(cache_ttl) = step.cache_ttl.filter(|_| is_external) { + if let Some(cache_ttl) = step.cache_ttl { match &mut job_payload { - JobPayload::ScriptHash { cache_ttl: ref mut ct, .. } => { + JobPayload::ScriptHash { cache_ttl: ref mut ct, .. } + | JobPayload::FlowScript { cache_ttl: ref mut ct, .. } => { *ct = Some(cache_ttl) } JobPayload::Code(ref mut code) => code.cache_ttl = Some(cache_ttl), @@ -3215,10 +3220,8 @@ pub async fn handle_wac_v2_output( step.tag.clone().or_else(|| Some(job.tag.clone())), step.timeout.or(job.timeout), 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 }), + .or(if own_runnable { None } else { job.priority }), None, // authed false, // running None, // end_user_email diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 0a5deb52d2..9c012fd117 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1559,7 +1559,7 @@ pub async fn cached_result_path( client: &AuthedClient, job: &MiniPulledJob, raw_data: Option<&RawData>, -) -> String { +) -> windmill_common::error::Result { let mut hasher = sha2::Sha256::new(); hasher.update(&[job.kind as u8]); if let Some(ScriptHash(hash)) = job.runnable_id { @@ -1574,6 +1574,13 @@ pub async fn cached_result_path( _ => {} } } + // A workflow-as-code task child runs its parent's code with the parent's + // arguments; the step it executes is what tells its result from the parent's + // and from its siblings'. + if let Some(step_key) = wac_executing_key(db, job).await? { + hasher.update(b"wac_step:"); + hasher.update(step_key.as_bytes()); + } hash_args( db, client, @@ -1584,7 +1591,26 @@ pub async fn cached_result_path( job.cache_ignore_s3_path.unwrap_or(false), ) .await; - format!("g/results/{:064x}", hasher.finalize()) + Ok(format!("g/results/{:064x}", hasher.finalize())) +} + +/// The checkpoint step key a workflow-as-code parent seeded for this child at push +/// time; `None` for any job that is not such a child. +async fn wac_executing_key( + db: &DB, + job: &MiniPulledJob, +) -> windmill_common::error::Result> { + if job.parent_job.is_none() || job.flow_step_id.is_some() { + return Ok(None); + } + let key: Option> = sqlx::query_scalar( + "SELECT workflow_as_code_status->'_checkpoint'->>'_executing_key' \ + FROM v2_job_status WHERE id = $1", + ) + .bind(job.id) + .fetch_optional(db) + .await?; + Ok(key.flatten()) } #[cfg(feature = "parquet")] diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 467b8dbf19..56b4b26fb0 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -4645,7 +4645,7 @@ pub async fn handle_queued_job( let cached_res_path = if job.cache_ttl.is_some() { match conn { Connection::Sql(db) => { - Some(cached_result_path(db, &client, &job, preview_data.as_ref()).await) + Some(cached_result_path(db, &client, &job, preview_data.as_ref()).await?) } Connection::Http(_) => None, } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index cbda50c7bc..ed543bd517 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -2005,7 +2005,8 @@ pub async fn update_flow_status_after_job_completion_internal( if flow_job.cache_ttl.is_some() && success { let flow = RawData::Flow(flow_data.clone()); - let cached_res_path = cached_result_path(db, client, &flow_job, Some(&flow)).await; + let cached_res_path = + cached_result_path(db, client, &flow_job, Some(&flow)).await?; save_in_cache( db,