feat(sdk,core,webapp): offload large batch payloads to object storage (#4165)

## Summary

`batchTrigger` and `batchTriggerAndWait` (and the by-id and by-task
variants) now offload any per-item payload over 128KB to object storage
before sending, the same way single `trigger`/`triggerAndWait` already
do since
[#3785](#3785). A batch
of large items no longer inflates the request body past the API limit.

## Demo

A live local run: `batchTriggerAndWait` of 5 items × 300KB (1.5MB
total). Each item offloads to object storage, so the receiver run rows
hold a 65-byte `application/store` pointer instead of the 300KB body,
and every item round-trips (received == sent).

<img width="1000" height="494" alt="batch large-payload offload demo"
src="https://github.com/user-attachments/assets/77ae3958-97d6-4b5c-ab25-39b217caefbc"
/>


## Design

Both the array and streaming batch paths funnel through
`executeBatchTwoPhase`, so offloading happens once there: each item is
measured, then offloaded through the existing
`conditionallyExportPacket` when it crosses 128KB, with bounded
concurrency so a big batch doesn't fire an unbounded number of presigned
PUTs.

Because items are offloaded before the request, SDK batches arrive as
small `application/store` references, so the server-side inline offload
during item ingest (parallelised in
[#3777](#3777)) mostly
no longer fires for them.

Every trigger and item also carries its pre-offload serialised size as
`options.payloadSize`. The trigger span records that value, so an
offloaded payload shows its real size instead of the size of the small
object-store reference (previously the span measured the reference).

Author: 1stvamp

.changeset/large-batch-payloads-and-size.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Large batch payloads now offload to object storage instead of riding inline in the trigger request. `batchTrigger` and `batchTriggerAndWait` (and the by-id and by-task variants) offload any per-item payload over 128KB before sending, the same way single `trigger` and `triggerAndWait` already do, so a big batch no longer blows past the API body limit.

apps/webapp/app/runEngine/concerns/payloads.server.ts

@@ -24,7 +24,11 @@ export class DefaultPayloadProcessor implements PayloadProcessor {
       );
 
       span.setAttribute("needsOffloading", needsOffloading);
-      span.setAttribute("size", size);
+      // When the caller already offloaded the payload (payloadType "application/store"), the
+      // packet here is just the small object-store reference, so `size` measures the reference,
+      // not the payload. Prefer the caller-reported pre-offload size when it's provided so the
+      // span reflects the real payload size. For inline payloads the two agree.
+      span.setAttribute("size", request.body.options?.payloadSize ?? size);
 
       if (!needsOffloading) {
         return packet;

packages/core/src/v3/schemas/api-type.test.ts

@@ -1,5 +1,5 @@
 import { describe, it, expect } from "vitest";
-import { InitializeDeploymentRequestBody, TriggerTaskRequestBody } from "./api.js";
+import { BatchItemNDJSON, InitializeDeploymentRequestBody, TriggerTaskRequestBody } from "./api.js";
 import type { InitializeDeploymentRequestBody as InitializeDeploymentRequestBodyType } from "./api.js";
 
 describe("InitializeDeploymentRequestBody", () => {
@@ -176,4 +176,47 @@ describe("TriggerTaskRequestBody", () => {
 
     expect(result.success).toBe(false);
   });
+
+  it("accepts an optional payloadSize on the options", () => {
+    const result = TriggerTaskRequestBody.safeParse({
+      payload: "packets/payloads/file.json",
+      context: {},
+      options: {
+        payloadType: "application/store",
+        payloadSize: 512 * 1024,
+      },
+    });
+
+    expect(result.success).toBe(true);
+    expect(result.success && result.data.options?.payloadSize).toBe(512 * 1024);
+  });
+
+  it("rejects a negative payloadSize", () => {
+    const result = TriggerTaskRequestBody.safeParse({
+      payload: { foo: "bar" },
+      context: {},
+      options: {
+        payloadSize: -1,
+      },
+    });
+
+    expect(result.success).toBe(false);
+  });
+});
+
+describe("BatchItemNDJSON", () => {
+  it("accepts an optional payloadSize on a batch item's options", () => {
+    const result = BatchItemNDJSON.safeParse({
+      index: 0,
+      task: "my-task",
+      payload: "packets/payloads/file.json",
+      options: {
+        payloadType: "application/store",
+        payloadSize: 256 * 1024,
+      },
+    });
+
+    expect(result.success).toBe(true);
+    expect(result.success && result.data.options?.payloadSize).toBe(256 * 1024);
+  });
 });

packages/core/src/v3/schemas/api.ts

@@ -238,6 +238,12 @@ export const TriggerTaskRequestBody = z
         metadata: z.any(),
         metadataType: z.string().optional(),
         payloadType: z.string().optional(),
+        /**
+         * Byte size of the (pre-offload) serialized payload, measured by the caller
+         * before any object-store offload. Lets the pipeline know how large a payload
+         * is without downloading an "application/store" reference.
+         */
+        payloadSize: z.number().int().nonnegative().optional(),
         tags: RunTags.optional(),
         test: z.boolean().optional(),
         ttl: z.string().or(z.number().nonnegative().int()).optional(),
@@ -311,6 +317,8 @@ export const BatchTriggerTaskItem = z.object({
       metadataType: z.string().optional(),
       parentAttempt: z.string().optional(),
       payloadType: z.string().optional(),
+      /** Byte size of the (pre-offload) serialized payload for this item. */
+      payloadSize: z.number().int().nonnegative().optional(),
       queue: z
         .object({
           name: z.string(),

packages/trigger-sdk/src/v3/shared.test.ts

@@ -1,5 +1,57 @@
+import { ApiClient } from "@trigger.dev/core/v3";
 import { describe, it, expect } from "vitest";
-import { readableStreamToAsyncIterable } from "./shared.js";
+import { offloadBatchItemPayloads, readableStreamToAsyncIterable } from "./shared.js";
+
+describe("offloadBatchItemPayloads", () => {
+  // A real client is required for conditionallyExportPacket's truthy check; small payloads
+  // short-circuit before any network call, so this never actually reaches the server.
+  const apiClient = new ApiClient("http://localhost:3030", "tr_dev_test");
+
+  it("returns an empty array unchanged", async () => {
+    expect(await offloadBatchItemPayloads([], apiClient)).toEqual([]);
+  });
+
+  it("passes small payloads through and records their pre-offload byte size", async () => {
+    const payload = JSON.stringify({ hello: "world" });
+    const result = await offloadBatchItemPayloads(
+      [{ index: 0, task: "my-task", payload, options: { payloadType: "application/json" } }],
+      apiClient
+    );
+    const item = result[0]!;
+
+    expect(item.payload).toBe(payload);
+    expect(item.options?.payloadType).toBe("application/json");
+    expect(item.options?.payloadSize).toBe(Buffer.byteLength(payload, "utf8"));
+  });
+
+  it("measures multi-byte payloads by byte length, not character count", async () => {
+    const payload = "€€€"; // 3 chars, 9 bytes in UTF-8
+    const result = await offloadBatchItemPayloads(
+      [{ index: 0, task: "my-task", payload, options: { payloadType: "application/json" } }],
+      apiClient
+    );
+
+    expect(result[0]!.options?.payloadSize).toBe(9);
+  });
+
+  it("leaves an already-offloaded (application/store) item untouched", async () => {
+    const item = {
+      index: 0,
+      task: "my-task",
+      payload: "trigger/my-task/123/payload.json",
+      options: { payloadType: "application/store" },
+    };
+
+    const result = await offloadBatchItemPayloads([item], apiClient);
+    expect(result[0]).toEqual(item);
+  });
+
+  it("leaves items without a string payload untouched", async () => {
+    const item = { index: 0, task: "my-task", options: {} };
+    const result = await offloadBatchItemPayloads([item], apiClient);
+    expect(result[0]).toEqual(item);
+  });
+});
 
 describe("readableStreamToAsyncIterable", () => {
   it("yields all values from the stream", async () => {

packages/trigger-sdk/src/v3/shared.ts

@@ -23,6 +23,7 @@ import {
   getSchemaParseFn,
   lifecycleHooks,
   makeIdempotencyKey,
+  packetRequiresOffloading,
   parsePacket,
   RateLimitError,
   resourceCatalog,
@@ -1655,6 +1656,21 @@ async function executeBatchTwoPhase(
   },
   requestOptions?: TriggerApiRequestOptions
 ): Promise<{ id: string; runCount: number; publicAccessToken: string; taskIdentifiers: string[] }> {
+  // Offload any oversized per-item payloads to object storage before streaming, so a batch
+  // of large items keeps the request body small — the same treatment single triggers get.
+  // Both the array and streaming batch paths funnel through here, so this is the one place
+  // batch offloading needs to happen. Wrap failures so a presign/upload error carries the
+  // same batch context (phase, itemCount, rate-limit info) as the create/stream phases.
+  try {
+    items = await offloadBatchItemPayloads(items, apiClient);
+  } catch (error) {
+    throw new BatchTriggerError(`Failed to offload payloads for batch with ${items.length} items`, {
+      cause: error,
+      phase: "offload",
+      itemCount: items.length,
+    });
+  }
+
   let batch: Awaited<ReturnType<typeof apiClient.createBatch>> | undefined;
 
   try {
@@ -1701,6 +1717,81 @@ async function executeBatchTwoPhase(
   };
 }
 
+/**
+ * Offload any oversized per-item payloads in a batch to object storage, mirroring the
+ * single-trigger offload path. Small payloads pass through untouched. Every returned item
+ * carries `options.payloadSize` (the pre-offload serialized byte size) so the pipeline can
+ * reason about payload size without downloading an "application/store" reference.
+ *
+ * Uploads run with bounded concurrency so a batch of large items doesn't fire an unbounded
+ * number of simultaneous presigned PUTs.
+ *
+ * @internal Exported for testing.
+ */
+export async function offloadBatchItemPayloads(
+  items: BatchItemNDJSON[],
+  apiClient: ApiClient,
+  concurrency = 10
+): Promise<BatchItemNDJSON[]> {
+  if (items.length === 0) {
+    return items;
+  }
+
+  const results = new Array<BatchItemNDJSON>(items.length);
+  let cursor = 0;
+
+  async function worker() {
+    while (cursor < items.length) {
+      const index = cursor++;
+      results[index] = await offloadBatchItemPayload(items[index]!, apiClient);
+    }
+  }
+
+  await Promise.all(Array.from({ length: Math.min(concurrency, items.length) }, () => worker()));
+
+  return results;
+}
+
+async function offloadBatchItemPayload(
+  item: BatchItemNDJSON,
+  apiClient: ApiClient
+): Promise<BatchItemNDJSON> {
+  // The batch builders serialize each payload to a string via stringifyIO before we get
+  // here; anything else has no inline body to offload or measure.
+  if (typeof item.payload !== "string" || item.payload.length === 0) {
+    return item;
+  }
+
+  const dataType = item.options?.payloadType ?? "application/json";
+
+  // Already an object-store reference — the caller pre-offloaded it, nothing to do.
+  if (dataType === "application/store") {
+    return item;
+  }
+
+  const packet: IOPacket = { data: item.payload, dataType };
+  // Measure before offload so the size reflects the real payload, not the small reference
+  // we may replace it with.
+  const { size: payloadSize } = packetRequiresOffloading(packet);
+
+  const exported = await conditionallyExportPacket(
+    packet,
+    createTriggerPayloadPathPrefix(item.task),
+    undefined,
+    apiClient
+  );
+
+  return {
+    ...item,
+    payload: exported.data,
+    options: {
+      ...item.options,
+      payloadType: exported.dataType,
+      payloadSize,
+    },
+  };
+}
+
 /**
  * Error thrown when batch trigger operations fail.
  * Includes context about which phase failed and the batch details.
@@ -1710,7 +1801,7 @@ async function executeBatchTwoPhase(
  * - `retryAfterMs`: milliseconds until the rate limit resets
  */
 export class BatchTriggerError extends Error {
-  readonly phase: "create" | "stream";
+  readonly phase: "offload" | "create" | "stream";
   readonly batchId?: string;
   readonly itemCount: number;
 
@@ -1730,7 +1821,7 @@ export class BatchTriggerError extends Error {
     message: string,
     options: {
       cause?: unknown;
-      phase: "create" | "stream";
+      phase: "offload" | "create" | "stream";
       batchId?: string;
       itemCount: number;
     }
@@ -2205,7 +2296,11 @@ async function trigger_internal<TRunTypes extends AnyRunTypes>(
   const apiClient = apiClientManager.clientOrThrow(requestOptions?.clientConfig);
 
   const parsedPayload = parsePayload ? await parsePayload(payload) : payload;
-  const triggerPayloadPacket = await prepareTriggerPayload(parsedPayload, apiClient, id);
+  const { packet: triggerPayloadPacket, payloadSize } = await prepareTriggerPayload(
+    parsedPayload,
+    apiClient,
+    id
+  );
 
   // Process idempotency key and extract options for storage
   const processedIdempotencyKey = await makeIdempotencyKey(options?.idempotencyKey);
@@ -2222,6 +2317,7 @@ async function trigger_internal<TRunTypes extends AnyRunTypes>(
         concurrencyKey: options?.concurrencyKey,
         test: taskContext.ctx?.run.isTest,
         payloadType: triggerPayloadPacket.dataType,
+        payloadSize,
         idempotencyKey: processedIdempotencyKey?.toString(),
         idempotencyKeyTTL: options?.idempotencyKeyTTL,
         idempotencyKeyOptions,
@@ -2460,7 +2556,11 @@ async function triggerAndWait_internal<TIdentifier extends string, TPayload, TOu
   const apiClient = apiClientManager.clientOrThrow(requestOptions?.clientConfig);
 
   const parsedPayload = parsePayload ? await parsePayload(payload) : payload;
-  const triggerPayloadPacket = await prepareTriggerPayload(parsedPayload, apiClient, id);
+  const { packet: triggerPayloadPacket, payloadSize } = await prepareTriggerPayload(
+    parsedPayload,
+    apiClient,
+    id
+  );
 
   // Process idempotency key and extract options for storage
   const processedIdempotencyKey = await makeIdempotencyKey(options?.idempotencyKey);
@@ -2481,6 +2581,7 @@ async function triggerAndWait_internal<TIdentifier extends string, TPayload, TOu
             concurrencyKey: options?.concurrencyKey,
             test: taskContext.ctx?.run.isTest,
             payloadType: triggerPayloadPacket.dataType,
+            payloadSize,
             delay: options?.delay,
             ttl: options?.ttl,
             tags: options?.tags,
@@ -2546,7 +2647,11 @@ async function triggerAndSubscribe_internal<TIdentifier extends string, TPayload
   const apiClient = apiClientManager.clientOrThrow(requestOptions?.clientConfig);
 
   const parsedPayload = parsePayload ? await parsePayload(payload) : payload;
-  const triggerPayloadPacket = await prepareTriggerPayload(parsedPayload, apiClient, id);
+  const { packet: triggerPayloadPacket, payloadSize } = await prepareTriggerPayload(
+    parsedPayload,
+    apiClient,
+    id
+  );
 
   const processedIdempotencyKey = await makeIdempotencyKey(options?.idempotencyKey);
   const idempotencyKeyOptions = processedIdempotencyKey
@@ -2566,6 +2671,7 @@ async function triggerAndSubscribe_internal<TIdentifier extends string, TPayload
             concurrencyKey: options?.concurrencyKey,
             test: taskContext.ctx?.run.isTest,
             payloadType: triggerPayloadPacket.dataType,
+            payloadSize,
             delay: options?.delay,
             ttl: options?.ttl,
             tags: options?.tags,
@@ -3070,14 +3176,18 @@ async function prepareTriggerPayload(
   payload: unknown,
   apiClient: ApiClient,
   taskId: string
-): Promise<IOPacket> {
+): Promise<{ packet: IOPacket; payloadSize: number }> {
   const payloadPacket = await stringifyIO(payload);
-  return await conditionallyExportPacket(
+  // Measure the serialized size before any offload, so it reflects the real payload
+  // size rather than the small "application/store" reference we may send instead.
+  const { size: payloadSize } = packetRequiresOffloading(payloadPacket);
+  const packet = await conditionallyExportPacket(
     payloadPacket,
     createTriggerPayloadPathPrefix(taskId),
     undefined,
     apiClient
   );
+  return { packet, payloadSize };
 }
 
 function createTriggerPayloadPathPrefix(taskId: string): string {