Preserve ConvexError data across workflow steps and replay

.changeset/preserve-convex-error-data.md

@@ -0,0 +1,8 @@
+---
+"@convex-dev/workflow": minor
+---
+
+Preserve `ConvexError` identity and structured error data when catching failed
+steps, including inline calls, queued functions, nested workflows, and journal
+replay. Raises the `@convex-dev/workpool` peer minimum to 0.4.16. Oversized
+failures are replaced with a bounded diagnostic so they can still be journaled.

README.md

@@ -313,6 +313,17 @@ that the "myAction" step is considered "in progress" while it is waiting, and it
 only enqueues one item into the Workpool (myAction@delay), instead of two
 (sleep@delay, myAction@now).
 
+### Catching step errors
+
+When a step throws a `ConvexError`, catching it in the workflow preserves
+`error instanceof ConvexError` and the original `error.data`. This applies to
+queries, mutations, actions, and nested workflows, including journal replay.
+Other failures throw an `Error` with the recorded error message.
+
+Failures larger than 800 KiB, counting the message, stack, and structured data
+together, are replaced with an `Error` containing a bounded diagnostic preview.
+The oversized structured data is omitted so the failure can be journaled.
+
 ### Specifying retry behavior
 
 Sometimes actions fail due to transient errors, whether it was an unreliable

example/convex/_generated/api.d.ts

@@ -17,6 +17,7 @@ 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_convexErrors from "../test/convexErrors.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";
@@ -38,6 +39,7 @@ declare const fullApi: ApiFromModules<{
   passingSignals: typeof passingSignals;
   "test/contextRoundtrip": typeof test_contextRoundtrip;
   "test/fanIn": typeof test_fanIn;
+  "test/convexErrors": typeof test_convexErrors;
   "test/inline": typeof test_inline;
   "test/oldSyntax": typeof test_oldSyntax;
   transcription: typeof transcription;

example/convex/test/convexErrors.test.ts

@@ -0,0 +1,95 @@
+import { getStatus, WorkflowManager } from "@convex-dev/workflow";
+import { afterEach, beforeEach, expect, test, vi } from "vitest";
+import { components, internal } from "../_generated/api.js";
+import { initConvexTest } from "../setup.test.js";
+import { getConvexSize, type Value } from "convex/values";
+
+const workflow = new WorkflowManager(components.workflow);
+
+beforeEach(() => vi.useFakeTimers());
+afterEach(() => vi.useRealTimers());
+
+test.each([
+  { functionType: "query" as const, inline: true },
+  { functionType: "mutation" as const, inline: true },
+  { functionType: "query" as const, inline: false },
+  { functionType: "mutation" as const, inline: false },
+  { functionType: "action" as const, inline: false },
+])(
+  "preserves ConvexError from $functionType (inline: $inline) through replay",
+  async (options) => {
+    const t = initConvexTest();
+    const data = {
+      code: "INVALID_INPUT",
+      message: "Please try again",
+      count: 7,
+    };
+    const workflowId = await t.mutation((ctx) =>
+      workflow.start(ctx, internal.test.convexErrors.catchingWorkflow, {
+        ...options,
+        data,
+      }),
+    );
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    const status = await t.query((ctx) =>
+      getStatus(ctx, components.workflow, workflowId),
+    );
+    expect(status).toEqual({
+      type: "completed",
+      result: { isConvexError: true, data },
+    });
+  },
+);
+
+test("preserves an uncaught ConvexError from a nested workflow", async () => {
+  const t = initConvexTest();
+  const data = { code: "CHILD_FAILED", message: "Please try again" };
+  const workflowId = await t.mutation((ctx) =>
+    workflow.start(ctx, internal.test.convexErrors.catchingNestedWorkflow, {
+      data,
+    }),
+  );
+  await t.finishAllScheduledFunctions(vi.runAllTimers);
+  expect(
+    await t.query((ctx) => getStatus(ctx, components.workflow, workflowId)),
+  ).toEqual({
+    type: "completed",
+    result: { isConvexError: true, data },
+  });
+});
+
+test.each(["query", "mutation"] as const)(
+  "oversized inline %s failures remain catchable through replay",
+  async (functionType) => {
+    const t = initConvexTest();
+    const workflowId = await t.mutation((ctx) =>
+      workflow.start(ctx, internal.test.convexErrors.catchingOversizedError, {
+        functionType,
+      }),
+    );
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    const status = await t.query((ctx) =>
+      getStatus(ctx, components.workflow, workflowId),
+    );
+    expect(status.type).toBe("completed");
+    if (status.type !== "completed")
+      throw new Error("Expected recovery to complete");
+    expect(status.result).toMatchObject({
+      isConvexError: false,
+      data: expect.stringContaining("Step failure too large"),
+    });
+    expect(getConvexSize(status.result as Value)).toBeLessThan(100 << 10);
+    const steps = await t.query((ctx) =>
+      workflow.listSteps(ctx, workflowId, {
+        paginationOpts: { numItems: 10, cursor: null },
+      }),
+    );
+    expect(steps.page).toHaveLength(2);
+    const failure = steps.page[0].runResult;
+    expect(failure).toMatchObject({ kind: "failed" });
+    if (failure?.kind !== "failed")
+      throw new Error("Expected a recorded failure");
+    expect(failure).not.toHaveProperty("errorData");
+    expect(getConvexSize(failure)).toBeLessThan(100 << 10);
+  },
+);

example/convex/test/convexErrors.ts

@@ -0,0 +1,156 @@
+import { WorkflowManager } from "@convex-dev/workflow";
+import { ConvexError, v, type Value } from "convex/values";
+import { components, internal } from "../_generated/api.js";
+import {
+  internalAction,
+  internalMutation,
+  internalQuery,
+} from "../_generated/server.js";
+
+const workflow = new WorkflowManager(components.workflow);
+const args = { data: v.any() };
+const fail = (_ctx: unknown, { data }: { data: Value }): never => {
+  throw new ConvexError(data);
+};
+
+export const failingQuery = internalQuery({
+  args,
+  returns: v.null(),
+  handler: fail,
+});
+export const failingMutation = internalMutation({
+  args,
+  returns: v.null(),
+  handler: fail,
+});
+export const failingAction = internalAction({
+  args,
+  returns: v.null(),
+  handler: fail,
+});
+
+export const oversizedQuery = internalQuery({
+  args: {},
+  returns: v.null(),
+  handler: (ctx) => fail(ctx, { data: "x".repeat(900_000) }),
+});
+
+export const oversizedMutation = internalMutation({
+  args: {},
+  returns: v.null(),
+  handler: (ctx) => fail(ctx, { data: "x".repeat(900_000) }),
+});
+
+export const catchingOversizedError = workflow
+  .define({
+    args: { functionType: v.union(v.literal("query"), v.literal("mutation")) },
+    returns: v.object({ isConvexError: v.boolean(), data: v.any() }),
+  })
+  .handler(async (step, { functionType }) => {
+    try {
+      if (functionType === "query") {
+        await step.runQuery(
+          internal.test.convexErrors.oversizedQuery,
+          {},
+          { inline: true },
+        );
+      } else {
+        await step.runMutation(
+          internal.test.convexErrors.oversizedMutation,
+          {},
+          { inline: true },
+        );
+      }
+    } catch (error) {
+      // Journal these caught values in another step so replay must agree.
+      return await step.runAction(internal.test.convexErrors.echo, {
+        isConvexError: error instanceof ConvexError,
+        data: error instanceof Error ? error.message : String(error),
+      });
+    }
+    throw new Error("Expected an oversized error");
+  });
+
+export const echo = internalAction({
+  args: { isConvexError: v.boolean(), data: v.any() },
+  returns: v.object({ isConvexError: v.boolean(), data: v.any() }),
+  handler: async (_ctx, args) => args,
+});
+
+export const failingWorkflow = workflow
+  .define({ args, returns: v.null() })
+  .handler(async (_step, { data }) => {
+    throw new ConvexError(data);
+  });
+
+export const catchingNestedWorkflow = workflow
+  .define({
+    args,
+    returns: v.object({ isConvexError: v.boolean(), data: v.any() }),
+  })
+  .handler(async (step, { data }) => {
+    try {
+      await step.runWorkflow(internal.test.convexErrors.failingWorkflow, {
+        data,
+      });
+    } catch (error) {
+      return await step.runAction(internal.test.convexErrors.echo, {
+        isConvexError: error instanceof ConvexError,
+        data: error instanceof ConvexError ? error.data : null,
+      });
+    }
+    throw new Error("Expected the child workflow to fail");
+  });
+
+export const catchingWorkflow = workflow
+  .define({
+    args: {
+      data: v.any(),
+      functionType: v.union(
+        v.literal("query"),
+        v.literal("mutation"),
+        v.literal("action"),
+      ),
+      inline: v.boolean(),
+    },
+    returns: v.object({ isConvexError: v.boolean(), data: v.any() }),
+  })
+  .handler(async (step, { data, functionType, inline }) => {
+    let caught: { isConvexError: boolean; data: Value } = {
+      isConvexError: false,
+      data: null,
+    };
+    try {
+      switch (functionType) {
+        case "query":
+          await step.runQuery(
+            internal.test.convexErrors.failingQuery,
+            { data },
+            { inline },
+          );
+          break;
+        case "mutation":
+          await step.runMutation(
+            internal.test.convexErrors.failingMutation,
+            { data },
+            { inline },
+          );
+          break;
+        case "action":
+          await step.runAction(
+            internal.test.convexErrors.failingAction,
+            { data },
+            { retry: false },
+          );
+          break;
+      }
+    } catch (error) {
+      caught = {
+        isConvexError: error instanceof ConvexError,
+        data: error instanceof ConvexError ? error.data : null,
+      };
+    }
+    // A queued recovery step forces replay of the failed step and validates
+    // that the caught data stays the same when used in subsequent step args.
+    return await step.runAction(internal.test.convexErrors.echo, caught);
+  });

package.json

@@ -58,7 +58,7 @@
     }
   },
   "peerDependencies": {
-    "@convex-dev/workpool": "^0.4.4",
+    "@convex-dev/workpool": "^0.4.16",
     "convex": "^1.46.0",
     "convex-helpers": "^0.1.127"
   },
@@ -68,7 +68,7 @@
   "devDependencies": {
     "@changesets/cli": "3.0.3",
     "@convex-dev/eslint-plugin": "5.0.1",
-    "@convex-dev/workpool": "0.4.15",
+    "@convex-dev/workpool": "0.4.16",
     "@edge-runtime/vm": "5.0.0",
     "@eslint/eslintrc": "3.3.7",
     "@eslint/js": "10.0.1",

src/client/step.ts

@@ -17,7 +17,7 @@ import type {
   TransactionLimits,
   WorkflowComponent,
 } from "./types.js";
-import { MAX_JOURNAL_SIZE, formatErrorWithStack } from "../shared.js";
+import { MAX_JOURNAL_SIZE, failedResult } from "../shared.js";
 import type { EventId, SchedulerOptions } from "../types.js";
 import { pick } from "convex-helpers";
 
@@ -190,7 +190,7 @@ export class StepExecutor {
                   );
             runResult = { kind: "success", returnValue: result ?? null };
           } catch (error: unknown) {
-            runResult = { kind: "failed", error: formatErrorWithStack(error) };
+            runResult = failedResult(error);
           }
         }