From 17f5319de70c162c1d29106d0fa8aa05e57f9e87 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 15 Sep 2026 18:11:21 +0200 Subject: [PATCH] fix: key a cached workflow-as-code task on a fingerprint of its code --- backend/windmill-common/src/wac.rs | 10 +++++--- backend/windmill-worker/src/bun_executor.rs | 5 +++- backend/windmill-worker/src/common.rs | 26 +++++++++++---------- backend/windmill-worker/src/wac_executor.rs | 4 ++++ python-client/wmill/wmill/client.py | 21 +++++++++++++++-- typescript-client/client.ts | 17 +++++++++++++- 6 files changed, 64 insertions(+), 19 deletions(-) diff --git a/backend/windmill-common/src/wac.rs b/backend/windmill-common/src/wac.rs index c994dcc4c9..2c1a9ccf59 100644 --- a/backend/windmill-common/src/wac.rs +++ b/backend/windmill-common/src/wac.rs @@ -40,9 +40,13 @@ pub struct WacCheckpoint { #[serde(skip_serializing_if = "Option::is_none")] #[serde(default)] pub _executing_key: Option, - /// With `_executing_key`: the arguments the task was called with. A cached - /// result is keyed on both; the key alone is a name and a position, which two - /// tasks can share, and the arguments alone say nothing about which task ran. + /// With `_executing_key`: the SDK's fingerprint of the task's code and the + /// arguments the task was called with, what its cached result is keyed on. + /// The step key stands in for the fingerprint when the SDK sent none; it is + /// a name and a position, which two tasks can share. + #[serde(skip_serializing_if = "Option::is_none")] + #[serde(default)] + pub _executing_fn: Option, #[serde(skip_serializing_if = "Option::is_none")] #[serde(default)] pub _executing_args: Option>, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 8380ab0ef2..4ea15d7b03 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -3234,11 +3234,14 @@ pub async fn handle_wac_v2_output( // _executing_key to know which step to run). External // scripts/flows don't need a WAC checkpoint. if !is_external { - let child_checkpoint_json = serde_json::json!({ + let mut child_checkpoint_json = serde_json::json!({ "completed_steps": &checkpoint.completed_steps, "_executing_key": &step.key, "_executing_args": &step.args, }); + if let Some(fn_id) = &step.fn_id { + child_checkpoint_json["_executing_fn"] = serde_json::json!(fn_id); + } sqlx::query( "INSERT INTO v2_job_status (id, workflow_as_code_status) VALUES ($1, jsonb_build_object('_checkpoint', $2::jsonb)) diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index b96416ad2a..b2b2db0b53 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1575,16 +1575,16 @@ 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 step it - // runs (a task's name and position, which is what tells two tasks of one - // name apart) and the arguments that task was called with. - let (step, args) = match wac_task_identity(db, job).await? { - Some((step, args)) => (Some(step), Some(Json(args))), + // nothing about the task's own inputs; its result is keyed on the task it + // runs (the SDK's fingerprint of its code, or its step key from an older + // SDK) and the arguments that task was called with. + 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(step) = step { - hasher.update(b"wac_step:"); - hasher.update(step.as_bytes()); + if let Some(task) = task { + hasher.update(b"wac_task:"); + hasher.update(task.as_bytes()); } hash_args( db, @@ -1599,8 +1599,9 @@ pub async fn cached_result_path( Ok(format!("g/results/{:064x}", hasher.finalize())) } -/// The step key 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. +/// The task fingerprint (or, from an SDK that sends none, the step key) 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, @@ -1610,7 +1611,8 @@ async fn wac_task_identity( } let identity: Option<(Option, Option>>>)> = sqlx::query_as( - "SELECT workflow_as_code_status->'_checkpoint'->>'_executing_key', \ + "SELECT COALESCE(workflow_as_code_status->'_checkpoint'->>'_executing_fn', \ + workflow_as_code_status->'_checkpoint'->>'_executing_key'), \ workflow_as_code_status->'_checkpoint'->'_executing_args' \ FROM v2_job_status WHERE id = $1", ) @@ -1618,7 +1620,7 @@ async fn wac_task_identity( .fetch_optional(db) .await?; Ok(match identity { - Some((Some(step), Some(Json(args)))) => Some((step, args)), + Some((Some(task), Some(Json(args)))) => Some((task, args)), _ => None, }) } diff --git a/backend/windmill-worker/src/wac_executor.rs b/backend/windmill-worker/src/wac_executor.rs index c188792a5e..ab86c0e2fa 100644 --- a/backend/windmill-worker/src/wac_executor.rs +++ b/backend/windmill-worker/src/wac_executor.rs @@ -85,6 +85,10 @@ pub struct WacStepDispatch { pub concurrency_key: Option, #[serde(default)] pub concurrency_time_window_s: Option, + /// The SDK's fingerprint of the task's code, the identity its cached result + /// is keyed on; absent from an SDK that predates it. + #[serde(default)] + pub fn_id: Option, } fn default_dispatch_type() -> String { diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index 5587cd5b1f..2b0371aac6 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -3026,7 +3026,7 @@ class WorkflowCtx: print(f"\n--- WAC: {key} ---") info = {"name": name or key, "script": script or key, "args": kwargs, "key": key, "dispatch_type": dispatch_type} if _task_options: - for opt_key in ("timeout", "tag", "cache_ttl", "priority", "concurrent_limit", "concurrency_key", "concurrency_time_window_s"): + for opt_key in ("timeout", "tag", "cache_ttl", "priority", "concurrent_limit", "concurrency_key", "concurrency_time_window_s", "fn_id"): if opt_key in _task_options and _task_options[opt_key] is not None: info[opt_key] = _task_options[opt_key] self._pending.append(info) @@ -3285,6 +3285,22 @@ class WorkflowCtx: }) +def _fn_fingerprint(func) -> str: + """A stable identity for a task's code, what its cached result is keyed on: a + name is shared by any two tasks called the same, and a step key by any two + tasks called at the same position, so neither can tell them apart.""" + import hashlib + import inspect + import marshal + + try: + src = inspect.getsource(func).encode() + except (OSError, TypeError): + # No source on disk: the whole code object, constants and names included. + src = marshal.dumps(func.__code__) + return hashlib.sha256(src).hexdigest() + + def task( _func=None, *, @@ -3361,6 +3377,7 @@ def task( def decorator(func) -> Callable[..., Any]: task_path = path task_name = func.__name__ + _fn_opts = {**(_task_opts or {}), "fn_id": _fn_fingerprint(func)} _params_list = list(_sig(func).parameters) @@ -3386,7 +3403,7 @@ def task( if ctx is not None: script = task_path if task_path else task_name merged = _merge_args(args, kwargs) - return ctx._next_step(task_name, script, func, _task_options=_task_opts, **merged) + return ctx._next_step(task_name, script, func, _task_options=_fn_opts, **merged) # WAC v1: running inside a Windmill job but not in a @workflow if ( diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 2daf4e2138..c983e503e0 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1853,6 +1853,7 @@ export class WorkflowCtx { args: Record = {}, dispatch_type: string = "inline", options?: TaskOptions, + fnId?: string, ): PromiseLike { this._rethrowSwallowed(); const stepName = name || script || "step"; @@ -1906,6 +1907,7 @@ export class WorkflowCtx { } const stepInfo: any = { name: name || key, script: script || key, args, key, dispatch_type }; + if (fnId) stepInfo.fn_id = fnId; if (options) { if (options.timeout !== undefined) stepInfo.timeout = options.timeout; if (options.tag !== undefined) stepInfo.tag = options.tag; @@ -2298,6 +2300,18 @@ export async function step( * decoded back before the caller sees it: a `Date` comes back as a string, a * `Map` as `{}`. {@link JsonifiedFn} is that shape. */ +/** A stable identity for a task's code, what its cached result is keyed on: a + * name is shared by any two tasks called the same, and a step key by any two + * tasks called at the same position, so neither can tell them apart. */ +function fnFingerprint(fn: Function): string { + const src = fn.toString(); + let h = 0xcbf29ce484222325n; + for (let i = 0; i < src.length; i++) { + h = ((h ^ BigInt(src.charCodeAt(i))) * 0x100000001b3n) & 0xffffffffffffffffn; + } + return h.toString(16); +} + export function task Promise>( fnOrPath: T | string, maybeFnOrOptions?: T | TaskOptions, @@ -2319,6 +2333,7 @@ export function task Promise>( assertUsableRetry(taskOptions?.retry); const taskName = fn.name || taskPath || ""; + const fnId = fnFingerprint(fn); // NOT async — in workflow context we return the thenable directly so that // unawaited task calls leave the step in ctx.pending (for _flushPending). @@ -2339,7 +2354,7 @@ export function task Promise>( kwargs[`arg${i}`] = args[i]; } } - const stepResult = ctx._nextStep(taskName, script, kwargs, "inline", taskOptions); + const stepResult = ctx._nextStep(taskName, script, kwargs, "inline", taskOptions, fnId); // If this step should execute directly (child job mode), run the inner function // and throw StepSuspend with mode "step_complete" to signal that we're done if ((stepResult as any)?._execute_directly) {