diff --git a/apps/dashboard/src/components/run-action.tsx b/apps/dashboard/src/components/run-action.tsx new file mode 100644 index 00000000..ba2b12b9 --- /dev/null +++ b/apps/dashboard/src/components/run-action.tsx @@ -0,0 +1,134 @@ +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, +} from "@/components/ui/alert-dialog"; +import { Button } from "@/components/ui/button"; +import { cancelWorkflowRunServerFn, rerunWorkflowRunServerFn } from "@/lib/api"; +import { isRunCancelableStatus, TERMINAL_RUN_STATUSES } from "@/lib/status"; +import { useNavigate } from "@tanstack/react-router"; +import type { WorkflowRunStatus } from "openworkflow/internal"; +import { useState } from "react"; + +interface RunActionProps { + action?: "cancel" | "rerun"; + fromStep?: string; + runId: string; + status: WorkflowRunStatus; + onDone?: (() => Promise) | (() => void); +} + +function getErrorMessage(cause: unknown): string { + if (cause instanceof Error && cause.message) { + return cause.message; + } + + return "Unable to update workflow run"; +} + +export function RunAction({ + action = "cancel", + fromStep, + runId, + status, + onDone, +}: RunActionProps) { + const navigate = useNavigate(); + const rerun = action === "rerun"; + const rerunLabel = fromStep === undefined ? "Rerun" : "Rerun from step"; + const actionLabel = rerun ? rerunLabel : "Cancel Run"; + const rerunDescription = + fromStep === undefined + ? "Creates a new run with the original input and version. All steps will execute again." + : `Creates a new run using successful results recorded before "${fromStep}". This step, later work, and earlier failed or unfinished steps will execute again if reached.`; + const [isOpen, setIsOpen] = useState(false); + const [isPending, setIsPending] = useState(false); + const [error, setError] = useState(null); + + if ( + rerun ? !TERMINAL_RUN_STATUSES.has(status) : !isRunCancelableStatus(status) + ) { + return null; + } + + async function performAction() { + setIsPending(true); + setError(null); + + try { + if (rerun) { + const run = await rerunWorkflowRunServerFn({ + data: { workflowRunId: runId, fromStep }, + }); + await navigate({ to: "/runs/$runId", params: { runId: run.id } }); + } else { + await cancelWorkflowRunServerFn({ data: { workflowRunId: runId } }); + } + await onDone?.(); + setIsOpen(false); + } catch (caughtError) { + setError(getErrorMessage(caughtError)); + } finally { + setIsPending(false); + } + } + + return ( + { + setIsOpen(nextOpen); + if (!nextOpen) { + setError(null); + } + }} + > + + + + + + {rerun ? `${rerunLabel}?` : "Cancel this run?"} + + + {rerun + ? rerunDescription + : "This will stop any future progress for this workflow run."} + + + + {error &&

{error}

} + + + + {rerun ? "Cancel" : "Keep Running"} + + { + event.preventDefault(); + void performAction(); + }} + disabled={isPending} + > + {isPending ? "Working..." : actionLabel} + + + + + ); +} diff --git a/apps/dashboard/src/components/run-cancel-action.tsx b/apps/dashboard/src/components/run-cancel-action.tsx deleted file mode 100644 index 4d5a017b..00000000 --- a/apps/dashboard/src/components/run-cancel-action.tsx +++ /dev/null @@ -1,111 +0,0 @@ -import { - AlertDialog, - AlertDialogAction, - AlertDialogCancel, - AlertDialogContent, - AlertDialogDescription, - AlertDialogFooter, - AlertDialogHeader, - AlertDialogTitle, -} from "@/components/ui/alert-dialog"; -import { Button } from "@/components/ui/button"; -import { cancelWorkflowRunServerFn } from "@/lib/api"; -import { isRunCancelableStatus } from "@/lib/status"; -import type { WorkflowRunStatus } from "openworkflow/internal"; -import { useState } from "react"; - -interface RunCancelActionProps { - runId: string; - status: WorkflowRunStatus; - onCanceled?: (() => Promise) | (() => void); -} - -function getErrorMessage(cause: unknown): string { - if (cause instanceof Error && cause.message) { - return cause.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"} - - - - - ); -} diff --git a/apps/dashboard/src/lib/api.ts b/apps/dashboard/src/lib/api.ts index 1e6eabd7..fa2cb945 100644 --- a/apps/dashboard/src/lib/api.ts +++ b/apps/dashboard/src/lib/api.ts @@ -1,5 +1,6 @@ import { getBackend } from "./backend"; import { createServerFn } from "@tanstack/react-start"; +import { rerunWorkflowRun } from "openworkflow/internal"; import type { PaginatedResponse, PaginationOptions, @@ -90,6 +91,19 @@ export const cancelWorkflowRunServerFn = createServerFn({ method: "POST" }) return backend.cancelWorkflowRun({ workflowRunId: data.workflowRunId }); }); +/** Rerun a finished workflow with its original input and version. */ +export const rerunWorkflowRunServerFn = createServerFn({ method: "POST" }) + .validator( + z.object({ workflowRunId: z.string(), fromStep: z.string().optional() }), + ) + .handler(async ({ data }): Promise => { + return await rerunWorkflowRun( + await getBackend(), + data.workflowRunId, + data.fromStep, + ); + }); + /** * List step attempts for a workflow run. */ diff --git a/apps/dashboard/src/routes/runs/$runId.tsx b/apps/dashboard/src/routes/runs/$runId.tsx index 275c449c..18a92a6d 100644 --- a/apps/dashboard/src/routes/runs/$runId.tsx +++ b/apps/dashboard/src/routes/runs/$runId.tsx @@ -1,6 +1,6 @@ import { AppLayout } from "@/components/app-layout"; import { CursorPaginationControls } from "@/components/cursor-pagination-controls"; -import { RunCancelAction } from "@/components/run-cancel-action"; +import { RunAction } from "@/components/run-action"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { Card } from "@/components/ui/card"; @@ -285,11 +285,12 @@ function RunDetailsPage() { /> )} -
- + + { + onDone={async () => { await router.invalidate(); }} /> @@ -476,6 +477,7 @@ function RunDetailsPage() {

Step Inspector

{step.stepName}

+
diff --git a/apps/docs/docs/retries.mdx b/apps/docs/docs/retries.mdx index 1f8b47db..0f71fe9f 100644 --- a/apps/docs/docs/retries.mdx +++ b/apps/docs/docs/retries.mdx @@ -28,6 +28,24 @@ To prevent runaway executions, each workflow run also has a hard cap of 1000 total step attempts. If the run reaches that cap, it fails immediately and is not retried. +## Rerun a Finished Workflow + +Use `rerunWorkflowRun` to create a new run from a failed, completed, or canceled +run. Optionally pass a recorded step name to reuse earlier results: + +```ts +const newRun = await ow.rerunWorkflowRun(originalRunId, { + fromStep: "send-email", // <-- optional +}); +``` + +The new run uses the original input and workflow version. It is ready to run +immediately, with fresh retry counts and no deadline, idempotency key, or parent +run. The source run stays unchanged. + +With `fromStep`, OpenWorkflow copies successful steps from before the selected +step. The new run reuses those results. + ## Step Retries Steps that throw are retried automatically: diff --git a/cspell.config.jsonc b/cspell.config.jsonc index 2faae4e4..51b202a0 100644 --- a/cspell.config.jsonc +++ b/cspell.config.jsonc @@ -30,9 +30,10 @@ "trivago", "tsgolint", - // postgres + // SQL "hashtextextended", "mydb", + "randomblob", "sslmode", "timestamptz", "xact", diff --git a/oxlint.config.ts b/oxlint.config.ts index ac3abf5f..bc02e347 100644 --- a/oxlint.config.ts +++ b/oxlint.config.ts @@ -197,7 +197,7 @@ export default defineConfig({ files: [ "apps/cli/commands.ts", "apps/dashboard/src/components/create-run-form.tsx", - "apps/dashboard/src/components/run-cancel-action.tsx", + "apps/dashboard/src/components/run-action.tsx", "apps/dashboard/src/components/run-list.tsx", "apps/dashboard/src/routes/index.tsx", "apps/dashboard/src/routes/runs/$runId.tsx", diff --git a/packages/openworkflow/client/client.ts b/packages/openworkflow/client/client.ts index 3947ccef..bd77473b 100644 --- a/packages/openworkflow/client/client.ts +++ b/packages/openworkflow/client/client.ts @@ -28,6 +28,7 @@ import { traceOperation, } from "../telemetry.js"; import { Worker } from "../worker/worker.js"; +import { rerunWorkflowRun } from "./rerun.js"; const DEFAULT_RESULT_POLL_INTERVAL_MS = 1000; // 1s const DEFAULT_RESULT_TIMEOUT_MS = 5 * 60 * 1000; // 5m @@ -199,6 +200,25 @@ export class OpenWorkflow { await cancelWorkflowRun(this.backend, workflowRunId); } + /** + * Rerun a finished workflow with its original input and version. + * @param workflowRunId - ID of a failed, completed, or canceled run + * @param options - Rerun options + * @param options.fromStep - Recorded step name to rerun, reusing earlier + * successful results; omit to run every step again + * @returns A new pending run, leaving the source run unchanged + */ + async rerunWorkflowRun( + workflowRunId: string, + options?: { fromStep?: string }, + ): Promise { + return await rerunWorkflowRun( + this.backend, + workflowRunId, + options?.fromStep, + ); + } + /** * 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/client/rerun.test.ts b/packages/openworkflow/client/rerun.test.ts new file mode 100644 index 00000000..fdf6e293 --- /dev/null +++ b/packages/openworkflow/client/rerun.test.ts @@ -0,0 +1,547 @@ +import { isTerminalStatus } from "../core/workflow-run.js"; +import { BackendPostgres } from "../postgres/backend.js"; +import { createTestBackend } from "../postgres/test-backend.testsuite.js"; +import { BackendSqlite } from "../sqlite/backend.js"; +import { OpenWorkflow } from "./client.js"; +import assert from "node:assert/strict"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; + +describe.each(["sqlite", "postgres"])("reruns (%s)", (database) => { + let backend: BackendSqlite | BackendPostgres; + let client: OpenWorkflow; + beforeEach(async () => { + backend = + database === "sqlite" + ? BackendSqlite.connect(":memory:") + : await createTestBackend(); + client = new OpenWorkflow({ backend }); + }); + afterEach(async () => { + vi.restoreAllMocks(); + if (database === "sqlite") await backend.stop(); + }); + + async function finish(workflowRunId: string) { + const worker = client.newWorker(); + for (let tick = 0; tick < 100; tick++) { + await worker.tick(); + const run = await backend.getWorkflowRun({ workflowRunId }); + if (run && isTerminalStatus(run.status)) return run; + await new Promise((resolve) => { + setTimeout(resolve, 5); + }); + } + throw new Error("Run did not finish"); + } + + async function history(workflowRunId: string) { + const result = await backend.listStepAttempts({ + workflowRunId, + limit: 1000, + }); + return result.data; + } + + test("copies successful results before execution, preserves the source, and gives fresh retries", async () => { + const calls = { read: 0, write: 0 }; + let broken = true; + const workflow = client.defineWorkflow<{ value: number }, number>( + { name: "recover", version: "v1" }, + async ({ step, input }) => { + const value = await step.run({ name: "read" }, () => { + calls.read++; + return input.value; + }); + return await step.run( + { + name: "write", + retryPolicy: { maximumAttempts: 2, initialInterval: "1ms" }, + }, + () => { + calls.write++; + if (broken) throw new Error("offline"); + return value; + }, + ); + }, + ); + const original = await workflow.run( + { value: 42 }, + { idempotencyKey: "request" }, + ); + const source = await finish(original.workflowRun.id); + expect(source.status).toBe("failed"); + const sourceHistory = await history(source.id); + const read = sourceHistory.find((step) => step.stepName === "read"); + assert.ok(read); + const rerun = await client.rerunWorkflowRun(source.id, { + fromStep: "write", + }); + expect(rerun).toMatchObject({ + input: { value: 42 }, + version: "v1", + status: "pending", + attempts: 0, + error: null, + context: null, + idempotencyKey: null, + deadlineAt: null, + parentStepAttemptNamespaceId: null, + parentStepAttemptId: null, + }); + const copied = await history(rerun.id); + expect(copied).toHaveLength(1); + expect(copied[0]).toEqual({ + ...read, + id: copied[0]?.id, + workflowRunId: rerun.id, + }); + expect(copied[0]?.id).not.toBe(read.id); + await expect(finish(rerun.id)).resolves.toMatchObject({ status: "failed" }); + expect(calls).toEqual({ read: 1, write: 4 }); + + broken = false; + const recovered = await client.rerunWorkflowRun(rerun.id, { + fromStep: "write", + }); + await expect(finish(recovered.id)).resolves.toMatchObject({ output: 42 }); + expect(calls).toEqual({ read: 1, write: 5 }); + const fresh = await client.rerunWorkflowRun(recovered.id); + expect(await history(fresh.id)).toEqual([]); + expect(fresh.context).toBeNull(); + await expect(finish(fresh.id)).resolves.toMatchObject({ output: 42 }); + expect(calls).toEqual({ read: 2, write: 6 }); + expect(await backend.getWorkflowRun({ workflowRunId: source.id })).toEqual( + source, + ); + expect(await history(source.id)).toEqual(sourceHistory); + }); + + test("copies successful steps by index", async () => { + const source = await backend.createWorkflowRun({ + workflowName: "snapshot", + version: null, + input: null, + config: {}, + context: null, + idempotencyKey: null, + parentStepAttemptNamespaceId: null, + parentStepAttemptId: null, + availableAt: null, + deadlineAt: null, + }); + const workerId = "seed"; + await backend.claimWorkflowRun({ workerId, leaseDurationMs: 60_000 }); + async function attempt( + stepName: string, + stepIndex: number, + status: "completed" | "failed" | "running", + ) { + const step = await backend.createStepAttempt({ + workflowRunId: source.id, + workerId, + stepName, + stepIndex, + kind: "function", + config: {}, + context: null, + }); + if (status === "completed") + await backend.completeStepAttempt({ + workflowRunId: source.id, + workerId, + stepAttemptId: step.id, + output: stepName, + }); + if (status === "failed") + await backend.failStepAttempt({ + workflowRunId: source.id, + workerId, + stepAttemptId: step.id, + error: { message: "failed" }, + }); + return step; + } + await attempt("caught", 0, "failed"); + await attempt("unfinished", 1, "running"); + await attempt("first", 2, "failed"); + const first = await attempt("first", 2, "completed"); + const second = await attempt("second", 3, "completed"); + const third = await attempt("third", 4, "completed"); + await attempt("target", 5, "failed"); + await attempt("later", 6, "completed"); + await attempt("target", 5, "completed"); + // Earlier writes, tied timestamps, and replacement UUIDs cannot change the prefix. + const early = new Date("2026-01-01T00:00:00Z"); + const late = new Date("2026-01-02T00:00:00Z"); + if (backend instanceof BackendSqlite) { + backend["db"] + .prepare( + 'UPDATE "step_attempts" SET "created_at" = ? WHERE "workflow_run_id" = ?', + ) + .run(late.toISOString(), source.id); + backend["db"] + .prepare( + 'UPDATE "step_attempts" SET "status" = \'succeeded\' WHERE "id" = ?', + ) + .run(first.id); + backend["db"] + .prepare( + 'UPDATE "step_attempts" SET "created_at" = ? WHERE "workflow_run_id" = ? AND "step_name" = \'target\'', + ) + .run(early.toISOString(), source.id); + } else { + const pg = backend["pg"]; + const table = backend["stepAttemptsTable"](); + await pg`UPDATE ${table} SET "created_at" = ${late} WHERE "workflow_run_id" = ${source.id}`; + await pg`UPDATE ${table} SET "status" = 'succeeded' WHERE "id" = ${first.id}`; + await pg`UPDATE ${table} SET "created_at" = ${early} WHERE "workflow_run_id" = ${source.id} AND "step_name" = 'target'`; + } + await backend.cancelWorkflowRun({ workflowRunId: source.id }); + const sourceHistory = await history(source.id); + const saved = sourceHistory.filter( + (step) => + step.id === first.id || step.id === second.id || step.id === third.id, + ); + const context = { traceContext: { traceparent: "new-trace" } }; + const rerun = await backend.rerunWorkflowRun({ + workflowRunId: source.id, + fromStep: "target", + context, + }); + expect(rerun.context).toEqual({ + ...context, + rerunStepIndices: { caught: 0, unfinished: 1 }, + }); + expect(context).toEqual({ traceContext: { traceparent: "new-trace" } }); + const copies = await history(rerun.id); + expect(copies.map((step) => step.stepName).toSorted()).toEqual( + saved.map((step) => step.stepName).toSorted(), + ); + for (const step of copies) { + const original = saved.find((entry) => entry.stepName === step.stepName); + assert.ok(original); + expect(step.id).not.toBe(original.id); + expect(step.id).toMatch( + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/, + ); + expect(step).toEqual({ + ...original, + id: step.id, + workflowRunId: rerun.id, + }); + } + await client.cancelWorkflowRun(rerun.id); + const chained = await client.rerunWorkflowRun(rerun.id, { + fromStep: "third", + }); + expect(chained.context).toEqual({ + rerunStepIndices: { caught: 0, unfinished: 1 }, + }); + const chainedHistory = await history(chained.id); + expect(chainedHistory.map((step) => step.stepName).toSorted()).toEqual([ + "first", + "second", + ]); + }); + + test("preserves caught-failure order across chained reruns", async () => { + let failures = 0; + let savedCalls = 0; + const workflow = client.defineWorkflow( + { name: "caught" }, + async ({ step }) => { + await step + .run({ name: "error", retryPolicy: { maximumAttempts: 1 } }, () => { + failures++; + throw new Error("caught"); + }) + .catch(() => null); + await step.run({ name: "saved" }, () => ++savedCalls); + return await step.run({ name: "target" }, () => "ok"); + }, + ); + const source = await workflow.run(); + await finish(source.workflowRun.id); + const rerun = await client.rerunWorkflowRun(source.workflowRun.id, { + fromStep: "target", + }); + expect(rerun.context).toEqual({ rerunStepIndices: { error: 0 } }); + await expect(finish(rerun.id)).resolves.toMatchObject({ output: "ok" }); + expect(failures).toBe(2); + expect(savedCalls).toBe(1); + + const chained = await client.rerunWorkflowRun(rerun.id, { + fromStep: "error", + }); + expect(await history(chained.id)).toEqual([]); + expect(chained.context).toBeNull(); + await expect(finish(chained.id)).resolves.toMatchObject({ output: "ok" }); + expect(failures).toBe(3); + expect(savedCalls).toBe(2); + }); + + test("replays saved signal results without deliveries or the source run", async () => { + const workflow = client.defineWorkflow( + { name: "signal" }, + async ({ step }) => { + const result = await step.waitForSignal<{ value: number }>({ + signal: "reply", + }); + await new Promise((resolve) => { + setTimeout(resolve, 2); + }); + return await step.run({ name: "target" }, () => result); + }, + ); + const original = await workflow.run(); + await client.newWorker().tick(); + await client.sendSignal({ + signal: "reply", + data: { value: 42 }, + idempotencyKey: "delivery", + }); + await finish(original.workflowRun.id); + const rerun = await client.rerunWorkflowRun(original.workflowRun.id, { + fromStep: "target", + }); + const copied = await history(rerun.id); + expect(copied).toHaveLength(1); + const savedWait = copied[0]; + assert.ok(savedWait); + await expect( + backend.getSignalDelivery({ stepAttemptId: savedWait.id }), + ).resolves.toBeUndefined(); + if (backend instanceof BackendSqlite) { + backend["db"] + .prepare('DELETE FROM "workflow_signals" WHERE "workflow_run_id" = ?') + .run(original.workflowRun.id); + backend["db"] + .prepare('DELETE FROM "workflow_runs" WHERE "id" = ?') + .run(original.workflowRun.id); + } else { + const pg = backend["pg"]; + await pg`DELETE FROM ${backend["workflowSignalsTable"]()} WHERE "workflow_run_id" = ${original.workflowRun.id}`; + await pg`DELETE FROM ${backend["workflowRunsTable"]()} WHERE "id" = ${original.workflowRun.id}`; + } + const deliveries = vi.spyOn(backend, "getSignalDelivery"); + await expect(finish(rerun.id)).resolves.toMatchObject({ + status: "completed", + output: { data: { value: 42 } }, + }); + expect(deliveries).not.toHaveBeenCalled(); + }); + + test("rolls back if rerun creation fails after inserting the new run", async () => { + const workflow = client.defineWorkflow( + { name: "atomic" }, + async ({ step }) => { + await step.run({ name: "prefix" }, async () => { + await new Promise((resolve) => { + setTimeout(resolve, 2); + }); + return "saved"; + }); + await step.run({ name: "target" }, () => null); + }, + ); + const original = await workflow.run(); + await finish(original.workflowRun.id); + const before = await backend.listWorkflowRuns({}); + let restore: () => void; + if (backend instanceof BackendSqlite) { + const sqlite = backend; + const insert = sqlite["insertWorkflowRun"]; + sqlite["insertWorkflowRun"] = (params) => { + insert.call(sqlite, params); + throw new Error("copy failed"); + }; + restore = () => { + sqlite["insertWorkflowRun"] = insert; + }; + } else { + const postgres = backend; + const insert = postgres["insertWorkflowRun"]; + postgres["insertWorkflowRun"] = async (tx, params) => { + await insert.call(postgres, tx, params); + throw new Error("copy failed"); + }; + restore = () => { + postgres["insertWorkflowRun"] = insert; + }; + } + try { + await expect( + client.rerunWorkflowRun(original.workflowRun.id, { + fromStep: "target", + }), + ).rejects.toThrow("copy failed"); + } finally { + restore(); + } + expect(await backend.listWorkflowRuns({})).toEqual(before); + const rerun = await client.rerunWorkflowRun(original.workflowRun.id, { + fromStep: "target", + }); + expect(await history(rerun.id)).toHaveLength(1); + }); + + test("starts fresh children at the restart point and reuses completed children before it", async () => { + let childCalls = 0; + let broken = true; + client.defineWorkflow({ name: "child" }, async ({ step }) => { + await step.run({ name: "earlier-child-work" }, () => ++childCalls); + return await step.run( + { name: "child-failure", retryPolicy: { maximumAttempts: 1 } }, + () => { + if (broken) throw new Error("child failed"); + return "child result"; + }, + ); + }); + const parent = client.defineWorkflow( + { name: "parent" }, + async ({ step }) => { + const result = await step.runWorkflow({ name: "child" }); + return await step.run({ name: "after-child" }, () => result); + }, + ); + const original = await parent.run(); + const failed = await finish(original.workflowRun.id); + expect(failed.status).toBe("failed"); + broken = false; + const rerun = await client.rerunWorkflowRun(failed.id, { + fromStep: "child", + }); + await expect(finish(rerun.id)).resolves.toMatchObject({ + output: "child result", + }); + expect(childCalls).toBe(2); + const priorSteps = await backend.listStepAttempts({ + workflowRunId: rerun.id, + }); + const later = await client.rerunWorkflowRun(rerun.id, { + fromStep: "after-child", + }); + await expect(finish(later.id)).resolves.toMatchObject({ + output: "child result", + }); + expect(childCalls).toBe(2); + const laterSteps = await backend.listStepAttempts({ + workflowRunId: later.id, + }); + expect( + laterSteps.data.find((step) => step.kind === "workflow") + ?.childWorkflowRunId, + ).toBe( + priorSteps.data.find((step) => step.kind === "workflow") + ?.childWorkflowRunId, + ); + expect(await backend.getWorkflowRun({ workflowRunId: failed.id })).toEqual( + failed, + ); + }); + + test("reuses waits and signal sends before the boundary and restarts them at the boundary", async () => { + const send = vi.spyOn(backend, "sendSignal"); + const workflow = client.defineWorkflow( + { name: "waits" }, + async ({ step }) => { + await step.sendSignal({ signal: "notice" }); + await step.sleep("sleep", "1ms"); + const value = await step.waitForSignal({ signal: "reply", timeout: 0 }); + return await step.run({ name: "last" }, () => value); + }, + ); + const original = await workflow.run(); + await expect(finish(original.workflowRun.id)).resolves.toMatchObject({ + status: "completed", + }); + const rerun = await client.rerunWorkflowRun(original.workflowRun.id, { + fromStep: "last", + }); + await expect(finish(rerun.id)).resolves.toMatchObject({ output: null }); + expect(send).toHaveBeenCalledTimes(1); + const again = await client.rerunWorkflowRun(rerun.id, { + fromStep: "notice", + }); + await expect(finish(again.id)).resolves.toMatchObject({ + status: "completed", + }); + expect(send).toHaveBeenCalledTimes(2); + }); + + test("rejects active runs, unknown runs and unknown steps, but accepts canceled runs", async () => { + const workflow = client.defineWorkflow({ name: "canceled" }, ({ step }) => + step.sleep("wait", "1h"), + ); + const original = await workflow.run(); + await expect( + client.rerunWorkflowRun(original.workflowRun.id), + ).rejects.toThrow("Only finished"); + await client.newWorker().tick(); + await expect( + client.rerunWorkflowRun(original.workflowRun.id), + ).rejects.toThrow("Only finished"); + await client.cancelWorkflowRun(original.workflowRun.id); + await expect( + client.rerunWorkflowRun(original.workflowRun.id, { fromStep: "missing" }), + ).rejects.toThrow("does not exist"); + await expect(client.rerunWorkflowRun("missing")).rejects.toThrow( + "does not exist", + ); + const rerun = await client.rerunWorkflowRun(original.workflowRun.id, { + fromStep: "wait", + }); + expect(rerun.status).toBe("pending"); + }); + + test.each(["prefix", "target"])( + "rejects partial reruns when %s has no recorded order, but permits a full rerun", + async (legacyStep) => { + let calls = 0; + const workflow = client.defineWorkflow( + { name: "legacy-step-order" }, + async ({ step }) => { + await step.run({ name: "prefix" }, () => ++calls); + return await step.run({ name: "target" }, () => ++calls); + }, + ); + const original = await workflow.run(); + const source = await finish(original.workflowRun.id); + expect(source.status).toBe("completed"); + if (backend instanceof BackendSqlite) { + backend["db"] + .prepare( + 'UPDATE "step_attempts" SET "step_index" = NULL WHERE "workflow_run_id" = ? AND "step_name" = ?', + ) + .run(source.id, legacyStep); + } else { + const pg = backend["pg"]; + await pg`UPDATE ${backend["stepAttemptsTable"]()} SET "step_index" = NULL + WHERE "workflow_run_id" = ${source.id} AND "step_name" = ${legacyStep}`; + } + const sourceHistory = await history(source.id); + const before = await backend.listWorkflowRuns({}); + + await expect( + client.rerunWorkflowRun(source.id, { fromStep: "target" }), + ).rejects.toThrow( + "Cannot rerun from a step without recorded step order; rerun the entire workflow instead", + ); + expect(await backend.listWorkflowRuns({})).toEqual(before); + + const rerun = await client.rerunWorkflowRun(source.id); + expect(await history(rerun.id)).toEqual([]); + await expect(finish(rerun.id)).resolves.toMatchObject({ + status: "completed", + output: 4, + }); + expect(calls).toBe(4); + expect( + await backend.getWorkflowRun({ workflowRunId: source.id }), + ).toEqual(source); + expect(await history(source.id)).toEqual(sourceHistory); + }, + ); +}); diff --git a/packages/openworkflow/client/rerun.ts b/packages/openworkflow/client/rerun.ts new file mode 100644 index 00000000..f30fb316 --- /dev/null +++ b/packages/openworkflow/client/rerun.ts @@ -0,0 +1,37 @@ +import type { Backend } from "../core/backend.js"; +import type { WorkflowRun } from "../core/workflow-run.js"; +import { + captureTraceContext, + getSpanKind, + setAttributes, + SPAN_NAMES, + traceOperation, + workflowRunAttributes, +} from "../telemetry.js"; + +/** + * Rerun a finished workflow, optionally copying earlier successful steps. + * @param backend - Backend containing the source run + * @param workflowRunId - Source run ID + * @param fromStep - Step to execute again; omit to rerun every step + * @returns A new pending run + */ +export async function rerunWorkflowRun( + backend: Readonly, + workflowRunId: string, + fromStep?: string, +): Promise { + return traceOperation( + SPAN_NAMES.WORKFLOW_RUN_CREATE, + { kind: await getSpanKind("PRODUCER") }, + async (span) => { + const run = await backend.rerunWorkflowRun({ + workflowRunId, + fromStep: fromStep ?? null, + context: captureTraceContext(), + }); + setAttributes(span, workflowRunAttributes(run)); + return run; + }, + ); +} diff --git a/packages/openworkflow/core/backend.ts b/packages/openworkflow/core/backend.ts index 926901b1..804487ab 100644 --- a/packages/openworkflow/core/backend.ts +++ b/packages/openworkflow/core/backend.ts @@ -19,6 +19,9 @@ export interface Backend { createWorkflowRun( params: Readonly, ): Promise; + rerunWorkflowRun( + params: Readonly, + ): Promise; getWorkflowRun( params: Readonly, ): Promise; @@ -95,6 +98,12 @@ export interface GetWorkflowRunParams { workflowRunId: string; } +export interface RerunWorkflowRunParams { + workflowRunId: string; + fromStep: string | null; + context: JsonValue | null; +} + export interface ListWorkflowRunsParams extends PaginationOptions { status?: WorkflowRunStatus; workflowName?: string; diff --git a/packages/openworkflow/core/rerun.test.ts b/packages/openworkflow/core/rerun.test.ts new file mode 100644 index 00000000..2e38206a --- /dev/null +++ b/packages/openworkflow/core/rerun.test.ts @@ -0,0 +1,46 @@ +import type { JsonValue } from "./json.js"; +import { getRerunStepIndices } from "./rerun.js"; +import { describe, expect, test } from "vitest"; + +describe("getRerunStepIndices", () => { + test.each<{ context: JsonValue }>([ + { context: null }, + { context: "legacy context" }, + { context: [] }, + { context: {} }, + { context: { rerunStepIndices: null } }, + { context: { rerunStepIndices: [0, 1] } }, + { context: { rerunStepIndices: "invalid" } }, + ])( + "ignores absent or malformed step-order metadata: $context", + ({ context }) => { + expect(getRerunStepIndices(context)).toEqual(new Map()); + }, + ); + + test("retains only nonnegative safe integer indices without changing the context", () => { + const indices = { + first: 0, + later: 5, + largest: Number.MAX_SAFE_INTEGER, + negative: -1, + fractional: 1.5, + unsafe: Number.MAX_SAFE_INTEGER + 1, + numericString: "2", + missing: null, + nested: { index: 3 }, + }; + const context = { rerunStepIndices: { ...indices } }; + const result = getRerunStepIndices(context); + + expect(result).toEqual( + new Map([ + ["first", 0], + ["later", 5], + ["largest", Number.MAX_SAFE_INTEGER], + ]), + ); + result.set("first", 10); + expect(context.rerunStepIndices).toEqual(indices); + }); +}); diff --git a/packages/openworkflow/core/rerun.ts b/packages/openworkflow/core/rerun.ts new file mode 100644 index 00000000..2c310493 --- /dev/null +++ b/packages/openworkflow/core/rerun.ts @@ -0,0 +1,111 @@ +import type { + CreateWorkflowRunParams, + RerunWorkflowRunParams, +} from "./backend.js"; +import type { JsonValue } from "./json.js"; +import type { StepAttempt } from "./step-attempt.js"; +import { isTerminalStatus, type WorkflowRun } from "./workflow-run.js"; + +interface PreparedWorkflowRerun { + params: CreateWorkflowRunParams; + stepIndex: number | null; +} + +/** + * Validate a rerun and reuse the source input, version, and configuration. + * @param source - Source run, read within the copying transaction + * @param request - Rerun request + * @param steps - Source attempt metadata, read in the same transaction + * @returns New run parameters and the exclusive step-index boundary + */ +export function prepareWorkflowRerun( + source: Readonly | null, + request: Readonly, + steps: readonly Pick[], +): PreparedWorkflowRerun { + if (!source) { + throw new Error(`Workflow run ${request.workflowRunId} does not exist`); + } + if (!isTerminalStatus(source.status)) { + throw new Error("Only finished workflow runs can be rerun"); + } + const boundary = steps.find((step) => step.stepName === request.fromStep); + if (request.fromStep !== null && !boundary) { + throw new Error( + `Step "${request.fromStep}" does not exist in workflow run ${source.id}`, + ); + } + const stepIndices = getRerunStepIndices(source.context); + const completed = new Set(); + if (boundary) { + for (const step of steps) { + if (step.stepIndex === null) { + throw new Error( + "Cannot rerun from a step without recorded step order; rerun the entire workflow instead", + ); + } + stepIndices.set(step.stepName, step.stepIndex); + if (step.status === "completed" || step.status === "succeeded") { + completed.add(step.stepName); + } + } + } + const stepIndex = boundary?.stepIndex ?? null; + let context = request.context; + if (stepIndex !== null) { + for (const [name, index] of stepIndices) { + if (index >= stepIndex || completed.has(name)) stepIndices.delete(name); + } + if (stepIndices.size > 0) { + if (!isJsonObject(context)) context = {}; + context = { + ...context, + rerunStepIndices: Object.fromEntries(stepIndices), + }; + } + } + return { + stepIndex, + params: { + workflowName: source.workflowName, + version: source.version, + input: source.input, + config: source.config, + context, + idempotencyKey: null, + parentStepAttemptNamespaceId: null, + parentStepAttemptId: null, + availableAt: null, + deadlineAt: null, + }, + }; +} + +/** + * Read step order retained independently of copied successful attempts. + * @param context - Persisted workflow execution metadata + * @returns Recorded indices, including steps omitted from a rerun's history + */ +export function getRerunStepIndices(context: JsonValue): Map { + if (!isJsonObject(context)) { + return new Map(); + } + const indices = context["rerunStepIndices"]; + if (!isJsonObject(indices)) { + return new Map(); + } + return new Map( + Object.entries(indices).filter( + (entry): entry is [string, number] => + typeof entry[1] === "number" && + Number.isSafeInteger(entry[1]) && + entry[1] >= 0, + ), + ); +} + +function isJsonObject( + value: JsonValue | undefined, +): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} diff --git a/packages/openworkflow/internal.ts b/packages/openworkflow/internal.ts index 57d1606f..b1f3ddc7 100644 --- a/packages/openworkflow/internal.ts +++ b/packages/openworkflow/internal.ts @@ -12,3 +12,5 @@ export type { StepAttemptStatus, StepKind, } from "./core/step-attempt.js"; + +export { rerunWorkflowRun } from "./client/rerun.js"; diff --git a/packages/openworkflow/postgres/backend.ts b/packages/openworkflow/postgres/backend.ts index f9d0b054..66942b07 100644 --- a/packages/openworkflow/postgres/backend.ts +++ b/packages/openworkflow/postgres/backend.ts @@ -8,6 +8,7 @@ import { ClaimWorkflowRunParams, CreateStepAttemptParams, CreateWorkflowRunParams, + RerunWorkflowRunParams, GetStepAttemptParams, GetWorkflowRunParams, ExtendWorkflowRunLeaseParams, @@ -33,6 +34,7 @@ import { } from "../core/cursor.js"; import { requireRow, wrapError } from "../core/error.js"; import { JsonValue } from "../core/json.js"; +import { prepareWorkflowRerun } from "../core/rerun.js"; import { StepAttempt } from "../core/step-attempt.js"; import { computeFailedWorkflowRunUpdate } from "../core/workflow-definition.js"; import { @@ -218,6 +220,49 @@ export class BackendPostgres implements Backend { }); } + async rerunWorkflowRun( + request: RerunWorkflowRunParams, + ): Promise { + return await this.withTransaction(async (tx) => { + const [source] = await tx` + SELECT * FROM ${this.workflowRunsTable(tx)} + WHERE "namespace_id" = ${this.namespaceId} AND "id" = ${request.workflowRunId} + FOR UPDATE + `; + const stepAttemptsTable = this.stepAttemptsTable(tx); + const steps = + request.fromStep === null + ? [] + : await tx[]>` + SELECT "step_name", "step_index", "status" FROM ${stepAttemptsTable} + WHERE "namespace_id" = ${this.namespaceId} AND "workflow_run_id" = ${request.workflowRunId} + `; + const { params, stepIndex } = prepareWorkflowRerun( + source ?? null, + request, + steps, + ); + const run = await this.insertWorkflowRun(tx, params); + if (stepIndex !== null) { + await tx` + INSERT INTO ${stepAttemptsTable} ( + "namespace_id", "id", "workflow_run_id", "step_name", "step_index", "kind", "status", + "config", "context", "output", "child_workflow_run_namespace_id", "child_workflow_run_id", + "started_at", "finished_at", "created_at", "updated_at" + ) + SELECT "namespace_id", gen_random_uuid(), ${run.id}, "step_name", "step_index", "kind", "status", + "config", "context", "output", "child_workflow_run_namespace_id", "child_workflow_run_id", + "started_at", "finished_at", "created_at", "updated_at" + FROM ${stepAttemptsTable} + WHERE "namespace_id" = ${this.namespaceId} AND "workflow_run_id" = ${request.workflowRunId} + AND "status" IN ('completed', 'succeeded') + AND "step_index" < ${stepIndex} + `; + } + return run; + }); + } + private async insertWorkflowRun( pg: Postgres, params: CreateWorkflowRunParams, diff --git a/packages/openworkflow/sqlite/backend.ts b/packages/openworkflow/sqlite/backend.ts index a9bc6cda..33732ac6 100644 --- a/packages/openworkflow/sqlite/backend.ts +++ b/packages/openworkflow/sqlite/backend.ts @@ -7,6 +7,7 @@ import { ClaimWorkflowRunParams, CreateStepAttemptParams, CreateWorkflowRunParams, + RerunWorkflowRunParams, GetStepAttemptParams, GetWorkflowRunParams, ExtendWorkflowRunLeaseParams, @@ -33,6 +34,7 @@ import { } from "../core/cursor.js"; import { requireRow, wrapError } from "../core/error.js"; import { JsonValue } from "../core/json.js"; +import { prepareWorkflowRerun } from "../core/rerun.js"; import { StepAttempt } from "../core/step-attempt.js"; import { computeFailedWorkflowRunUpdate } from "../core/workflow-definition.js"; import { @@ -176,6 +178,79 @@ export class BackendSqlite implements Backend { } } + // oxlint-disable-next-line typescript/require-await -- keep the transaction synchronous + async rerunWorkflowRun( + request: RerunWorkflowRunParams, + ): Promise { + this.db.exec("BEGIN IMMEDIATE"); + try { + // safety: these queries select rows from the corresponding tables. + const source = this.db + .prepare( + ` + SELECT * FROM "workflow_runs" WHERE "namespace_id" = ? AND "id" = ? + `, + ) + .get(this.namespaceId, request.workflowRunId) as + WorkflowRunRow | undefined; + // safety: the query selects step names, nullable indices, and statuses. + const steps = + request.fromStep === null + ? [] + : (this.db + .prepare( + ` + SELECT "step_name", "step_index", "status" FROM "step_attempts" + WHERE "namespace_id" = ? AND "workflow_run_id" = ? + `, + ) + .all(this.namespaceId, request.workflowRunId) as Pick< + StepAttemptRow, + "step_name" | "step_index" | "status" + >[]); + const { params, stepIndex } = prepareWorkflowRerun( + source ? rowToWorkflowRun(source) : null, + request, + steps.map((step) => ({ + stepName: step.step_name, + stepIndex: step.step_index, + // safety: backend transitions write domain status values. + status: step.status as StepAttempt["status"], + })), + ); + const run = this.insertWorkflowRun(params); + if (stepIndex !== null) { + this.db + .prepare( + ` + INSERT INTO "step_attempts" ( + "namespace_id", "id", "workflow_run_id", "step_name", "step_index", "kind", "status", + "config", "context", "output", "child_workflow_run_namespace_id", "child_workflow_run_id", + "started_at", "finished_at", "created_at", "updated_at" + ) + SELECT "namespace_id", + lower(hex(randomblob(4)) || '-' || hex(randomblob(2)) || '-4' || + substr(hex(randomblob(2)), 2) || '-' || substr('89ab', (random() & 3) + 1, 1) || + substr(hex(randomblob(2)), 2) || '-' || hex(randomblob(6))), + ?, "step_name", "step_index", "kind", "status", + "config", "context", "output", "child_workflow_run_namespace_id", "child_workflow_run_id", + "started_at", "finished_at", "created_at", "updated_at" + FROM "step_attempts" + WHERE "namespace_id" = ? AND "workflow_run_id" = ? + AND "status" IN ('completed', 'succeeded') + AND "step_index" < ? + `, + ) + .run(run.id, this.namespaceId, request.workflowRunId, stepIndex); + } + this.db.exec("COMMIT"); + return run; + } catch (error) { + this.db.exec("ROLLBACK"); + throw error; + } + } + private insertWorkflowRun(params: CreateWorkflowRunParams): WorkflowRun { const id = generateUUID(); const currentTime = now(); diff --git a/packages/openworkflow/telemetry.test.ts b/packages/openworkflow/telemetry.test.ts index 22f3c084..ea58f328 100644 --- a/packages/openworkflow/telemetry.test.ts +++ b/packages/openworkflow/telemetry.test.ts @@ -321,6 +321,76 @@ describe("native OpenTelemetry instrumentation", () => { ); }); + test.each([undefined, "target"])( + "links a rerun from %s to its own submission context after reopening storage", + async (fromStep) => { + const ow = new OpenWorkflow({ backend }); + let baggage: string | undefined; + const workflow = ow.defineWorkflow( + { name: "rerun", version: "v1" }, + async ({ step }) => { + baggage = propagation + .getBaggage(context.active()) + ?.getEntry("example")?.value; + await step.run({ name: "prefix" }, () => "saved"); + return await step.run({ name: "target" }, () => 42); + }, + ); + const source = await workflow.run(); + await executeNext(workflow.workflow); + const origin = propagation.setBaggage( + ROOT_CONTEXT, + propagation.createBaggage({ example: { value: "rerun-request" } }), + ); + const rerun = await context.with(origin, () => + tracer.startActiveSpan("rerun-request", async (span) => { + try { + return await ow.rerunWorkflowRun( + source.workflowRun.id, + fromStep === undefined ? undefined : { fromStep }, + ); + } finally { + span.end(); + } + }), + ); + await backend.stop(); + backend = BackendSqlite.connect(databasePath); + await executeNext(workflow.workflow); + + const spans = exporter.getFinishedSpans(); + const request = spans.find((span) => span.name === "rerun-request"); + const submission = spans.find( + (span) => + span.name === SPAN_NAMES.WORKFLOW_RUN_CREATE && + span.attributes[ATTRIBUTE_NAMES.WORKFLOW_RUN_ID] === rerun.id, + ); + const execution = spans.find( + (span) => + span.name === SPAN_NAMES.WORKFLOW_RUN_EXECUTE && + span.attributes[ATTRIBUTE_NAMES.WORKFLOW_RUN_ID] === rerun.id, + ); + assert.ok(request); + assert.ok(submission); + assert.ok(execution); + expect(submission.parentSpanContext?.spanId).toBe( + request.spanContext().spanId, + ); + expect(submission.kind).toBe(SpanKind.PRODUCER); + expect(submission.attributes).toMatchObject({ + [ATTRIBUTE_NAMES.WORKFLOW_NAME]: "rerun", + [ATTRIBUTE_NAMES.WORKFLOW_VERSION]: "v1", + [ATTRIBUTE_NAMES.NAMESPACE_ID]: "default", + }); + expect(execution.parentSpanContext).toBeUndefined(); + expect(execution.links[0]?.context).toMatchObject({ + traceId: submission.spanContext().traceId, + spanId: submission.spanContext().spanId, + }); + expect(baggage).toBe("rerun-request"); + }, + ); + test("traces signal sends and suspended executions without exposing payloads", async () => { const ow = new OpenWorkflow({ backend }); const workflow = ow.defineWorkflow( diff --git a/packages/openworkflow/testing/backend-stub.testsuite.ts b/packages/openworkflow/testing/backend-stub.testsuite.ts index 0fe76b7f..e3b3157b 100644 --- a/packages/openworkflow/testing/backend-stub.testsuite.ts +++ b/packages/openworkflow/testing/backend-stub.testsuite.ts @@ -8,6 +8,7 @@ import type { Backend } from "../core/backend.js"; export function createStubBackend(overrides: Partial): Backend { return { createWorkflowRun: unexpectedBackendCall, + rerunWorkflowRun: unexpectedBackendCall, getWorkflowRun: unexpectedBackendCall, listWorkflowRuns: unexpectedBackendCall, countWorkflowRuns: unexpectedBackendCall, diff --git a/packages/openworkflow/worker/execution.ts b/packages/openworkflow/worker/execution.ts index a290f2c3..6c40c536 100644 --- a/packages/openworkflow/worker/execution.ts +++ b/packages/openworkflow/worker/execution.ts @@ -6,6 +6,7 @@ import { type SerializedError, } from "../core/error.js"; import type { JsonValue } from "../core/json.js"; +import { getRerunStepIndices } from "../core/rerun.js"; import type { StandardSchemaV1 } from "../core/standard-schema.js"; import type { StepAttempt, @@ -1167,7 +1168,10 @@ async function executeWorkflowAttempt( backend, workflowRun.id, ); - const history = new StepHistory({ attempts }); + const history = new StepHistory({ + attempts, + stepIndices: getRerunStepIndices(workflowRun.context), + }); // Complete any elapsed sleep waits first, then park on the earliest // remaining running wait (sleep, signal, or child workflow). diff --git a/packages/openworkflow/worker/step-history.ts b/packages/openworkflow/worker/step-history.ts index 19304435..fc55f270 100644 --- a/packages/openworkflow/worker/step-history.ts +++ b/packages/openworkflow/worker/step-history.ts @@ -169,6 +169,7 @@ function getEarliestRunningWait( */ export interface StepHistoryOptions { attempts: readonly StepAttempt[]; + stepIndices?: ReadonlyMap; stepLimit?: number; } @@ -184,7 +185,7 @@ export class StepHistory { private readonly failedCountsByStepName: Map; private readonly failedByStepName: Map; private readonly runningByStepName: Map; - private readonly persistedStepIndices = new Map(); + private readonly persistedStepIndices: Map; private readonly resolvedStepNames = new Map(); private readonly expectedNextStepIndexByName = new Map(); private nextStepIndex = 0; @@ -194,11 +195,14 @@ export class StepHistory { constructor(options: Readonly) { this.stepLimit = Math.max(1, options.stepLimit ?? WORKFLOW_STEP_LIMIT); this.stepCount = options.attempts.length; + this.persistedStepIndices = new Map(options.stepIndices); for (const attempt of options.attempts) { if (attempt.stepIndex === null) continue; this.persistedStepIndices.set(attempt.stepName, attempt.stepIndex); - this.nextStepIndex = Math.max(this.nextStepIndex, attempt.stepIndex + 1); + } + for (const index of this.persistedStepIndices.values()) { + this.nextStepIndex = Math.max(this.nextStepIndex, index + 1); } const state = createStepExecutionStateFromAttempts(options.attempts);