diff --git a/backend/windmill-common/src/wac.rs b/backend/windmill-common/src/wac.rs index fe9b6c154f..7a42ba413d 100644 --- a/backend/windmill-common/src/wac.rs +++ b/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, - /// 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, #[serde(skip_serializing_if = "Option::is_none")] #[serde(default)] - pub _executing_args_hash: Option, + pub _executing_args: Option>, /// `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 { - 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::>(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 diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index d4aa7b4fb8..a0d25f6025 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/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) diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 748cafe776..714652a484 100644 --- a/backend/windmill-worker/src/common.rs +++ b/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> { +) -> windmill_common::error::Result>)>> { if job.parent_job.is_none() || job.flow_step_id.is_some() { return Ok(None); } - let identity: Option<(Option, Option)> = 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, Option>>>)> = + 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, }) } diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index 918a16f2b0..5587cd5b1f 100644 --- a/python-client/wmill/wmill/client.py +++ b/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 diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 7b10c732be..2daf4e2138 100644 --- a/typescript-client/client.ts +++ b/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;