From b0d71bdcac9670d0fd9564ee18fbb4b26fa220e3 Mon Sep 17 00:00:00 2001 From: vudc Date: Mon, 18 May 2026 23:35:53 +0700 Subject: [PATCH 1/5] 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/5] 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/5] 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/5] 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/5] 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); }