diff --git a/.changeset/large-batch-payloads-and-size.md b/.changeset/large-batch-payloads-and-size.md new file mode 100644 index 0000000000..b20224e286 --- /dev/null +++ b/.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. diff --git a/apps/webapp/app/runEngine/concerns/payloads.server.ts b/apps/webapp/app/runEngine/concerns/payloads.server.ts index 77176ed038..c364cbd1df 100644 --- a/apps/webapp/app/runEngine/concerns/payloads.server.ts +++ b/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; diff --git a/packages/core/src/v3/schemas/api-type.test.ts b/packages/core/src/v3/schemas/api-type.test.ts index 87f755206d..39fb3131ce 100644 --- a/packages/core/src/v3/schemas/api-type.test.ts +++ b/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); + }); }); diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index e9b5b03cc6..b81d75c750 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -237,6 +237,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(), @@ -310,6 +316,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(), diff --git a/packages/trigger-sdk/src/v3/shared.test.ts b/packages/trigger-sdk/src/v3/shared.test.ts index 16191e9826..3eb7e821b0 100644 --- a/packages/trigger-sdk/src/v3/shared.test.ts +++ b/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 () => { diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts index 203f83bc77..13d402be6a 100644 --- a/packages/trigger-sdk/src/v3/shared.ts +++ b/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> | 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 { + if (items.length === 0) { + return items; + } + + const results = new Array(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 { + // 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( 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( 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 { +): 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 {