diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 0f9f73ad1d..77bdff64e0 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -1799,6 +1799,10 @@ pub async fn handle_bun_job( format!("argsObjToArr(args)") }; + // Kept comment-free — this string is written out per job. + // `_takePendingSuspend` returns a StepSuspend the body caught and swallowed + // (it is an `Error`), so honour it instead of reporting a `complete` whose + // step never reached the checkpoint. Optional: npm clients may predate it. let wrapper_content = if is_wac_v2 { format!( r#" @@ -1843,6 +1847,10 @@ async function run() {{ try {{ const result = await workflowFn(...argsArr); setWorkflowCtx(null); + const swallowed = ctx._takePendingSuspend?.(); + if (swallowed) {{ + throw swallowed; + }} // Flush any unawaited tasks (e.g. forgotten await on last statement) const trailing = ctx._flushPending(); if (trailing.length > 0) {{ diff --git a/python-client/wmill/tests/test_workflow.py b/python-client/wmill/tests/test_workflow.py index a8c97a31d2..5e49f7f595 100644 --- a/python-client/wmill/tests/test_workflow.py +++ b/python-client/wmill/tests/test_workflow.py @@ -3,7 +3,7 @@ import asyncio import pytest -from wmill.client import WorkflowCtx, _StepSuspend, TaskError, workflow, task, step, sleep, parallel, wait_for_approval, _run_workflow +from wmill.client import WorkflowCtx, _StepSuspend, TaskError, workflow, task, step, sleep, parallel, wait_for_approval, _run_workflow, _run_workflow_async @task @@ -914,6 +914,105 @@ class TestErrorPropagation: assert "step failed" in r["result"]["caught"] +class TestRaisingInlineStepIsCheckpointed: + """A ``step()`` whose body raises must still land in ``completed_steps``. + + Otherwise a workflow that catches the exception and later dispatches a task + replays with ``_executing_key`` set, reaches the unrecorded key, and parks + on the never-resolving future forever. + """ + + MARKER = { + "__wmill_error": True, + "message": "boom", + "step_key": "risky", + "result": {"error": "boom", "type": "ValueError"}, + } + + @staticmethod + def _boom(): + raise ValueError("boom") + + @classmethod + def _wf(cls): + @workflow + async def wf(x: int): + try: + await step("risky", cls._boom) + except Exception: + pass + return await double(x=x) + + return wf + + def test_first_run_emits_error_checkpoint(self): + r = _run_workflow(self._wf(), {}, {"x": 5}) + assert r["type"] == "inline_checkpoint" + assert r["key"] == "risky" + assert r["result"] == self.MARKER + + def test_fast_path_posts_error_and_raises_the_replay_exception(self, monkeypatch): + """The default path: the checkpoint is POSTed and the workflow body gets + the same ``TaskError`` a replay rebuilds from the marker — raising the + original ``ValueError`` here would make ``except ValueError:`` catch on + this run and miss on the next one.""" + for var, val in ( + ("WM_JOB_ID", "job-1"), + ("WM_WORKSPACE", "admins"), + ("BASE_INTERNAL_URL", "http://localhost:8000"), + ("WM_TOKEN", "tok"), + ): + monkeypatch.setenv(var, val) + + posted = [] + + class _StubResponse: + def raise_for_status(self): + pass + + class _StubClient: + async def post(self, url, json=None): + posted.append(json) + return _StubResponse() + + async def aclose(self): + pass + + async def run(): + ctx = WorkflowCtx({}) + ctx._inline_http_client = _StubClient() + with pytest.raises(TaskError, match="boom") as live: + await ctx._run_inline_step("risky", self._boom) + # ...and the replay of that very checkpoint raises the same thing. + replayed = WorkflowCtx({"completed_steps": {"risky": self.MARKER}}) + with pytest.raises(TaskError, match="boom") as replay: + await replayed._run_inline_step("risky", self._boom) + assert type(live.value) is type(replay.value) + assert live.value.args == replay.value.args + assert live.value.result == replay.value.result == self.MARKER["result"] + assert isinstance(live.value.__cause__, ValueError) + + asyncio.run(run()) + assert len(posted) == 1 + assert posted[0]["key"] == "risky" + assert posted[0]["result"] == self.MARKER + + def test_replay_reraises_and_does_not_hang(self): + checkpoint = { + "completed_steps": {"risky": self.MARKER}, + "_executing_key": "double", + } + + async def run(): + return await asyncio.wait_for( + _run_workflow_async(self._wf(), checkpoint, {"x": 5}), timeout=5 + ) + + r = asyncio.run(run()) + assert r["type"] == "complete" + assert r["result"] == 10 + + # ===================================================================== # TASK OPTIONS TESTS # ===================================================================== diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index f18cffc8e7..a4b8e9c476 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -2689,6 +2689,29 @@ class TaskError(Exception): self.result = result +def _step_error_marker(key: str, exc: BaseException) -> dict: + """Serialize a failed ``step()`` body into the ``__wmill_error`` marker that + task failures also use, so it can be stored in ``completed_steps``.""" + return { + "__wmill_error": True, + "message": str(exc), + "step_key": key, + "result": {"error": str(exc), "type": type(exc).__name__}, + } + + +def _step_error_from_marker(marker: dict, name: str) -> TaskError: + """Rebuild the exception a failed step raises. Both the run that produced the + failure and every later replay go through here, so a workflow's ``except`` + clauses see the same type either way.""" + return TaskError( + marker.get("message", f"Step '{name}' failed"), + step_key=marker.get("step_key", ""), + child_job_id=marker.get("child_job_id", ""), + result=marker.get("result"), + ) + + _workflow_ctx: _contextvars.ContextVar["WorkflowCtx"] = _contextvars.ContextVar( "_workflow_ctx" ) @@ -2858,12 +2881,7 @@ class WorkflowCtx: if key in self._completed: val = self._completed[key] if isinstance(val, dict) and val.get("__wmill_error"): - raise TaskError( - val.get("message", f"Step '{name}' failed"), - step_key=val.get("step_key", ""), - child_job_id=val.get("child_job_id", ""), - result=val.get("result"), - ) + raise _step_error_from_marker(val, name) return val if self._executing_key is not None: @@ -2873,9 +2891,18 @@ class WorkflowCtx: started_at = _dt.now(_tz.utc).isoformat() print(f"WM_WAC_STEP: {_json_mod.dumps({'key': key, 'started_at': started_at})}") t0 = _time_mod.monotonic() - result = fn() - if _asyncio.iscoroutine(result): - result = await result + # A raised step still has to reach ``completed_steps``, or a replay with + # ``_executing_key`` set finds nothing recorded and parks forever on the + # ``_asyncio.Future()`` above. ``_StepSuspend`` and ``CancelledError`` are + # ``BaseException``, so they pass through untouched. + step_error: Optional[Exception] = None + try: + result = fn() + if _asyncio.iscoroutine(result): + result = await result + except Exception as _exc: + step_error = _exc + result = _step_error_marker(key, _exc) duration_ms = int((_time_mod.monotonic() - t0) * 1000) # Fast path: POST the delta to the new per-job API endpoint and return @@ -2892,6 +2919,7 @@ class WorkflowCtx: _base = os.environ.get("BASE_INTERNAL_URL") _token = os.environ.get("WM_TOKEN") if _fast_path_enabled and _job_id and _workspace and _base and _token: + _fast_path_ok = False try: if self._inline_lock is None: self._inline_lock = _asyncio.Lock() @@ -2917,7 +2945,7 @@ class WorkflowCtx: }, ) _resp.raise_for_status() - return result + _fast_path_ok = True except Exception as _e: logger.info( "WAC v2 inline fast path failed for key %s, falling back to suspend: %s", @@ -2925,6 +2953,14 @@ class WorkflowCtx: _e, ) # fall through to the legacy suspend path + if _fast_path_ok: + # Raise what a replay would rebuild from the marker, never the + # original: a replay cannot reconstruct the original type, so + # raising it here would make ``except ValueError:`` catch on this + # run and miss on the next. ``__cause__`` is for tracebacks only. + if step_error is not None: + raise _step_error_from_marker(result, name) from step_error + return result raise _StepSuspend({ "mode": "inline_checkpoint", diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 1b5d8473f6..a17e7e361f 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1520,6 +1520,32 @@ export class StepSuspend extends Error { } } +/** Serialize a failed `step()` body into the `__wmill_error` marker that task + * failures also use, so it can be stored in `completed_steps`. */ +function stepErrorMarker(key: string, e: unknown): Record { + const message = e instanceof Error ? e.message : String(e); + // Constructor name, not `e.name`: a `class MyError extends Error {}` that + // never assigns `this.name` reports "Error", which would make the same + // failure read as `MyError` in the python client and `Error` here. + const type = e instanceof Error ? (e.constructor?.name ?? e.name) : typeof e; + return { __wmill_error: true, message, step_key: key, result: { error: message, type } }; +} + +/** Rebuild the error a failed step throws. Both the run that produced the + * failure and every later replay go through here, so a workflow's catch block + * sees the same shape either way. */ +function stepErrorFromMarker(marker: any, name: string): Error { + const err = new Error(marker?.message || `Step '${name}' failed`); + // Matches the python client, which raises TaskError here; the failed body's + // own type stays in `result.type`. Keeps a failed job's serialized error + // identical across the two languages. + err.name = "TaskError"; + (err as any).result = marker?.result; + (err as any).step_key = marker?.step_key; + (err as any).child_job_id = marker?.child_job_id; + return err; +} + export interface TaskOptions { timeout?: number; tag?: string; @@ -1563,6 +1589,11 @@ export class WorkflowCtx { [k: string]: any; }> = []; private _suspended = false; + /** The last suspend this ctx raised. `StepSuspend` is an `Error`, so any + * `catch` in the workflow body swallows it and the run would report a + * `complete` whose step never reached `completed_steps`. Python is immune: + * `_StepSuspend` derives from `BaseException`. */ + private _pendingSuspend: StepSuspend | null = null; /** When set, the task matching this key executes its inner function directly */ _executingKey: string | null; /** Serializes fast-path POSTs across concurrent step() calls within one @@ -1604,12 +1635,14 @@ export class WorkflowCtx { dispatch_type: string = "inline", options?: TaskOptions, ): PromiseLike { + this._rethrowSwallowedSuspend(); const key = this._allocKey(name || script || "step"); if (key in this.completed) { const value = this.completed[key]; if (value && typeof value === "object" && (value as any).__wmill_error) { const err = new Error((value as any).message || `Task '${name}' failed`); + err.name = "TaskError"; (err as any).result = (value as any).result; (err as any).step_key = (value as any).step_key; (err as any).child_job_id = (value as any).child_job_id; @@ -1654,7 +1687,7 @@ export class WorkflowCtx { this.pending = []; const names = steps.map(s => s.name).join(", "); console.log(`\n--- WAC: ${names} ---`); - throw new StepSuspend({ + this._raiseSuspend({ mode: steps.length > 1 ? "parallel" : "sequential", steps, }); @@ -1674,6 +1707,7 @@ export class WorkflowCtx { selfApproval?: boolean; key?: string; }): PromiseLike<{ value: any; approver: string; approved: boolean }> { + this._rethrowSwallowedSuspend(); if (options?.key !== undefined) assertUsableStepKey(options.key, "waitForApproval key"); const key = this._allocKey(options?.key || "approval"); @@ -1700,7 +1734,7 @@ export class WorkflowCtx { // Throw immediately — approval is always a blocking step console.log(`\n--- WAC: approval(${key}) ---`); - throw new StepSuspend({ + this._raiseSuspend({ mode: "approval", key, timeout: options?.timeout ?? 1800, @@ -1711,6 +1745,7 @@ export class WorkflowCtx { } _sleep(seconds: number): PromiseLike { + this._rethrowSwallowedSuspend(); const key = this._allocKey("sleep"); if (key in this.completed) { @@ -1722,7 +1757,7 @@ export class WorkflowCtx { } console.log(`\n--- WAC: sleep(${key}, ${seconds}s) ---`); - throw new StepSuspend({ + this._raiseSuspend({ mode: "sleep", key, seconds: Math.max(1, Math.round(seconds)), @@ -1731,16 +1766,13 @@ export class WorkflowCtx { } async _runInlineStep(name: string, fn: () => T | Promise): Promise { + this._rethrowSwallowedSuspend(); const key = this._allocKey(name || "step"); if (key in this.completed) { const value = this.completed[key]; if (value && typeof value === "object" && (value as any).__wmill_error) { - const err = new Error((value as any).message || `Step '${name}' failed`); - (err as any).result = (value as any).result; - (err as any).step_key = (value as any).step_key; - (err as any).child_job_id = (value as any).child_job_id; - throw err; + throw stepErrorFromMarker(value, name); } return value as T; } @@ -1753,7 +1785,21 @@ export class WorkflowCtx { const startedAt = new Date().toISOString(); console.log(`WM_WAC_STEP: ${JSON.stringify({ key, started_at: startedAt })}`); const t0 = Date.now(); - const result = await fn(); + // A thrown step still has to reach `completed_steps`, or a replay with + // `_executingKey` set finds nothing recorded and parks forever on the + // never-resolving promise above. A nested StepSuspend is control flow, + // not a step failure. + let result: T; + let stepError: unknown; + let errored = false; + try { + result = await fn(); + } catch (e) { + if ((e as any)?.name === "StepSuspend" || e instanceof StepSuspend) throw e; + errored = true; + stepError = e; + result = stepErrorMarker(key, e) as any; + } const durationMs = Date.now() - t0; // Fast path: POST the delta to the new per-job API endpoint and return the @@ -1810,18 +1856,54 @@ export class WorkflowCtx { }); // Swallow chain errors so a past failure does not poison future awaits. this._inlineChain = chainTail.catch(() => {}); + let fastPathOk = false; try { await chainTail; - return result as T; + fastPathOk = true; } catch (e) { console.log( `WAC v2 inline fast path failed for key ${key}, falling back to suspend: ${e}`, ); // fall through to the legacy suspend path below } + if (fastPathOk) { + // Throw what a replay would rebuild from the marker, never the + // original: a replay cannot reconstruct the original type, so throwing + // it here would match `e instanceof TypeError` on this run and miss on + // the next. `cause` is for logging only — absent on replay. + if (errored) { + throw Object.assign(stepErrorFromMarker(result, name), { cause: stepError }); + } + return result as T; + } } - throw new StepSuspend({ mode: "inline_checkpoint", steps: [], key, result, started_at: startedAt, duration_ms: durationMs }); + this._raiseSuspend({ mode: "inline_checkpoint", steps: [], key, result, started_at: startedAt, duration_ms: durationMs }); + } + + /** Raise a suspend, parking it so a body that catches it cannot make it + * vanish. Every suspend raised for this ctx must go through here. */ + _raiseSuspend(dispatchInfo: Record): never { + const suspend = new StepSuspend(dispatchInfo); + this._pendingSuspend = suspend; + throw suspend; + } + + /** Re-throw a swallowed suspend at the next SDK call. It happened before + * whatever the body is doing now, so it wins: the run is unwinding either + * way and everything after it re-runs on the replay. Left set, so a body + * that catches in a loop can't swallow it a second time. */ + private _rethrowSwallowedSuspend(): void { + if (this._pendingSuspend) throw this._pendingSuspend; + } + + /** Hand the runner a suspend the workflow body caught and swallowed, so it is + * honoured instead of silently turning into a `complete`. Returns null when + * the suspend propagated normally. */ + _takePendingSuspend(): StepSuspend | null { + const s = this._pendingSuspend; + this._pendingSuspend = null; + return s; } } @@ -1897,7 +1979,7 @@ export function task Promise>( if ((stepResult as any)?._execute_directly) { return (async () => { const result = await fn(...args); - throw new StepSuspend({ mode: "step_complete", steps: [], result }); + ctx._raiseSuspend({ mode: "step_complete", steps: [], result }); })(); } return stepResult; diff --git a/typescript-client/tests/workflow.test.ts b/typescript-client/tests/workflow.test.ts index a42895ccbb..6a4fed30c4 100644 --- a/typescript-client/tests/workflow.test.ts +++ b/typescript-client/tests/workflow.test.ts @@ -26,8 +26,25 @@ class WorkflowCtx { key: string; }> = []; private _suspended = false; + private _pendingSuspend: StepSuspend | null = null; _executingKey: string | null; + _raiseSuspend(dispatchInfo: Record): never { + const suspend = new StepSuspend(dispatchInfo); + this._pendingSuspend = suspend; + throw suspend; + } + + private _rethrowSwallowedSuspend(): void { + if (this._pendingSuspend) throw this._pendingSuspend; + } + + _takePendingSuspend(): StepSuspend | null { + const s = this._pendingSuspend; + this._pendingSuspend = null; + return s; + } + constructor(checkpoint: Record = {}) { this.completed = checkpoint?.completed_steps ?? {}; this._executingKey = checkpoint?._executing_key ?? null; @@ -43,12 +60,14 @@ class WorkflowCtx { args: Record = {}, options?: Record, ): PromiseLike { + this._rethrowSwallowedSuspend(); const key = this._allocKey(); if (key in this.completed) { const value = this.completed[key]; if (value && typeof value === "object" && (value as any).__wmill_error) { const err = new Error((value as any).message || `Task '${name}' failed`); + err.name = "TaskError"; (err as any).result = (value as any).result; (err as any).step_key = (value as any).step_key; (err as any).child_job_id = (value as any).child_job_id; @@ -79,7 +98,7 @@ class WorkflowCtx { this._suspended = true; const steps = [...this.pending]; this.pending = []; - throw new StepSuspend({ + this._raiseSuspend({ mode: steps.length > 1 ? "parallel" : "sequential", steps, }); @@ -99,6 +118,7 @@ class WorkflowCtx { } _sleep(seconds: number): PromiseLike { + this._rethrowSwallowedSuspend(); const key = this._allocKey(); if (key in this.completed) { return { then: (resolve: any) => resolve(undefined) }; @@ -106,7 +126,7 @@ class WorkflowCtx { if (this._executingKey !== null) { return { then: () => new Promise(() => {}) }; } - throw new StepSuspend({ + this._raiseSuspend({ mode: "sleep", key, seconds: Math.max(1, Math.round(seconds)), @@ -118,12 +138,14 @@ class WorkflowCtx { name: string, fn: () => T | Promise ): Promise { + this._rethrowSwallowedSuspend(); const key = this._allocKey(); if (key in this.completed) { const value = this.completed[key]; if (value && typeof value === "object" && (value as any).__wmill_error) { const err = new Error((value as any).message || `Step '${name}' failed`); + err.name = "TaskError"; (err as any).result = (value as any).result; throw err; } @@ -134,8 +156,25 @@ class WorkflowCtx { return new Promise(() => {}); } - const result = await fn(); - throw new StepSuspend({ + let result: any; + let errored = false; + try { + result = await fn(); + } catch (e: any) { + if (e?.name === "StepSuspend" || e instanceof StepSuspend) throw e; + errored = true; + const message = e instanceof Error ? e.message : String(e); + result = { + __wmill_error: true, + message, + step_key: key, + result: { + error: message, + type: e instanceof Error ? (e.constructor?.name ?? e.name) : typeof e, + }, + }; + } + this._raiseSuspend({ mode: "inline_checkpoint", steps: [], key, @@ -193,7 +232,7 @@ function task Promise>( if ((stepResult as any)?._execute_directly) { return (async () => { const result = await fn(...args); - throw new StepSuspend({ + ctx._raiseSuspend({ mode: "step_complete", steps: [], result, @@ -263,6 +302,9 @@ async function runWorkflow( _workflowCtx = ctx; try { const result = await fn(...args); + // Mirrors bun_executor.rs: honour a suspend the body caught and swallowed. + const swallowed = ctx._takePendingSuspend?.(); + if (swallowed) throw swallowed; // Flush unawaited tasks const pending = ctx._flushPending(); if (pending.length > 0) { @@ -1231,6 +1273,7 @@ describe("error propagation via __wmill_error marker", () => { await runWorkflow(wf, checkpoint, [5]); expect(true).toBe(false); // should not reach here } catch (e: any) { + expect(e.name).toBe("TaskError"); expect(e.message).toContain("double"); expect(e.result).toEqual({ message: "division by zero" }); expect(e.child_job_id).toBe("abc-123"); @@ -1343,6 +1386,128 @@ describe("error propagation via __wmill_error marker", () => { }); }); +// A step() whose body throws must still land in completed_steps. Otherwise a +// workflow that catches the error and later dispatches a task replays with +// _executingKey set, reaches the unrecorded key, and parks on the +// never-resolving promise forever. +describe("throwing inline step is checkpointed", () => { + const marker = { + __wmill_error: true, + message: "boom", + step_key: "step_0", + result: { error: "boom", type: "TypeError" }, + }; + + // The workflow body catches — the shape a failing step is written for, and + // the one that makes StepSuspend (an Error) swallowable in TS. + const catchingWf = () => + workflow(async (x: number) => { + let caught = null; + try { + await step("risky", () => { + throw new TypeError("boom"); + }); + } catch (e: any) { + caught = `${e.name}: ${e.message}`; + } + return { caught, doubled: await double(x) }; + }); + + test("a throwing step suspends with an error checkpoint", async () => { + const ctx = new WorkflowCtx({}); + let caught: any; + try { + await ctx._runInlineStep("risky", () => { + throw new TypeError("boom"); + }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(StepSuspend); + expect(caught.dispatchInfo.mode).toBe("inline_checkpoint"); + expect(caught.dispatchInfo.key).toBe("step_0"); + expect(caught.dispatchInfo.result).toEqual(marker); + }); + + test("a swallowed suspend still reaches the runner", async () => { + // Without _pendingSuspend the catch eats the suspend and the run reports a + // dispatch (or a complete) with `risky` missing from completed_steps. + const result = await runWorkflow(catchingWf(), {}, [5]); + expect(result.type).toBe("inline_checkpoint"); + expect(result.key).toBe("step_0"); + expect(result.result).toEqual(marker); + }); + + test("a swallowed suspend from a succeeding step still reaches the runner", async () => { + const wf = workflow(async () => { + try { + await step("fine", () => 42); + } catch { + // a body that catches broadly must not be able to erase the suspend + } + return "never reached on the first run"; + }); + const result = await runWorkflow(wf, {}, []); + expect(result.type).toBe("inline_checkpoint"); + expect(result.result).toBe(42); + }); + + test("a replayed step failure is named TaskError, like the python client", async () => { + const ctx = new WorkflowCtx({ completed_steps: { step_0: marker } }); + let caught: any; + try { + await ctx._runInlineStep("risky", () => 1); + } catch (e) { + caught = e; + } + expect(`${caught.name}: ${caught.message}`).toBe("TaskError: boom"); + // the failing body's own type stays addressable here + expect(caught.result).toEqual({ error: "boom", type: "TypeError" }); + }); + + test("a child job cannot swallow its own completion signal", async () => { + // The catch below is reached only if step_complete escapes the parking + // mechanism; the child would then report the catch branch as the result. + const wf = workflow(async (x: number) => { + try { + await double(x); + } catch { + return "swallowed"; + } + return "unreachable"; + }); + const result = await runWorkflow(wf, { _executing_key: "step_0" }, [5]); + expect(result.type).toBe("complete"); + expect(result.result).toBe(10); + }); + + test("a swallowed suspend from a task dispatch still reaches the runner", async () => { + const wf = workflow(async (x: number) => { + try { + await double(x); + } catch { + // ditto for task steps + } + return "never reached on the first run"; + }); + const result = await runWorkflow(wf, {}, [5]); + expect(result.type).toBe("dispatch"); + expect(result.steps[0].key).toBe("step_0"); + }); + + test("replay rethrows the error and does not hang", async () => { + const result = await runWorkflow( + catchingWf(), + { completed_steps: { step_0: marker }, _executing_key: "step_1" }, + [5], + ); + // The child runs only the dispatched task, so its result is that task's — + // what matters is that it got there instead of parking on `risky`. + expect(result.type).toBe("complete"); + expect(result.result).toBe(10); + }); +}); + // ===================================================================== // TASK OPTIONS TESTS // =====================================================================