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 new file mode 100644 index 00000000..f3f5f3a0 --- /dev/null +++ b/apps/dashboard/src/components/run-resume-action.tsx @@ -0,0 +1,37 @@ +import { RunActionDialog } from "@/components/run-action-dialog"; +import { resumeWorkflowRunServerFn } from "@/lib/api"; +import { isRunResumableStatus } from "@/lib/status"; +import type { WorkflowRunStatus } from "openworkflow/internal"; + +interface RunResumeActionProps { + runId: string; + status: WorkflowRunStatus; + onResumed?: (() => Promise) | (() => void); +} + +export function RunResumeAction({ + runId, + status, + onResumed, +}: RunResumeActionProps) { + if (!isRunResumableStatus(status)) { + return null; + } + + return ( + + resumeWorkflowRunServerFn({ data: { workflowRunId: runId } }) + } + onDone={onResumed} + /> + ); +} diff --git a/apps/dashboard/src/lib/api.ts b/apps/dashboard/src/lib/api.ts index d399ef81..c1e38de8 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 + * 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() })) + .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(); + }} + /> { + return 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.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 ae967f3c..dbbeae5d 100644 --- a/packages/openworkflow/core/workflow-run.ts +++ b/packages/openworkflow/core/workflow-run.ts @@ -58,6 +58,39 @@ 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'` 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 + * @throws {Error} If the run is missing, not `failed`, or past its deadline + */ +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`); + } + + 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`, + ); +} + /** * WorkflowRun represents a single execution instance of a workflow. */ @@ -81,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 77292a78..d3ee31a5 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,42 @@ export class BackendPostgres implements Backend { return updated; } + async resumeWorkflowRun( + params: ResumeWorkflowRunParams, + ): Promise { + const workflowRunsTable = this.workflowRunsTable(); + + // 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 * + `; + + if (!updated) { + const existing = await this.getWorkflowRun({ + workflowRunId: params.workflowRunId, + }); + resolveResumeWorkflowRunConflict(params.workflowRunId, existing); + } + + return updated; + } + private async wakeParentWorkflowRun( childWorkflowRun: Readonly, ): Promise { 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 fa610357..7e76e523 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,54 @@ export class BackendSqlite implements Backend { return updated; } + async resumeWorkflowRun( + params: ResumeWorkflowRunParams, + ): Promise { + const currentTime = now(); + + // 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, + "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" > ?) + `, + ) + .run( + currentTime, + currentTime, + currentTime, + this.namespaceId, + params.workflowRunId, + currentTime, + ); + + const updated = await this.getWorkflowRun({ + workflowRunId: params.workflowRunId, + }); + + if (updateResult.changes === 0) { + resolveResumeWorkflowRunConflict(params.workflowRunId, updated); + } + + 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. @@ -1108,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; } @@ -1182,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 22e5291b..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(), // - }; @@ -2573,6 +2574,271 @@ export function testBackend(options: TestBackendOptions): void { }); }); + describe("resumeWorkflowRun()", () => { + 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.workerId).toBeNull(); + expect(resumed.startedAt).toBeNull(); + expect(resumed.finishedAt).toBeNull(); + 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("preserves every step attempt and the run error on resume", 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" }, + }); + + // In-flight durable wait: preserved so replay resumes it rather than + // restarting the timer from zero. + await backend.createStepAttempt({ + workflowRunId: claimed.id, + workerId, + stepName: "running-sleep", + 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"); + + 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(3); + expect(survivingNames).toContain("completed-step"); + expect(survivingNames).toContain("failed-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); + }); + + 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 5a30ab58..acc118a3 100644 --- a/packages/openworkflow/worker/execution.test.ts +++ b/packages/openworkflow/worker/execution.test.ts @@ -2994,6 +2994,213 @@ 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, + { maxWaitMs: 20_000 }, + ); + 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).toHaveLength(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?.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, + }); + // history is preserved: the two failed "flaky" attempts still exist + expect( + stepsAfterResume.data.filter( + (s) => s.stepName === "flaky" && s.status === "failed", + ), + ).toHaveLength(2); + expect( + stepsAfterResume.data.some( + (s) => s.stepName === "validate" && s.status === "completed", + ), + ).toBe(true); + + const finalStatus = await tickUntilTerminal( + backend, + worker, + handle.workflowRun.id, + 40, + 25, + { maxWaitMs: 20_000 }, + ); + 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" }); + }, 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 }); + + 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/); + }); + + 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", () => { @@ -4067,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);