fix: hash a cached workflow-as-code task's arguments like any job's

backend/windmill-common/src/wac.rs

@@ -40,14 +40,14 @@ pub struct WacCheckpoint {
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(default)]
     pub _executing_key: Option<String>,
-    /// With `_executing_key`: the task the child runs and `task_args_hash` of the
-    /// arguments it was called with, the identity its cached result is keyed on.
+    /// With `_executing_key`: the task the child runs and the arguments it was
+    /// called with, the identity its cached result is keyed on.
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(default)]
     pub _executing_task: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(default)]
-    pub _executing_args_hash: Option<String>,
+    pub _executing_args: Option<serde_json::Map<String, Value>>,
     /// `resume_job.id` values already consumed by earlier approval steps (the
     /// row primary key, not the distinct integer `resume_id` column). Rows are
     /// never deleted, so a workflow with several sequential wait_for_approval()
@@ -80,34 +80,9 @@ pub fn approval_resume_id(step_key: &str) -> u32 {
     u32::from_be_bytes([digest[0], digest[1], digest[2], digest[3]])
 }
 
-/// Hash of a task call's arguments, with the keys in a fixed order so the order a
-/// caller passes them in cannot split its cached results.
-pub fn task_args_hash(args: &serde_json::Map<String, Value>) -> String {
-    use sha2::{Digest, Sha256};
-    let ordered: std::collections::BTreeMap<&String, &Value> = args.iter().collect();
-    let json = serde_json::to_string(&ordered).unwrap_or_default();
-    format!("{:x}", Sha256::digest(json.as_bytes()))
-}
-
 #[cfg(test)]
 mod tests {
     use super::approval_resume_id;
-    use super::task_args_hash;
-
-    #[test]
-    fn task_args_hash_ignores_key_order_and_sees_values() {
-        let parse = |s: &str| {
-            serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(s).unwrap()
-        };
-        assert_eq!(
-            task_args_hash(&parse(r#"{"a":1,"b":2}"#)),
-            task_args_hash(&parse(r#"{"b":2,"a":1}"#))
-        );
-        assert_ne!(
-            task_args_hash(&parse(r#"{"a":1,"b":2}"#)),
-            task_args_hash(&parse(r#"{"a":1,"b":3}"#))
-        );
-    }
 
     /// Golden values: worker and API must agree on this mapping, and they can run
     /// different builds during a rolling deploy. Changing it strands every resume

backend/windmill-worker/src/bun_executor.rs

@@ -3238,7 +3238,7 @@ pub async fn handle_wac_v2_output(
                             "completed_steps": &checkpoint.completed_steps,
                             "_executing_key": &step.key,
                             "_executing_task": &step.name,
-                            "_executing_args_hash": windmill_common::wac::task_args_hash(&step.args),
+                            "_executing_args": &step.args,
                         });
                         sqlx::query(
                             "INSERT INTO v2_job_status (id, workflow_as_code_status)

backend/windmill-worker/src/common.rs

@@ -1577,45 +1577,47 @@ pub async fn cached_result_path(
     // A workflow-as-code task child carries its parent's arguments, which say
     // nothing about the task's own inputs; its result is keyed on the task it
     // runs and the arguments that task was called with.
-    if let Some((task, args_hash)) = wac_task_identity(db, job).await? {
+    let (task, args) = match wac_task_identity(db, job).await? {
+        Some((task, args)) => (Some(task), Some(Json(args))),
+        None => (None, job.args.clone()),
+    };
+    if let Some(task) = task {
         hasher.update(b"wac_task:");
         hasher.update(task.as_bytes());
-        hasher.update(b":");
-        hasher.update(args_hash.as_bytes());
-    } else {
-        hash_args(
-            db,
-            client,
-            &job.workspace_id,
-            &job.args,
-            &mut hasher,
-            &job.id,
-            job.cache_ignore_s3_path.unwrap_or(false),
-        )
-        .await;
     }
+    hash_args(
+        db,
+        client,
+        &job.workspace_id,
+        &args,
+        &mut hasher,
+        &job.id,
+        job.cache_ignore_s3_path.unwrap_or(false),
+    )
+    .await;
     Ok(format!("g/results/{:064x}", hasher.finalize()))
 }
 
-/// The task name and arguments hash a workflow-as-code parent seeded in this
+/// The task name and call arguments a workflow-as-code parent seeded in this
 /// child's checkpoint at push time; `None` for any job that is not such a child.
 async fn wac_task_identity(
     db: &DB,
     job: &MiniPulledJob,
-) -> windmill_common::error::Result<Option<(String, String)>> {
+) -> windmill_common::error::Result<Option<(String, HashMap<String, Box<RawValue>>)>> {
     if job.parent_job.is_none() || job.flow_step_id.is_some() {
         return Ok(None);
     }
-    let identity: Option<(Option<String>, Option<String>)> = sqlx::query_as(
-        "SELECT workflow_as_code_status->'_checkpoint'->>'_executing_task', \
-                workflow_as_code_status->'_checkpoint'->>'_executing_args_hash' \
-         FROM v2_job_status WHERE id = $1",
-    )
-    .bind(job.id)
-    .fetch_optional(db)
-    .await?;
+    let identity: Option<(Option<String>, Option<Json<HashMap<String, Box<RawValue>>>>)> =
+        sqlx::query_as(
+            "SELECT workflow_as_code_status->'_checkpoint'->>'_executing_task', \
+                    workflow_as_code_status->'_checkpoint'->'_executing_args' \
+             FROM v2_job_status WHERE id = $1",
+        )
+        .bind(job.id)
+        .fetch_optional(db)
+        .await?;
     Ok(match identity {
-        Some((Some(task), Some(args_hash))) => Some((task, args_hash)),
+        Some((Some(task), Some(Json(args)))) => Some((task, args)),
         _ => None,
     })
 }

python-client/wmill/wmill/client.py

@@ -3327,6 +3327,11 @@ def task(
     it grows with both the width of the fan-out and ``attempts``. Retries with
     no ``delay`` all go out in a single round.
 
+    ``cache_ttl`` serves a previous result of the task for that many seconds
+    instead of running it again. The result is keyed on the task and the
+    arguments it is called with, so anything a cached task reads from its
+    closure must be passed in as an argument.
+
     Usage::
 
         @task

typescript-client/client.ts

@@ -1712,6 +1712,10 @@ export interface TaskRetry {
 export interface TaskOptions {
   timeout?: number;
   tag?: string;
+  /** Seconds during which a previous result of this task is served instead of
+   *  running it again. The result is keyed on the task and the arguments it is
+   *  called with, so anything a cached task reads from its closure must be
+   *  passed in as an argument. */
   cache_ttl?: number;
   priority?: number;
   concurrency_limit?: number;