Ignore stale workflow handler runs and failure notifications

.changeset/failed-fan-in-duplicates.md

@@ -0,0 +1,11 @@
+---
+"@convex-dev/workflow": patch
+---
+
+A workflow that fails after parallel steps is no longer replayed by the extra
+handler runs queued by those steps. Those runs now return early instead of
+failing with "Workflow not running" and being counted as Workpool failures.
+(#327)
+
+Handler runs and handler failure notifications for a workflow that already
+finished, was restarted or was cleaned up now return without throwing. (#328)

.changeset/journal-load-short-circuit.md

@@ -0,0 +1,6 @@
+---
+"@convex-dev/workflow": patch
+---
+
+A handler run for a workflow that already finished no longer reads its journal
+before returning.

example/convex/_generated/api.d.ts

@@ -16,6 +16,7 @@ import type * as nestedWorkflow from "../nestedWorkflow.js";
 import type * as oversized from "../oversized.js";
 import type * as passingSignals from "../passingSignals.js";
 import type * as test_contextRoundtrip from "../test/contextRoundtrip.js";
+import type * as test_fanIn from "../test/fanIn.js";
 import type * as test_inline from "../test/inline.js";
 import type * as test_oldSyntax from "../test/oldSyntax.js";
 import type * as transcription from "../transcription.js";
@@ -36,6 +37,7 @@ declare const fullApi: ApiFromModules<{
   oversized: typeof oversized;
   passingSignals: typeof passingSignals;
   "test/contextRoundtrip": typeof test_contextRoundtrip;
+  "test/fanIn": typeof test_fanIn;
   "test/inline": typeof test_inline;
   "test/oldSyntax": typeof test_oldSyntax;
   transcription: typeof transcription;

example/convex/test/fanIn.test.ts

@@ -0,0 +1,84 @@
+/// <reference types="vite/client" />
+
+import { getStatus } from "@convex-dev/workflow";
+import { makeFunctionReference } from "convex/server";
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+import { handlerOnComplete } from "../../../src/component/pool.js";
+import { components, internal } from "../_generated/api";
+import { initConvexTest } from "../setup.test";
+import { handlerRuns, workflow } from "./fanIn";
+
+type WithHandler<T> = T & { _handler: (...args: unknown[]) => unknown };
+
+describe("fan-in of parallel steps", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    handlerRuns.count = 0;
+  });
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.useRealTimers();
+  });
+
+  test.each([{ fail: true }, { fail: false }])(
+    "replays the handler once when all steps finish (fail: $fail)",
+    async ({ fail }) => {
+      const t = initConvexTest();
+      // Called once per handler work item that the workpool records as failed.
+      const failureCallback = vi.spyOn(
+        handlerOnComplete as WithHandler<typeof handlerOnComplete>,
+        "_handler",
+      );
+      const workflowId = await t.run((ctx) =>
+        workflow.start(ctx, internal.test.fanIn.fanIn, { width: 8, fail }),
+      );
+      await t.finishAllScheduledFunctions(vi.runAllTimers);
+      const status = await t.run((ctx) =>
+        getStatus(ctx, components.workflow, workflowId),
+      );
+      if (fail) {
+        expect(status).toMatchObject({
+          type: "failed",
+          error: expect.stringContaining("boom"),
+        });
+      } else {
+        expect(status).toEqual({ type: "completed", result: 8 });
+      }
+      // One run to start the steps and one replay once they have all finished.
+      // The other queued runs return without replaying.
+      expect(handlerRuns.count).toBe(2);
+      expect(failureCallback).not.toHaveBeenCalled();
+    },
+  );
+
+  test.each([{ fail: true }, { fail: false }])(
+    "a duplicate run of a finished workflow does not replay (fail: $fail)",
+    async ({ fail }) => {
+      const t = initConvexTest();
+      const workflowId = await t.run((ctx) =>
+        workflow.start(ctx, internal.test.fanIn.fanIn, { width: 3, fail }),
+      );
+      await t.finishAllScheduledFunctions(vi.runAllTimers);
+      const before = await t.run((ctx) =>
+        getStatus(ctx, components.workflow, workflowId),
+      );
+      const runsBefore = handlerRuns.count;
+
+      // What a second handler work item for the same generation would do.
+      const result = await t.mutation(
+        makeFunctionReference<
+          "mutation",
+          { workflowId: string; generationNumber: number }
+        >("test/fanIn:fanIn"),
+        { workflowId, generationNumber: 0 },
+      );
+
+      expect(result).toBe(workflowId);
+      expect(handlerRuns.count).toBe(runsBefore);
+      const after = await t.run((ctx) =>
+        getStatus(ctx, components.workflow, workflowId),
+      );
+      expect(after).toEqual(before);
+    },
+  );
+});

example/convex/test/fanIn.ts

@@ -0,0 +1,35 @@
+import { WorkflowManager } from "@convex-dev/workflow";
+import { v } from "convex/values";
+import { components, internal } from "../_generated/api.js";
+import { internalMutation } from "../_generated/server.js";
+
+export const workflow = new WorkflowManager(components.workflow);
+
+// Counts handler invocations (including replays), so tests can observe
+// redundant replays. Only meaningful in convex-test, where modules are shared.
+export const handlerRuns = { count: 0 };
+
+export const noop = internalMutation({
+  args: {},
+  returns: v.null(),
+  handler: async () => null,
+});
+
+// Many steps complete together, then the handler fails or succeeds.
+export const fanIn = workflow
+  .define({
+    args: { width: v.number(), fail: v.boolean() },
+    returns: v.number(),
+  })
+  .handler(async (step, args): Promise<number> => {
+    handlerRuns.count++;
+    await Promise.all(
+      Array.from({ length: args.width }, () =>
+        step.runMutation(internal.test.fanIn.noop, {}),
+      ),
+    );
+    if (args.fail) {
+      throw new Error("boom");
+    }
+    return args.width;
+  });

src/client/workflowMutation.ts

@@ -190,6 +190,20 @@ export function workflowMutation<
       );
       const inProgress = journalEntries.filter(({ step }) => step.inProgress);
       const console = createLogger(logLevel);
+      if (workflow.generationNumber !== generationNumber) {
+        console.error(
+          `Invalid generation number: ${generationNumber} running workflow ${workflow.name} (${workflowId})`,
+        );
+        return workflowId;
+      }
+      if (workflow.runResult !== undefined) {
+        // A duplicate run of a generation that already finished. Completing it
+        // again can never succeed, so don't replay the handler.
+        console.log(
+          `Workflow ${workflowId} already ${workflow.runResult.kind}, returning.`,
+        );
+        return workflowId;
+      }
       if (!ok) {
         console.error(`Failed to load journal for ${workflowId}`);
         await ctx.runMutation(component.workflow.complete, {
@@ -199,16 +213,6 @@ export function workflowMutation<
         });
         return workflowId;
       }
-      if (workflow.generationNumber !== generationNumber) {
-        console.error(
-          `Invalid generation number: ${generationNumber} running workflow ${workflow.name} (${workflowId})`,
-        );
-        return workflowId;
-      }
-      if (workflow.runResult?.kind === "success") {
-        console.log(`Workflow ${workflowId} completed, returning.`);
-        return workflowId;
-      }
       if (inProgress.length > 0) {
         console.log(
           `Workflow ${workflowId} blocked by ` +

src/component/journal.ts

@@ -40,6 +40,10 @@ export const load = query({
     const { logLevel } = await getDefaultLogger(ctx);
     const journalEntries: JournalEntry[] = [];
     let journalSize = 0;
+    if (shortCircuit && workflow.runResult !== undefined) {
+      // Nothing left to replay, so skip reading the journal.
+      return { journalEntries, workflow, logLevel, ok: true };
+    }
     if (shortCircuit) {
       const inProgress = await ctx.db
         .query("steps")

src/component/pool.ts

@@ -245,6 +245,29 @@ export const handlerOnComplete = internalMutation({
       return;
     }
     const { workflowId, generationNumber } = args.context;
+    // The handler's mutation already rolled back. If its run can no longer be
+    // completed, this notification never can be either, so don't throw.
+    // A malformed id falls through, and fails loudly in completeHandler.
+    const id = ctx.db.normalizeId("workflows", workflowId);
+    const workflow = id && (await ctx.db.get("workflows", id));
+    if (id && !workflow) {
+      console.warn(
+        `Workflow ${workflowId} not found when its handler failed. It was likely cleaned up.`,
+      );
+      return;
+    }
+    if (workflow && workflow.generationNumber !== generationNumber) {
+      console.warn(
+        `Workflow ${workflowId} is at generation ${workflow.generationNumber} when its handler failed at generation ${generationNumber}`,
+      );
+      return;
+    }
+    if (workflow && workflow.runResult !== undefined) {
+      console.warn(
+        `Workflow ${workflowId} already ${workflow.runResult.kind} when its handler failed`,
+      );
+      return;
+    }
     await completeHandler(ctx, {
       workflowId,
       generationNumber,