From 359188b7c81924ad37af4ef72cea46ad8dbbaaf7 Mon Sep 17 00:00:00 2001 From: Clifton Cunningham Date: Mon, 12 Oct 2026 03:43:51 +0100 Subject: [PATCH 1/2] feat(openworkflow): configurable worker lease duration Add a `leaseDurationMs` worker option, also accepted by `OpenWorkflow#newWorker` and the CLI `worker` config. It replaces the fixed 30s lease used when claiming and extending workflow runs. Heartbeats stay at half the lease, and missing or invalid values fall back to the 30s default. --- apps/cli/config.ts | 5 +- apps/docs/docs/configuration.mdx | 2 + apps/docs/docs/workers.mdx | 15 +++++ packages/openworkflow/client/client.ts | 8 ++- packages/openworkflow/worker/worker.test.ts | 69 +++++++++++++++++++++ packages/openworkflow/worker/worker.ts | 33 +++++++++- 6 files changed, 127 insertions(+), 5 deletions(-) diff --git a/apps/cli/config.ts b/apps/cli/config.ts index 7051bbb8..f9f8e822 100644 --- a/apps/cli/config.ts +++ b/apps/cli/config.ts @@ -25,7 +25,10 @@ export interface OpenWorkflowConfig { ignorePatterns?: string[]; } -export type WorkerConfig = Pick; +export type WorkerConfig = Pick< + WorkerOptions, + "concurrency" | "leaseDurationMs" +>; /** * Create a typed OpenWorkflow configuration. diff --git a/apps/docs/docs/configuration.mdx b/apps/docs/docs/configuration.mdx index 06a46cb8..2e466ef5 100644 --- a/apps/docs/docs/configuration.mdx +++ b/apps/docs/docs/configuration.mdx @@ -127,6 +127,8 @@ Worker configuration options: worker: { // Number of concurrent workflow executions (default: 1) concurrency: 4, + // How long a claimed run stays leased without a heartbeat (default: 30000) + leaseDurationMs: 30_000, }, ``` diff --git a/apps/docs/docs/workers.mdx b/apps/docs/docs/workers.mdx index da4077f7..62ac061a 100644 --- a/apps/docs/docs/workers.mdx +++ b/apps/docs/docs/workers.mdx @@ -101,6 +101,21 @@ Workers maintain their claim on workflow runs through a heartbeat mechanism: This ensures workflows are never stuck if a worker crashes. +The lease lasts 30 seconds by default, and workers heartbeat at half the lease +duration. Configure it with `leaseDurationMs`: + +```ts +export default defineConfig({ + backend, + worker: { + leaseDurationMs: 60_000, // Reclaim runs from crashed workers after 60s + }, +}); +``` + +A shorter lease recovers crashed runs sooner, at the cost of more frequent +heartbeat writes to the database. + ## High Availability For production, run multiple worker instances: diff --git a/packages/openworkflow/client/client.ts b/packages/openworkflow/client/client.ts index bd77473b..8e5cb4cc 100644 --- a/packages/openworkflow/client/client.ts +++ b/packages/openworkflow/client/client.ts @@ -27,7 +27,7 @@ import { SPAN_NAMES, traceOperation, } from "../telemetry.js"; -import { Worker } from "../worker/worker.js"; +import { Worker, type WorkerOptions } from "../worker/worker.js"; import { rerunWorkflowRun } from "./rerun.js"; const DEFAULT_RESULT_POLL_INTERVAL_MS = 1000; // 1s @@ -61,13 +61,17 @@ export class OpenWorkflow { * Create a new Worker with this client's backend and workflows. * @param options - Worker options * @param options.concurrency - Max concurrent workflow runs + * @param options.leaseDurationMs - Workflow run lease duration * @returns Worker instance */ - newWorker(options?: { concurrency?: number | undefined }): Worker { + newWorker( + options?: Pick, + ): Worker { return new Worker({ backend: this.backend, workflows: this.registry.getAll(), concurrency: options?.concurrency, + leaseDurationMs: options?.leaseDurationMs, }); } diff --git a/packages/openworkflow/worker/worker.test.ts b/packages/openworkflow/worker/worker.test.ts index f3a684c6..86ef6a18 100644 --- a/packages/openworkflow/worker/worker.test.ts +++ b/packages/openworkflow/worker/worker.test.ts @@ -577,6 +577,75 @@ describe("Worker", () => { }); }); + test("claims workflow runs with a custom lease duration", async () => { + const claimWorkflowRun = vi.fn().mockResolvedValue(null); + + const worker = new Worker({ + backend: createStubBackend({ + claimWorkflowRun, + }), + workflows: [], + leaseDurationMs: 5000, + }); + + await worker.tick(); + + expect(claimWorkflowRun.mock.calls[0]?.[0]).toMatchObject({ + leaseDurationMs: 5000, + }); + }); + + test.each([0, -1, Number.NaN, Number.POSITIVE_INFINITY])( + "falls back to the default lease duration for invalid value %s", + async (leaseDurationMs) => { + const claimWorkflowRun = vi.fn().mockResolvedValue(null); + + const worker = new Worker({ + backend: createStubBackend({ + claimWorkflowRun, + }), + workflows: [], + leaseDurationMs, + }); + + await worker.tick(); + + expect(claimWorkflowRun.mock.calls[0]?.[0]).toMatchObject({ + leaseDurationMs: 30 * 1000, + }); + }, + ); + + test("heartbeats at half the custom lease duration", async () => { + const backend = await createTestBackend(); + const extendWorkflowRunLease = vi.spyOn(backend, "extendWorkflowRunLease"); + const client = new OpenWorkflow({ backend }); + + const workflow = client.defineWorkflow( + { name: "custom-lease-heartbeat" }, + async ({ step }) => { + await step.run({ name: "slow" }, async () => { + await sleep(300); + }); + return "done"; + }, + ); + const worker = client.newWorker({ leaseDurationMs: 100 }); + + const handle = await workflow.run(); + await worker.tick(); + await expect(handle.result()).resolves.toBe("done"); + await worker.stop(); + + expect(extendWorkflowRunLease.mock.calls.length).toBeGreaterThanOrEqual(2); + for (const [params] of extendWorkflowRunLease.mock.calls) { + expect(params).toMatchObject({ + workflowRunId: handle.workflowRun.id, + leaseDurationMs: 100, + }); + } + }); + test( "worker only sleeps between claims when no work is available", { timeout: 15_000 }, diff --git a/packages/openworkflow/worker/worker.ts b/packages/openworkflow/worker/worker.ts index 8eae0ab8..38e4bbfd 100644 --- a/packages/openworkflow/worker/worker.ts +++ b/packages/openworkflow/worker/worker.ts @@ -35,6 +35,14 @@ export interface WorkerOptions { backend: Backend; workflows: Workflow[]; concurrency?: number | undefined; + /** + * How long a claimed workflow run stays leased to this worker without a + * heartbeat. Heartbeats run at half this interval. If the worker stops + * heartbeating (e.g. it crashes), another worker can reclaim the run once + * the lease expires. + * @default 30000 (30 seconds) + */ + leaseDurationMs?: number | undefined; } /** @@ -44,6 +52,7 @@ export interface WorkerOptions { export class Worker { private readonly backend: Backend; private readonly workerIds: string[]; + private readonly leaseDurationMs: number; private readonly registry = new WorkflowRegistry(); private readonly activeExecutions = new Set(); private running = false; @@ -62,6 +71,8 @@ export class Worker { options.concurrency ?? DEFAULT_CONCURRENCY, ); + this.leaseDurationMs = resolveLeaseDurationMs(options.leaseDurationMs); + // generate worker IDs for every concurrency slot this.workerIds = Array.from({ length: concurrency }, () => randomUUID()); } @@ -161,7 +172,7 @@ export class Worker { // claim workflow run const workflowRun = await this.backend.claimWorkflowRun({ workerId, - leaseDurationMs: DEFAULT_LEASE_DURATION_MS, + leaseDurationMs: this.leaseDurationMs, }); if (!workflowRun) return null; @@ -191,6 +202,7 @@ export class Worker { backend: this.backend, workflowRun, workerId, + leaseDurationMs: this.leaseDurationMs, }); this.activeExecutions.add(execution); @@ -247,6 +259,7 @@ interface WorkflowExecutionOptions { backend: Backend; workflowRun: WorkflowRun; workerId: string; + leaseDurationMs: number; } /** @@ -257,12 +270,14 @@ class WorkflowExecution { private backend: Backend; workflowRun: WorkflowRun; workerId: string; + private readonly leaseDurationMs: number; private heartbeatTimer: NodeJS.Timeout | null = null; constructor(options: WorkflowExecutionOptions) { this.backend = options.backend; this.workflowRun = options.workflowRun; this.workerId = options.workerId; + this.leaseDurationMs = options.leaseDurationMs; } /** @@ -270,7 +285,7 @@ class WorkflowExecution { * duration. */ startHeartbeat(): void { - const leaseDurationMs = DEFAULT_LEASE_DURATION_MS; + const leaseDurationMs = this.leaseDurationMs; const heartbeatIntervalMs = leaseDurationMs / 2; this.heartbeatTimer = setInterval( @@ -330,6 +345,20 @@ function getPollBackoffDelayMs(backoffAttempts: number): number { return Math.max(1, Math.round((cappedBackoffMs * jitterScale) / 1000)); } +/** + * Resolve the worker lease duration, falling back to the default when the + * value is missing or not a positive finite number. + * @param leaseDurationMs - Lease duration from worker options + * @returns A valid lease duration in milliseconds + */ +function resolveLeaseDurationMs(leaseDurationMs: number | undefined): number { + return leaseDurationMs !== undefined && + Number.isFinite(leaseDurationMs) && + leaseDurationMs > 0 + ? leaseDurationMs + : DEFAULT_LEASE_DURATION_MS; +} + /** * Resolve a partial retry policy by merging it with the default policy. * @param partial - Optional partial retry policy from a workflow spec From ad93c4434cbaa6b48280293ff02661896ed2b07e Mon Sep 17 00:00:00 2001 From: Clifton Cunningham Date: Mon, 12 Oct 2026 04:09:11 +0100 Subject: [PATCH 2/2] fix(openworkflow): cap heartbeat interval at the maximum timer delay Node runs timer delays above 2^31-1 ms after 1 ms, so a very long lease would heartbeat continuously. Cap the interval, and check the heartbeat timing with controlled timers. --- packages/openworkflow/worker/worker.test.ts | 61 ++++++++++++++++++--- packages/openworkflow/worker/worker.ts | 9 ++- 2 files changed, 59 insertions(+), 11 deletions(-) diff --git a/packages/openworkflow/worker/worker.test.ts b/packages/openworkflow/worker/worker.test.ts index 86ef6a18..e07bb845 100644 --- a/packages/openworkflow/worker/worker.test.ts +++ b/packages/openworkflow/worker/worker.test.ts @@ -617,32 +617,75 @@ describe("Worker", () => { ); test("heartbeats at half the custom lease duration", async () => { + vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); + const backend = await createTestBackend(); const extendWorkflowRunLease = vi.spyOn(backend, "extendWorkflowRunLease"); const client = new OpenWorkflow({ backend }); + const stepStarted = Promise.withResolvers(); + const releaseStep = Promise.withResolvers(); const workflow = client.defineWorkflow( { name: "custom-lease-heartbeat" }, async ({ step }) => { - await step.run({ name: "slow" }, async () => { - await sleep(300); + await step.run({ name: "wait" }, async () => { + stepStarted.resolve(null); + await releaseStep.promise; }); return "done"; }, ); const worker = client.newWorker({ leaseDurationMs: 100 }); - const handle = await workflow.run(); - await worker.tick(); - await expect(handle.result()).resolves.toBe("done"); - await worker.stop(); + try { + const handle = await workflow.run(); + await worker.tick(); + await stepStarted.promise; - expect(extendWorkflowRunLease.mock.calls.length).toBeGreaterThanOrEqual(2); - for (const [params] of extendWorkflowRunLease.mock.calls) { - expect(params).toMatchObject({ + await vi.advanceTimersByTimeAsync(49); + expect(extendWorkflowRunLease).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + expect(extendWorkflowRunLease).toHaveBeenCalledTimes(1); + expect(extendWorkflowRunLease.mock.calls[0]?.[0]).toMatchObject({ workflowRunId: handle.workflowRun.id, leaseDurationMs: 100, }); + + releaseStep.resolve(null); + await expect(handle.result()).resolves.toBe("done"); + } finally { + releaseStep.resolve(null); + await worker.stop(); + vi.useRealTimers(); + } + }); + + test("caps the heartbeat interval at the maximum timer delay", async () => { + const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); + const backend = await createTestBackend(); + const client = new OpenWorkflow({ backend }); + + const workflow = client.defineWorkflow( + { name: "long-lease-heartbeat" }, + () => "done", + ); + const worker = client.newWorker({ + leaseDurationMs: 60 * 24 * 60 * 60 * 1000, // 60 days + }); + + try { + const handle = await workflow.run(); + await worker.tick(); + await expect(handle.result()).resolves.toBe("done"); + + const heartbeatDelays = setIntervalSpy.mock.calls.map( + ([, delay]) => delay, + ); + expect(heartbeatDelays).toContain(2_147_483_647); + } finally { + await worker.stop(); + setIntervalSpy.mockRestore(); } }); diff --git a/packages/openworkflow/worker/worker.ts b/packages/openworkflow/worker/worker.ts index 38e4bbfd..3878a4bc 100644 --- a/packages/openworkflow/worker/worker.ts +++ b/packages/openworkflow/worker/worker.ts @@ -11,6 +11,8 @@ import { randomUUID } from "node:crypto"; import * as nodeCrypto from "node:crypto"; const DEFAULT_LEASE_DURATION_MS = 30 * 1000; // 30s +// Node runs any longer timer delay after 1ms instead +const MAX_TIMER_DELAY_MS = 2_147_483_647; const DEFAULT_POLL_BACKOFF_POLICY: BackoffPolicy = { initialInterval: "100ms", backoffCoefficient: 2, @@ -282,11 +284,14 @@ class WorkflowExecution { /** * Start the heartbeat loop for this execution, heartbeating at half the lease - * duration. + * duration (capped at the maximum timer delay). */ startHeartbeat(): void { const leaseDurationMs = this.leaseDurationMs; - const heartbeatIntervalMs = leaseDurationMs / 2; + const heartbeatIntervalMs = Math.min( + leaseDurationMs / 2, + MAX_TIMER_DELAY_MS, + ); this.heartbeatTimer = setInterval( bindActiveTraceContext(() => {