feat(openworkflow): configurable worker lease duration

apps/cli/config.ts

@@ -25,7 +25,10 @@ export interface OpenWorkflowConfig {
   ignorePatterns?: string[];
 }
 
-export type WorkerConfig = Pick<WorkerOptions, "concurrency">;
+export type WorkerConfig = Pick<
+  WorkerOptions,
+  "concurrency" | "leaseDurationMs"
+>;
 
 /**
  * Create a typed OpenWorkflow configuration.

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,
 },
 ```
 

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:

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<WorkerOptions, "concurrency" | "leaseDurationMs">,
+  ): Worker {
     return new Worker({
       backend: this.backend,
       workflows: this.registry.getAll(),
       concurrency: options?.concurrency,
+      leaseDurationMs: options?.leaseDurationMs,
     });
   }
 

packages/openworkflow/worker/worker.test.ts

@@ -577,6 +577,118 @@ 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 () => {
+    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
+
+    const backend = await createTestBackend();
+    const extendWorkflowRunLease = vi.spyOn(backend, "extendWorkflowRunLease");
+    const client = new OpenWorkflow({ backend });
+    const stepStarted = Promise.withResolvers<null>();
+    const releaseStep = Promise.withResolvers<null>();
+
+    const workflow = client.defineWorkflow(
+      { name: "custom-lease-heartbeat" },
+      async ({ step }) => {
+        await step.run({ name: "wait" }, async () => {
+          stepStarted.resolve(null);
+          await releaseStep.promise;
+        });
+        return "done";
+      },
+    );
+    const worker = client.newWorker({ leaseDurationMs: 100 });
+
+    try {
+      const handle = await workflow.run();
+      await worker.tick();
+      await stepStarted.promise;
+
+      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();
+    }
+  });
+
   test(
     "worker only sleeps between claims when no work is available",
     { timeout: 15_000 },

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,
@@ -35,6 +37,14 @@ export interface WorkerOptions {
   backend: Backend;
   workflows: Workflow<unknown, unknown, unknown>[];
   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 +54,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<WorkflowExecution>();
   private running = false;
@@ -62,6 +73,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 +174,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 +204,7 @@ export class Worker {
       backend: this.backend,
       workflowRun,
       workerId,
+      leaseDurationMs: this.leaseDurationMs,
     });
     this.activeExecutions.add(execution);
 
@@ -247,6 +261,7 @@ interface WorkflowExecutionOptions {
   backend: Backend;
   workflowRun: WorkflowRun;
   workerId: string;
+  leaseDurationMs: number;
 }
 
 /**
@@ -257,21 +272,26 @@ 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;
   }
 
   /**
    * Start the heartbeat loop for this execution, heartbeating at half the lease
-   * duration.
+   * duration (capped at the maximum timer delay).
    */
   startHeartbeat(): void {
-    const leaseDurationMs = DEFAULT_LEASE_DURATION_MS;
-    const heartbeatIntervalMs = leaseDurationMs / 2;
+    const leaseDurationMs = this.leaseDurationMs;
+    const heartbeatIntervalMs = Math.min(
+      leaseDurationMs / 2,
+      MAX_TIMER_DELAY_MS,
+    );
 
     this.heartbeatTimer = setInterval(
       bindActiveTraceContext(() => {
@@ -330,6 +350,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