diff --git a/.env.example b/.env.example index 963c0c58d1..9ab3c777ba 100644 --- a/.env.example +++ b/.env.example @@ -239,6 +239,26 @@ ALLOW_API_KEY_REVEAL=false # Default: 10485760 (10 MB) # MAX_BODY_SIZE_BYTES=10485760 +# Heap-pressure-aware admission for POST /v1/chat/completions (#5152). A large +# coding-agent "compact" body amplifies into hundreds of MB of transient JS objects +# on the combo path; concurrent compacts can stack past the V8 heap ceiling and OOM +# the process. These shed a LARGE body with 503 (Retry-After) only while the heap is +# already under pressure — healthy heap admits every body untouched. +# Used by: src/shared/middleware/chatBodyAdmission.ts +# Bodies below this size skip the guard entirely (heap not even sampled). Default 262144 (256 KB). +# OMNIROUTE_CHAT_LARGE_BODY_BYTES=262144 +# Hard cap — bodies above this are rejected with 413 before any clone/parse. Default 52428800 (50 MB). +# OMNIROUTE_CHAT_HARD_MAX_BODY_BYTES=52428800 +# Shed large bodies once heapUsed/heap_size_limit reaches this ratio (0 { + const parsed = Number.parseInt( + String(process.env.OMNIROUTE_MAX_NONSTREAMING_RESPONSE_BYTES), + 10 + ); + return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_MAX_NONSTREAMING_RESPONSE_BYTES; +})(); + export async function readNonStreamingResponseBody( response: Response, contentType: string, - upstreamStream: boolean + upstreamStream: boolean, + maxBytes: number = MAX_NONSTREAMING_RESPONSE_BYTES ): Promise { if ( !upstreamStream || !response.body || (!contentType.includes("text/event-stream") && !contentType.includes("application/x-ndjson")) ) { + // Reject before buffering when the upstream declares an over-cap Content-Length. + const declared = Number.parseInt(response.headers.get("content-length") ?? "", 10); + if (Number.isFinite(declared) && declared > maxBytes) { + throw new NonStreamingResponseTooLargeError(declared, maxBytes); + } return withBodyTimeout(response.text()); } - const reader = response.body.getReader(); + return drainNonStreamingSseBody(response.body, maxBytes); +} + +/** + * Drain an SSE/NDJSON stream consumed in non-streaming mode into a single string, bounded + * by `maxBytes` (cancels the upstream and throws {@link NonStreamingResponseTooLargeError} + * past the cap) and by the body timeout, cancelling early on a terminal SSE signal. + */ +type NonStreamingChunk = + | { kind: "done" } + | { kind: "skip" } + | { kind: "chunk"; value: Uint8Array }; + +/** Read the next chunk under the body-timeout deadline, normalizing end/empty cases. */ +async function readNextNonStreamingChunk( + reader: ReadableStreamDefaultReader, + deadline: number +): Promise { + const timeoutMs = deadline > 0 ? deadline - Date.now() : 0; + if (deadline > 0 && timeoutMs <= 0) { + throw createBodyTimeoutError(FETCH_BODY_TIMEOUT_MS); + } + const { done, value } = await readStreamChunkWithTimeout(reader, timeoutMs); + if (done) return { kind: "done" }; + if (!value) return { kind: "skip" }; + return { kind: "chunk", value }; +} + +async function drainNonStreamingSseBody( + body: ReadableStream, + maxBytes: number +): Promise { + const reader = body.getReader(); const decoder = new TextDecoder(); const terminalState: NonStreamingSseTerminalState = { currentEvent: "", pendingLine: "", }; let rawBody = ""; + let bytesSeen = 0; const deadline = FETCH_BODY_TIMEOUT_MS > 0 ? Date.now() + FETCH_BODY_TIMEOUT_MS : 0; try { while (true) { - const timeoutMs = deadline > 0 ? deadline - Date.now() : 0; - if (deadline > 0 && timeoutMs <= 0) { - throw createBodyTimeoutError(FETCH_BODY_TIMEOUT_MS); + const next = await readNextNonStreamingChunk(reader, deadline); + if (next.kind === "done") break; + if (next.kind === "skip") continue; + + // Bound the buffer: cancel the upstream and fail fast past the cap rather than + // growing `rawBody` until the V8 heap is exhausted. + bytesSeen += next.value.byteLength; + if (bytesSeen > maxBytes) { + await reader.cancel("non-streaming response exceeded byte cap").catch(() => {}); + throw new NonStreamingResponseTooLargeError(bytesSeen, maxBytes); } - const { done, value } = await readStreamChunkWithTimeout(reader, timeoutMs); - if (done) break; - if (!value) continue; - - const decodedChunk = decoder.decode(value, { stream: true }); + const decodedChunk = decoder.decode(next.value, { stream: true }); rawBody += decodedChunk; if (appendNonStreamingSseTerminalSignal(terminalState, decodedChunk)) { await reader.cancel("non-streaming bridge consumed terminal SSE event").catch(() => {}); diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 1cfdc5131b..e445fac026 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -1881,8 +1881,11 @@ export async function handleComboChat({ }); // Deep clone the body to ensure context preservation and prevent mutations - // from affecting other targets in the combo - let attemptBody = JSON.parse(JSON.stringify(body)); + // from affecting other targets in the combo. structuredClone avoids the + // full intermediate JSON string that JSON.parse(JSON.stringify(...)) builds + // (a second multi-hundred-KB allocation per target on large agent payloads), + // halving the per-target transient heap on the hot path (#5152). + let attemptBody = structuredClone(body); // Proactive Context Compression for fallbacks (Zero-Latency optimization) if ( diff --git a/src/app/api/v1/chat/completions/route.ts b/src/app/api/v1/chat/completions/route.ts index 27afc67da2..50becee9c9 100644 --- a/src/app/api/v1/chat/completions/route.ts +++ b/src/app/api/v1/chat/completions/route.ts @@ -3,6 +3,7 @@ import { callCloudWithMachineId } from "@/shared/utils/cloud"; import { handleChat } from "@/sse/handlers/chat"; import { initTranslators } from "@omniroute/open-sse/translator/index.ts"; import { createInjectionGuard } from "@/middleware/promptInjectionGuard"; +import { checkChatAdmission } from "@/shared/middleware/chatBodyAdmission"; let initPromise = null; @@ -31,6 +32,15 @@ export async function OPTIONS() { export async function POST(request) { await ensureInitialized(); + // Heap-pressure-aware admission: shed a large body with 503 (or 413 if pathological) + // BEFORE the request is cloned + JSON-parsed below. A large coding-agent compact body + // amplifies into hundreds of MB of transient JS objects on the combo path; under a + // burst of concurrent compacts that stacks past the V8 heap ceiling and OOM-crashes the + // whole process. Shedding the marginal request here turns a pod-wide crash into a single + // client retry. Healthy heap (the normal case) admits every body untouched. (#5152) + const admissionRejection = checkChatAdmission(request); + if (admissionRejection) return admissionRejection; + // One-line marker for diagnosing 413 / Server-Action interceptions. // Logs only when Content-Length is present so debug noise stays low for // typical chat payloads. Toggle off via OMNIROUTE_LOG_REQUEST_SHAPE=0. diff --git a/src/shared/middleware/chatBodyAdmission.ts b/src/shared/middleware/chatBodyAdmission.ts new file mode 100644 index 0000000000..f6cc6a34e3 --- /dev/null +++ b/src/shared/middleware/chatBodyAdmission.ts @@ -0,0 +1,161 @@ +/** + * Heap-pressure-aware admission guard for POST /v1/chat/completions. + * + * Root cause (homelab 3.8.40 OOM crash-loop): a forced-GC heap inspection of the live + * pod showed a HEALTHY ~350 MB live heap — there is no baseline leak. The crash is a + * per-request transient: a large coding-agent "compact" body (~750 KB) is cloned + + * JSON-parsed + fanned out across a round-robin combo, allocating hundreds of MB of JS + * objects; several concurrent compacts stack those transients past the V8 heap ceiling + * (`FATAL ERROR: Reached heap limit … heap out of memory`), which kills EVERY in-flight + * request and restarts the pod. + * + * A fixed body-size cap is the wrong tool — those large compacts are LEGITIMATE traffic. + * Instead this guard sheds a large body with a 503 (Retry-After) ONLY when the V8 heap is + * ALREADY under pressure, converting a process-wide OOM crash into a single graceful + * client retry. When the heap is healthy (the normal case) large bodies pass through + * unchanged, so it is invisible to ordinary traffic. A separate hard cap rejects only + * pathological (multi-MB) bodies before they are cloned/parsed. + * + * @module shared/middleware/chatBodyAdmission + */ + +import v8 from "node:v8"; +import { CORS_HEADERS } from "../utils/cors"; + +function parsePositiveInt(value: string | undefined, fallback: number): number { + const parsed = Number.parseInt(String(value), 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; +} + +function parseRatio(value: string | undefined, fallback: number): number { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed > 0 && parsed < 1 ? parsed : fallback; +} + +/** + * Bodies below this size cannot drive the transient amplification that causes the OOM, so + * they are always admitted and the heap is never even sampled for them (hot-path cheap). + * Matches the route's existing large-body log threshold (256 KB). + */ +export const CHAT_LARGE_BODY_BYTES = parsePositiveInt( + process.env.OMNIROUTE_CHAT_LARGE_BODY_BYTES, + 256 * 1024 +); + +/** Pathological bodies above this are rejected (413) before any clone/parse. Generous by + * default so real compacts are never rejected; only absurd payloads are. */ +export const CHAT_HARD_MAX_BODY_BYTES = parsePositiveInt( + process.env.OMNIROUTE_CHAT_HARD_MAX_BODY_BYTES, + 50 * 1024 * 1024 +); + +/** Shed large bodies once heapUsed/heap_size_limit reaches this fraction. 0.75 leaves + * headroom for the in-flight request to finish + GC. Healthy idle (~0.11) always admits. */ +export const CHAT_HEAP_SHED_RATIO = parseRatio(process.env.OMNIROUTE_CHAT_HEAP_SHED_RATIO, 0.75); + +export type ChatAdmissionDecision = + | { admit: true } + | { admit: false; status: 413 | 503; code: string; message: string }; + +/** + * Pure admission decision — no I/O, fully unit-testable. The route wrapper feeds it live + * Content-Length + heap figures. + */ +export function evaluateChatBodyAdmission(input: { + contentLength: number | null; + heapUsedBytes: number; + heapLimitBytes: number; + largeBodyBytes?: number; + hardMaxBytes?: number; + shedRatio?: number; +}): ChatAdmissionDecision { + const largeBodyBytes = input.largeBodyBytes ?? CHAT_LARGE_BODY_BYTES; + const hardMaxBytes = input.hardMaxBytes ?? CHAT_HARD_MAX_BODY_BYTES; + const shedRatio = input.shedRatio ?? CHAT_HEAP_SHED_RATIO; + const cl = input.contentLength; + + // Unknown or small bodies: always admit (cannot cause the crash). + if (cl === null || !Number.isFinite(cl) || cl < largeBodyBytes) return { admit: true }; + + // Pathological body: reject before cloning/parsing, independent of heap state. + if (cl > hardMaxBytes) { + return { + admit: false, + status: 413, + code: "PAYLOAD_TOO_LARGE", + message: `Request body too large for chat completions (max ${Math.floor( + hardMaxBytes / (1024 * 1024) + )} MB).`, + }; + } + + // Large but legitimate body: shed with 503 only when the heap is already under pressure, + // so a burst of concurrent compacts degrades to per-request retries, not a pod-wide OOM. + if (input.heapLimitBytes > 0 && input.heapUsedBytes / input.heapLimitBytes >= shedRatio) { + return { + admit: false, + status: 503, + code: "heap_pressure", + message: "Service temporarily under memory pressure. Retry shortly.", + }; + } + + return { admit: true }; +} + +/** Resolved once at module load (mirrors heapPressure.ts) — heap_size_limit is constant. */ +const HEAP_LIMIT_BYTES = v8.getHeapStatistics().heap_size_limit; + +/** + * Route-level guard. Returns a ready 413/503 Response when the request must be shed, or + * null to proceed. Only samples the heap for large bodies, so ordinary requests pay + * nothing. The heap figure is logged for INTERNAL telemetry only and never placed in the + * client response (Hard Rule #12). + * + * @param heapOverride test-only seam to inject heap figures; production omits it and the + * live heap is sampled lazily (never for small bodies). + */ +export function checkChatAdmission( + request: Request, + heapOverride?: { heapUsedBytes: number; heapLimitBytes: number } +): Response | null { + const clHeader = request.headers.get("content-length"); + const contentLength = clHeader ? Number.parseInt(clHeader, 10) : null; + + // Fast path: small/unknown bodies skip the heap sample entirely. + if ( + contentLength === null || + !Number.isFinite(contentLength) || + contentLength < CHAT_LARGE_BODY_BYTES + ) { + return null; + } + + const { + heapUsedBytes = process.memoryUsage().heapUsed, + heapLimitBytes = HEAP_LIMIT_BYTES, + } = heapOverride ?? {}; + const decision = evaluateChatBodyAdmission({ contentLength, heapUsedBytes, heapLimitBytes }); + + if (decision.admit) return null; + + const headers: Record = { + ...CORS_HEADERS, + "Content-Type": "application/json", + }; + if (decision.status === 503) { + headers["Retry-After"] = "2"; + console.warn( + `[chat-admission] shedding large body (${contentLength}B) under heap pressure: ` + + `heapUsed=${Math.round(heapUsedBytes / 1048576)}MB / limit=${Math.round( + heapLimitBytes / 1048576 + )}MB` + ); + } + + const type = decision.status === 413 ? "payload_too_large" : "server_error"; + return new Response( + JSON.stringify({ error: { message: decision.message, type, code: decision.code } }), + { status: decision.status, headers } + ); +} diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index a12a65fd21..a40a074377 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -224,8 +224,6 @@ export async function handleChat( return errorResponse(HTTP_STATUS.BAD_REQUEST, "Invalid JSON body"); } - const rawClientBody = cloneLogPayload(body); - // Early guard: an explicitly empty `messages` array is invalid for every // upstream (Anthropic/OpenAI both reject "at least one message is required"). // Forwarding it produced a confusing raw upstream 400/502; reject it here with @@ -240,9 +238,10 @@ export async function handleChat( return errorResponse(HTTP_STATUS.BAD_REQUEST, "messages: at least one message is required"); } - // Build clientRawRequest for logging (if not provided) + // buildClientRawRequest already deep-clones the body, so pass `body` directly — the + // prior local clone was a redundant second full-body copy on the hot path (#5152). if (!clientRawRequest) { - clientRawRequest = buildClientRawRequest(request, rawClientBody); + clientRawRequest = buildClientRawRequest(request, body); } // T01 — Accept-header streaming opt-in (#302 / #5305). A bare `Accept: diff --git a/tests/unit/chat-body-admission.test.ts b/tests/unit/chat-body-admission.test.ts new file mode 100644 index 0000000000..c2ef0b3d7b --- /dev/null +++ b/tests/unit/chat-body-admission.test.ts @@ -0,0 +1,136 @@ +// #5152: heap-pressure-aware admission for POST /v1/chat/completions. +// +// The homelab OOM crash-loop was a per-request transient explosion — a large coding-agent +// compact body cloned/parsed/fanned-out across a combo allocates hundreds of MB of JS +// objects, and concurrent compacts stack past the V8 heap ceiling, OOM-crashing the whole +// process. A fixed size cap is wrong (those bodies are legitimate); instead we shed a large +// body with 503 only when the heap is ALREADY under pressure, and 413 only for pathological +// bodies. These tests pin that policy and the cheap fast-path for ordinary traffic. +import test from "node:test"; +import assert from "node:assert/strict"; + +const { + evaluateChatBodyAdmission, + checkChatAdmission, + CHAT_LARGE_BODY_BYTES, +} = await import("../../src/shared/middleware/chatBodyAdmission.ts"); + +const MB = 1024 * 1024; +const HEAP_LIMIT = 3072 * MB; // mirror the homelab --max-old-space-size=3072 + +test("small body is always admitted, even under heap pressure", () => { + const decision = evaluateChatBodyAdmission({ + contentLength: 10 * 1024, + heapUsedBytes: 0.95 * HEAP_LIMIT, + heapLimitBytes: HEAP_LIMIT, + }); + assert.equal(decision.admit, true); +}); + +test("unknown content-length is admitted", () => { + const decision = evaluateChatBodyAdmission({ + contentLength: null, + heapUsedBytes: 0.95 * HEAP_LIMIT, + heapLimitBytes: HEAP_LIMIT, + }); + assert.equal(decision.admit, true); +}); + +test("large body on a HEALTHY heap is admitted (normal case — guard is invisible)", () => { + const decision = evaluateChatBodyAdmission({ + contentLength: 746578, // the production compact body + heapUsedBytes: 0.11 * HEAP_LIMIT, // ~350 MB live baseline + heapLimitBytes: HEAP_LIMIT, + }); + assert.equal(decision.admit, true); +}); + +test("large body under heap PRESSURE is shed with 503 + retry", () => { + const decision = evaluateChatBodyAdmission({ + contentLength: 746578, + heapUsedBytes: 0.8 * HEAP_LIMIT, + heapLimitBytes: HEAP_LIMIT, + }); + assert.equal(decision.admit, false); + assert.equal(decision.status, 503); + assert.equal(decision.code, "heap_pressure"); +}); + +test("pathological body is rejected with 413 regardless of heap state", () => { + const decision = evaluateChatBodyAdmission({ + contentLength: 200 * MB, + heapUsedBytes: 0.05 * HEAP_LIMIT, // heap totally healthy + heapLimitBytes: HEAP_LIMIT, + }); + assert.equal(decision.admit, false); + assert.equal(decision.status, 413); +}); + +test("shed threshold is exactly at the ratio boundary", () => { + const atBoundary = evaluateChatBodyAdmission({ + contentLength: CHAT_LARGE_BODY_BYTES, + heapUsedBytes: 0.75 * HEAP_LIMIT, + heapLimitBytes: HEAP_LIMIT, + shedRatio: 0.75, + }); + assert.equal(atBoundary.admit, false, "at the ratio it sheds"); + + const justBelow = evaluateChatBodyAdmission({ + contentLength: CHAT_LARGE_BODY_BYTES, + heapUsedBytes: 0.7499 * HEAP_LIMIT, + heapLimitBytes: HEAP_LIMIT, + shedRatio: 0.75, + }); + assert.equal(justBelow.admit, true, "just below the ratio it admits"); +}); + +test("checkChatAdmission returns null for a small-body request (fast path, no heap sample)", () => { + const request = new Request("http://x/v1/chat/completions", { + method: "POST", + headers: { "content-length": "1024" }, + }); + assert.equal(checkChatAdmission(request), null); +}); + +test("checkChatAdmission sheds a large body with 503 + Retry-After under injected heap pressure", async () => { + const request = new Request("http://x/v1/chat/completions", { + method: "POST", + headers: { "content-length": "746578" }, + }); + const res = checkChatAdmission(request, { + heapUsedBytes: 0.9 * HEAP_LIMIT, + heapLimitBytes: HEAP_LIMIT, + }); + assert.ok(res, "expected a 503 rejection"); + assert.equal(res.status, 503); + assert.equal(res.headers.get("Retry-After"), "2"); + const body = await res.json(); + assert.equal(body.error.code, "heap_pressure"); + assert.ok(!String(body.error.message).includes("at /"), "must not leak a stack trace"); +}); + +test("checkChatAdmission admits a large body when injected heap is healthy", () => { + const request = new Request("http://x/v1/chat/completions", { + method: "POST", + headers: { "content-length": "746578" }, + }); + const res = checkChatAdmission(request, { + heapUsedBytes: 0.1 * HEAP_LIMIT, + heapLimitBytes: HEAP_LIMIT, + }); + assert.equal(res, null); +}); + +test("checkChatAdmission 413 response does not leak a stack trace", async () => { + // Force a pathological size via the hard-cap env so we exercise the real wrapper. + const request = new Request("http://x/v1/chat/completions", { + method: "POST", + headers: { "content-length": String(500 * MB) }, + }); + const res = checkChatAdmission(request); + assert.ok(res, "expected a rejection Response"); + assert.equal(res.status, 413); + const body = await res.json(); + assert.equal(body.error.code, "PAYLOAD_TOO_LARGE"); + assert.ok(!String(body.error.message).includes("at /"), "must not leak a stack trace"); +}); diff --git a/tests/unit/chat-build-client-raw-request.test.ts b/tests/unit/chat-build-client-raw-request.test.ts new file mode 100644 index 0000000000..303854948f --- /dev/null +++ b/tests/unit/chat-build-client-raw-request.test.ts @@ -0,0 +1,40 @@ +// #5152: handleChat used to clone the body twice for logging — once into a local +// `rawClientBody` and again inside buildClientRawRequest — doubling per-request heap +// residency on the hot path (and cloning even when clientRawRequest was already provided). +// The outer clone was removed; buildClientRawRequest still owns the (single) deep clone. +// These tests pin that the logging snapshot remains an ISOLATED copy so dropping the outer +// clone cannot leak a shared reference that downstream mutation would corrupt. +import test from "node:test"; +import assert from "node:assert/strict"; +import { buildClientRawRequest } from "../../src/sse/handlers/chat.ts"; + +function req(body: unknown) { + return new Request("http://x/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); +} + +test("buildClientRawRequest deep-clones the body (not the same reference)", () => { + const body = { model: "m", messages: [{ role: "user", content: "hi" }] }; + const out = buildClientRawRequest(req(body), body); + assert.deepEqual(out.body, body); + assert.notEqual(out.body, body, "must be a distinct object"); + assert.notEqual(out.body.messages, body.messages, "nested arrays must be cloned too"); +}); + +test("mutating the original body after capture does not corrupt the snapshot", () => { + const body = { model: "m", messages: [{ role: "user", content: "original" }] }; + const out = buildClientRawRequest(req(body), body); + body.messages[0].content = "MUTATED"; + body.messages.push({ role: "user", content: "added" }); + assert.equal(out.body.messages.length, 1, "snapshot length is frozen at capture time"); + assert.equal(out.body.messages[0].content, "original", "snapshot content is isolated"); +}); + +test("endpoint and headers are captured from the request", () => { + const out = buildClientRawRequest(req({ model: "m" }), { model: "m" }); + assert.equal(out.endpoint, "/v1/chat/completions"); + assert.equal(out.headers["content-type"], "application/json"); +}); diff --git a/tests/unit/chatcore-non-streaming-response-body.test.ts b/tests/unit/chatcore-non-streaming-response-body.test.ts index 41d52a2e37..7e9807cd1d 100644 --- a/tests/unit/chatcore-non-streaming-response-body.test.ts +++ b/tests/unit/chatcore-non-streaming-response-body.test.ts @@ -5,7 +5,10 @@ // stream closes. import { test } from "node:test"; import assert from "node:assert/strict"; -import { readNonStreamingResponseBody } from "../../open-sse/handlers/chatCore/nonStreamingResponseBody.ts"; +import { + readNonStreamingResponseBody, + NonStreamingResponseTooLargeError, +} from "../../open-sse/handlers/chatCore/nonStreamingResponseBody.ts"; test("falls back to response.text() when upstream is not streaming", async () => { const out = await readNonStreamingResponseBody(new Response("hello"), "application/json", false); @@ -31,3 +34,49 @@ test("drains an SSE stream chunk-by-chunk and concatenates until close", async ( assert.ok(out.includes('"a":1')); assert.ok(out.includes('"b":2')); }); + +// #5152: bound the non-streaming buffer so a runaway upstream body cannot fill the V8 heap. + +test("aborts and throws when an SSE stream exceeds the byte cap (no unbounded string)", async () => { + const enc = new TextEncoder(); + let cancelled = false; + const body = new ReadableStream({ + pull(controller) { + // 1 KB chunks with no terminal signal — would grow forever without the cap. + controller.enqueue(enc.encode("data: " + "x".repeat(1024) + "\n")); + }, + cancel() { + cancelled = true; + }, + }); + const response = new Response(body, { headers: { "Content-Type": "text/event-stream" } }); + + await assert.rejects( + () => readNonStreamingResponseBody(response, "text/event-stream", true, 8 * 1024), + (err) => err instanceof NonStreamingResponseTooLargeError && err.maxBytes === 8 * 1024 + ); + assert.equal(cancelled, true, "upstream reader must be cancelled on cap exceed"); +}); + +test("an SSE stream within the cap still drains normally", async () => { + const enc = new TextEncoder(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(enc.encode('data: {"ok":1}\n\n')); + controller.close(); + }, + }); + const response = new Response(body, { headers: { "Content-Type": "text/event-stream" } }); + const out = await readNonStreamingResponseBody(response, "text/event-stream", true, 1024 * 1024); + assert.ok(out.includes('"ok":1')); +}); + +test("rejects a non-SSE response whose declared Content-Length exceeds the cap (no buffering)", async () => { + const response = new Response("body-not-read", { + headers: { "Content-Type": "application/json", "content-length": String(100 * 1024 * 1024) }, + }); + await assert.rejects( + () => readNonStreamingResponseBody(response, "application/json", false, 1024 * 1024), + (err) => err instanceof NonStreamingResponseTooLargeError + ); +});