From b0d71bdcac9670d0fd9564ee18fbb4b26fa220e3 Mon Sep 17 00:00:00 2001 From: vudc Date: Mon, 18 May 2026 23:35:53 +0700 Subject: [PATCH 1/6] feat: add resume workflow run functionality and UI components --- .../src/components/run-resume-action.tsx | 110 +++++++++++++++++ apps/dashboard/src/lib/api.ts | 12 ++ apps/dashboard/src/lib/status.ts | 9 ++ apps/dashboard/src/routes/runs/$runId.tsx | 10 +- openworkflow/flaky-payment.run.ts | 32 +++++ openworkflow/flaky-payment.ts | 95 +++++++++++++++ packages/openworkflow/client/client.ts | 17 +++ packages/openworkflow/core/backend.ts | 7 ++ packages/openworkflow/core/workflow-run.ts | 24 ++++ packages/openworkflow/postgres/backend.ts | 46 +++++++ packages/openworkflow/sqlite/backend.ts | 67 +++++++++++ .../openworkflow/worker/execution.test.ts | 113 ++++++++++++++++++ 12 files changed, 541 insertions(+), 1 deletion(-) create mode 100644 apps/dashboard/src/components/run-resume-action.tsx create mode 100644 openworkflow/flaky-payment.run.ts create mode 100644 openworkflow/flaky-payment.ts diff --git a/apps/dashboard/src/components/run-resume-action.tsx b/apps/dashboard/src/components/run-resume-action.tsx new file mode 100644 index 00000000..452d4c20 --- /dev/null +++ b/apps/dashboard/src/components/run-resume-action.tsx @@ -0,0 +1,110 @@ +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, +} from "@/components/ui/alert-dialog"; +import { Button } from "@/components/ui/button"; +import { resumeWorkflowRunServerFn } from "@/lib/api"; +import { isRunResumableStatus } from "@/lib/status"; +import type { WorkflowRunStatus } from "openworkflow/internal"; +import { useState } from "react"; + +interface RunResumeActionProps { + runId: string; + status: WorkflowRunStatus; + onResumed?: (() => Promise) | (() => void); +} + +function getErrorMessage(error: unknown): string { + if (error instanceof Error && error.message) { + return error.message; + } + + return "Unable to resume workflow run"; +} + +export function RunResumeAction({ + runId, + status, + onResumed, +}: RunResumeActionProps) { + const [isOpen, setIsOpen] = useState(false); + const [isResuming, setIsResuming] = useState(false); + const [error, setError] = useState(null); + + if (!isRunResumableStatus(status)) { + return null; + } + + async function resumeRun() { + setIsResuming(true); + setError(null); + + try { + await resumeWorkflowRunServerFn({ + data: { + workflowRunId: runId, + }, + }); + await onResumed?.(); + setIsOpen(false); + } catch (caughtError) { + setError(getErrorMessage(caughtError)); + } finally { + setIsResuming(false); + } + } + + return ( + { + setIsOpen(nextOpen); + if (!nextOpen) { + setError(null); + } + }} + > + + + + + Resume this failed run? + + Completed steps stay cached and won't re-run. The failing step will + be retried with a fresh retry budget. Previous failed attempts will + be discarded. + + + + {error &&

{error}

} + + + Cancel + { + void resumeRun(); + }} + disabled={isResuming} + > + {isResuming ? "Resuming..." : "Resume Run"} + + + + + ); +} diff --git a/apps/dashboard/src/lib/api.ts b/apps/dashboard/src/lib/api.ts index d399ef81..e604f7ee 100644 --- a/apps/dashboard/src/lib/api.ts +++ b/apps/dashboard/src/lib/api.ts @@ -90,6 +90,18 @@ export const cancelWorkflowRunServerFn = createServerFn({ method: "POST" }) return backend.cancelWorkflowRun({ workflowRunId: data.workflowRunId }); }); +/** + * Resume a failed workflow run by ID. Flips the run back to `pending` and + * drops failed step attempts so the failing step starts with a fresh retry + * budget; completed steps stay cached. + */ +export const resumeWorkflowRunServerFn = createServerFn({ method: "POST" }) + .inputValidator(z.object({ workflowRunId: z.string() })) + .handler(async ({ data }): Promise => { + const backend = await getBackend(); + return backend.resumeWorkflowRun({ workflowRunId: data.workflowRunId }); + }); + /** * List step attempts for a workflow run. */ diff --git a/apps/dashboard/src/lib/status.ts b/apps/dashboard/src/lib/status.ts index af762394..be95d797 100644 --- a/apps/dashboard/src/lib/status.ts +++ b/apps/dashboard/src/lib/status.ts @@ -169,6 +169,11 @@ const CANCELABLE_RUN_STATUSES: ReadonlySet = new Set([ "sleeping", ]); +/** Run statuses that can be resumed from the dashboard. */ +const RESUMABLE_RUN_STATUSES: ReadonlySet = new Set([ + "failed", +]); + const fallbackStatusConfig = STATUS_CONFIG.pending; export function getRunStatusConfig(status: string): StatusConfig { @@ -198,3 +203,7 @@ export function getStatusStatIconClass(status: string): string { export function isRunCancelableStatus(status: string): boolean { return CANCELABLE_RUN_STATUSES.has(status as WorkflowRunStatus); } + +export function isRunResumableStatus(status: string): boolean { + return RESUMABLE_RUN_STATUSES.has(status as WorkflowRunStatus); +} diff --git a/apps/dashboard/src/routes/runs/$runId.tsx b/apps/dashboard/src/routes/runs/$runId.tsx index e9bf8115..288ba772 100644 --- a/apps/dashboard/src/routes/runs/$runId.tsx +++ b/apps/dashboard/src/routes/runs/$runId.tsx @@ -1,6 +1,7 @@ import { AppLayout } from "@/components/app-layout"; import { CursorPaginationControls } from "@/components/cursor-pagination-controls"; import { RunCancelAction } from "@/components/run-cancel-action"; +import { RunResumeAction } from "@/components/run-resume-action"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { Card } from "@/components/ui/card"; @@ -292,7 +293,14 @@ function RunDetailsPage() { /> )} -
+
+ { + await router.invalidate(); + }} + /> workflow marked failed."); +console.log( + "Then open http://localhost:3000, click 'Resume Run', and watch it complete.\n", +); + +const handle = await ow.runWorkflow(flakyPayment.spec, { + cartId: "cart_demo_1", + amountCents: 4200, +}); + +console.log(`Run id: ${handle.workflowRun.id}`); +console.log("Waiting for first terminal state..."); + +try { + const result = await handle.result(); + console.log(`Workflow completed: ${JSON.stringify(result, null, 2)}`); +} catch (error) { + console.log("\nWorkflow failed (as expected on first pass):"); + console.log(error instanceof Error ? error.message : String(error)); + console.log( + "\nNow click 'Resume Run' on the run detail page in the dashboard.", + ); + console.log( + "Leave the worker running so the in-memory attempt counter persists.", + ); +} + +await backend.stop(); diff --git a/openworkflow/flaky-payment.ts b/openworkflow/flaky-payment.ts new file mode 100644 index 00000000..d4d719ec --- /dev/null +++ b/openworkflow/flaky-payment.ts @@ -0,0 +1,95 @@ +import { defineWorkflow } from "openworkflow"; + +interface FlakyPaymentInput { + cartId: string; + amountCents: number; +} + +interface FlakyPaymentOutput { + cartId: string; + authorizationId: string; + receiptId: string; + attemptsForReserve: number; +} + +const RESERVE_MAX_ATTEMPTS = 3; + +// Module-scoped counter. Persists across the failure-then-resume cycle as long +// as the worker process is alive, so the step succeeds on the first attempt +// after `ow.resumeWorkflowRun` is called. +let reserveAttempt = 0; + +/** + * Demo workflow for the Resume feature. + * + * Flow: + * 1. First run: "reserve-funds" throws on every attempt; the step's retry + * budget (RESERVE_MAX_ATTEMPTS) is exhausted, so the workflow run ends in + * `failed`. The downstream steps never run. + * 2. Click "Resume Run" in the dashboard (or call `ow.resumeWorkflowRun(id)`). + * The failed step_attempt rows are dropped and the run is requeued. + * 3. On the next worker tick, "validate-cart" is served from the cache (not + * re-executed); "reserve-funds" runs once more, the counter is now past + * RESERVE_MAX_ATTEMPTS so it returns successfully, and "confirm-payment" + * plus "send-receipt" proceed to completion. + * + * Note: this relies on `reserveAttempt` persisting in the worker process. If + * you restart the worker between the failure and the resume, the counter + * resets, so resume will fail again and you'll need to resume once more. + */ +export const flakyPayment = defineWorkflow( + { name: "flaky-payment" }, + async ({ input, step, run }) => { + console.log(`[run ${run.id}] flaky-payment for cart ${input.cartId}`); + + await step.run({ name: "validate-cart" }, () => { + if (input.amountCents <= 0) { + throw new Error("amountCents must be positive"); + } + }); + + const { authorizationId, attempts } = await step.run( + { + name: "reserve-funds", + retryPolicy: { + maximumAttempts: RESERVE_MAX_ATTEMPTS, + initialInterval: "500ms", + }, + }, + () => { + reserveAttempt++; + console.log(`reserve-funds attempt ${String(reserveAttempt)}`); + + if (reserveAttempt <= RESERVE_MAX_ATTEMPTS) { + throw new Error( + `simulated upstream 503 (attempt ${String(reserveAttempt)})`, + ); + } + + console.log( + `reserve-funds recovered on attempt ${String(reserveAttempt)} (after resume)`, + ); + return { + authorizationId: `auth_${input.cartId}_${String(Date.now())}`, + attempts: reserveAttempt, + }; + }, + ); + + const receiptId = await step.run({ name: "confirm-payment" }, () => { + console.log(`confirming with ${authorizationId}`); + return `rcpt_${input.cartId}`; + }); + + await step.run({ name: "send-receipt" }, () => { + console.log(`receipt ${receiptId} mailed`); + }); + + return { + cartId: input.cartId, + authorizationId, + receiptId, + attemptsForReserve: attempts, + }; + }, +); diff --git a/packages/openworkflow/client/client.ts b/packages/openworkflow/client/client.ts index e5f1f9fd..ac3f7930 100644 --- a/packages/openworkflow/client/client.ts +++ b/packages/openworkflow/client/client.ts @@ -175,6 +175,23 @@ export class OpenWorkflow { await this.backend.cancelWorkflowRun({ workflowRunId }); } + /** + * Resume a failed workflow run. The run's status flips back to `pending` + * so the next worker tick picks it up. Already-completed steps are served + * from history without re-executing; failed step attempts are discarded so + * the failing step starts with a fresh retry budget. + * @param workflowRunId - The ID of the failed workflow run to resume + * @returns Promise + * @throws {Error} If the run does not exist or is not in `failed` status + * @example + * ```ts + * await ow.resumeWorkflowRun("123"); + * ``` + */ + async resumeWorkflowRun(workflowRunId: string): Promise { + await this.backend.resumeWorkflowRun({ workflowRunId }); + } + /** * Send a signal to all workflows currently waiting on the given signal * string. If no workflow is waiting, the signal is silently dropped. diff --git a/packages/openworkflow/core/backend.ts b/packages/openworkflow/core/backend.ts index efc4b07f..ad59acb4 100644 --- a/packages/openworkflow/core/backend.ts +++ b/packages/openworkflow/core/backend.ts @@ -47,6 +47,9 @@ export interface Backend { cancelWorkflowRun( params: Readonly, ): Promise; + resumeWorkflowRun( + params: Readonly, + ): Promise; // Step Attempts createStepAttempt( @@ -143,6 +146,10 @@ export interface CancelWorkflowRunParams { workflowRunId: string; } +export interface ResumeWorkflowRunParams { + workflowRunId: string; +} + export interface CreateStepAttemptParams { workflowRunId: string; workerId: string; diff --git a/packages/openworkflow/core/workflow-run.ts b/packages/openworkflow/core/workflow-run.ts index ae967f3c..942c3b8a 100644 --- a/packages/openworkflow/core/workflow-run.ts +++ b/packages/openworkflow/core/workflow-run.ts @@ -58,6 +58,30 @@ export function resolveCancelWorkflowRunConflict( throw new Error("Failed to cancel workflow run"); } +/** + * Resolve the outcome when a resumeWorkflowRun UPDATE affected no rows. The + * UPDATE is gated on `status = 'failed'`, so a zero-row outcome means either + * the run doesn't exist or it isn't in a resumable state. + * @param workflowRunId - ID of the workflow run (used in error messages) + * @param existing - Current workflow run, or null if not found + * @returns Never; always throws describing why the resume is impossible + * @throws {Error} If the run does not exist or is not in `failed` status + */ +export function resolveResumeWorkflowRunConflict( + workflowRunId: string, + existing: Readonly | null, +): never { + if (!existing) { + // eslint-disable-next-line functional/no-throw-statements + throw new Error(`Workflow run ${workflowRunId} does not exist`); + } + + // eslint-disable-next-line functional/no-throw-statements + throw new Error( + `Cannot resume workflow run ${workflowRunId} with status ${existing.status}; only failed runs can be resumed`, + ); +} + /** * WorkflowRun represents a single execution instance of a workflow. */ diff --git a/packages/openworkflow/postgres/backend.ts b/packages/openworkflow/postgres/backend.ts index 77292a78..e4d128ac 100644 --- a/packages/openworkflow/postgres/backend.ts +++ b/packages/openworkflow/postgres/backend.ts @@ -5,6 +5,7 @@ import { Backend, WorkflowRunCounts, CancelWorkflowRunParams, + ResumeWorkflowRunParams, ClaimWorkflowRunParams, CreateStepAttemptParams, CreateWorkflowRunParams, @@ -37,6 +38,7 @@ import { StepAttempt } from "../core/step-attempt.js"; import { computeFailedWorkflowRunUpdate } from "../core/workflow-definition.js"; import { resolveCancelWorkflowRunConflict, + resolveResumeWorkflowRunConflict, WorkflowRun, } from "../core/workflow-run.js"; import { @@ -763,6 +765,50 @@ export class BackendPostgres implements Backend { return updated; } + async resumeWorkflowRun( + params: ResumeWorkflowRunParams, + ): Promise { + return await this.pg.begin(async (sql): Promise => { + const tx = sql as unknown as Postgres; + const workflowRunsTable = this.workflowRunsTable(tx); + const stepAttemptsTable = this.stepAttemptsTable(tx); + + const [updated] = await tx` + UPDATE ${workflowRunsTable} + SET + "status" = 'pending', + "worker_id" = NULL, + "error" = NULL, + "finished_at" = NULL, + "available_at" = NOW(), + "updated_at" = NOW() + WHERE "namespace_id" = ${this.namespaceId} + AND "id" = ${params.workflowRunId} + AND "status" = 'failed' + RETURNING * + `; + + if (!updated) { + const existing = await this.getWorkflowRun({ + workflowRunId: params.workflowRunId, + }); + resolveResumeWorkflowRunConflict(params.workflowRunId, existing); + } + + // Drop the prior failed attempts so the next worker pass starts the + // failed step with a fresh retry budget and the existing completed + // attempts remain in cache (replay skips re-execution). + await tx` + DELETE FROM ${stepAttemptsTable} + WHERE "namespace_id" = ${this.namespaceId} + AND "workflow_run_id" = ${params.workflowRunId} + AND "status" = 'failed' + `; + + return updated; + }); + } + private async wakeParentWorkflowRun( childWorkflowRun: Readonly, ): Promise { diff --git a/packages/openworkflow/sqlite/backend.ts b/packages/openworkflow/sqlite/backend.ts index fa610357..fa3005a0 100644 --- a/packages/openworkflow/sqlite/backend.ts +++ b/packages/openworkflow/sqlite/backend.ts @@ -4,6 +4,7 @@ import { DEFAULT_RUN_IDEMPOTENCY_PERIOD_MS, Backend, CancelWorkflowRunParams, + ResumeWorkflowRunParams, ClaimWorkflowRunParams, CreateStepAttemptParams, CreateWorkflowRunParams, @@ -37,6 +38,7 @@ import { StepAttempt } from "../core/step-attempt.js"; import { computeFailedWorkflowRunUpdate } from "../core/workflow-definition.js"; import { resolveCancelWorkflowRunConflict, + resolveResumeWorkflowRunConflict, WorkflowRun, } from "../core/workflow-run.js"; import { @@ -742,6 +744,71 @@ export class BackendSqlite implements Backend { return updated; } + async resumeWorkflowRun( + params: ResumeWorkflowRunParams, + ): Promise { + const currentTime = now(); + + try { + this.db.exec("BEGIN IMMEDIATE"); + + const updateStmt = this.db.prepare(` + UPDATE "workflow_runs" + SET + "status" = 'pending', + "worker_id" = NULL, + "error" = NULL, + "finished_at" = NULL, + "available_at" = ?, + "updated_at" = ? + WHERE "namespace_id" = ? + AND "id" = ? + AND "status" = 'failed' + `); + + const updateResult = updateStmt.run( + currentTime, + currentTime, + this.namespaceId, + params.workflowRunId, + ); + + if (updateResult.changes === 0) { + this.db.exec("ROLLBACK"); + const existing = await this.getWorkflowRun({ + workflowRunId: params.workflowRunId, + }); + resolveResumeWorkflowRunConflict(params.workflowRunId, existing); + } + + this.db + .prepare( + ` + DELETE FROM "step_attempts" + WHERE "namespace_id" = ? + AND "workflow_run_id" = ? + AND "status" = 'failed' + `, + ) + .run(this.namespaceId, params.workflowRunId); + + this.db.exec("COMMIT"); + } catch (error) { + try { + this.db.exec("ROLLBACK"); + } catch { + // ignore + } + throw error; + } + + const updated = await this.getWorkflowRun({ + workflowRunId: params.workflowRunId, + }); + requireRow(updated, "resume workflow run"); + return updated; + } + /** * Return positional placeholders for {@link RUNNING_WORKFLOW_RUN_OWNED_WHERE} * in the order the fragment expects: namespace, run id, worker id. diff --git a/packages/openworkflow/worker/execution.test.ts b/packages/openworkflow/worker/execution.test.ts index 5a30ab58..1fe71707 100644 --- a/packages/openworkflow/worker/execution.test.ts +++ b/packages/openworkflow/worker/execution.test.ts @@ -2994,6 +2994,119 @@ describe("StepExecutor", () => { expect(status).toBe("failed"); sendSignalSpy.mockRestore(); }); + + test("resumeWorkflowRun re-runs the failed step without re-executing completed steps", async () => { + const backend = await createTestBackend(); + const client = new OpenWorkflow({ backend }); + + let validateRuns = 0; + let flakyRuns = 0; + let shouldFail = true; + + const workflow = client.defineWorkflow( + { name: `resume-from-failure-${randomUUID()}` }, + async ({ step }) => { + const validated = await step.run({ name: "validate" }, () => { + validateRuns++; + return "ok"; + }); + + const flaky = await step.run( + { name: "flaky", retryPolicy: { maximumAttempts: 2 } }, + () => { + flakyRuns++; + if (shouldFail) { + throw new Error("simulated upstream failure"); + } + return "recovered"; + }, + ); + + return { validated, flaky }; + }, + ); + + const worker = client.newWorker({ concurrency: 1 }); + const handle = await workflow.run(); + + const failedStatus = await tickUntilTerminal( + backend, + worker, + handle.workflowRun.id, + 40, + 25, + ); + expect(failedStatus).toBe("failed"); + expect(validateRuns).toBe(1); + expect(flakyRuns).toBe(2); + + const stepsBeforeResume = await backend.listStepAttempts({ + workflowRunId: handle.workflowRun.id, + limit: 100, + }); + const failedBefore = stepsBeforeResume.data.filter( + (s) => s.status === "failed", + ); + expect(failedBefore.length).toBe(2); + + shouldFail = false; + await backend.resumeWorkflowRun({ workflowRunId: handle.workflowRun.id }); + + const resumedRun = await backend.getWorkflowRun({ + workflowRunId: handle.workflowRun.id, + }); + expect(resumedRun?.status).toBe("pending"); + expect(resumedRun?.error).toBeNull(); + expect(resumedRun?.finishedAt).toBeNull(); + expect(resumedRun?.workerId).toBeNull(); + + const stepsAfterResume = await backend.listStepAttempts({ + workflowRunId: handle.workflowRun.id, + limit: 100, + }); + expect( + stepsAfterResume.data.some((s) => s.status === "failed"), + ).toBe(false); + expect( + stepsAfterResume.data.some( + (s) => s.stepName === "validate" && s.status === "completed", + ), + ).toBe(true); + + const finalStatus = await tickUntilTerminal( + backend, + worker, + handle.workflowRun.id, + 40, + 25, + ); + expect(finalStatus).toBe("completed"); + // "validate" was cached from the original run, not re-executed + expect(validateRuns).toBe(1); + // "flaky" ran 2 times on the original run + 1 on resume + expect(flakyRuns).toBe(3); + + const result = await handle.result(); + expect(result).toEqual({ validated: "ok", flaky: "recovered" }); + }); + + test("resumeWorkflowRun throws when the run is not in failed status", async () => { + const backend = await createTestBackend(); + const client = new OpenWorkflow({ backend }); + + const workflow = client.defineWorkflow( + { name: `resume-invalid-${randomUUID()}` }, + async ({ step }) => { + return await step.run({ name: "noop" }, () => "ok"); + }, + ); + + const handle = await workflow.run(); + + await expect( + backend.resumeWorkflowRun({ workflowRunId: handle.workflowRun.id }), + ).rejects.toThrow(/Cannot resume workflow run.*pending/); + }); }); describe("executeWorkflow", () => { From 3be38eb076264b7e19ac735ec796ce669ee7e61e Mon Sep 17 00:00:00 2001 From: Duong Cong Vu <32474388+vudc@users.noreply.github.com> Date: Mon, 18 May 2026 23:50:01 +0700 Subject: [PATCH 2/6] fix(openworkflow): return the resumed run from client resumeWorkflowRun Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- packages/openworkflow/client/client.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/openworkflow/client/client.ts b/packages/openworkflow/client/client.ts index ac3f7930..411f4814 100644 --- a/packages/openworkflow/client/client.ts +++ b/packages/openworkflow/client/client.ts @@ -181,15 +181,15 @@ export class OpenWorkflow { * from history without re-executing; failed step attempts are discarded so * the failing step starts with a fresh retry budget. * @param workflowRunId - The ID of the failed workflow run to resume - * @returns Promise + * @returns The updated workflow run * @throws {Error} If the run does not exist or is not in `failed` status * @example * ```ts * await ow.resumeWorkflowRun("123"); * ``` */ - async resumeWorkflowRun(workflowRunId: string): Promise { - await this.backend.resumeWorkflowRun({ workflowRunId }); + async resumeWorkflowRun(workflowRunId: string): Promise { + return await this.backend.resumeWorkflowRun({ workflowRunId }); } /** From a0ac0834ce99db65053bddba6711276578e90a7b Mon Sep 17 00:00:00 2001 From: vudc Date: Wed, 22 Jul 2026 22:14:05 +0700 Subject: [PATCH 3/6] fix(openworkflow): address review feedback on resume workflow run Backend correctness: - Delete every non-successful step attempt on resume, not just failed ones. Sleep, signal-wait and child-workflow attempts sit in 'running' and could survive a resume, making replay treat the step as in-flight instead of re-executing it. - Clear started_at so a resumed run reports its new lifecycle, matching rescheduleWorkflowRunAfterFailedStepAttempt. - Restructure the SQLite transaction so the conflict lookup happens outside it, removing the double ROLLBACK on an already-closed transaction. - Read the conflicting run through the transaction in Postgres instead of the outer connection, and use the shared withTransaction helper. Tests: - Add a resumeWorkflowRun() block to the shared backend testsuite so both SQLite and Postgres are covered; the SQLite path had none. - Cover the non-existent-run branch of resolveResumeWorkflowRunConflict. - Assert started_at is cleared and only successful attempts survive. Remove the flaky-payment demo, which drove failures from module-scoped mutable state in a workflow handler. --- openworkflow/flaky-payment.run.ts | 32 ---- openworkflow/flaky-payment.ts | 95 ------------ packages/openworkflow/postgres/backend.ts | 30 ++-- packages/openworkflow/sqlite/backend.ts | 60 ++++---- .../openworkflow/testing/backend.testsuite.ts | 145 ++++++++++++++++++ .../openworkflow/worker/execution.test.ts | 16 +- 6 files changed, 212 insertions(+), 166 deletions(-) delete mode 100644 openworkflow/flaky-payment.run.ts delete mode 100644 openworkflow/flaky-payment.ts diff --git a/openworkflow/flaky-payment.run.ts b/openworkflow/flaky-payment.run.ts deleted file mode 100644 index 8810c151..00000000 --- a/openworkflow/flaky-payment.run.ts +++ /dev/null @@ -1,32 +0,0 @@ -import { backend, ow } from "./client.js"; -import { flakyPayment } from "./flaky-payment.js"; - -console.log("Running flaky-payment workflow..."); -console.log("Expected: 3 failed attempts -> workflow marked failed."); -console.log( - "Then open http://localhost:3000, click 'Resume Run', and watch it complete.\n", -); - -const handle = await ow.runWorkflow(flakyPayment.spec, { - cartId: "cart_demo_1", - amountCents: 4200, -}); - -console.log(`Run id: ${handle.workflowRun.id}`); -console.log("Waiting for first terminal state..."); - -try { - const result = await handle.result(); - console.log(`Workflow completed: ${JSON.stringify(result, null, 2)}`); -} catch (error) { - console.log("\nWorkflow failed (as expected on first pass):"); - console.log(error instanceof Error ? error.message : String(error)); - console.log( - "\nNow click 'Resume Run' on the run detail page in the dashboard.", - ); - console.log( - "Leave the worker running so the in-memory attempt counter persists.", - ); -} - -await backend.stop(); diff --git a/openworkflow/flaky-payment.ts b/openworkflow/flaky-payment.ts deleted file mode 100644 index d4d719ec..00000000 --- a/openworkflow/flaky-payment.ts +++ /dev/null @@ -1,95 +0,0 @@ -import { defineWorkflow } from "openworkflow"; - -interface FlakyPaymentInput { - cartId: string; - amountCents: number; -} - -interface FlakyPaymentOutput { - cartId: string; - authorizationId: string; - receiptId: string; - attemptsForReserve: number; -} - -const RESERVE_MAX_ATTEMPTS = 3; - -// Module-scoped counter. Persists across the failure-then-resume cycle as long -// as the worker process is alive, so the step succeeds on the first attempt -// after `ow.resumeWorkflowRun` is called. -let reserveAttempt = 0; - -/** - * Demo workflow for the Resume feature. - * - * Flow: - * 1. First run: "reserve-funds" throws on every attempt; the step's retry - * budget (RESERVE_MAX_ATTEMPTS) is exhausted, so the workflow run ends in - * `failed`. The downstream steps never run. - * 2. Click "Resume Run" in the dashboard (or call `ow.resumeWorkflowRun(id)`). - * The failed step_attempt rows are dropped and the run is requeued. - * 3. On the next worker tick, "validate-cart" is served from the cache (not - * re-executed); "reserve-funds" runs once more, the counter is now past - * RESERVE_MAX_ATTEMPTS so it returns successfully, and "confirm-payment" - * plus "send-receipt" proceed to completion. - * - * Note: this relies on `reserveAttempt` persisting in the worker process. If - * you restart the worker between the failure and the resume, the counter - * resets, so resume will fail again and you'll need to resume once more. - */ -export const flakyPayment = defineWorkflow( - { name: "flaky-payment" }, - async ({ input, step, run }) => { - console.log(`[run ${run.id}] flaky-payment for cart ${input.cartId}`); - - await step.run({ name: "validate-cart" }, () => { - if (input.amountCents <= 0) { - throw new Error("amountCents must be positive"); - } - }); - - const { authorizationId, attempts } = await step.run( - { - name: "reserve-funds", - retryPolicy: { - maximumAttempts: RESERVE_MAX_ATTEMPTS, - initialInterval: "500ms", - }, - }, - () => { - reserveAttempt++; - console.log(`reserve-funds attempt ${String(reserveAttempt)}`); - - if (reserveAttempt <= RESERVE_MAX_ATTEMPTS) { - throw new Error( - `simulated upstream 503 (attempt ${String(reserveAttempt)})`, - ); - } - - console.log( - `reserve-funds recovered on attempt ${String(reserveAttempt)} (after resume)`, - ); - return { - authorizationId: `auth_${input.cartId}_${String(Date.now())}`, - attempts: reserveAttempt, - }; - }, - ); - - const receiptId = await step.run({ name: "confirm-payment" }, () => { - console.log(`confirming with ${authorizationId}`); - return `rcpt_${input.cartId}`; - }); - - await step.run({ name: "send-receipt" }, () => { - console.log(`receipt ${receiptId} mailed`); - }); - - return { - cartId: input.cartId, - authorizationId, - receiptId, - attemptsForReserve: attempts, - }; - }, -); diff --git a/packages/openworkflow/postgres/backend.ts b/packages/openworkflow/postgres/backend.ts index e4d128ac..88534353 100644 --- a/packages/openworkflow/postgres/backend.ts +++ b/packages/openworkflow/postgres/backend.ts @@ -768,8 +768,7 @@ export class BackendPostgres implements Backend { async resumeWorkflowRun( params: ResumeWorkflowRunParams, ): Promise { - return await this.pg.begin(async (sql): Promise => { - const tx = sql as unknown as Postgres; + return await this.withTransaction(async (tx): Promise => { const workflowRunsTable = this.workflowRunsTable(tx); const stepAttemptsTable = this.stepAttemptsTable(tx); @@ -779,6 +778,7 @@ export class BackendPostgres implements Backend { "status" = 'pending', "worker_id" = NULL, "error" = NULL, + "started_at" = NULL, "finished_at" = NULL, "available_at" = NOW(), "updated_at" = NOW() @@ -789,20 +789,30 @@ export class BackendPostgres implements Backend { `; if (!updated) { - const existing = await this.getWorkflowRun({ - workflowRunId: params.workflowRunId, - }); - resolveResumeWorkflowRunConflict(params.workflowRunId, existing); + const [existing] = await tx` + SELECT * + FROM ${workflowRunsTable} + WHERE "namespace_id" = ${this.namespaceId} + AND "id" = ${params.workflowRunId} + LIMIT 1 + `; + + resolveResumeWorkflowRunConflict( + params.workflowRunId, + existing ?? null, + ); } - // Drop the prior failed attempts so the next worker pass starts the - // failed step with a fresh retry budget and the existing completed - // attempts remain in cache (replay skips re-execution). + // Drop every attempt that did not succeed. Failed attempts go so the + // failing step gets a fresh retry budget; still-'running' attempts + // (sleep, signal-wait, child workflow) go so replay does not mistake + // them for in-flight work. Successful attempts stay and are replayed + // from cache without re-executing. await tx` DELETE FROM ${stepAttemptsTable} WHERE "namespace_id" = ${this.namespaceId} AND "workflow_run_id" = ${params.workflowRunId} - AND "status" = 'failed' + AND "status" NOT IN ('completed', 'succeeded') `; return updated; diff --git a/packages/openworkflow/sqlite/backend.ts b/packages/openworkflow/sqlite/backend.ts index fa3005a0..c7d22b3d 100644 --- a/packages/openworkflow/sqlite/backend.ts +++ b/packages/openworkflow/sqlite/backend.ts @@ -275,7 +275,8 @@ export class BackendSqlite implements Backend { `); const row = stmt.get(this.namespaceId, params.workflowRunId) as - WorkflowRunRow | undefined; + | WorkflowRunRow + | undefined; return Promise.resolve(row ? rowToWorkflowRun(row) : null); } @@ -394,7 +395,8 @@ export class BackendSqlite implements Backend { LIMIT 1 `); const row = stmt.get(this.namespaceId, params.stepAttemptId) as - { data: string | null } | undefined; + | { data: string | null } + | undefined; if (!row) return Promise.resolve(undefined); return Promise.resolve( @@ -748,16 +750,17 @@ export class BackendSqlite implements Backend { params: ResumeWorkflowRunParams, ): Promise { const currentTime = now(); + let resumed = false; + this.db.exec("BEGIN IMMEDIATE"); try { - this.db.exec("BEGIN IMMEDIATE"); - const updateStmt = this.db.prepare(` UPDATE "workflow_runs" SET "status" = 'pending', "worker_id" = NULL, "error" = NULL, + "started_at" = NULL, "finished_at" = NULL, "available_at" = ?, "updated_at" = ? @@ -773,38 +776,40 @@ export class BackendSqlite implements Backend { params.workflowRunId, ); - if (updateResult.changes === 0) { - this.db.exec("ROLLBACK"); - const existing = await this.getWorkflowRun({ - workflowRunId: params.workflowRunId, - }); - resolveResumeWorkflowRunConflict(params.workflowRunId, existing); + resumed = updateResult.changes > 0; + + if (resumed) { + // Drop every attempt that did not succeed. Failed attempts go so the + // failing step gets a fresh retry budget; still-'running' attempts + // (sleep, signal-wait, child workflow) go so replay does not mistake + // them for in-flight work. Successful attempts stay and are replayed + // from cache without re-executing. + this.db + .prepare( + ` + DELETE FROM "step_attempts" + WHERE "namespace_id" = ? + AND "workflow_run_id" = ? + AND "status" NOT IN ('completed', 'succeeded') + `, + ) + .run(this.namespaceId, params.workflowRunId); } - this.db - .prepare( - ` - DELETE FROM "step_attempts" - WHERE "namespace_id" = ? - AND "workflow_run_id" = ? - AND "status" = 'failed' - `, - ) - .run(this.namespaceId, params.workflowRunId); - this.db.exec("COMMIT"); } catch (error) { - try { - this.db.exec("ROLLBACK"); - } catch { - // ignore - } + this.db.exec("ROLLBACK"); throw error; } const updated = await this.getWorkflowRun({ workflowRunId: params.workflowRunId, }); + + if (!resumed) { + resolveResumeWorkflowRunConflict(params.workflowRunId, updated); + } + requireRow(updated, "resume workflow run"); return updated; } @@ -1080,7 +1085,8 @@ export class BackendSqlite implements Backend { `); const row = stmt.get(this.namespaceId, params.stepAttemptId) as - StepAttemptRow | undefined; + | StepAttemptRow + | undefined; return Promise.resolve(row ? rowToStepAttempt(row) : null); } diff --git a/packages/openworkflow/testing/backend.testsuite.ts b/packages/openworkflow/testing/backend.testsuite.ts index 22e5291b..be12cbd7 100644 --- a/packages/openworkflow/testing/backend.testsuite.ts +++ b/packages/openworkflow/testing/backend.testsuite.ts @@ -2573,6 +2573,151 @@ export function testBackend(options: TestBackendOptions): void { }); }); + describe("resumeWorkflowRun()", () => { + test("flips a failed run back to pending and clears failure fields", async () => { + const backend = await setup(); + + await createPendingWorkflowRun(backend); + const failedId = await claimAndFailNextPendingRun(backend); + + const resumed = await backend.resumeWorkflowRun({ + workflowRunId: failedId, + }); + + expect(resumed.status).toBe("pending"); + expect(resumed.error).toBeNull(); + expect(resumed.workerId).toBeNull(); + expect(resumed.startedAt).toBeNull(); + expect(resumed.finishedAt).toBeNull(); + expect(resumed.availableAt).not.toBeNull(); + expect(deltaSeconds(resumed.availableAt)).toBeLessThan(1); + + await teardown(backend); + }); + + test("drops every unsuccessful step attempt but keeps successful ones", async () => { + const backend = await setup(); + + const claimed = await createClaimedWorkflowRun(backend); + const workerId = claimed.workerId!; // eslint-disable-line @typescript-eslint/no-non-null-assertion + + const completed = await backend.createStepAttempt({ + workflowRunId: claimed.id, + workerId, + stepName: "completed-step", + kind: "function", + config: {}, + context: null, + }); + await backend.completeStepAttempt({ + workflowRunId: claimed.id, + stepAttemptId: completed.id, + workerId, + output: { ok: true }, + }); + + const failed = await backend.createStepAttempt({ + workflowRunId: claimed.id, + workerId, + stepName: "failed-step", + kind: "function", + config: {}, + context: null, + }); + await backend.failStepAttempt({ + workflowRunId: claimed.id, + stepAttemptId: failed.id, + workerId, + error: { message: "boom" }, + }); + + // left in 'running': mirrors a sleep/signal-wait/child-workflow + // attempt that never reached a terminal state before the run failed + await backend.createStepAttempt({ + workflowRunId: claimed.id, + workerId, + stepName: "running-step", + kind: "sleep", + config: {}, + context: { + kind: "sleep", + resumeAt: new Date(Date.now() + 60_000).toISOString(), + }, + }); + + await backend.failWorkflowRun({ + workflowRunId: claimed.id, + workerId, + error: { message: "run failed" }, + retryPolicy: { + ...DEFAULT_WORKFLOW_RETRY_POLICY, + maximumAttempts: 1, + }, + }); + + const failedRun = await backend.getWorkflowRun({ + workflowRunId: claimed.id, + }); + expect(failedRun?.status).toBe("failed"); + + await backend.resumeWorkflowRun({ workflowRunId: claimed.id }); + + const attempts = await backend.listStepAttempts({ + workflowRunId: claimed.id, + limit: 100, + }); + expect(attempts.data.map((a) => a.stepName)).toEqual([ + "completed-step", + ]); + expect(attempts.data[0]?.status).toBe("completed"); + + await teardown(backend); + }); + + test("throws when resuming a run that is not failed", async () => { + const backend = await setup(); + + const created = await createPendingWorkflowRun(backend); + + await expect( + backend.resumeWorkflowRun({ workflowRunId: created.id }), + ).rejects.toThrow(/Cannot resume workflow run .* with status pending/); + + await teardown(backend); + }); + + test("throws when resuming a non-existent workflow run", async () => { + const backend = await setup(); + + const nonExistentId = randomUUID(); + + await expect( + backend.resumeWorkflowRun({ workflowRunId: nonExistentId }), + ).rejects.toThrow(`Workflow run ${nonExistentId} does not exist`); + + await teardown(backend); + }); + + test("a resumed run is claimable by workers again", async () => { + const backend = await setup(); + + await createPendingWorkflowRun(backend); + const failedId = await claimAndFailNextPendingRun(backend); + + await backend.resumeWorkflowRun({ workflowRunId: failedId }); + + const claimed = await backend.claimWorkflowRun({ + workerId: randomUUID(), + leaseDurationMs: 100, + }); + + expect(claimed?.id).toBe(failedId); + expect(claimed?.status).toBe("running"); + + await teardown(backend); + }); + }); + describe("sendSignal()", () => { test("returns empty when no active waiters", async () => { const result = await backend.sendSignal({ diff --git a/packages/openworkflow/worker/execution.test.ts b/packages/openworkflow/worker/execution.test.ts index 1fe71707..5ef85392 100644 --- a/packages/openworkflow/worker/execution.test.ts +++ b/packages/openworkflow/worker/execution.test.ts @@ -3057,6 +3057,7 @@ describe("StepExecutor", () => { }); expect(resumedRun?.status).toBe("pending"); expect(resumedRun?.error).toBeNull(); + expect(resumedRun?.startedAt).toBeNull(); expect(resumedRun?.finishedAt).toBeNull(); expect(resumedRun?.workerId).toBeNull(); @@ -3064,9 +3065,12 @@ describe("StepExecutor", () => { workflowRunId: handle.workflowRun.id, limit: 100, }); + // only successful attempts survive; failed and still-running rows are gone expect( - stepsAfterResume.data.some((s) => s.status === "failed"), - ).toBe(false); + stepsAfterResume.data.every( + (s) => s.status === "completed" || s.status === "succeeded", + ), + ).toBe(true); expect( stepsAfterResume.data.some( (s) => s.stepName === "validate" && s.status === "completed", @@ -3107,6 +3111,14 @@ describe("StepExecutor", () => { backend.resumeWorkflowRun({ workflowRunId: handle.workflowRun.id }), ).rejects.toThrow(/Cannot resume workflow run.*pending/); }); + + test("resumeWorkflowRun throws when the run does not exist", async () => { + const backend = await createTestBackend(); + + await expect( + backend.resumeWorkflowRun({ workflowRunId: randomUUID() }), + ).rejects.toThrow(/does not exist/); + }); }); describe("executeWorkflow", () => { From c18fa2890b90fb368efce9549fbf2af60f3a924f Mon Sep 17 00:00:00 2001 From: vudc Date: Thu, 23 Jul 2026 11:00:07 +0700 Subject: [PATCH 4/6] fix(openworkflow, dashboard): address resume review findings Correctness: - Narrow the step-attempt cleanup to failed attempts plus inert running attempts (function, signal-send). Running sleep, signal-wait and workflow attempts are now preserved: deleting them orphaned live child runs via ON DELETE SET NULL, dropped already-delivered signals, and restarted durable timers from zero. - Reset the run-level attempts counter on resume so a resumed run gets the fresh retry budget the UI promises. - Gate resume on an unexpired deadline and reject deadline-expired runs with a clear error instead of destroying their failure history for a run the next tick would immediately re-fail. Dashboard: - Extract the shared RunActionDialog so resume and cancel stop tripping the zero-tolerance jscpd duplication gate. - Use .validator instead of the removed .inputValidator, matching the sibling server functions. Tests: - Rewrite the attempt-cleanup test to assert in-flight attempts survive, and add a child-workflow test proving the linked child is not orphaned (both fail against the previous broad DELETE). - Add deadline-rejection and attempts-reset coverage. - Give the end-to-end resume test a 20s wait budget so the flaky step's 1s retry backoff fits inside it. --- .../src/components/run-action-dialog.tsx | 118 ++++++++++++++++ .../src/components/run-cancel-action.tsx | 105 +++----------- .../src/components/run-resume-action.tsx | 103 ++------------ apps/dashboard/src/lib/api.ts | 2 +- packages/openworkflow/core/workflow-run.ts | 20 ++- packages/openworkflow/postgres/backend.ts | 18 ++- packages/openworkflow/sqlite/backend.ts | 19 ++- .../openworkflow/testing/backend.testsuite.ts | 132 ++++++++++++++++-- .../openworkflow/worker/execution.test.ts | 4 +- 9 files changed, 317 insertions(+), 204 deletions(-) create mode 100644 apps/dashboard/src/components/run-action-dialog.tsx diff --git a/apps/dashboard/src/components/run-action-dialog.tsx b/apps/dashboard/src/components/run-action-dialog.tsx new file mode 100644 index 00000000..0a15be45 --- /dev/null +++ b/apps/dashboard/src/components/run-action-dialog.tsx @@ -0,0 +1,118 @@ +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, +} from "@/components/ui/alert-dialog"; +import { Button } from "@/components/ui/button"; +import type { ReactNode } from "react"; +import { useState } from "react"; + +type ButtonVariant = React.ComponentProps["variant"]; + +interface RunActionDialogProps { + triggerLabel: string; + triggerVariant?: ButtonVariant; + title: string; + description: ReactNode; + cancelLabel: string; + confirmLabel: string; + pendingLabel: string; + confirmVariant?: ButtonVariant; + fallbackErrorMessage: string; + action: () => Promise; + onDone?: (() => Promise) | (() => void); +} + +function getErrorMessage(error: unknown, fallback: string): string { + if (error instanceof Error && error.message) { + return error.message; + } + + return fallback; +} + +// Confirmation dialog shared by the run action buttons (cancel, resume). +export function RunActionDialog({ + triggerLabel, + triggerVariant = "default", + title, + description, + cancelLabel, + confirmLabel, + pendingLabel, + confirmVariant, + fallbackErrorMessage, + action, + onDone, +}: RunActionDialogProps) { + const [isOpen, setIsOpen] = useState(false); + const [isPending, setIsPending] = useState(false); + const [error, setError] = useState(null); + + async function runAction() { + setIsPending(true); + setError(null); + + try { + await action(); + await onDone?.(); + setIsOpen(false); + } catch (caughtError) { + setError(getErrorMessage(caughtError, fallbackErrorMessage)); + } finally { + setIsPending(false); + } + } + + return ( + { + setIsOpen(nextOpen); + if (!nextOpen) { + setError(null); + } + }} + > + + + + + {title} + {description} + + + {error &&

{error}

} + + + + {cancelLabel} + + { + void runAction(); + }} + disabled={isPending} + > + {isPending ? pendingLabel : confirmLabel} + + + + + ); +} diff --git a/apps/dashboard/src/components/run-cancel-action.tsx b/apps/dashboard/src/components/run-cancel-action.tsx index 77b94021..d8dd2727 100644 --- a/apps/dashboard/src/components/run-cancel-action.tsx +++ b/apps/dashboard/src/components/run-cancel-action.tsx @@ -1,18 +1,7 @@ -import { - AlertDialog, - AlertDialogAction, - AlertDialogCancel, - AlertDialogContent, - AlertDialogDescription, - AlertDialogFooter, - AlertDialogHeader, - AlertDialogTitle, -} from "@/components/ui/alert-dialog"; -import { Button } from "@/components/ui/button"; +import { RunActionDialog } from "@/components/run-action-dialog"; import { cancelWorkflowRunServerFn } from "@/lib/api"; import { isRunCancelableStatus } from "@/lib/status"; import type { WorkflowRunStatus } from "openworkflow/internal"; -import { useState } from "react"; interface RunCancelActionProps { runId: string; @@ -20,92 +9,30 @@ interface RunCancelActionProps { onCanceled?: (() => Promise) | (() => void); } -function getErrorMessage(error: unknown): string { - if (error instanceof Error && error.message) { - return error.message; - } - - return "Unable to cancel workflow run"; -} - export function RunCancelAction({ runId, status, onCanceled, }: RunCancelActionProps) { - const [isOpen, setIsOpen] = useState(false); - const [isCanceling, setIsCanceling] = useState(false); - const [error, setError] = useState(null); - if (!isRunCancelableStatus(status)) { return null; } - async function cancelRun() { - setIsCanceling(true); - setError(null); - - try { - await cancelWorkflowRunServerFn({ - data: { - workflowRunId: runId, - }, - }); - await onCanceled?.(); - setIsOpen(false); - } catch (caughtError) { - setError(getErrorMessage(caughtError)); - } finally { - setIsCanceling(false); - } - } - return ( - { - setIsOpen(nextOpen); - if (!nextOpen) { - setError(null); - } - }} - > - - - - - Cancel this run? - - This will stop any future progress for this workflow run. - - - - {error &&

{error}

} - - - - Keep Running - - { - void cancelRun(); - }} - disabled={isCanceling} - > - {isCanceling ? "Canceling..." : "Cancel Run"} - - - - + + cancelWorkflowRunServerFn({ data: { workflowRunId: runId } }) + } + onDone={onCanceled} + /> ); } diff --git a/apps/dashboard/src/components/run-resume-action.tsx b/apps/dashboard/src/components/run-resume-action.tsx index 452d4c20..5795a691 100644 --- a/apps/dashboard/src/components/run-resume-action.tsx +++ b/apps/dashboard/src/components/run-resume-action.tsx @@ -1,18 +1,7 @@ -import { - AlertDialog, - AlertDialogAction, - AlertDialogCancel, - AlertDialogContent, - AlertDialogDescription, - AlertDialogFooter, - AlertDialogHeader, - AlertDialogTitle, -} from "@/components/ui/alert-dialog"; -import { Button } from "@/components/ui/button"; +import { RunActionDialog } from "@/components/run-action-dialog"; import { resumeWorkflowRunServerFn } from "@/lib/api"; import { isRunResumableStatus } from "@/lib/status"; import type { WorkflowRunStatus } from "openworkflow/internal"; -import { useState } from "react"; interface RunResumeActionProps { runId: string; @@ -20,91 +9,29 @@ interface RunResumeActionProps { onResumed?: (() => Promise) | (() => void); } -function getErrorMessage(error: unknown): string { - if (error instanceof Error && error.message) { - return error.message; - } - - return "Unable to resume workflow run"; -} - export function RunResumeAction({ runId, status, onResumed, }: RunResumeActionProps) { - const [isOpen, setIsOpen] = useState(false); - const [isResuming, setIsResuming] = useState(false); - const [error, setError] = useState(null); - if (!isRunResumableStatus(status)) { return null; } - async function resumeRun() { - setIsResuming(true); - setError(null); - - try { - await resumeWorkflowRunServerFn({ - data: { - workflowRunId: runId, - }, - }); - await onResumed?.(); - setIsOpen(false); - } catch (caughtError) { - setError(getErrorMessage(caughtError)); - } finally { - setIsResuming(false); - } - } - return ( - { - setIsOpen(nextOpen); - if (!nextOpen) { - setError(null); - } - }} - > - - - - - Resume this failed run? - - Completed steps stay cached and won't re-run. The failing step will - be retried with a fresh retry budget. Previous failed attempts will - be discarded. - - - - {error &&

{error}

} - - - Cancel - { - void resumeRun(); - }} - disabled={isResuming} - > - {isResuming ? "Resuming..." : "Resume Run"} - - - - + + resumeWorkflowRunServerFn({ data: { workflowRunId: runId } }) + } + onDone={onResumed} + /> ); } diff --git a/apps/dashboard/src/lib/api.ts b/apps/dashboard/src/lib/api.ts index e604f7ee..59dba33a 100644 --- a/apps/dashboard/src/lib/api.ts +++ b/apps/dashboard/src/lib/api.ts @@ -96,7 +96,7 @@ export const cancelWorkflowRunServerFn = createServerFn({ method: "POST" }) * budget; completed steps stay cached. */ export const resumeWorkflowRunServerFn = createServerFn({ method: "POST" }) - .inputValidator(z.object({ workflowRunId: z.string() })) + .validator(z.object({ workflowRunId: z.string() })) .handler(async ({ data }): Promise => { const backend = await getBackend(); return backend.resumeWorkflowRun({ workflowRunId: data.workflowRunId }); diff --git a/packages/openworkflow/core/workflow-run.ts b/packages/openworkflow/core/workflow-run.ts index 942c3b8a..c74ad47b 100644 --- a/packages/openworkflow/core/workflow-run.ts +++ b/packages/openworkflow/core/workflow-run.ts @@ -60,12 +60,12 @@ export function resolveCancelWorkflowRunConflict( /** * Resolve the outcome when a resumeWorkflowRun UPDATE affected no rows. The - * UPDATE is gated on `status = 'failed'`, so a zero-row outcome means either - * the run doesn't exist or it isn't in a resumable state. + * UPDATE is gated on `status = 'failed'` AND an unexpired deadline, so a + * zero-row outcome means the run doesn't exist, isn't in a resumable state, + * or has already blown its deadline (in which case resuming it is futile). * @param workflowRunId - ID of the workflow run (used in error messages) * @param existing - Current workflow run, or null if not found - * @returns Never; always throws describing why the resume is impossible - * @throws {Error} If the run does not exist or is not in `failed` status + * @throws {Error} If the run is missing, not `failed`, or past its deadline */ export function resolveResumeWorkflowRunConflict( workflowRunId: string, @@ -76,6 +76,15 @@ export function resolveResumeWorkflowRunConflict( throw new Error(`Workflow run ${workflowRunId} does not exist`); } + if (existing.status === "failed") { + // The UPDATE also gates on the deadline, so a still-`failed` run that did + // not resume is one whose deadline has already elapsed. + // eslint-disable-next-line functional/no-throw-statements + throw new Error( + `Cannot resume workflow run ${workflowRunId}; its deadline has already passed`, + ); + } + // eslint-disable-next-line functional/no-throw-statements throw new Error( `Cannot resume workflow run ${workflowRunId} with status ${existing.status}; only failed runs can be resumed`, @@ -128,7 +137,8 @@ export type SchemaOutput = TSchema extends StandardSchemaV1 * error message. */ export type ValidationResult = - { success: true; value: T } | { success: false; error: string }; + | { success: true; value: T } + | { success: false; error: string }; /** * Validate input against a Standard Schema. Pure async function that validates diff --git a/packages/openworkflow/postgres/backend.ts b/packages/openworkflow/postgres/backend.ts index 88534353..f473c5d2 100644 --- a/packages/openworkflow/postgres/backend.ts +++ b/packages/openworkflow/postgres/backend.ts @@ -778,6 +778,7 @@ export class BackendPostgres implements Backend { "status" = 'pending', "worker_id" = NULL, "error" = NULL, + "attempts" = 0, "started_at" = NULL, "finished_at" = NULL, "available_at" = NOW(), @@ -785,6 +786,7 @@ export class BackendPostgres implements Backend { WHERE "namespace_id" = ${this.namespaceId} AND "id" = ${params.workflowRunId} AND "status" = 'failed' + AND ("deadline_at" IS NULL OR "deadline_at" > NOW()) RETURNING * `; @@ -803,16 +805,20 @@ export class BackendPostgres implements Backend { ); } - // Drop every attempt that did not succeed. Failed attempts go so the - // failing step gets a fresh retry budget; still-'running' attempts - // (sleep, signal-wait, child workflow) go so replay does not mistake - // them for in-flight work. Successful attempts stay and are replayed - // from cache without re-executing. + // Drop failed attempts so the failing step gets a fresh retry budget, + // plus inert running attempts whose kinds hold no external state + // (function, signal-send). Running sleep, signal-wait and workflow + // attempts are preserved: replay resumes them, and deleting them would + // orphan linked child runs and already-delivered signals. Successful + // attempts stay and are replayed from cache. await tx` DELETE FROM ${stepAttemptsTable} WHERE "namespace_id" = ${this.namespaceId} AND "workflow_run_id" = ${params.workflowRunId} - AND "status" NOT IN ('completed', 'succeeded') + AND ( + "status" = 'failed' + OR ("status" = 'running' AND "kind" IN ('function', 'signal-send')) + ) `; return updated; diff --git a/packages/openworkflow/sqlite/backend.ts b/packages/openworkflow/sqlite/backend.ts index c7d22b3d..699c8a88 100644 --- a/packages/openworkflow/sqlite/backend.ts +++ b/packages/openworkflow/sqlite/backend.ts @@ -760,6 +760,7 @@ export class BackendSqlite implements Backend { "status" = 'pending', "worker_id" = NULL, "error" = NULL, + "attempts" = 0, "started_at" = NULL, "finished_at" = NULL, "available_at" = ?, @@ -767,6 +768,7 @@ export class BackendSqlite implements Backend { WHERE "namespace_id" = ? AND "id" = ? AND "status" = 'failed' + AND ("deadline_at" IS NULL OR "deadline_at" > ?) `); const updateResult = updateStmt.run( @@ -774,23 +776,28 @@ export class BackendSqlite implements Backend { currentTime, this.namespaceId, params.workflowRunId, + currentTime, ); resumed = updateResult.changes > 0; if (resumed) { - // Drop every attempt that did not succeed. Failed attempts go so the - // failing step gets a fresh retry budget; still-'running' attempts - // (sleep, signal-wait, child workflow) go so replay does not mistake - // them for in-flight work. Successful attempts stay and are replayed - // from cache without re-executing. + // Drop failed attempts so the failing step gets a fresh retry budget, + // plus inert running attempts whose kinds hold no external state + // (function, signal-send). Running sleep, signal-wait and workflow + // attempts are preserved: replay resumes them, and deleting them + // would orphan linked child runs and already-delivered signals. + // Successful attempts stay and are replayed from cache. this.db .prepare( ` DELETE FROM "step_attempts" WHERE "namespace_id" = ? AND "workflow_run_id" = ? - AND "status" NOT IN ('completed', 'succeeded') + AND ( + "status" = 'failed' + OR ("status" = 'running' AND "kind" IN ('function', 'signal-send')) + ) `, ) .run(this.namespaceId, params.workflowRunId); diff --git a/packages/openworkflow/testing/backend.testsuite.ts b/packages/openworkflow/testing/backend.testsuite.ts index be12cbd7..f8749568 100644 --- a/packages/openworkflow/testing/backend.testsuite.ts +++ b/packages/openworkflow/testing/backend.testsuite.ts @@ -2589,13 +2589,15 @@ export function testBackend(options: TestBackendOptions): void { expect(resumed.workerId).toBeNull(); expect(resumed.startedAt).toBeNull(); expect(resumed.finishedAt).toBeNull(); + // run-level retry budget is reset so the resumed run gets fresh retries + expect(resumed.attempts).toBe(0); expect(resumed.availableAt).not.toBeNull(); expect(deltaSeconds(resumed.availableAt)).toBeLessThan(1); await teardown(backend); }); - test("drops every unsuccessful step attempt but keeps successful ones", async () => { + test("keeps successful and in-flight attempts, drops failed and inert running ones", async () => { const backend = await setup(); const claimed = await createClaimedWorkflowRun(backend); @@ -2631,12 +2633,23 @@ export function testBackend(options: TestBackendOptions): void { error: { message: "boom" }, }); - // left in 'running': mirrors a sleep/signal-wait/child-workflow - // attempt that never reached a terminal state before the run failed + // Inert running function attempt (e.g. a worker that died mid-step): + // safe to drop because nothing references it. await backend.createStepAttempt({ workflowRunId: claimed.id, workerId, - stepName: "running-step", + stepName: "running-fn", + kind: "function", + config: {}, + context: null, + }); + + // In-flight durable wait: must be preserved so replay resumes it + // instead of restarting the timer from zero. + await backend.createStepAttempt({ + workflowRunId: claimed.id, + workerId, + stepName: "running-sleep", kind: "sleep", config: {}, context: { @@ -2666,10 +2679,113 @@ export function testBackend(options: TestBackendOptions): void { workflowRunId: claimed.id, limit: 100, }); - expect(attempts.data.map((a) => a.stepName)).toEqual([ - "completed-step", - ]); - expect(attempts.data[0]?.status).toBe("completed"); + const survivingNames = attempts.data.map((a) => a.stepName); + expect(survivingNames).toHaveLength(2); + expect(survivingNames).toContain("completed-step"); + expect(survivingNames).toContain("running-sleep"); + + await teardown(backend); + }); + + test("preserves an in-flight child-workflow attempt so the child is not orphaned", async () => { + const backend = await setup(); + + const parent = await createClaimedWorkflowRun(backend); + const workerId = parent.workerId!; // eslint-disable-line @typescript-eslint/no-non-null-assertion + + // Running kind='workflow' attempt with a child run linked back to it. + const workflowAttempt = await backend.createStepAttempt({ + workflowRunId: parent.id, + workerId, + stepName: "invoke-child", + kind: "workflow", + config: {}, + context: { kind: "workflow", timeoutAt: null }, + }); + + const child = await backend.createWorkflowRun({ + workflowName: randomUUID(), + version: null, + idempotencyKey: null, + input: null, + config: {}, + context: null, + parentStepAttemptNamespaceId: workflowAttempt.namespaceId, + parentStepAttemptId: workflowAttempt.id, + availableAt: null, + deadlineAt: null, + }); + expect(child.parentStepAttemptId).toBe(workflowAttempt.id); + + await backend.failWorkflowRun({ + workflowRunId: parent.id, + workerId, + error: { message: "sibling failed" }, + retryPolicy: { + ...DEFAULT_WORKFLOW_RETRY_POLICY, + maximumAttempts: 1, + }, + }); + + await backend.resumeWorkflowRun({ workflowRunId: parent.id }); + + // The running workflow attempt must survive the resume... + const attempts = await backend.listStepAttempts({ + workflowRunId: parent.id, + limit: 100, + }); + expect(attempts.data.some((a) => a.id === workflowAttempt.id)).toBe( + true, + ); + + // ...so the child's parent pointer is not nulled by ON DELETE SET NULL. + const childAfter = await backend.getWorkflowRun({ + workflowRunId: child.id, + }); + expect(childAfter?.parentStepAttemptId).toBe(workflowAttempt.id); + + await teardown(backend); + }); + + test("throws and preserves history when the deadline has passed", async () => { + const backend = await setup(); + + const created = await backend.createWorkflowRun({ + workflowName: randomUUID(), + version: null, + idempotencyKey: null, + input: null, + config: {}, + context: null, + parentStepAttemptNamespaceId: null, + parentStepAttemptId: null, + availableAt: null, + deadlineAt: new Date(Date.now() - 1000), + }); + + // Claiming triggers the deadline sweep, which flips the run to failed; + // the run itself is then excluded from the claim, so this returns null. + await backend.claimWorkflowRun({ + workerId: randomUUID(), + leaseDurationMs: 100, + }); + + const failedRun = await backend.getWorkflowRun({ + workflowRunId: created.id, + }); + expect(failedRun?.status).toBe("failed"); + expect(failedRun?.error).not.toBeNull(); + + await expect( + backend.resumeWorkflowRun({ workflowRunId: created.id }), + ).rejects.toThrow(/deadline has already passed/); + + // Resume must not have destroyed the run's failure diagnostics. + const afterResume = await backend.getWorkflowRun({ + workflowRunId: created.id, + }); + expect(afterResume?.status).toBe("failed"); + expect(afterResume?.error).not.toBeNull(); await teardown(backend); }); diff --git a/packages/openworkflow/worker/execution.test.ts b/packages/openworkflow/worker/execution.test.ts index 5ef85392..78048224 100644 --- a/packages/openworkflow/worker/execution.test.ts +++ b/packages/openworkflow/worker/execution.test.ts @@ -3035,6 +3035,7 @@ describe("StepExecutor", () => { handle.workflowRun.id, 40, 25, + { maxWaitMs: 20_000 }, ); expect(failedStatus).toBe("failed"); expect(validateRuns).toBe(1); @@ -3083,6 +3084,7 @@ describe("StepExecutor", () => { handle.workflowRun.id, 40, 25, + { maxWaitMs: 20_000 }, ); expect(finalStatus).toBe("completed"); // "validate" was cached from the original run, not re-executed @@ -3092,7 +3094,7 @@ describe("StepExecutor", () => { const result = await handle.result(); expect(result).toEqual({ validated: "ok", flaky: "recovered" }); - }); + }, 30_000); test("resumeWorkflowRun throws when the run is not in failed status", async () => { const backend = await createTestBackend(); From e69a02f75d6022a1a176ddb6ab97b4cc2afba36c Mon Sep 17 00:00:00 2001 From: vudc Date: Mon, 27 Jul 2026 01:40:07 +0700 Subject: [PATCH 5/6] style(openworkflow): match prettier 3.9.6 union formatting An earlier local prettier 3.8.3 pass expanded short union types that the pinned 3.9.6 keeps on one line, breaking the format check in CI. Reformat with the pinned version. --- packages/openworkflow/core/workflow-run.ts | 3 +-- packages/openworkflow/sqlite/backend.ts | 9 +++------ 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/packages/openworkflow/core/workflow-run.ts b/packages/openworkflow/core/workflow-run.ts index c74ad47b..7575ab59 100644 --- a/packages/openworkflow/core/workflow-run.ts +++ b/packages/openworkflow/core/workflow-run.ts @@ -137,8 +137,7 @@ export type SchemaOutput = TSchema extends StandardSchemaV1 * error message. */ export type ValidationResult = - | { success: true; value: T } - | { success: false; error: string }; + { success: true; value: T } | { success: false; error: string }; /** * Validate input against a Standard Schema. Pure async function that validates diff --git a/packages/openworkflow/sqlite/backend.ts b/packages/openworkflow/sqlite/backend.ts index 699c8a88..7162a21a 100644 --- a/packages/openworkflow/sqlite/backend.ts +++ b/packages/openworkflow/sqlite/backend.ts @@ -275,8 +275,7 @@ export class BackendSqlite implements Backend { `); const row = stmt.get(this.namespaceId, params.workflowRunId) as - | WorkflowRunRow - | undefined; + WorkflowRunRow | undefined; return Promise.resolve(row ? rowToWorkflowRun(row) : null); } @@ -395,8 +394,7 @@ export class BackendSqlite implements Backend { LIMIT 1 `); const row = stmt.get(this.namespaceId, params.stepAttemptId) as - | { data: string | null } - | undefined; + { data: string | null } | undefined; if (!row) return Promise.resolve(undefined); return Promise.resolve( @@ -1092,8 +1090,7 @@ export class BackendSqlite implements Backend { `); const row = stmt.get(this.namespaceId, params.stepAttemptId) as - | StepAttemptRow - | undefined; + StepAttemptRow | undefined; return Promise.resolve(row ? rowToStepAttempt(row) : null); } From e7fe4c7af1c110a05c253c83655c9f3402f55a48 Mon Sep 17 00:00:00 2001 From: vudc Date: Thu, 30 Jul 2026 00:37:07 +0700 Subject: [PATCH 6/6] refactor(openworkflow, dashboard): resume by marking, not deleting Deleting failed step attempts and clearing the run error on resume erased the record of why a run failed and broke parent/child linkage (children are linked through their parent's step attempt). Resume now preserves all history and resets the retry budget with a marker instead. - Add nullable `resumed_at` to workflow_runs (migration 6, both backends) and `WorkflowRun.resumedAt`. - resumeWorkflowRun stamps `resumed_at` and requeues. It no longer deletes step attempts, clears `error`, or resets `attempts`. The deadline gate and conflict resolver are unchanged. - Step replay counts a failed attempt toward the step retry budget only when it finished at/after `resumed_at`, so the failing step retries fresh while every prior attempt stays as history. Wired through StepHistory and createStepExecutionStateFromAttempts. - Update the resume dialog copy and client/server JSDoc to match. Tests: - Step-history unit tests for the marker filter. - Shared backend testsuite asserts resume preserves every attempt plus the error and stamps resumed_at; deadline rejection and child-workflow preservation covered. - End-to-end tests: the failed attempts are kept, and a mid-workflow failure resumes and continues to the downstream steps. - Populate resumedAt in the WorkflowRun test mocks. --- .../src/components/run-resume-action.tsx | 2 +- apps/dashboard/src/lib/api.ts | 4 +- packages/openworkflow/client/client.test.ts | 1 + packages/openworkflow/client/client.ts | 5 +- .../openworkflow/core/workflow-run.test.ts | 1 + packages/openworkflow/core/workflow-run.ts | 1 + packages/openworkflow/postgres/backend.ts | 80 ++++++---------- packages/openworkflow/postgres/postgres.ts | 12 +++ packages/openworkflow/sqlite/backend.ts | 55 +++-------- packages/openworkflow/sqlite/sqlite.ts | 10 ++ .../openworkflow/testing/backend.testsuite.ts | 45 +++++---- .../openworkflow/worker/execution.test.ts | 93 +++++++++++++++++-- packages/openworkflow/worker/execution.ts | 5 +- .../openworkflow/worker/step-history.test.ts | 39 ++++++++ packages/openworkflow/worker/step-history.ts | 17 +++- 15 files changed, 245 insertions(+), 125 deletions(-) diff --git a/apps/dashboard/src/components/run-resume-action.tsx b/apps/dashboard/src/components/run-resume-action.tsx index 5795a691..f3f5f3a0 100644 --- a/apps/dashboard/src/components/run-resume-action.tsx +++ b/apps/dashboard/src/components/run-resume-action.tsx @@ -23,7 +23,7 @@ export function RunResumeAction({ triggerLabel="Resume Run" triggerVariant="default" title="Resume this failed run?" - description="Completed steps stay cached and won't re-run. The failing step will be retried with a fresh retry budget. Previous failed attempts will be discarded." + description="Completed steps stay cached and won't re-run. The failing step is retried with a fresh retry budget, and the run's history is kept." cancelLabel="Cancel" confirmLabel="Resume Run" pendingLabel="Resuming..." diff --git a/apps/dashboard/src/lib/api.ts b/apps/dashboard/src/lib/api.ts index 59dba33a..c1e38de8 100644 --- a/apps/dashboard/src/lib/api.ts +++ b/apps/dashboard/src/lib/api.ts @@ -92,8 +92,8 @@ export const cancelWorkflowRunServerFn = createServerFn({ method: "POST" }) /** * Resume a failed workflow run by ID. Flips the run back to `pending` and - * drops failed step attempts so the failing step starts with a fresh retry - * budget; completed steps stay cached. + * gives the failing step a fresh retry budget by only counting failures after + * the resume; completed steps stay cached and history is preserved. */ export const resumeWorkflowRunServerFn = createServerFn({ method: "POST" }) .validator(z.object({ workflowRunId: z.string() })) diff --git a/packages/openworkflow/client/client.test.ts b/packages/openworkflow/client/client.test.ts index 80bc7bb1..5ceb2b21 100644 --- a/packages/openworkflow/client/client.test.ts +++ b/packages/openworkflow/client/client.test.ts @@ -698,6 +698,7 @@ function createMockWorkflowRun( deadlineAt: null, startedAt: null, finishedAt: null, + resumedAt: null, createdAt: currentTime, updatedAt: currentTime, ...overrides, diff --git a/packages/openworkflow/client/client.ts b/packages/openworkflow/client/client.ts index 411f4814..c849c150 100644 --- a/packages/openworkflow/client/client.ts +++ b/packages/openworkflow/client/client.ts @@ -178,8 +178,9 @@ export class OpenWorkflow { /** * Resume a failed workflow run. The run's status flips back to `pending` * so the next worker tick picks it up. Already-completed steps are served - * from history without re-executing; failed step attempts are discarded so - * the failing step starts with a fresh retry budget. + * from history without re-executing. Nothing is deleted: the failing step + * starts with a fresh retry budget because only failures recorded after the + * resume count against it. * @param workflowRunId - The ID of the failed workflow run to resume * @returns The updated workflow run * @throws {Error} If the run does not exist or is not in `failed` status diff --git a/packages/openworkflow/core/workflow-run.test.ts b/packages/openworkflow/core/workflow-run.test.ts index 3fa5da9a..9f980009 100644 --- a/packages/openworkflow/core/workflow-run.test.ts +++ b/packages/openworkflow/core/workflow-run.test.ts @@ -169,6 +169,7 @@ describe("resolveCancelWorkflowRunConflict", () => { deadlineAt: null, startedAt: null, finishedAt: null, + resumedAt: null, createdAt: new Date(0), updatedAt: new Date(0), }; diff --git a/packages/openworkflow/core/workflow-run.ts b/packages/openworkflow/core/workflow-run.ts index 7575ab59..dbbeae5d 100644 --- a/packages/openworkflow/core/workflow-run.ts +++ b/packages/openworkflow/core/workflow-run.ts @@ -114,6 +114,7 @@ export interface WorkflowRun { deadlineAt: Date | null; startedAt: Date | null; finishedAt: Date | null; + resumedAt: Date | null; // Timestamp of the most recent resume createdAt: Date; updatedAt: Date; } diff --git a/packages/openworkflow/postgres/backend.ts b/packages/openworkflow/postgres/backend.ts index f473c5d2..d3ee31a5 100644 --- a/packages/openworkflow/postgres/backend.ts +++ b/packages/openworkflow/postgres/backend.ts @@ -768,61 +768,37 @@ export class BackendPostgres implements Backend { async resumeWorkflowRun( params: ResumeWorkflowRunParams, ): Promise { - return await this.withTransaction(async (tx): Promise => { - const workflowRunsTable = this.workflowRunsTable(tx); - const stepAttemptsTable = this.stepAttemptsTable(tx); - - const [updated] = await tx` - UPDATE ${workflowRunsTable} - SET - "status" = 'pending', - "worker_id" = NULL, - "error" = NULL, - "attempts" = 0, - "started_at" = NULL, - "finished_at" = NULL, - "available_at" = NOW(), - "updated_at" = NOW() - WHERE "namespace_id" = ${this.namespaceId} - AND "id" = ${params.workflowRunId} - AND "status" = 'failed' - AND ("deadline_at" IS NULL OR "deadline_at" > NOW()) - RETURNING * - `; - - if (!updated) { - const [existing] = await tx` - SELECT * - FROM ${workflowRunsTable} - WHERE "namespace_id" = ${this.namespaceId} - AND "id" = ${params.workflowRunId} - LIMIT 1 - `; + const workflowRunsTable = this.workflowRunsTable(); - resolveResumeWorkflowRunConflict( - params.workflowRunId, - existing ?? null, - ); - } + // Stamp the resume marker and requeue. Nothing is deleted and neither + // `error` nor `attempts` is touched: step attempts stay (preserving the + // failure record and parent/child linkage), and the retry budget is reset + // by only counting failures after `resumed_at` during replay. + const [updated] = await this.pg` + UPDATE ${workflowRunsTable} + SET + "status" = 'pending', + "worker_id" = NULL, + "started_at" = NULL, + "finished_at" = NULL, + "available_at" = NOW(), + "resumed_at" = NOW(), + "updated_at" = NOW() + WHERE "namespace_id" = ${this.namespaceId} + AND "id" = ${params.workflowRunId} + AND "status" = 'failed' + AND ("deadline_at" IS NULL OR "deadline_at" > NOW()) + RETURNING * + `; - // Drop failed attempts so the failing step gets a fresh retry budget, - // plus inert running attempts whose kinds hold no external state - // (function, signal-send). Running sleep, signal-wait and workflow - // attempts are preserved: replay resumes them, and deleting them would - // orphan linked child runs and already-delivered signals. Successful - // attempts stay and are replayed from cache. - await tx` - DELETE FROM ${stepAttemptsTable} - WHERE "namespace_id" = ${this.namespaceId} - AND "workflow_run_id" = ${params.workflowRunId} - AND ( - "status" = 'failed' - OR ("status" = 'running' AND "kind" IN ('function', 'signal-send')) - ) - `; + if (!updated) { + const existing = await this.getWorkflowRun({ + workflowRunId: params.workflowRunId, + }); + resolveResumeWorkflowRunConflict(params.workflowRunId, existing); + } - return updated; - }); + return updated; } private async wakeParentWorkflowRun( diff --git a/packages/openworkflow/postgres/postgres.ts b/packages/openworkflow/postgres/postgres.ts index f52f6523..92960758 100644 --- a/packages/openworkflow/postgres/postgres.ts +++ b/packages/openworkflow/postgres/postgres.ts @@ -240,6 +240,18 @@ export function migrations(schema: string): string[] { ON CONFLICT DO NOTHING; COMMIT;`, + + // 6 - resume marker + `BEGIN; + + ALTER TABLE ${quotedSchema}."workflow_runs" + ADD COLUMN IF NOT EXISTS "resumed_at" TIMESTAMPTZ; + + INSERT INTO ${quotedSchema}."openworkflow_migrations"("version") + VALUES (6) + ON CONFLICT DO NOTHING; + + COMMIT;`, ]; } diff --git a/packages/openworkflow/sqlite/backend.ts b/packages/openworkflow/sqlite/backend.ts index 7162a21a..7e76e523 100644 --- a/packages/openworkflow/sqlite/backend.ts +++ b/packages/openworkflow/sqlite/backend.ts @@ -748,28 +748,31 @@ export class BackendSqlite implements Backend { params: ResumeWorkflowRunParams, ): Promise { const currentTime = now(); - let resumed = false; - this.db.exec("BEGIN IMMEDIATE"); - try { - const updateStmt = this.db.prepare(` + // Stamp the resume marker and requeue. Nothing is deleted and neither + // `error` nor `attempts` is touched: step attempts stay (preserving the + // failure record and parent/child linkage), and the retry budget is reset + // by only counting failures after `resumed_at` during replay. + const updateResult = this.db + .prepare( + ` UPDATE "workflow_runs" SET "status" = 'pending', "worker_id" = NULL, - "error" = NULL, - "attempts" = 0, "started_at" = NULL, "finished_at" = NULL, "available_at" = ?, + "resumed_at" = ?, "updated_at" = ? WHERE "namespace_id" = ? AND "id" = ? AND "status" = 'failed' AND ("deadline_at" IS NULL OR "deadline_at" > ?) - `); - - const updateResult = updateStmt.run( + `, + ) + .run( + currentTime, currentTime, currentTime, this.namespaceId, @@ -777,41 +780,11 @@ export class BackendSqlite implements Backend { currentTime, ); - resumed = updateResult.changes > 0; - - if (resumed) { - // Drop failed attempts so the failing step gets a fresh retry budget, - // plus inert running attempts whose kinds hold no external state - // (function, signal-send). Running sleep, signal-wait and workflow - // attempts are preserved: replay resumes them, and deleting them - // would orphan linked child runs and already-delivered signals. - // Successful attempts stay and are replayed from cache. - this.db - .prepare( - ` - DELETE FROM "step_attempts" - WHERE "namespace_id" = ? - AND "workflow_run_id" = ? - AND ( - "status" = 'failed' - OR ("status" = 'running' AND "kind" IN ('function', 'signal-send')) - ) - `, - ) - .run(this.namespaceId, params.workflowRunId); - } - - this.db.exec("COMMIT"); - } catch (error) { - this.db.exec("ROLLBACK"); - throw error; - } - const updated = await this.getWorkflowRun({ workflowRunId: params.workflowRunId, }); - if (!resumed) { + if (updateResult.changes === 0) { resolveResumeWorkflowRunConflict(params.workflowRunId, updated); } @@ -1185,6 +1158,7 @@ interface WorkflowRunRow { deadline_at: string | null; started_at: string | null; finished_at: string | null; + resumed_at: string | null; created_at: string; updated_at: string; } @@ -1259,6 +1233,7 @@ function rowToWorkflowRun(row: WorkflowRunRow): WorkflowRun { deadlineAt: fromISO(row.deadline_at), startedAt: fromISO(row.started_at), finishedAt: fromISO(row.finished_at), + resumedAt: fromISO(row.resumed_at), createdAt, updatedAt, }; diff --git a/packages/openworkflow/sqlite/sqlite.ts b/packages/openworkflow/sqlite/sqlite.ts index 8d11736f..71d96f89 100644 --- a/packages/openworkflow/sqlite/sqlite.ts +++ b/packages/openworkflow/sqlite/sqlite.ts @@ -224,6 +224,16 @@ export function migrations(): string[] { VALUES (5); COMMIT;`, + + // 6 - resume marker + `BEGIN; + + ALTER TABLE "workflow_runs" ADD COLUMN "resumed_at" TEXT; + + INSERT OR IGNORE INTO "openworkflow_migrations" ("version") + VALUES (6); + + COMMIT;`, ]; } diff --git a/packages/openworkflow/testing/backend.testsuite.ts b/packages/openworkflow/testing/backend.testsuite.ts index f8749568..27f6ffc7 100644 --- a/packages/openworkflow/testing/backend.testsuite.ts +++ b/packages/openworkflow/testing/backend.testsuite.ts @@ -66,6 +66,7 @@ export function testBackend(options: TestBackendOptions): void { deadlineAt: newDateInOneYear(), startedAt: null, finishedAt: null, + resumedAt: null, createdAt: new Date(), // - updatedAt: new Date(), // - }; @@ -2574,30 +2575,39 @@ export function testBackend(options: TestBackendOptions): void { }); describe("resumeWorkflowRun()", () => { - test("flips a failed run back to pending and clears failure fields", async () => { + test("requeues a failed run and stamps resumed_at without erasing history", async () => { const backend = await setup(); await createPendingWorkflowRun(backend); const failedId = await claimAndFailNextPendingRun(backend); + const failedRun = await backend.getWorkflowRun({ + workflowRunId: failedId, + }); + expect(failedRun?.error).not.toBeNull(); + const resumed = await backend.resumeWorkflowRun({ workflowRunId: failedId, }); expect(resumed.status).toBe("pending"); - expect(resumed.error).toBeNull(); expect(resumed.workerId).toBeNull(); expect(resumed.startedAt).toBeNull(); expect(resumed.finishedAt).toBeNull(); - // run-level retry budget is reset so the resumed run gets fresh retries - expect(resumed.attempts).toBe(0); expect(resumed.availableAt).not.toBeNull(); expect(deltaSeconds(resumed.availableAt)).toBeLessThan(1); + // resume marker is stamped; the budget is reset by counting failures + // after it, not by mutating the run + expect(resumed.resumedAt).not.toBeNull(); + expect(deltaSeconds(resumed.resumedAt)).toBeLessThan(1); + // error and the claim counter are left as the record of what happened + expect(resumed.error).not.toBeNull(); + expect(resumed.attempts).toBe(failedRun?.attempts); await teardown(backend); }); - test("keeps successful and in-flight attempts, drops failed and inert running ones", async () => { + test("preserves every step attempt and the run error on resume", async () => { const backend = await setup(); const claimed = await createClaimedWorkflowRun(backend); @@ -2633,19 +2643,8 @@ export function testBackend(options: TestBackendOptions): void { error: { message: "boom" }, }); - // Inert running function attempt (e.g. a worker that died mid-step): - // safe to drop because nothing references it. - await backend.createStepAttempt({ - workflowRunId: claimed.id, - workerId, - stepName: "running-fn", - kind: "function", - config: {}, - context: null, - }); - - // In-flight durable wait: must be preserved so replay resumes it - // instead of restarting the timer from zero. + // In-flight durable wait: preserved so replay resumes it rather than + // restarting the timer from zero. await backend.createStepAttempt({ workflowRunId: claimed.id, workerId, @@ -2673,15 +2672,21 @@ export function testBackend(options: TestBackendOptions): void { }); expect(failedRun?.status).toBe("failed"); - await backend.resumeWorkflowRun({ workflowRunId: claimed.id }); + const resumed = await backend.resumeWorkflowRun({ + workflowRunId: claimed.id, + }); + // the failure record survives the resume + expect(resumed.error).not.toBeNull(); + // nothing is deleted: completed, failed and running attempts all remain const attempts = await backend.listStepAttempts({ workflowRunId: claimed.id, limit: 100, }); const survivingNames = attempts.data.map((a) => a.stepName); - expect(survivingNames).toHaveLength(2); + expect(survivingNames).toHaveLength(3); expect(survivingNames).toContain("completed-step"); + expect(survivingNames).toContain("failed-step"); expect(survivingNames).toContain("running-sleep"); await teardown(backend); diff --git a/packages/openworkflow/worker/execution.test.ts b/packages/openworkflow/worker/execution.test.ts index 78048224..acc118a3 100644 --- a/packages/openworkflow/worker/execution.test.ts +++ b/packages/openworkflow/worker/execution.test.ts @@ -3048,7 +3048,7 @@ describe("StepExecutor", () => { const failedBefore = stepsBeforeResume.data.filter( (s) => s.status === "failed", ); - expect(failedBefore.length).toBe(2); + expect(failedBefore).toHaveLength(2); shouldFail = false; await backend.resumeWorkflowRun({ workflowRunId: handle.workflowRun.id }); @@ -3057,21 +3057,23 @@ describe("StepExecutor", () => { workflowRunId: handle.workflowRun.id, }); expect(resumedRun?.status).toBe("pending"); - expect(resumedRun?.error).toBeNull(); expect(resumedRun?.startedAt).toBeNull(); expect(resumedRun?.finishedAt).toBeNull(); expect(resumedRun?.workerId).toBeNull(); + // resume marker is stamped; the failure record is left intact + expect(resumedRun?.resumedAt).not.toBeNull(); + expect(resumedRun?.error).not.toBeNull(); const stepsAfterResume = await backend.listStepAttempts({ workflowRunId: handle.workflowRun.id, limit: 100, }); - // only successful attempts survive; failed and still-running rows are gone + // history is preserved: the two failed "flaky" attempts still exist expect( - stepsAfterResume.data.every( - (s) => s.status === "completed" || s.status === "succeeded", + stepsAfterResume.data.filter( + (s) => s.stepName === "flaky" && s.status === "failed", ), - ).toBe(true); + ).toHaveLength(2); expect( stepsAfterResume.data.some( (s) => s.stepName === "validate" && s.status === "completed", @@ -3096,6 +3098,84 @@ describe("StepExecutor", () => { expect(result).toEqual({ validated: "ok", flaky: "recovered" }); }, 30_000); + test("resumeWorkflowRun continues past the fixed step to downstream steps", async () => { + const backend = await createTestBackend(); + const client = new OpenWorkflow({ backend }); + + const ran: string[] = []; + let gatewayDown = true; + + const workflow = client.defineWorkflow( + { name: `resume-middle-step-${randomUUID()}` }, + async ({ step }) => { + await step.run({ name: "validate" }, () => { + ran.push("validate"); + return "ok"; + }); + + await step.run( + { name: "reserve", retryPolicy: { maximumAttempts: 2 } }, + () => { + ran.push("reserve"); + if (gatewayDown) { + throw new Error("gateway down"); + } + return "auth"; + }, + ); + + await step.run({ name: "confirm" }, () => { + ran.push("confirm"); + return "receipt"; + }); + + await step.run({ name: "send-receipt" }, () => { + ran.push("send-receipt"); + }); + + return "done"; + }, + ); + + const worker = client.newWorker({ concurrency: 1 }); + const handle = await workflow.run(); + + const failedStatus = await tickUntilTerminal( + backend, + worker, + handle.workflowRun.id, + 40, + 25, + { maxWaitMs: 20_000 }, + ); + expect(failedStatus).toBe("failed"); + // validate completed once; reserve exhausted its 2 attempts; the steps + // after the failing one never ran + expect(ran.filter((s) => s === "validate")).toHaveLength(1); + expect(ran.filter((s) => s === "reserve")).toHaveLength(2); + expect(ran).not.toContain("confirm"); + expect(ran).not.toContain("send-receipt"); + + gatewayDown = false; + await backend.resumeWorkflowRun({ workflowRunId: handle.workflowRun.id }); + + const finalStatus = await tickUntilTerminal( + backend, + worker, + handle.workflowRun.id, + 40, + 25, + { maxWaitMs: 20_000 }, + ); + expect(finalStatus).toBe("completed"); + // validate stayed cached (not re-run); reserve retried once more with a + // fresh budget and succeeded; the downstream steps now run + expect(ran.filter((s) => s === "validate")).toHaveLength(1); + expect(ran.filter((s) => s === "reserve")).toHaveLength(3); + expect(ran).toContain("confirm"); + expect(ran).toContain("send-receipt"); + }, 30_000); + test("resumeWorkflowRun throws when the run is not in failed status", async () => { const backend = await createTestBackend(); const client = new OpenWorkflow({ backend }); @@ -4194,6 +4274,7 @@ function createMockWorkflowRun( deadlineAt: null, startedAt: new Date("2026-01-01T00:00:00.000Z"), finishedAt: null, + resumedAt: null, createdAt: new Date("2026-01-01T00:00:00.000Z"), updatedAt: new Date("2026-01-01T00:00:00.000Z"), ...overrides, diff --git a/packages/openworkflow/worker/execution.ts b/packages/openworkflow/worker/execution.ts index b3031b02..35917254 100644 --- a/packages/openworkflow/worker/execution.ts +++ b/packages/openworkflow/worker/execution.ts @@ -1038,7 +1038,10 @@ export async function executeWorkflow( backend, workflowRun.id, ); - const history = new StepHistory({ attempts }); + const history = new StepHistory({ + attempts, + resumedAt: workflowRun.resumedAt, + }); // Complete any elapsed sleep waits first, then park on the earliest // remaining running wait (sleep or runWorkflow timeout). diff --git a/packages/openworkflow/worker/step-history.test.ts b/packages/openworkflow/worker/step-history.test.ts index a218c6f3..7be501e5 100644 --- a/packages/openworkflow/worker/step-history.test.ts +++ b/packages/openworkflow/worker/step-history.test.ts @@ -151,6 +151,45 @@ describe("StepHistory", () => { expect(history.findRunning("a")).toBeUndefined(); }); + test("failures before resumedAt are excluded from the retry budget", () => { + const resumedAt = new Date("2026-01-01T12:00:00.000Z"); + const beforeResume = createMockStepAttempt({ + stepName: "a", + status: "failed", + finishedAt: new Date("2026-01-01T11:59:59.000Z"), + }); + const afterResume = createMockStepAttempt({ + stepName: "a", + status: "failed", + finishedAt: new Date("2026-01-01T12:00:01.000Z"), + }); + + const history = new StepHistory({ + attempts: [beforeResume, afterResume], + resumedAt, + }); + + // only the post-resume failure counts, but both rows remain history + expect(history.failedAttemptCount("a")).toBe(1); + }); + + test("without resumedAt every failure counts", () => { + const first = createMockStepAttempt({ + stepName: "a", + status: "failed", + finishedAt: new Date("2026-01-01T11:00:00.000Z"), + }); + const second = createMockStepAttempt({ + stepName: "a", + status: "failed", + finishedAt: new Date("2026-01-01T12:00:00.000Z"), + }); + + const history = new StepHistory({ attempts: [first, second] }); + + expect(history.failedAttemptCount("a")).toBe(2); + }); + test("replaceRunningAttempt updates the running entry in place", () => { const initial = createMockStepAttempt({ id: "attempt-1", diff --git a/packages/openworkflow/worker/step-history.ts b/packages/openworkflow/worker/step-history.ts index 39532890..24999c99 100644 --- a/packages/openworkflow/worker/step-history.ts +++ b/packages/openworkflow/worker/step-history.ts @@ -41,10 +41,13 @@ export interface StepExecutionState { /** * Build step execution state from loaded attempts in one pass. * @param attempts - Loaded step attempts for the workflow run + * @param resumedAt - Most recent resume timestamp; failures that finished + * before it are kept as history but not counted against the retry budget * @returns Successful cache plus failed-attempt counts by step name */ export function createStepExecutionStateFromAttempts( attempts: readonly StepAttempt[], + resumedAt: Readonly | null = null, ): StepExecutionState { const cache = new Map(); const failedCountsByStepName = new Map(); @@ -58,6 +61,14 @@ export function createStepExecutionStateFromAttempts( } if (attempt.status === "failed") { + // Failures from before the latest resume stay in history (linkage, + // diagnostics) but don't count toward the step's retry budget. + if ( + resumedAt !== null && + (attempt.finishedAt === null || attempt.finishedAt < resumedAt) + ) { + continue; + } const previousCount = failedCountsByStepName.get(attempt.stepName) ?? 0; failedCountsByStepName.set(attempt.stepName, previousCount + 1); failedByStepName.set(attempt.stepName, attempt); @@ -169,6 +180,7 @@ function getEarliestRunningWaitResumeAt( export interface StepHistoryOptions { attempts: readonly StepAttempt[]; stepLimit?: number; + resumedAt?: Readonly | null; } /** @@ -192,7 +204,10 @@ export class StepHistory { this.stepLimit = Math.max(1, options.stepLimit ?? WORKFLOW_STEP_LIMIT); this.stepCount = options.attempts.length; - const state = createStepExecutionStateFromAttempts(options.attempts); + const state = createStepExecutionStateFromAttempts( + options.attempts, + options.resumedAt ?? null, + ); this.cache = state.cache; this.failedCountsByStepName = new Map(state.failedCountsByStepName); this.failedByStepName = new Map(state.failedByStepName);