From 8003923598e8a3112a167a1156bab3c50ca2c363 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 27 Feb 2026 10:11:42 +0000 Subject: [PATCH] feat(server): Gracefully handle oversized batch items instead of aborting the stream (#3137) Gracefully handle oversized batch items instead of aborting the stream. When an NDJSON batch item exceeds the maximum size, the parser now emits an error marker instead of throwing, allowing the batch to seal normally. The oversized item becomes a pre-failed run with `PAYLOAD_TOO_LARGE` error code, while other items in the batch process successfully. This prevents `batchTriggerAndWait` from seeing connection errors and retrying with exponential backoff. Also fixes the NDJSON parser not consuming the remainder of an oversized line split across multiple chunks, which caused "Invalid JSON" errors on subsequent lines. --- .changeset/modern-boxes-watch.md | 5 + .../graceful-oversized-batch-items.md | 10 + ailogger-output.log | 0 .../app/presenters/v3/SpanPresenter.server.ts | 3 +- .../routes/api.v3.batches.$batchId.items.ts | 7 +- .../route.tsx | 104 +++---- .../services/streamBatchItems.server.ts | 263 +++++++++++++++++- .../webapp/app/v3/runEngineHandlers.server.ts | 40 +++ .../test/engine/streamBatchItems.test.ts | 141 +++++++++- .../run-engine/src/batch-queue/index.ts | 12 +- .../run-engine/src/engine/errors.ts | 1 + packages/core/src/v3/errors.ts | 1 + packages/core/src/v3/schemas/common.ts | 1 + references/hello-world/src/trigger/batches.ts | 39 +++ 14 files changed, 541 insertions(+), 86 deletions(-) create mode 100644 .changeset/modern-boxes-watch.md create mode 100644 .server-changes/graceful-oversized-batch-items.md create mode 100644 ailogger-output.log diff --git a/.changeset/modern-boxes-watch.md b/.changeset/modern-boxes-watch.md new file mode 100644 index 000000000..e9539e210 --- /dev/null +++ b/.changeset/modern-boxes-watch.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Add PAYLOAD_TOO_LARGE error to handle graceful recovery of sending batch trigger items with payloads that exceed the maximum payload size diff --git a/.server-changes/graceful-oversized-batch-items.md b/.server-changes/graceful-oversized-batch-items.md new file mode 100644 index 000000000..980dd33e5 --- /dev/null +++ b/.server-changes/graceful-oversized-batch-items.md @@ -0,0 +1,10 @@ +--- +area: webapp +type: fix +--- + +Gracefully handle oversized batch items instead of aborting the stream. + +When an NDJSON batch item exceeds the maximum size, the parser now emits an error marker instead of throwing, allowing the batch to seal normally. The oversized item becomes a pre-failed run with `PAYLOAD_TOO_LARGE` error code, while other items in the batch process successfully. This prevents `batchTriggerAndWait` from seeing connection errors and retrying with exponential backoff. + +Also fixes the NDJSON parser not consuming the remainder of an oversized line split across multiple chunks, which caused "Invalid JSON" errors on subsequent lines. diff --git a/ailogger-output.log b/ailogger-output.log new file mode 100644 index 000000000..e69de29bb diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts index 59e717f7c..a85d8b20d 100644 --- a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts @@ -509,8 +509,7 @@ export class SpanPresenter extends BasePresenter { taskIdentifier: true, spanId: true, createdAt: true, - number: true, - taskVersion: true, + status: true, }, where: { parentSpanId: spanId, diff --git a/apps/webapp/app/routes/api.v3.batches.$batchId.items.ts b/apps/webapp/app/routes/api.v3.batches.$batchId.items.ts index 8307f34af..b3ed1c224 100644 --- a/apps/webapp/app/routes/api.v3.batches.$batchId.items.ts +++ b/apps/webapp/app/routes/api.v3.batches.$batchId.items.ts @@ -99,11 +99,8 @@ export async function action({ request, params }: ActionFunctionArgs) { if (error instanceof ServiceValidationError) { return json({ error: error.message }, { status: 422 }); } else if (error instanceof Error) { - // Check for stream parsing errors - if ( - error.message.includes("Invalid JSON") || - error.message.includes("exceeds maximum size") - ) { + // Check for stream parsing errors (e.g. invalid JSON) + if (error.message.includes("Invalid JSON")) { return json({ error: error.message }, { status: 400 }); } diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx index ae8bdaa70..a78c95d60 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx @@ -58,6 +58,7 @@ import { RunTimeline, RunTimelineEvent, SpanTimeline } from "~/components/run/Ru import { PacketDisplay } from "~/components/runs/v3/PacketDisplay"; import { RunIcon } from "~/components/runs/v3/RunIcon"; import { RunTag } from "~/components/runs/v3/RunTag"; +import { TruncatedCopyableValue } from "~/components/primitives/TruncatedCopyableValue"; import { SpanEvents } from "~/components/runs/v3/SpanEvents"; import { SpanTitle } from "~/components/runs/v3/SpanTitle"; import { TaskRunAttemptStatusCombo } from "~/components/runs/v3/TaskRunAttemptStatus"; @@ -133,9 +134,10 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { name: error.name, message: error.message, stack: error.stack, - cause: error.cause instanceof Error - ? { name: error.cause.name, message: error.cause.message } - : error.cause, + cause: + error.cause instanceof Error + ? { name: error.cause.name, message: error.cause.message } + : error.cause, } : error, }); @@ -1003,7 +1005,7 @@ function RunBody({ )} -
+
{run.friendlyId !== runParam && ( Message {span.message} - {span.triggeredRuns.length > 0 && ( - -
- Triggered runs - - - - Run # - Task - Version - Created at - - - - {span.triggeredRuns.map((run) => { - const path = v3RunSpanPath( - organization, - project, - environment, - { friendlyId: run.friendlyId }, - { spanId: run.spanId } - ); - return ( - - - {run.number} - - - {run.taskIdentifier} - - - {run.taskVersion ?? "–"} - - - - - - ); - })} - -
-
- - )} {span.events.length > 0 && } {span.properties !== undefined ? ( @@ -1268,6 +1228,48 @@ function SpanEntity({ span }: { span: Span }) { showOpenInModal /> ) : null} + {span.triggeredRuns.length > 0 && ( +
+ Runs + + + + ID + Task + Status + Created + + + + {span.triggeredRuns.map((run) => { + const path = v3RunSpanPath( + organization, + project, + environment, + { friendlyId: run.friendlyId }, + { spanId: run.spanId } + ); + return ( + + + + + + {run.taskIdentifier} + + + + + + + + + ); + })} + +
+
+ )}
); } diff --git a/apps/webapp/app/runEngine/services/streamBatchItems.server.ts b/apps/webapp/app/runEngine/services/streamBatchItems.server.ts index 8206760f4..859dfe2e6 100644 --- a/apps/webapp/app/runEngine/services/streamBatchItems.server.ts +++ b/apps/webapp/app/runEngine/services/streamBatchItems.server.ts @@ -14,6 +14,14 @@ export type StreamBatchItemsServiceOptions = { maxItemBytes: number; }; +export type OversizedItemMarker = { + __batchItemError: "OVERSIZED"; + index: number; + task: string; + actualSize: number; + maxSize: number; +}; + export type StreamBatchItemsServiceConstructorOptions = { prisma?: PrismaClientOrTransaction; engine?: RunEngine; @@ -110,6 +118,41 @@ export class StreamBatchItemsService extends WithRunEngine { // Process items from the stream for await (const rawItem of itemsIterator) { + // Check for oversized item markers from the NDJSON parser + if (rawItem && typeof rawItem === "object" && "__batchItemError" in rawItem) { + const marker = rawItem as OversizedItemMarker; + const itemIndex = marker.index >= 0 ? marker.index : lastIndex + 1; + + const errorMessage = `Batch item payload is too large (${(marker.actualSize / 1024).toFixed(1)} KB). Maximum allowed size is ${(marker.maxSize / 1024).toFixed(1)} KB. Reduce the payload size or offload large data to external storage.`; + + // Enqueue with __error metadata - processItemCallback will detect this + // and use TriggerFailedTaskService to create a pre-failed run + const batchItem: BatchItem = { + task: marker.task, + payload: "{}", + payloadType: "application/json", + options: { + __error: errorMessage, + __errorCode: "PAYLOAD_TOO_LARGE", + }, + }; + + const result = await this._engine.enqueueBatchItem( + batchId, + environment.id, + itemIndex, + batchItem + ); + + if (result.enqueued) { + itemsAccepted++; + } else { + itemsDeduplicated++; + } + lastIndex = itemIndex; + continue; + } + // Parse and validate the item const parseResult = BatchItemNDJSONSchema.safeParse(rawItem); if (!parseResult.success) { @@ -168,6 +211,34 @@ export class StreamBatchItemsService extends WithRunEngine { // Validate we received the expected number of items if (enqueuedCount !== batch.runCount) { + // The batch queue consumers may have already processed all items and + // cleaned up the Redis keys before we got here (especially likely when + // items include pre-failed runs that complete instantly). Check if the + // batch was already sealed/completed in Postgres. + const currentBatch = await this._prisma.batchTaskRun.findUnique({ + where: { id: batchId }, + select: { sealed: true, status: true }, + }); + + if (currentBatch?.sealed) { + logger.info("Batch already sealed before count check (fast completion)", { + batchId: batchFriendlyId, + itemsAccepted, + itemsDeduplicated, + enqueuedCount, + expectedCount: batch.runCount, + batchStatus: currentBatch.status, + }); + + return { + id: batchFriendlyId, + itemsAccepted, + itemsDeduplicated, + sealed: true, + runCount: batch.runCount, + }; + } + logger.warn("Batch item count mismatch", { batchId: batchFriendlyId, expected: batch.runCount, @@ -281,6 +352,121 @@ export class StreamBatchItemsService extends WithRunEngine { } } +/** + * Extract `index` and `task` from raw JSON bytes without decoding the full line. + * Scans at most 512 bytes, tracking JSON nesting depth to only match top-level keys. + */ +export function extractIndexAndTask(bytes: Uint8Array): { index: number; task: string } { + let index = -1; + let task = "unknown"; + let depth = 0; + let foundIndex = false; + let foundTask = false; + const limit = Math.min(bytes.byteLength, 512); + + const QUOTE = 0x22; // " + const COLON = 0x3a; // : + const LBRACE = 0x7b; // { + const RBRACE = 0x7d; // } + const LBRACKET = 0x5b; // [ + const RBRACKET = 0x5d; // ] + const BACKSLASH = 0x5c; // \ + + // Byte patterns for "index" and "task" (without quotes) + const INDEX_BYTES = [0x69, 0x6e, 0x64, 0x65, 0x78]; // index + const TASK_BYTES = [0x74, 0x61, 0x73, 0x6b]; // task + + let i = 0; + while (i < limit && !(foundIndex && foundTask)) { + const b = bytes[i]; + + if (b === LBRACE || b === LBRACKET) { + depth++; + i++; + continue; + } + if (b === RBRACE || b === RBRACKET) { + depth--; + i++; + continue; + } + + // Only match keys at depth 1 (top-level object) + if (b === QUOTE && depth === 1) { + // Read the key inside quotes + const keyStart = i + 1; + let keyEnd = keyStart; + while (keyEnd < limit && bytes[keyEnd] !== QUOTE) { + if (bytes[keyEnd] === BACKSLASH) keyEnd++; // skip escaped char + keyEnd++; + } + + const keyLen = keyEnd - keyStart; + + // Check if this key matches "index" or "task" + const isIndex = + !foundIndex && + keyLen === INDEX_BYTES.length && + INDEX_BYTES.every((b, j) => bytes[keyStart + j] === b); + const isTask = + !foundTask && + keyLen === TASK_BYTES.length && + TASK_BYTES.every((b, j) => bytes[keyStart + j] === b); + + if (isIndex || isTask) { + // Skip past closing quote and find colon + let pos = keyEnd + 1; + while (pos < limit && bytes[pos] !== COLON) pos++; + pos++; // skip colon + // Skip whitespace + while (pos < limit && (bytes[pos] === 0x20 || bytes[pos] === 0x09)) pos++; + + if (isIndex) { + // Parse digits + let num = 0; + let hasDigit = false; + while (pos < limit && bytes[pos] >= 0x30 && bytes[pos] <= 0x39) { + num = num * 10 + (bytes[pos] - 0x30); + hasDigit = true; + pos++; + } + if (hasDigit) { + index = num; + foundIndex = true; + } + } else { + // Parse quoted string value + if (pos < limit && bytes[pos] === QUOTE) { + const valStart = pos + 1; + let valEnd = valStart; + while (valEnd < limit && bytes[valEnd] !== QUOTE) { + if (bytes[valEnd] === BACKSLASH) valEnd++; + valEnd++; + } + // Decode just this slice + try { + task = new TextDecoder("utf-8", { fatal: true }).decode( + bytes.slice(valStart, valEnd) + ); + foundTask = true; + } catch { + // Leave as "unknown" + } + } + } + } + + // Skip past the key's closing quote + i = keyEnd + 1; + continue; + } + + i++; + } + + return { index, task }; +} + /** * Create an NDJSON parser transform stream. * @@ -305,6 +491,9 @@ export function createNdjsonParserStream( let chunks: Uint8Array[] = []; let totalBytes = 0; let lineNumber = 0; + // When an oversized incomplete line is detected (Case 2), we must discard + // all remaining bytes of that line until the next newline delimiter. + let skipUntilNewline = false; const NEWLINE_BYTE = 0x0a; // '\n' @@ -398,6 +587,24 @@ export function createNdjsonParserStream( return new TransformStream({ transform(chunk, controller) { + // If we're skipping the remainder of an oversized line, scan for the + // next newline in this chunk and discard everything before it. + if (skipUntilNewline) { + const nlPos = chunk.indexOf(NEWLINE_BYTE); + if (nlPos === -1) { + // Entire chunk is still part of the oversized line — discard it + return; + } + // Found the newline — keep everything after it + skipUntilNewline = false; + const remaining = chunk.slice(nlPos + 1); + if (remaining.byteLength === 0) { + return; + } + // Replace chunk with the remainder and fall through to normal processing + chunk = remaining; + } + // Append chunk to buffer chunks.push(chunk); totalBytes += chunk.byteLength; @@ -407,11 +614,19 @@ export function createNdjsonParserStream( while ((newlineIndex = findNewlineIndex()) !== -1) { // Check size limit BEFORE extracting/decoding (bytes up to newline) if (newlineIndex > maxItemBytes) { - throw new Error( - `Item at line ${ - lineNumber + 1 - } exceeds maximum size of ${maxItemBytes} bytes (actual: ${newlineIndex})` - ); + // Case 1: Complete line exceeds limit - emit marker instead of throwing + const lineBytes = extractLine(newlineIndex); + const extracted = extractIndexAndTask(lineBytes); + const marker: OversizedItemMarker = { + __batchItemError: "OVERSIZED", + index: extracted.index, + task: extracted.task, + actualSize: newlineIndex, + maxSize: maxItemBytes, + }; + controller.enqueue(marker); + lineNumber++; + continue; } const lineBytes = extractLine(newlineIndex); @@ -421,11 +636,23 @@ export function createNdjsonParserStream( // Check if the remaining buffer (incomplete line) exceeds the limit // This prevents OOM from a single huge line without newlines if (totalBytes > maxItemBytes) { - throw new Error( - `Item at line ${ - lineNumber + 1 - } exceeds maximum size of ${maxItemBytes} bytes (buffered: ${totalBytes}, no newline found)` - ); + // Case 2: Incomplete line exceeds limit - emit marker instead of throwing + const extracted = extractIndexAndTask(concatenateChunks()); + const marker: OversizedItemMarker = { + __batchItemError: "OVERSIZED", + index: extracted.index, + task: extracted.task, + actualSize: totalBytes, + maxSize: maxItemBytes, + }; + controller.enqueue(marker); + lineNumber++; + // Clear buffer and skip remaining bytes of this oversized line + // until the next newline delimiter is found in a subsequent chunk + chunks = []; + totalBytes = 0; + skipUntilNewline = true; + return; } }, @@ -441,11 +668,17 @@ export function createNdjsonParserStream( // Check size limit before processing final line if (totalBytes > maxItemBytes) { - throw new Error( - `Item at line ${ - lineNumber + 1 - } exceeds maximum size of ${maxItemBytes} bytes (actual: ${totalBytes})` - ); + // Case 3: Flush with oversized remaining - emit marker instead of throwing + const extracted = extractIndexAndTask(concatenateChunks()); + const marker: OversizedItemMarker = { + __batchItemError: "OVERSIZED", + index: extracted.index, + task: extracted.task, + actualSize: totalBytes, + maxSize: maxItemBytes, + }; + controller.enqueue(marker); + return; } const finalBytes = concatenateChunks(); diff --git a/apps/webapp/app/v3/runEngineHandlers.server.ts b/apps/webapp/app/v3/runEngineHandlers.server.ts index f0cf449d3..46fe5eaa7 100644 --- a/apps/webapp/app/v3/runEngineHandlers.server.ts +++ b/apps/webapp/app/v3/runEngineHandlers.server.ts @@ -669,6 +669,46 @@ export function setupBatchQueueCallbacks() { engine, }); + // Check for pre-marked error items (e.g. oversized payloads) + const itemError = item.options?.__error as string | undefined; + if (itemError) { + const errorCode = (item.options?.__errorCode as string) ?? "ITEM_ERROR"; + + let environment: AuthenticatedEnvironment | undefined; + try { + environment = (await findEnvironmentById(meta.environmentId)) ?? undefined; + } catch { + // Best-effort environment lookup + } + + if (environment) { + const failedRunId = await triggerFailedTaskService.call({ + taskId: item.task, + environment, + payload: item.payload ?? "{}", + payloadType: item.payloadType as string, + errorMessage: itemError, + errorCode: errorCode as TaskRunErrorCodes, + parentRunId: meta.parentRunId, + resumeParentOnCompletion: meta.resumeParentOnCompletion, + batch: { id: batchId, index: itemIndex }, + traceContext: meta.traceContext as Record | undefined, + spanParentAsLink: meta.spanParentAsLink, + }); + + if (failedRunId) { + span.setAttribute("batch.result.pre_failed", true); + span.setAttribute("batch.result.run_id", failedRunId); + span.end(); + return { success: true as const, runId: failedRunId }; + } + } + + // Fallback if TriggerFailedTaskService or environment lookup fails + span.end(); + return { success: false as const, error: itemError, errorCode }; + } + let environment: AuthenticatedEnvironment | undefined; try { environment = (await findEnvironmentById(meta.environmentId)) ?? undefined; diff --git a/apps/webapp/test/engine/streamBatchItems.test.ts b/apps/webapp/test/engine/streamBatchItems.test.ts index 9e3b3aafe..2dee86687 100644 --- a/apps/webapp/test/engine/streamBatchItems.test.ts +++ b/apps/webapp/test/engine/streamBatchItems.test.ts @@ -24,6 +24,8 @@ import { StreamBatchItemsService, createNdjsonParserStream, streamToAsyncIterable, + extractIndexAndTask, + type OversizedItemMarker, } from "../../app/runEngine/services/streamBatchItems.server"; import { ServiceValidationError } from "../../app/v3/services/baseService.server"; @@ -641,6 +643,25 @@ describe("createNdjsonParserStream", () => { expect(results).toEqual([{ id: 1 }, { id: 2 }]); }); + it("should handle escaped newlines in JSON string values", async () => { + // JSON.stringify escapes newlines as \n (two chars: backslash + n), + // so they don't break NDJSON line boundaries. This is the normal case + // when the SDK serializes payloads containing newlines. + const item1 = JSON.stringify({ payload: "line1\nline2\nline3" }); + const item2 = JSON.stringify({ payload: "no newlines" }); + const ndjson = item1 + "\n" + item2 + "\n"; + const encoder = new TextEncoder(); + const stream = chunksToStream([encoder.encode(ndjson)]); + + const parser = createNdjsonParserStream(1024); + const results = await collectStream(stream.pipeThrough(parser)); + + expect(results).toEqual([ + { payload: "line1\nline2\nline3" }, + { payload: "no newlines" }, + ]); + }); + it("should skip empty lines", async () => { const ndjson = '{"a":1}\n\n{"b":2}\n \n{"c":3}\n'; const encoder = new TextEncoder(); @@ -705,33 +726,76 @@ describe("createNdjsonParserStream", () => { expect(results).toEqual([{ greeting: "こんにちは" }]); }); - it("should reject lines exceeding maxItemBytes", async () => { + it("should emit OversizedItemMarker for lines exceeding maxItemBytes", async () => { const maxBytes = 50; - // Create a line that exceeds the limit - const largeJson = JSON.stringify({ data: "x".repeat(100) }) + "\n"; + // Create a line that exceeds the limit with index and task fields + const largeJson = JSON.stringify({ index: 3, task: "my-task", data: "x".repeat(100) }) + "\n"; const encoder = new TextEncoder(); const stream = chunksToStream([encoder.encode(largeJson)]); const parser = createNdjsonParserStream(maxBytes); + const results = await collectStream(stream.pipeThrough(parser)); - await expect(collectStream(stream.pipeThrough(parser))).rejects.toThrow(/exceeds maximum size/); + expect(results).toHaveLength(1); + const marker = results[0] as OversizedItemMarker; + expect(marker.__batchItemError).toBe("OVERSIZED"); + expect(marker.index).toBe(3); + expect(marker.task).toBe("my-task"); + expect(marker.maxSize).toBe(maxBytes); + expect(marker.actualSize).toBeGreaterThan(maxBytes); }); - it("should reject unbounded accumulation without newlines", async () => { + it("should emit OversizedItemMarker for unbounded accumulation without newlines", async () => { const maxBytes = 50; // Send data without any newlines that exceeds the buffer limit const encoder = new TextEncoder(); const chunks = [ - encoder.encode('{"start":"'), + encoder.encode('{"index":7,"task":"big-task","start":"'), encoder.encode("x".repeat(60)), // This will push buffer over 50 bytes ]; const stream = chunksToStream(chunks); const parser = createNdjsonParserStream(maxBytes); + const results = await collectStream(stream.pipeThrough(parser)); - await expect(collectStream(stream.pipeThrough(parser))).rejects.toThrow( - /exceeds maximum size.*no newline found/ - ); + expect(results).toHaveLength(1); + const marker = results[0] as OversizedItemMarker; + expect(marker.__batchItemError).toBe("OVERSIZED"); + expect(marker.index).toBe(7); + expect(marker.task).toBe("big-task"); + expect(marker.maxSize).toBe(maxBytes); + }); + + it("should skip remaining bytes of oversized line arriving in subsequent chunks", async () => { + const maxBytes = 50; + const encoder = new TextEncoder(); + // Simulate a normal item, then an oversized item split across many chunks, + // then another normal item after the newline. + // The oversized line is: {"index":1,"task":"t","data":"xxxx...120 x's...xxxx"}\n + const normalItem1 = '{"index":0,"task":"t","x":1}\n'; + const oversizedStart = '{"index":1,"task":"t","data":"'; + const oversizedMiddle = "x".repeat(120); // way over 50 bytes + const oversizedEnd = '"}\n'; + const normalItem2 = '{"index":2,"task":"t","x":2}\n'; + + // Send as separate chunks to trigger Case 2 (no newline, buffer > limit) + const chunks = [ + encoder.encode(normalItem1 + oversizedStart), + encoder.encode(oversizedMiddle.slice(0, 60)), + encoder.encode(oversizedMiddle.slice(60)), + encoder.encode(oversizedEnd + normalItem2), + ]; + const stream = chunksToStream(chunks); + + const parser = createNdjsonParserStream(maxBytes); + const results = await collectStream(stream.pipeThrough(parser)); + + // Should get: normal item 1, oversized marker, normal item 2 + expect(results).toHaveLength(3); + expect(results[0]).toEqual({ index: 0, task: "t", x: 1 }); + expect((results[1] as OversizedItemMarker).__batchItemError).toBe("OVERSIZED"); + expect((results[1] as OversizedItemMarker).index).toBe(1); + expect(results[2]).toEqual({ index: 2, task: "t", x: 2 }); }); it("should check byte size before decoding to prevent OOM", async () => { @@ -756,10 +820,12 @@ describe("createNdjsonParserStream", () => { const results1 = await collectStream(stream1.pipeThrough(parser1)); expect(results1).toHaveLength(1); - // Large one should fail + // Large one should emit an OversizedItemMarker const stream2 = chunksToStream([largeBytes]); const parser2 = createNdjsonParserStream(maxBytes); - await expect(collectStream(stream2.pipeThrough(parser2))).rejects.toThrow(/exceeds maximum/); + const results2 = await collectStream(stream2.pipeThrough(parser2)); + expect(results2).toHaveLength(1); + expect((results2[0] as OversizedItemMarker).__batchItemError).toBe("OVERSIZED"); }); it("should handle final line in flush without trailing newline", async () => { @@ -837,6 +903,28 @@ describe("createNdjsonParserStream", () => { expect(results).toEqual([]); }); + it("should pass normal items and emit markers for oversized items in the same stream", async () => { + const maxBytes = 50; + const encoder = new TextEncoder(); + // Normal item, then oversized item, then another normal item + const normalItem1 = '{"index":0,"task":"t","x":1}\n'; + const oversizedItem = JSON.stringify({ index: 1, task: "t", data: "x".repeat(100) }) + "\n"; + const normalItem2 = '{"index":2,"task":"t","x":2}\n'; + const stream = chunksToStream([encoder.encode(normalItem1 + oversizedItem + normalItem2)]); + + const parser = createNdjsonParserStream(maxBytes); + const results = await collectStream(stream.pipeThrough(parser)); + + expect(results).toHaveLength(3); + // First: normal parsed object + expect(results[0]).toEqual({ index: 0, task: "t", x: 1 }); + // Second: oversized marker + expect((results[1] as OversizedItemMarker).__batchItemError).toBe("OVERSIZED"); + expect((results[1] as OversizedItemMarker).index).toBe(1); + // Third: normal parsed object + expect(results[2]).toEqual({ index: 2, task: "t", x: 2 }); + }); + it("should handle stream with only whitespace", async () => { const encoder = new TextEncoder(); const stream = chunksToStream([encoder.encode(" \n\n \n")]); @@ -847,3 +935,34 @@ describe("createNdjsonParserStream", () => { expect(results).toEqual([]); }); }); + +describe("extractIndexAndTask", () => { + const encoder = new TextEncoder(); + + it("should extract index and task from JSON bytes", () => { + const bytes = encoder.encode('{"index":42,"task":"my-task","data":"x"}'); + const result = extractIndexAndTask(bytes); + expect(result.index).toBe(42); + expect(result.task).toBe("my-task"); + }); + + it("should return defaults for empty or malformed bytes", () => { + const result = extractIndexAndTask(new Uint8Array(0)); + expect(result.index).toBe(-1); + expect(result.task).toBe("unknown"); + }); + + it("should handle keys in any order", () => { + const bytes = encoder.encode('{"task":"other-task","data":"y","index":99}'); + const result = extractIndexAndTask(bytes); + expect(result.index).toBe(99); + expect(result.task).toBe("other-task"); + }); + + it("should not match nested keys", () => { + const bytes = encoder.encode('{"nested":{"index":999,"task":"inner"},"index":5,"task":"outer"}'); + const result = extractIndexAndTask(bytes); + expect(result.index).toBe(5); + expect(result.task).toBe("outer"); + }); +}); diff --git a/internal-packages/run-engine/src/batch-queue/index.ts b/internal-packages/run-engine/src/batch-queue/index.ts index 98bdacc05..312bf4772 100644 --- a/internal-packages/run-engine/src/batch-queue/index.ts +++ b/internal-packages/run-engine/src/batch-queue/index.ts @@ -848,7 +848,11 @@ export class BatchQueue { "BatchQueue.serializePayload", async (innerSpan) => { const str = - typeof item.payload === "string" ? item.payload : JSON.stringify(item.payload); + item.payload === undefined || item.payload === null + ? "{}" + : typeof item.payload === "string" + ? item.payload + : JSON.stringify(item.payload); innerSpan?.setAttribute("batch.payloadSize", str.length); return str; } @@ -912,7 +916,11 @@ export class BatchQueue { "BatchQueue.serializePayload", async (innerSpan) => { const str = - typeof item.payload === "string" ? item.payload : JSON.stringify(item.payload); + item.payload === undefined || item.payload === null + ? "{}" + : typeof item.payload === "string" + ? item.payload + : JSON.stringify(item.payload); innerSpan?.setAttribute("batch.payloadSize", str.length); return str; } diff --git a/internal-packages/run-engine/src/engine/errors.ts b/internal-packages/run-engine/src/engine/errors.ts index 373f9daa1..772282deb 100644 --- a/internal-packages/run-engine/src/engine/errors.ts +++ b/internal-packages/run-engine/src/engine/errors.ts @@ -61,6 +61,7 @@ export function runStatusFromError( case "TASK_PROCESS_SIGTERM": case "TASK_DID_CONCURRENT_WAIT": case "BATCH_ITEM_COULD_NOT_TRIGGER": + case "PAYLOAD_TOO_LARGE": case "UNSPECIFIED_ERROR": return "SYSTEM_FAILURE"; default: diff --git a/packages/core/src/v3/errors.ts b/packages/core/src/v3/errors.ts index 914832513..87fff767d 100644 --- a/packages/core/src/v3/errors.ts +++ b/packages/core/src/v3/errors.ts @@ -308,6 +308,7 @@ export function shouldRetryError(error: TaskRunError): boolean { case "TASK_HAS_N0_EXECUTION_SNAPSHOT": case "TASK_RUN_DEQUEUED_MAX_RETRIES": case "BATCH_ITEM_COULD_NOT_TRIGGER": + case "PAYLOAD_TOO_LARGE": case "UNSPECIFIED_ERROR": return false; diff --git a/packages/core/src/v3/schemas/common.ts b/packages/core/src/v3/schemas/common.ts index d489a5939..f37572083 100644 --- a/packages/core/src/v3/schemas/common.ts +++ b/packages/core/src/v3/schemas/common.ts @@ -188,6 +188,7 @@ export const TaskRunInternalError = z.object({ "TASK_DID_CONCURRENT_WAIT", "RECURSIVE_WAIT_DEADLOCK", "BATCH_ITEM_COULD_NOT_TRIGGER", + "PAYLOAD_TOO_LARGE", "UNSPECIFIED_ERROR", ]), message: z.string().optional(), diff --git a/references/hello-world/src/trigger/batches.ts b/references/hello-world/src/trigger/batches.ts index 6bbdf9461..b6a3f79e7 100644 --- a/references/hello-world/src/trigger/batches.ts +++ b/references/hello-world/src/trigger/batches.ts @@ -999,6 +999,45 @@ export const largePayloadTask = task({ }, }); +// ============================================================================ +// Oversized Payload Graceful Handling +// ============================================================================ + +/** + * Test: Batch with oversized item should complete gracefully + * + * Sends 2 items: one normal, one oversized (~3.2MB). + * The oversized item should result in a pre-failed run (ok: false) + * while the normal item processes successfully (ok: true). + */ +export const batchSealFailureOversizedPayload = task({ + id: "batch-seal-failure-oversized", + maxDuration: 60, + retry: { + maxAttempts: 1, + }, + run: async () => { + const results = await fixedLengthTask.batchTriggerAndWait([ + { payload: { waitSeconds: 1, output: "normal" } }, + { payload: { waitSeconds: 1, output: "x".repeat(3_200_000) } }, // ~3.2MB oversized + ]); + + const normal = results.runs[0]; + const oversized = results.runs[1]; + + logger.info("Batch results", { + normalOk: normal?.ok, + oversizedOk: oversized?.ok, + }); + + return { + normalOk: normal?.ok === true, + oversizedOk: oversized?.ok === false, + oversizedError: !oversized?.ok ? oversized?.error : undefined, + }; + }, +}); + type Payload = { waitSeconds: number; error?: string;