fix: give a workflow-as-code task its own result-cache key

backend/tests/bun_jobs.rs

@@ -1256,6 +1256,53 @@ async fn test_bun_wac_task_dispatch_from_flow_script(db: Pool<Postgres>) -> 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<Postgres>) -> 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
 // ============================================================================

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 };
+});'
+);

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<Uuid> = 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

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<String> {
     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<Option<String>> {
+    if job.parent_job.is_none() || job.flow_step_id.is_some() {
+        return Ok(None);
+    }
+    let key: Option<Option<String>> = 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")]

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,
         }

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,