Wes Mason
76c37ecd24
## 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](https://github.com/triggerdotdev/trigger.dev/pull/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](https://github.com/triggerdotdev/trigger.dev/pull/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).
72 lines
2.5 KiB
TypeScript
72 lines
2.5 KiB
TypeScript
import type { IOPacket } from "@trigger.dev/core/v3";
|
|
import { packetRequiresOffloading, tryCatch } from "@trigger.dev/core/v3";
|
|
import type { PayloadProcessor, TriggerTaskRequest } from "../types";
|
|
import { env } from "~/env.server";
|
|
import { startActiveSpan } from "~/v3/tracer.server";
|
|
import { uploadPacketToObjectStore } from "~/v3/objectStore.server";
|
|
import { ServiceValidationError } from "~/v3/services/common.server";
|
|
|
|
export class DefaultPayloadProcessor implements PayloadProcessor {
|
|
async process(request: TriggerTaskRequest): Promise<IOPacket> {
|
|
return await startActiveSpan("handlePayloadPacket()", async (span) => {
|
|
const payload = request.body.payload;
|
|
const payloadType = request.body.options?.payloadType ?? "application/json";
|
|
|
|
const packet = this.#createPayloadPacket(payload, payloadType);
|
|
|
|
if (!packet.data) {
|
|
return packet;
|
|
}
|
|
|
|
const { needsOffloading, size } = packetRequiresOffloading(
|
|
packet,
|
|
env.TASK_PAYLOAD_OFFLOAD_THRESHOLD
|
|
);
|
|
|
|
span.setAttribute("needsOffloading", needsOffloading);
|
|
// 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;
|
|
}
|
|
|
|
const filename = `${request.friendlyId}/payload.json`;
|
|
|
|
const [uploadError, uploadedFilename] = await tryCatch(
|
|
uploadPacketToObjectStore(
|
|
filename,
|
|
packet.data,
|
|
packet.dataType,
|
|
request.environment,
|
|
env.OBJECT_STORE_DEFAULT_PROTOCOL
|
|
)
|
|
);
|
|
|
|
if (uploadError) {
|
|
throw new ServiceValidationError("Failed to upload large payload to object store", 500); // This is retryable
|
|
}
|
|
|
|
return {
|
|
data: uploadedFilename!,
|
|
dataType: "application/store",
|
|
};
|
|
});
|
|
}
|
|
|
|
#createPayloadPacket(payload: any, payloadType: string): IOPacket {
|
|
if (payloadType === "application/json") {
|
|
return { data: JSON.stringify(payload), dataType: "application/json" };
|
|
}
|
|
|
|
if (typeof payload === "string") {
|
|
return { data: payload, dataType: payloadType };
|
|
}
|
|
|
|
return { dataType: payloadType };
|
|
}
|
|
}
|