Store oversized step results using mutation file storage

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## Unreleased
+
+- When a step fails due to an oversized result, the result is saved in file storage. The value can be read with `oversizedValues.read` for debugging. Workflow cleanup and restart delete stored values.
+- Requires Convex 1.48 or newer for file storage access in mutations.
+
 ## 0.4.10
 
 - Updates the /test entrypoint for compatibility with convex-test's new

README.md

@@ -838,7 +838,7 @@ Here are a few limitations to keep in mind:
 - Steps can only take in and return a total of _1 MB_ of data within a single
   workflow execution. If you run into journal size limits, you can work around
   this by storing results in the database or file storage from your step
-  functions and passing IDs around within the the workflow.
+  functions and passing IDs around within the the workflow. Oversized step results are stored in file storage when possible and can be read with `oversizedValues.read` for debugging.
 - The workflow body is internally a mutation, with each step's return value read
   from the database on each subsequent step. As a result, the limits for a
   mutation apply and limit the number and size of steps you can perform

example/convex/oversized.test.ts

@@ -7,6 +7,12 @@ import { assert } from "convex-helpers";
 import { getStatus } from "@convex-dev/workflow";
 import { workflow } from "./example";
 
+function storedStepId(error: string) {
+  const stepId = error.match(/stepId: "([^"]+)"/)?.[1];
+  assert(stepId, `No retrieval hint in: ${error}`);
+  return stepId;
+}
+
 describe("oversized values", () => {
   beforeEach(() => {
     vi.useFakeTimers();
@@ -55,6 +61,16 @@ describe("oversized values", () => {
     expect(flow.out).not.toBeNull();
     expect(flow.out.kind).toBe("failed");
     expect(flow.out.error).toContain("Step return value too large");
+
+    const stepId = storedStepId(status.error);
+    expect(
+      await t.mutation(components.workflow.oversizedValues.read, { stepId }),
+    ).toBe("x".repeat(900_000));
+    await t.mutation(components.workflow.oversizedValues.remove, { stepId });
+    await expect(
+      t.mutation(components.workflow.oversizedValues.read, { stepId }),
+    ).rejects.toThrow("No oversized value found");
+    await t.mutation(components.workflow.oversizedValues.remove, { stepId });
   });
 
   test("large event value fails workflow and calls onComplete", async () => {
@@ -107,5 +123,100 @@ describe("oversized values", () => {
     expect(flow!.out).not.toBeNull();
     expect(flow!.out.kind).toBe("failed");
     expect(flow!.out.error).toContain("Step return value too large");
+
+    const stepId = storedStepId(status2.error);
+    expect(
+      await t.mutation(components.workflow.oversizedValues.read, { stepId }),
+    ).toBe("y".repeat(900_000));
+    expect(
+      await t.mutation(components.workflow.workflow.cleanup, { workflowId }),
+    ).toBe(true);
+  });
+
+  test("stores an oversized pre-sent event value", async () => {
+    const t = initConvexTest();
+    const workflowId = await t.mutation(async (ctx) => {
+      const id = await workflow.start(
+        ctx,
+        internal.oversized.eventWorkflow,
+        {},
+        {
+          startAsync: true,
+        },
+      );
+      await ctx.runMutation(internal.oversized.sendBigEvent, {
+        workflowId: id,
+      });
+      return id;
+    });
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    const status = await t.run((ctx) =>
+      getStatus(ctx, components.workflow, workflowId),
+    );
+    assert(status.type === "failed");
+    expect(status.error).toContain("Step return value too large");
+    const stepId = storedStepId(status.error);
+    expect(
+      await t.mutation(components.workflow.oversizedValues.read, { stepId }),
+    ).toBe("y".repeat(900_000));
+  });
+
+  test("a batch of 11 oversized pre-sent events fails the workflow", async () => {
+    const t = initConvexTest();
+    const workflowId = await t.mutation(async (ctx) => {
+      const id = await workflow.start(
+        ctx,
+        internal.oversized.eventBatchWorkflow,
+        {},
+        { startAsync: true },
+      );
+      await ctx.runMutation(internal.oversized.sendEventBatch, {
+        workflowId: id,
+      });
+      return id;
+    });
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    const status = await t.run((ctx) =>
+      getStatus(ctx, components.workflow, workflowId),
+    );
+    assert(status.type === "failed");
+    expect(status.error).toContain("Step return value too large");
+  });
+
+  test("a queued result exceeding the file byte budget fails and calls onComplete", async () => {
+    const t = initConvexTest();
+    const workflowId = await t.mutation(async (ctx) => {
+      const id = await workflow.start(
+        ctx,
+        internal.oversized.overStorageLimitWorkflow,
+        {},
+        {
+          startAsync: true,
+          onComplete: internal.oversized.onComplete,
+          context: {},
+        },
+      );
+      await ctx.db.insert("flows", {
+        workflowId: id,
+        in: "over storage limit",
+        out: null,
+      });
+      return id;
+    });
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    const status = await t.run((ctx) =>
+      getStatus(ctx, components.workflow, workflowId),
+    );
+    assert(status.type === "failed");
+    expect(status.error).toContain("3000002 bytes");
+    expect(status.error).toContain("Preview:");
+    expect(status.error).not.toContain("Retrieve it");
+    const flow = await t.query((ctx) =>
+      ctx.db
+        .query("flows")
+        .withIndex("workflowId", (q) => q.eq("workflowId", workflowId))
+        .unique(),
+    );
+    expect(flow?.out.kind).toBe("failed");
   });
 });

example/convex/oversized.ts

@@ -27,6 +27,19 @@ export const largeReturnWorkflow = workflow
     return result;
   });
 
+export const overStorageLimitAction = internalAction({
+  args: {},
+  returns: v.string(),
+  // Escaping NUL characters produces an 18 MB JSON blob.
+  handler: async (): Promise<string> => "\0".repeat(3_000_000),
+});
+
+export const overStorageLimitWorkflow = workflow
+  .define({ args: {} })
+  .handler(async (step) => {
+    return await step.runAction(internal.oversized.overStorageLimitAction, {});
+  });
+
 export const bigEvent = defineEvent({
   name: "bigEvent",
   validator: v.string(),
@@ -41,6 +54,29 @@ export const eventWorkflow = workflow
     return result;
   });
 
+export const eventBatchWorkflow = workflow
+  .define({ args: {} })
+  .handler(async (step) => {
+    await Promise.all(
+      Array.from({ length: 11 }, (_, i) =>
+        step.awaitEvent({ name: `batch-${i}` }),
+      ),
+    );
+  });
+
+export const sendEventBatch = internalMutation({
+  args: { workflowId: vWorkflowId },
+  handler: async (ctx, { workflowId }) => {
+    for (let i = 0; i < 11; i++) {
+      await sendEvent(ctx, components.workflow, {
+        workflowId,
+        name: `batch-${i}`,
+        value: "z".repeat(900_000),
+      });
+    }
+  },
+});
+
 export const sendBigEvent = internalMutation({
   args: {
     workflowId: vWorkflowId,

package.json

@@ -60,7 +60,7 @@
   },
   "peerDependencies": {
     "@convex-dev/workpool": "^0.4.4",
-    "convex": "^1.46.0",
+    "convex": "^1.48.0",
     "convex-helpers": "^0.1.127"
   },
   "dependencies": {
@@ -76,9 +76,9 @@
     "@typescript-eslint/eslint-plugin": "8.71.1",
     "@typescript-eslint/parser": "8.71.1",
     "chokidar-cli": "3.0.0",
-    "convex": "1.46.0",
+    "convex": "1.48.0",
     "convex-helpers": "0.1.128",
-    "convex-test": "0.0.64",
+    "convex-test": "0.0.67",
     "cpy-cli": "7.0.0",
     "eslint": "10.12.0",
     "eslint-plugin-react-hooks": "7.1.1",

src/component/_generated/component.ts

@@ -303,6 +303,22 @@ export type ComponentApi<Name extends string | undefined = string | undefined> =
         Name
       >;
     };
+    oversizedValues: {
+      read: FunctionReference<
+        "mutation",
+        "internal",
+        { stepId: string },
+        any,
+        Name
+      >;
+      remove: FunctionReference<
+        "mutation",
+        "internal",
+        { stepId: string },
+        null,
+        Name
+      >;
+    };
     workflow: {
       cancel: FunctionReference<
         "mutation",

src/component/event.ts

@@ -6,7 +6,7 @@ import { vResultValidator } from "@convex-dev/workpool";
 import type { Doc, Id } from "./_generated/dataModel.js";
 import { assert } from "convex-helpers";
 import { enqueueWorkflow, getWorkpool, workpoolOptions } from "./pool.js";
-import { checkForOversizedResult } from "./oversizedValues.js";
+import { storeOversizedResult } from "./oversizedValues.js";
 
 export async function awaitEvent(
   ctx: MutationCtx,
@@ -41,7 +41,11 @@ export async function awaitEvent(
           stepId: entry._id,
         },
       });
-      entry.step.runResult = checkForOversizedResult(event.state.result);
+      entry.step.runResult = await storeOversizedResult(
+        ctx,
+        event.state.result,
+        { stepId: entry._id },
+      );
       entry.step.inProgress = false;
       entry.step.completedAt = Date.now();
       break;
@@ -147,7 +151,9 @@ export const send = mutation({
         );
         assert(step.step.kind === "event", "Step is not an event");
         step.step.eventId = event._id;
-        step.step.runResult = checkForOversizedResult(args.result);
+        step.step.runResult = await storeOversizedResult(ctx, args.result, {
+          stepId: step._id,
+        });
         step.step.inProgress = false;
         step.step.completedAt = Date.now();
         await ctx.db.replace("steps", step._id, step);