From 4585410254dcf3bb41d324059714ac045a92d7b4 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 15 Sep 2026 23:59:40 +0200 Subject: [PATCH] fix: key a cached inline task on its step key and the workflow input --- backend/tests/bun_jobs.rs | 69 ++++++++----------- backend/tests/fixtures/wac_flow_script.sql | 20 ------ backend/windmill-common/src/wac.rs | 10 --- backend/windmill-worker/src/bun_executor.rs | 6 +- backend/windmill-worker/src/common.rs | 48 +++++-------- backend/windmill-worker/src/wac_executor.rs | 4 -- cli/src/guidance/skills.gen.ts | 24 ++++--- python-client/wmill/wmill/client.py | 29 ++------ system_prompts/auto-generated/prompts.ts | 24 ++++--- system_prompts/auto-generated/script.md | 8 ++- system_prompts/auto-generated/sdks/python.md | 8 ++- .../auto-generated/sdks/wac-python.md | 8 ++- .../auto-generated/sdks/wac-typescript.md | 8 ++- .../skills/write-script-python3/SKILL.md | 8 ++- .../skills/write-workflow-as-code/SKILL.md | 16 +++-- typescript-client/client.ts | 25 ++----- 16 files changed, 126 insertions(+), 189 deletions(-) diff --git a/backend/tests/bun_jobs.rs b/backend/tests/bun_jobs.rs index 67391a6134..63d5341ba3 100644 --- a/backend/tests/bun_jobs.rs +++ b/backend/tests/bun_jobs.rs @@ -1257,10 +1257,8 @@ async fn test_bun_wac_task_dispatch_from_flow_script(db: Pool) -> anyh } /// `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 is keyed on the task -/// and the arguments it was called with, or the parent and every sibling would read -/// its result back as their own, and a task fed by an earlier step would be served a -/// result computed for another input. +/// 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; @@ -1268,49 +1266,40 @@ async fn test_bun_wac_inline_task_cache_is_per_task(db: Pool) -> anyho let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); - // Runs the step of node `node` with n=5; returns its result and how many of its - // task children were served from the result cache. - async fn run( - db: &Pool, - port: u16, - node: i64, - ) -> anyhow::Result<(serde_json::Value, i64)> { - let job = RunJob::from(JobPayload::FlowScript { - id: FlowNodeId(node), - 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( - ), - }) - .arg("n", serde_json::json!(5)) - .run_until_complete(db, false, port) - .await; + 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) + .fetch_one(&db) .await?; - Ok((job.json_result().unwrap(), from_cache)) - } - - // Two cached tasks on the same input: the second run serves each from its own entry. - for expected_from_cache in [0, 2] { - let (result, from_cache) = run(&db, port, 3000000000000012).await?; - assert_eq!(result, serde_json::json!({"doubled": 10, "tripled": 15})); - assert_eq!(from_cache, expected_from_cache); - } - - // A cached task called with a fresh value each run is never served a stale result. - for _ in 0..2 { - let (result, from_cache) = run(&db, port, 3000000000000013).await?; - assert_eq!(result, serde_json::json!({"fresh": true})); - assert_eq!(from_cache, 0); + 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(()) } diff --git a/backend/tests/fixtures/wac_flow_script.sql b/backend/tests/fixtures/wac_flow_script.sql index d284a500c0..1b780c6cf4 100644 --- a/backend/tests/fixtures/wac_flow_script.sql +++ b/backend/tests/fixtures/wac_flow_script.sql @@ -51,23 +51,3 @@ export const main = workflow(async (n: number) => { return { doubled: d, tripled: t }; });' ); - --- The same flow's step with a cached task fed by a value that differs on every run. -INSERT INTO public.flow_node(id, workspace_id, path, hash_v2, lock, code) VALUES ( -3000000000000013, -'test-workspace', -'f/system/wac_flow_script', -'0000000000000000000000000000000000000000000000000000000000000013', -NULL, -E'import { workflow, task, step } from "windmill-client"; - -const double = task(async (n: number) => { - return n * 2; -}, { cache_ttl: 60 }); - -export const main = workflow(async (n: number) => { - const r = await step("pick", async () => Math.floor(Math.random() * 1e9)); - const d = await double(r); - return { fresh: d === r * 2 }; -});' -); diff --git a/backend/windmill-common/src/wac.rs b/backend/windmill-common/src/wac.rs index 2c1a9ccf59..05c343b427 100644 --- a/backend/windmill-common/src/wac.rs +++ b/backend/windmill-common/src/wac.rs @@ -40,16 +40,6 @@ pub struct WacCheckpoint { #[serde(skip_serializing_if = "Option::is_none")] #[serde(default)] pub _executing_key: Option, - /// 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>, /// `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() diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 4ea15d7b03..117fbda086 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -3234,14 +3234,10 @@ 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 mut child_checkpoint_json = serde_json::json!({ + let 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 b2b2db0b53..9c012fd117 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1574,23 +1574,18 @@ 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 (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(task) = task { - hasher.update(b"wac_task:"); - hasher.update(task.as_bytes()); + // 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, &job.workspace_id, - &args, + &job.args, &mut hasher, &job.id, job.cache_ignore_s3_path.unwrap_or(false), @@ -1599,30 +1594,23 @@ pub async fn cached_result_path( Ok(format!("g/results/{:064x}", hasher.finalize())) } -/// 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 +/// 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_task_identity( +async fn wac_executing_key( 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 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", - ) - .bind(job.id) - .fetch_optional(db) - .await?; - Ok(match identity { - Some((Some(task), Some(Json(args)))) => Some((task, args)), - _ => 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/wac_executor.rs b/backend/windmill-worker/src/wac_executor.rs index ab86c0e2fa..c188792a5e 100644 --- a/backend/windmill-worker/src/wac_executor.rs +++ b/backend/windmill-worker/src/wac_executor.rs @@ -85,10 +85,6 @@ 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/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index 6eb378060a..fab03923b6 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -4609,9 +4609,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A \`\`task_script\`\` or \`\`task_flow\`\` target +# is keyed on the arguments it is called with. # # Usage:: # @@ -6747,9 +6749,11 @@ 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. */ + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A \`taskScript\` + * or \`taskFlow\` target is keyed on the arguments it is called with. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -6942,9 +6946,11 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A \`\`task_script\`\` or \`\`task_flow\`\` target +# is keyed on the arguments it is called with. # # Usage:: # diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index 2b0371aac6..51c37640c4 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", "fn_id"): + for opt_key in ("timeout", "tag", "cache_ttl", "priority", "concurrent_limit", "concurrency_key", "concurrency_time_window_s"): 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,22 +3285,6 @@ 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, *, @@ -3344,9 +3328,11 @@ def task( 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. + instead of running it again. A task is keyed on its step key (its name and + call order) and the workflow's input, not on the arguments it is called + with, so cache one only when whether it runs, and what it receives, follow + from the workflow's input alone. A ``task_script`` or ``task_flow`` target + is keyed on the arguments it is called with. Usage:: @@ -3377,7 +3363,6 @@ 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) @@ -3403,7 +3388,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=_fn_opts, **merged) + return ctx._next_step(task_name, script, func, _task_options=_task_opts, **merged) # WAC v1: running inside a Windmill job but not in a @workflow if ( diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index c114c77676..e7fd0ebc67 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -2588,9 +2588,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A \`\`task_script\`\` or \`\`task_flow\`\` target +# is keyed on the arguments it is called with. # # Usage:: # @@ -2743,9 +2745,11 @@ 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. */ + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A \`taskScript\` + * or \`taskFlow\` target is keyed on the arguments it is called with. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -2938,9 +2942,11 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A \`\`task_script\`\` or \`\`task_flow\`\` target +# is keyed on the arguments it is called with. # # Usage:: # diff --git a/system_prompts/auto-generated/script.md b/system_prompts/auto-generated/script.md index d75d06abf2..8380464642 100644 --- a/system_prompts/auto-generated/script.md +++ b/system_prompts/auto-generated/script.md @@ -2729,9 +2729,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` or ``task_flow`` target +# is keyed on the arguments it is called with. # # Usage:: # diff --git a/system_prompts/auto-generated/sdks/python.md b/system_prompts/auto-generated/sdks/python.md index cf473d5b27..bf72d8aef9 100644 --- a/system_prompts/auto-generated/sdks/python.md +++ b/system_prompts/auto-generated/sdks/python.md @@ -673,9 +673,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` or ``task_flow`` target +# is keyed on the arguments it is called with. # # Usage:: # diff --git a/system_prompts/auto-generated/sdks/wac-python.md b/system_prompts/auto-generated/sdks/wac-python.md index 3a2dcdfa8f..9014d26764 100644 --- a/system_prompts/auto-generated/sdks/wac-python.md +++ b/system_prompts/auto-generated/sdks/wac-python.md @@ -59,9 +59,11 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` or ``task_flow`` target +# is keyed on the arguments it is called with. # # Usage:: # diff --git a/system_prompts/auto-generated/sdks/wac-typescript.md b/system_prompts/auto-generated/sdks/wac-typescript.md index 4677a6d983..1bba72abca 100644 --- a/system_prompts/auto-generated/sdks/wac-typescript.md +++ b/system_prompts/auto-generated/sdks/wac-typescript.md @@ -35,9 +35,11 @@ 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. */ + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A `taskScript` + * or `taskFlow` target is keyed on the arguments it is called with. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; diff --git a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md index a837dc75d6..5dc388e5f3 100644 --- a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md @@ -858,9 +858,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]] # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` or ``task_flow`` target +# is keyed on the arguments it is called with. # # Usage:: # diff --git a/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md b/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md index 8ca425faca..c44c28a3ec 100644 --- a/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md +++ b/system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md @@ -278,9 +278,11 @@ 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. */ + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A `taskScript` + * or `taskFlow` target is keyed on the arguments it is called with. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -473,9 +475,11 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict # 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. +# instead of running it again. A task is keyed on its step key (its name and +# call order) and the workflow's input, not on the arguments it is called +# with, so cache one only when whether it runs, and what it receives, follow +# from the workflow's input alone. A ``task_script`` or ``task_flow`` target +# is keyed on the arguments it is called with. # # Usage:: # diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 153ae55295..0e6e8e3ab9 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1713,9 +1713,11 @@ 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. */ + * running it again. A task written inline in the workflow is keyed on its + * step key (its name and call order) and the workflow's input, not on the + * arguments it is called with, so cache one only when whether it runs, and + * what it receives, follow from the workflow's input alone. A `taskScript` + * or `taskFlow` target is keyed on the arguments it is called with. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -1853,7 +1855,6 @@ export class WorkflowCtx { args: Record = {}, dispatch_type: string = "inline", options?: TaskOptions, - fnId?: string, ): PromiseLike { this._rethrowSwallowed(); const stepName = name || script || "step"; @@ -1907,7 +1908,6 @@ 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; @@ -2284,18 +2284,6 @@ export async function step( return jsonRoundTrip(await fn()); } -// 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); -} - /** * Wrap an async function as a workflow task. * @@ -2333,7 +2321,6 @@ 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). @@ -2354,7 +2341,7 @@ export function task Promise>( kwargs[`arg${i}`] = args[i]; } } - const stepResult = ctx._nextStep(taskName, script, kwargs, "inline", taskOptions, fnId); + const stepResult = ctx._nextStep(taskName, script, kwargs, "inline", taskOptions); // 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) {