fix(openworkflow): handle stale-write races in workflow run transitions

Author: jamescmartinez

packages/openworkflow/worker/execution.test.ts

@@ -2726,6 +2726,56 @@ describe("executeWorkflow", () => {
       expect(failCall.error.message).toMatch(/exceeded the step limit/i);
     });
 
+    test("ignores fail-transition races after run cancel", async () => {
+      const listStepAttempts = vi.fn(() =>
+        Promise.resolve({
+          data: [],
+          pagination: { next: null, prev: null },
+        }),
+      );
+      const failWorkflowRun = vi.fn(() =>
+        Promise.reject(new Error("Failed to mark workflow run failed")),
+      );
+      const getWorkflowRun = vi.fn(() =>
+        Promise.resolve(
+          createMockWorkflowRun({
+            id: "workflow-cancel-race-run",
+            status: "canceled",
+            workerId: null,
+            finishedAt: new Date("2026-01-01T00:00:02.000Z"),
+          }),
+        ),
+      );
+      const workflowFn = vi.fn(() => {
+        throw new Error("workflow failed");
+      });
+      const workflowRun = createMockWorkflowRun({
+        id: "workflow-cancel-race-run",
+        workerId: "worker-cancel-race",
+      });
+
+      await expect(
+        executeWorkflow({
+          backend: {
+            listStepAttempts,
+            failWorkflowRun,
+            getWorkflowRun,
+          } as unknown as Backend,
+          workflowRun,
+          workflowFn,
+          workflowVersion: null,
+          workerId: "worker-cancel-race",
+          retryPolicy: DEFAULT_WORKFLOW_RETRY_POLICY,
+        }),
+      ).resolves.toBeUndefined();
+
+      expect(workflowFn).toHaveBeenCalledTimes(1);
+      expect(failWorkflowRun).toHaveBeenCalledTimes(1);
+      expect(getWorkflowRun).toHaveBeenCalledWith({
+        workflowRunId: workflowRun.id,
+      });
+    });
+
     test("handles workflow errors with deadline exceeded", async () => {
       const backend = await createBackend();
       const client = new OpenWorkflow({ backend });

packages/openworkflow/worker/execution.ts

@@ -935,6 +935,44 @@ class StepExecutor implements StepApi {
   }
 }
 
+/**
+ * Execute a workflow-run transition and swallow expected stale-write races when
+ * this worker no longer owns an actively running execution.
+ * @param options - Transition execution options
+ */
+async function executeWorkflowRunTransition(
+  options: Readonly<{
+    backend: Backend;
+    workflowRunId: string;
+    workerId: string;
+    transition: () => Promise<unknown>;
+  }>,
+): Promise<void> {
+  try {
+    await options.transition();
+  } catch (error) {
+    let currentRun: WorkflowRun | null = null;
+
+    try {
+      currentRun = await options.backend.getWorkflowRun({
+        workflowRunId: options.workflowRunId,
+      });
+    } catch {
+      throw error;
+    }
+
+    if (
+      currentRun &&
+      (currentRun.status !== "running" ||
+        currentRun.workerId !== options.workerId)
+    ) {
+      return;
+    }
+
+    throw error;
+  }
+}
+
 /**
  * Parameters for the workflow execution use case.
  */
@@ -1030,31 +1068,52 @@ export async function executeWorkflow(
 
     // mark success
     executionFence.deactivate();
-    await backend.completeWorkflowRun({
+    await executeWorkflowRunTransition({
+      backend,
       workflowRunId: workflowRun.id,
       workerId,
-      output: (output ?? null) as JsonValue,
+      transition: async () => {
+        await backend.completeWorkflowRun({
+          workflowRunId: workflowRun.id,
+          workerId,
+          output: (output ?? null) as JsonValue,
+        });
+      },
     });
   } catch (error) {
     executionFence.deactivate();
 
     // handle sleep signal by parking the workflow in running status
     if (error instanceof SleepSignal) {
-      await backend.sleepWorkflowRun({
+      await executeWorkflowRunTransition({
+        backend,
         workflowRunId: workflowRun.id,
         workerId,
-        availableAt: error.resumeAt,
+        transition: async () => {
+          await backend.sleepWorkflowRun({
+            workflowRunId: workflowRun.id,
+            workerId,
+            availableAt: error.resumeAt,
+          });
+        },
       });
 
       return;
     }
 
     if (error instanceof StepLimitExceededError) {
-      await backend.failWorkflowRun({
+      await executeWorkflowRunTransition({
+        backend,
         workflowRunId: workflowRun.id,
         workerId,
-        error: serializeStepLimitExceededError(error),
-        retryPolicy: DEFAULT_WORKFLOW_RETRY_POLICY,
+        transition: async () => {
+          await backend.failWorkflowRun({
+            workflowRunId: workflowRun.id,
+            workerId,
+            error: serializeStepLimitExceededError(error),
+            retryPolicy: DEFAULT_WORKFLOW_RETRY_POLICY,
+          });
+        },
       });
       return;
     }
@@ -1071,11 +1130,18 @@ export async function executeWorkflow(
       );
 
       if (retryDecision.status === "failed") {
-        await backend.failWorkflowRun({
+        await executeWorkflowRunTransition({
+          backend,
           workflowRunId: workflowRun.id,
           workerId,
-          error: serializedError,
-          retryPolicy: DEFAULT_WORKFLOW_RETRY_POLICY,
+          transition: async () => {
+            await backend.failWorkflowRun({
+              workflowRunId: workflowRun.id,
+              workerId,
+              error: serializedError,
+              retryPolicy: DEFAULT_WORKFLOW_RETRY_POLICY,
+            });
+          },
         });
         return;
       }
@@ -1089,11 +1155,20 @@ export async function executeWorkflow(
       }
       /* v8 ignore stop */
 
-      await backend.rescheduleWorkflowRunAfterFailedStepAttempt({
+      const availableAt = retryDecision.availableAt;
+
+      await executeWorkflowRunTransition({
+        backend,
         workflowRunId: workflowRun.id,
         workerId,
-        error: serializedError,
-        availableAt: retryDecision.availableAt,
+        transition: async () => {
+          await backend.rescheduleWorkflowRunAfterFailedStepAttempt({
+            workflowRunId: workflowRun.id,
+            workerId,
+            error: serializedError,
+            availableAt,
+          });
+        },
       });
       return;
     }
@@ -1103,11 +1178,18 @@ export async function executeWorkflow(
     }
 
     // mark failure
-    await backend.failWorkflowRun({
+    await executeWorkflowRunTransition({
+      backend,
       workflowRunId: workflowRun.id,
       workerId,
-      error: serializeError(error),
-      retryPolicy: params.retryPolicy,
+      transition: async () => {
+        await backend.failWorkflowRun({
+          workflowRunId: workflowRun.id,
+          workerId,
+          error: serializeError(error),
+          retryPolicy: params.retryPolicy,
+        });
+      },
     });
   }
 }

packages/openworkflow/worker/worker.test.ts

@@ -1078,6 +1078,9 @@ describe("Worker", () => {
   test("worker handles when canceled workflow during execution", async () => {
     const backend = await createBackend();
     const client = new OpenWorkflow({ backend });
+    const consoleErrorSpy = vi
+      .spyOn(console, "error")
+      .mockImplementation(() => 0);
 
     let stepExecuted = false;
     const workflow = client.defineWorkflow(
@@ -1094,25 +1097,33 @@ describe("Worker", () => {
     );
     const worker = client.newWorker();
 
-    const handle = await workflow.run();
+    try {
+      const handle = await workflow.run();
 
-    // start processing in the background
-    const tickPromise = worker.tick();
-    await sleep(25);
+      // start processing in the background
+      const tickPromise = worker.tick();
+      await sleep(25);
 
-    // cancel while step is executing
-    await handle.cancel();
+      // cancel while step is executing
+      await handle.cancel();
 
-    // wait for tick to complete
-    await tickPromise;
-    await worker.stop();
+      // wait for tick to complete
+      await tickPromise;
+      await worker.stop();
 
-    // step should have been executed but workflow should be canceled
-    expect(stepExecuted).toBe(true);
-    const canceled = await backend.getWorkflowRun({
-      workflowRunId: handle.workflowRun.id,
-    });
-    expect(canceled?.status).toBe("canceled");
+      // step should have been executed but workflow should be canceled
+      expect(stepExecuted).toBe(true);
+      const canceled = await backend.getWorkflowRun({
+        workflowRunId: handle.workflowRun.id,
+      });
+      expect(canceled?.status).toBe("canceled");
+      expect(consoleErrorSpy).not.toHaveBeenCalledWith(
+        expect.stringContaining("Critical error during workflow execution"),
+        expect.anything(),
+      );
+    } finally {
+      consoleErrorSpy.mockRestore();
+    }
   });
 
   test("result() rejects for canceled workflows", async () => {