fix(openworkflow): park replay pre-pass on earliest running wait

Author: jamescmartinez

packages/openworkflow/worker/execution.test.ts

@@ -1636,6 +1636,81 @@ describe("StepExecutor", () => {
     expect(runningWorkflowStepNames).toEqual(["wait-long", "wait-short"]);
   });
 
+  test("replay pre-pass parks on earliest running wait across sleep and runWorkflow history", async () => {
+    const backend = await createBackend();
+
+    const workflowRun = await backend.createWorkflowRun({
+      workflowName: `workflow-replay-earliest-wait-${randomUUID()}`,
+      version: null,
+      idempotencyKey: null,
+      config: {},
+      context: null,
+      input: null,
+      parentStepAttemptNamespaceId: null,
+      parentStepAttemptId: null,
+      availableAt: null,
+      deadlineAt: null,
+    });
+
+    const claimedWorkflowRun = await backend.claimWorkflowRun({
+      workerId: randomUUID(),
+      leaseDurationMs: 5000,
+    });
+    if (!claimedWorkflowRun) {
+      throw new Error("Expected workflow run to be claimed");
+    }
+    expect(claimedWorkflowRun.id).toBe(workflowRun.id);
+    if (!claimedWorkflowRun.workerId) {
+      throw new Error("Expected claimed workflow run worker id");
+    }
+
+    await backend.createStepAttempt({
+      workflowRunId: claimedWorkflowRun.id,
+      workerId: claimedWorkflowRun.workerId,
+      stepName: "sleep-late",
+      kind: "sleep",
+      config: {},
+      context: {
+        kind: "sleep",
+        resumeAt: new Date(Date.now() + 120_000).toISOString(),
+      },
+    });
+    await backend.createStepAttempt({
+      workflowRunId: claimedWorkflowRun.id,
+      workerId: claimedWorkflowRun.workerId,
+      stepName: "wait-early",
+      kind: "workflow",
+      config: {},
+      context: {
+        kind: "workflow",
+        timeoutAt: new Date(Date.now() + 5000).toISOString(),
+      },
+    });
+
+    await executeWorkflow({
+      backend,
+      workflowRun: claimedWorkflowRun,
+      workflowFn: () => "unreachable",
+      workflowVersion: null,
+      workerId: claimedWorkflowRun.workerId,
+      retryPolicy: DEFAULT_WORKFLOW_RETRY_POLICY,
+    });
+
+    const parkedRun = await backend.getWorkflowRun({
+      workflowRunId: workflowRun.id,
+    });
+    expect(parkedRun?.status).toBe("running");
+    expect(parkedRun?.workerId).toBeNull();
+    expect(parkedRun?.availableAt).not.toBeNull();
+    if (!parkedRun?.availableAt) {
+      throw new Error("Expected parked workflow availableAt");
+    }
+
+    const millisecondsUntilWake = parkedRun.availableAt.getTime() - Date.now();
+    expect(millisecondsUntilWake).toBeGreaterThan(1000);
+    expect(millisecondsUntilWake).toBeLessThan(20_000);
+  });
+
   test("best-effort fences late parallel branches after parent parks", async () => {
     const backend = await createBackend();
     const client = new OpenWorkflow({ backend });

packages/openworkflow/worker/execution.ts

@@ -376,6 +376,55 @@ function getEarliestRunningWaitResumeAt(
   return earliest;
 }
 
+/**
+ * Complete running sleep step attempts whose resume timestamp has elapsed.
+ * Malformed historical resume timestamps are treated as elapsed for backward
+ * compatibility.
+ * @param options - Sleep pre-pass options
+ * @returns Whether any running sleep remains pending after completion pass
+ */
+async function completeElapsedRunningSleepAttempts(
+  options: Readonly<{
+    backend: Backend;
+    workflowRunId: string;
+    workerId: string;
+    attempts: StepAttempt[];
+  }>,
+): Promise<boolean> {
+  let hasPendingRunningSleep = false;
+
+  for (let i = 0; i < options.attempts.length; i += 1) {
+    const attempt = options.attempts[i];
+    if (!attempt) continue;
+
+    if (
+      attempt.status !== "running" ||
+      attempt.kind !== "sleep" ||
+      attempt.context?.kind !== "sleep"
+    ) {
+      continue;
+    }
+
+    const resumeAt = new Date(attempt.context.resumeAt);
+    const resumeAtMs = resumeAt.getTime();
+    if (Number.isFinite(resumeAtMs) && Date.now() < resumeAtMs) {
+      hasPendingRunningSleep = true;
+      continue;
+    }
+
+    const completed = await options.backend.completeStepAttempt({
+      workflowRunId: options.workflowRunId,
+      stepAttemptId: attempt.id,
+      workerId: options.workerId,
+      output: null,
+    });
+
+    options.attempts[i] = completed;
+  }
+
+  return hasPendingRunningSleep;
+}
+
 /**
  * Load all step attempts for a workflow run.
  * @param backend - Backend instance
@@ -1009,37 +1058,23 @@ export async function executeWorkflow(
       workflowRun.id,
     );
 
-    // mark any sleep steps as completed if their sleep duration has elapsed,
-    // or rethrow SleepSignal if still sleeping
-    for (let i = 0; i < attempts.length; i++) {
-      const attempt = attempts[i];
-      if (!attempt) continue;
+    // complete any elapsed sleep waits first, then park on the earliest
+    // remaining running wait (sleep or runWorkflow timeout).
+    const hasPendingRunningSleep = await completeElapsedRunningSleepAttempts({
+      backend,
+      workflowRunId: workflowRun.id,
+      workerId,
+      attempts,
+    });
 
+    if (hasPendingRunningSleep) {
+      const earliestRunningWaitResumeAt =
+        getEarliestRunningWaitResumeAt(attempts);
       if (
-        attempt.status === "running" &&
-        attempt.kind === "sleep" &&
-        attempt.context?.kind === "sleep"
+        earliestRunningWaitResumeAt &&
+        Date.now() < earliestRunningWaitResumeAt.getTime()
       ) {
-        const now = Date.now();
-        const resumeAt = new Date(attempt.context.resumeAt);
-        const resumeAtMs = resumeAt.getTime();
-
-        if (now < resumeAtMs) {
-          // sleep duration HAS NOT elapsed yet, throw signal to put workflow
-          // back to sleep
-          throw new SleepSignal(resumeAt);
-        }
-
-        // sleep duration HAS elapsed, mark the step as completed and continue
-        const completed = await backend.completeStepAttempt({
-          workflowRunId: workflowRun.id,
-          stepAttemptId: attempt.id,
-          workerId,
-          output: null,
-        });
-
-        // update cache w/ completed attempt
-        attempts[i] = completed;
+        throw new SleepSignal(earliestRunningWaitResumeAt);
       }
     }