diff --git a/backend/tests/wac_task_cache_key.rs b/backend/tests/wac_task_cache_key.rs index 0cf04816d2..f871b35c6e 100644 --- a/backend/tests/wac_task_cache_key.rs +++ b/backend/tests/wac_task_cache_key.rs @@ -104,9 +104,15 @@ async fn a_task_child_is_cached_under_its_fingerprint_or_else_its_step_key( "two tasks called at one position: the fingerprint tells them apart" ); assert_eq!( + fingerprinted, + child_cache_path(&db, parent_args(), "step", Some("f1"), 1).await?, + "one task at one step with one set of arguments is one entry" + ); + assert_ne!( fingerprinted, child_cache_path(&db, parent_args(), "step_2", Some("f1"), 1).await?, - "one task called at two positions keeps one identity" + "the step key stays in the key: a fingerprint cannot separate two bound \ + functions of one name, or two lambdas sharing a source line" ); assert_eq!( fingerprinted, diff --git a/backend/windmill-common/src/wac.rs b/backend/windmill-common/src/wac.rs index 5e06e650e7..6b49301fb3 100644 --- a/backend/windmill-common/src/wac.rs +++ b/backend/windmill-common/src/wac.rs @@ -41,9 +41,9 @@ pub struct WacCheckpoint { #[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. - /// A child whose SDK sent no fingerprint is keyed on its step key and its - /// parent's arguments instead (`cached_result_path`). + /// arguments it was called with. A cached result is keyed on all three, or, + /// for a child whose SDK sent no fingerprint, on the step key and the + /// parent's arguments (`cached_result_path`). #[serde(skip_serializing_if = "Option::is_none")] #[serde(default)] pub _executing_fn: Option, diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 9d01cf8b07..ab086b7da6 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1575,13 +1575,18 @@ pub async fn cached_result_path( } } // A task child runs its parent's code with its parent's arguments. With the SDK's - // fingerprint it is keyed on that and its own call arguments. Without one it is keyed - // on its step key and the parent's arguments: a step key names a position, not a - // task, and only the parent's arguments tell apart tasks a branch puts at one. + // fingerprint it is keyed on that, the step it runs as and its own call arguments. + // Without one, on its step key and the parent's arguments, which is what tells apart + // tasks a branch puts at one step. let args = match wac_task_identity(db, job).await? { - Some(WacTaskIdentity::Fingerprint { fn_id, args }) => { + Some(WacTaskIdentity::Fingerprint { key, fn_id, args }) => { + // The step key stays in: a fingerprint cannot separate two tasks whose + // difference it never sees, such as two bound functions of one name or + // two lambdas sharing a source line, and the step key can. hasher.update(b"wac_fn:"); hasher.update(fn_id.as_bytes()); + hasher.update(b"@"); + hasher.update(key.as_bytes()); Some(Json(args)) } Some(WacTaskIdentity::StepKey(key)) => { @@ -1607,7 +1612,7 @@ pub async fn cached_result_path( /// What a workflow-as-code parent seeded in a task child's checkpoint at push time /// to key the child's cached result on. enum WacTaskIdentity { - Fingerprint { fn_id: String, args: HashMap> }, + Fingerprint { key: String, fn_id: String, args: HashMap> }, StepKey(String), } @@ -1633,8 +1638,8 @@ async fn wac_task_identity( .fetch_optional(db) .await?; Ok(match identity { - Some((Some(fn_id), _, Some(Json(args)))) => { - Some(WacTaskIdentity::Fingerprint { fn_id, args }) + Some((Some(fn_id), Some(key), Some(Json(args)))) => { + Some(WacTaskIdentity::Fingerprint { key, fn_id, args }) } Some((_, Some(key), _)) => Some(WacTaskIdentity::StepKey(key)), _ => None, diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index 1dc4bea8c1..acf55574a9 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -4609,10 +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. It has no effect on a -# \`\`task_flow\`\` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a \`\`task_flow\`\` target, which keeps its +# flow's own cache policy. # # Usage:: # @@ -6748,10 +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. It has no effect on a \`taskFlow\` target, which - * keeps its flow's own cache policy. */ + * running it again. The result is keyed on the task, the step it runs as and + * the arguments it is called with, so anything a cached task reads from its + * closure, the receiver of a bound method included, must be passed in as an + * argument. It has no effect on a \`taskFlow\` target, which keeps its flow's + * own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -6944,10 +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. It has no effect on a -# \`\`task_flow\`\` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a \`\`task_flow\`\` target, which keeps its +# flow's own cache policy. # # Usage:: # diff --git a/python-client/wmill/tests/test_workflow.py b/python-client/wmill/tests/test_workflow.py index 3502cb9d47..07059fcf91 100644 --- a/python-client/wmill/tests/test_workflow.py +++ b/python-client/wmill/tests/test_workflow.py @@ -1863,3 +1863,28 @@ class TestTaskFingerprint: ) assert [s["key"] for s in second["steps"]] == ["add_one", "add_one_2"] assert second["steps"][0]["fn_id"] != first["steps"][0]["fn_id"] + + def test_a_builtin_task_still_decorates_and_dispatches(self): + """The fingerprint is taken for every task, cached or not, so a callable + with neither source nor code object must not break the decorator.""" + builtin_task = task(pow) + + @workflow + async def wf(): + return await builtin_task(2, 3) + + result = _run_workflow(wf, {}, {}) + assert result["type"] == "dispatch" + assert result["steps"][0]["key"] == "pow" + + def test_two_lambdas_on_one_line_are_told_apart(self): + """``inspect.getsource`` gives each the whole line, so the code's shape is + what separates them, and it must not depend on where the line sits.""" + first, second = task(lambda x: x + 1), task(lambda x: x + 2) + + @workflow + async def wf(): + return await asyncio.gather(first(x=1), second(x=1)) + + result = _run_workflow(wf, {}, {}) + assert result["steps"][0]["fn_id"] != result["steps"][1]["fn_id"] diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index a5ee1bb151..eb3d524b84 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -3291,14 +3291,23 @@ def _fn_fingerprint(func) -> str: tasks called at the same position, so neither can tell them apart.""" import hashlib import inspect - import marshal + # Runs for every task at decoration, cached or not, so it must never raise: a + # builtin has neither source nor code object. Source alone cannot separate two + # lambdas written on one line, so the code's shape goes in as well, without the + # line numbers that would move whenever the file is edited above it. + parts = [] 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() + parts.append(inspect.getsource(func).encode()) + except Exception: + pass + code = getattr(func, "__code__", None) + if code is not None: + consts = tuple(c for c in code.co_consts if not isinstance(c, type(code))) + parts.append(repr((code.co_code, code.co_names, code.co_varnames, consts)).encode()) + if not parts: + parts.append(repr(func).encode()) + return hashlib.sha256(b"\x1f".join(parts)).hexdigest() def task( @@ -3344,10 +3353,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. It has no effect on a - ``task_flow`` target, which keeps its flow's own cache policy. + instead of running it again. The result is keyed on the task, the step it + runs as and the arguments it is called with, so anything a cached task reads + from its closure, the receiver of a bound method included, must be passed in + as an argument. It has no effect on a ``task_flow`` target, which keeps its + flow's own cache policy. Usage:: diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index e2cb1c93cb..8676d7e29e 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -2588,10 +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. It has no effect on a -# \`\`task_flow\`\` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a \`\`task_flow\`\` target, which keeps its +# flow's own cache policy. # # Usage:: # @@ -2744,10 +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. It has no effect on a \`taskFlow\` target, which - * keeps its flow's own cache policy. */ + * running it again. The result is keyed on the task, the step it runs as and + * the arguments it is called with, so anything a cached task reads from its + * closure, the receiver of a bound method included, must be passed in as an + * argument. It has no effect on a \`taskFlow\` target, which keeps its flow's + * own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -2940,10 +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. It has no effect on a -# \`\`task_flow\`\` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a \`\`task_flow\`\` target, which keeps its +# flow's own cache policy. # # Usage:: # diff --git a/system_prompts/auto-generated/script.md b/system_prompts/auto-generated/script.md index e4171a7150..87ed3d8c6a 100644 --- a/system_prompts/auto-generated/script.md +++ b/system_prompts/auto-generated/script.md @@ -2729,10 +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. It has no effect on a -# ``task_flow`` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a ``task_flow`` target, which keeps its +# flow's own cache policy. # # Usage:: # diff --git a/system_prompts/auto-generated/sdks/python.md b/system_prompts/auto-generated/sdks/python.md index c3b34b69db..9b97f7c9f0 100644 --- a/system_prompts/auto-generated/sdks/python.md +++ b/system_prompts/auto-generated/sdks/python.md @@ -673,10 +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. It has no effect on a -# ``task_flow`` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a ``task_flow`` target, which keeps its +# flow's own cache policy. # # Usage:: # diff --git a/system_prompts/auto-generated/sdks/wac-python.md b/system_prompts/auto-generated/sdks/wac-python.md index 2e54c5bd06..a199c9d792 100644 --- a/system_prompts/auto-generated/sdks/wac-python.md +++ b/system_prompts/auto-generated/sdks/wac-python.md @@ -59,10 +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. It has no effect on a -# ``task_flow`` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a ``task_flow`` target, which keeps its +# flow's own cache policy. # # Usage:: # diff --git a/system_prompts/auto-generated/sdks/wac-typescript.md b/system_prompts/auto-generated/sdks/wac-typescript.md index cc59e74954..fc0dcc7241 100644 --- a/system_prompts/auto-generated/sdks/wac-typescript.md +++ b/system_prompts/auto-generated/sdks/wac-typescript.md @@ -35,10 +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. It has no effect on a `taskFlow` target, which - * keeps its flow's own cache policy. */ + * running it again. The result is keyed on the task, the step it runs as and + * the arguments it is called with, so anything a cached task reads from its + * closure, the receiver of a bound method included, must be passed in as an + * argument. It has no effect on a `taskFlow` target, which keeps its flow's + * own cache policy. */ 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 f6a2bbe70e..b82af4a0c0 100644 --- a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md @@ -858,10 +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. It has no effect on a -# ``task_flow`` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a ``task_flow`` target, which keeps its +# flow's own cache policy. # # 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 f30acc0e47..2d727c302d 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,10 +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. It has no effect on a `taskFlow` target, which - * keeps its flow's own cache policy. */ + * running it again. The result is keyed on the task, the step it runs as and + * the arguments it is called with, so anything a cached task reads from its + * closure, the receiver of a bound method included, must be passed in as an + * argument. It has no effect on a `taskFlow` target, which keeps its flow's + * own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; @@ -474,10 +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. It has no effect on a -# ``task_flow`` target, which keeps its flow's own cache policy. +# instead of running it again. The result is keyed on the task, the step it +# runs as and the arguments it is called with, so anything a cached task reads +# from its closure, the receiver of a bound method included, must be passed in +# as an argument. It has no effect on a ``task_flow`` target, which keeps its +# flow's own cache policy. # # Usage:: # diff --git a/typescript-client/client.ts b/typescript-client/client.ts index faf82e8e42..f6c9f50d0e 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1713,10 +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. It has no effect on a `taskFlow` target, which - * keeps its flow's own cache policy. */ + * running it again. The result is keyed on the task, the step it runs as and + * the arguments it is called with, so anything a cached task reads from its + * closure, the receiver of a bound method included, must be passed in as an + * argument. It has no effect on a `taskFlow` target, which keeps its flow's + * own cache policy. */ cache_ttl?: number; priority?: number; concurrency_limit?: number; diff --git a/typescript-client/tests/workflow_task_identity.test.ts b/typescript-client/tests/workflow_task_identity.test.ts index 5f4f644eba..2cbfa9c1c9 100644 --- a/typescript-client/tests/workflow_task_identity.test.ts +++ b/typescript-client/tests/workflow_task_identity.test.ts @@ -68,7 +68,25 @@ describe("task fingerprint", () => { expect(a.fn_id).not.toBe(b.fn_id); }); - test("one task keeps its fingerprint across positions", async () => { + test("two bound methods of one name share a fingerprint, and their step keys separate them", async () => { + const a = { + async read() { + return "A"; + }, + }; + const b = { + async read() { + return "B"; + }, + }; + const steps = await dispatched(async () => { + await Promise.all([task(a.read.bind(a))(), task(b.read.bind(b))()]); + }); + expect(steps[0].fn_id).toBe(steps[1].fn_id); + expect(steps.map((s) => s.key)).toEqual(["bound read", "bound read_2"]); + }); + + test("one task carries one fingerprint at every position", async () => { const double = task(async (n: number) => n * 2, { cache_ttl: 60 }); const steps = await dispatched(async () => { await Promise.all([double(1), double(2)]);