starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-21 00:02:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: key a cached inline task on its step key and the workflow input

Browse Source
This commit is contained in:
Ruben Fiszel
2026-09-15 23:59:40 +02:00
parent ab2b5ef4e9
commit 4585410254
16 changed files with 126 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/bun_jobs.rs
+29 -40
View File
@@ -1257,10 +1257,8 @@ async fn test_bun_wac_task_dispatch_from_flow_script(db: Pool<Postgres>) -> 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<Postgres>) -> anyhow::Result<()> {
use windmill_common::flows::FlowNodeId;
@@ -1268,49 +1266,40 @@ async fn test_bun_wac_inline_task_cache_is_per_task(db: Pool<Postgres>) -> 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<Postgres>,
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(())
}
backend/tests/fixtures/wac_flow_script.sql
Vendored
-20
View File
@@ -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 };
});'
);
backend/windmill-common/src/wac.rs
-10
View File
@@ -40,16 +40,6 @@ pub struct WacCheckpoint {
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
pub _executing_key: Option<String>,
/// 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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
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()
backend/windmill-worker/src/bun_executor.rs
+1 -5
View File
@@ -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))
backend/windmill-worker/src/common.rs
+18 -30
View File
@@ -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<Option<(String, HashMap<String, Box<RawValue>>)>> {
) -> windmill_common::error::Result<Option<String>> {
if job.parent_job.is_none() || job.flow_step_id.is_some() {
return Ok(None);
}
let identity: Option<(Option<String>, Option<Json<HashMap<String, Box<RawValue>>>>)> =
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<Option<String>> = 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")]
backend/windmill-worker/src/wac_executor.rs
-4
View File
@@ -85,10 +85,6 @@ pub struct WacStepDispatch {
pub concurrency_key: Option<String>,
#[serde(default)]
pub concurrency_time_window_s: Option<i32>,
/// 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<String>,
}
fn default_dispatch_type() -> String {
cli/src/guidance/skills.gen.ts
Generated
+15 -9
View File
@@ -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::
#
python-client/wmill/wmill/client.py
+7 -22
View File
@@ -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 (
system_prompts/auto-generated/prompts.ts
+15 -9
View File
@@ -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::
#
system_prompts/auto-generated/script.md
+5 -3
View File
@@ -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::
#
system_prompts/auto-generated/sdks/python.md
+5 -3
View File
@@ -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::
#
system_prompts/auto-generated/sdks/wac-python.md
+5 -3
View File
@@ -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::
#
system_prompts/auto-generated/sdks/wac-typescript.md
+5 -3
View File
@@ -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;
system_prompts/auto-generated/skills/write-script-python3/SKILL.md
+5 -3
View File
@@ -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::
#
system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md
+10 -6
View File
@@ -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::
#
typescript-client/client.ts
+6 -19
View File
@@ -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<string, any> = {},
dispatch_type: string = "inline",
options?: TaskOptions,
fnId?: string,
): PromiseLike<any> {
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<T>(
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<T extends (...args: any[]) => Promise<any>>(
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<T extends (...args: any[]) => Promise<any>>(
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) {