diff --git a/src/server/relay.ts b/src/server/relay.ts index f480ace68a..2f3ccc5fd0 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -26,6 +26,7 @@ import { MAX_CLIENT_SSE_FRAME_BYTES, } from "./sse-frame-buffer"; import { replaceSseDataPayload } from "./sse-payload-rewrite"; +import { relayResponseLogBody } from "./response-log-body"; const nativePassthroughSseResponses = new WeakSet(); const eagerRelaySseResponses = new WeakSet(); @@ -718,24 +719,21 @@ export function responseWithDeferredRequestLog( } if (!response.body || !contentType.includes("text/event-stream")) { if (response.body && (contentType.includes("application/json") || response.status >= 400)) { - const finalizeJsonLog = async () => { - const text = await response.text(); - // Non-JSON error bodies: inspect/log only a bounded prefix (the stored - // upstreamError is 500 chars anyway); the FULL text is still forwarded to the - // client below, unchanged. JSON bodies keep full inspection (usage parsing). - const isJson = contentType.includes("application/json"); - inspectResponseLogJson(logCtx, isJson ? text : text.slice(0, 8192)); - addFinalRequestLog(requestId, start, logCtx, response.status, { closeReason: "non_stream" }, addLog); - return text; - }; - const body = new ReadableStream({ - async start(controller) { + const body = relayResponseLogBody(response.body, { + isJson: contentType.includes("application/json"), + onFinalize(outcome, text) { try { - controller.enqueue(new TextEncoder().encode(await finalizeJsonLog())); - controller.close(); - } catch (err) { - addFinalRequestLog(requestId, start, logCtx, 502, { closeReason: "non_stream" }, addLog); - try { controller.error(err); } catch { /* already torn down */ } + // Only complete, within-budget JSON reaches the existing usage parser. + // Error/cancel prefixes must not masquerade as reported JSON usage. + if (text !== undefined) inspectResponseLogJson(logCtx, text); + } finally { + if (outcome === "error" && logCtx.activeAttempt) { + logCtx.activeAttempt.streamAborted = true; + } + const status = outcome === "cancel" ? 499 : outcome === "error" ? 502 : response.status; + addFinalRequestLog(requestId, start, logCtx, status, { + closeReason: outcome === "cancel" ? "client_cancel" : "non_stream", + }, addLog); } }, }); diff --git a/src/server/response-log-body.ts b/src/server/response-log-body.ts new file mode 100644 index 0000000000..58fe8016ec --- /dev/null +++ b/src/server/response-log-body.ts @@ -0,0 +1,134 @@ +export const MAX_RESPONSE_LOG_INSPECTION_BYTES = 32 * 1024 * 1024; +export const MAX_NON_JSON_ERROR_INSPECTION_BYTES = 8 * 1024; + +export type ResponseLogBodyOutcome = "eof" | "error" | "cancel"; + +export type ResponseLogBodyOptions = { + isJson: boolean; + onFinalize(outcome: ResponseLogBodyOutcome, text: string | undefined): void; + /** Test seam; production uses the JSON / non-JSON byte ceilings above. */ + inspectionLimit?: number; +}; + +/** + * Forward original bytes on demand, retaining only a bounded log-inspection copy. + * JSON is inspected only after clean EOF and only when the complete body fits. + * Other error bodies keep a diagnostic prefix, including on error/cancellation. + * This is not an SSE observer and never owns a second, eagerly drained tee branch. + */ +export function relayResponseLogBody( + source: ReadableStream, + options: ResponseLogBodyOptions, +): ReadableStream { + const limit = options.inspectionLimit ?? (options.isJson + ? MAX_RESPONSE_LOG_INSPECTION_BYTES + : MAX_NON_JSON_ERROR_INSPECTION_BYTES); + if (!Number.isSafeInteger(limit) || limit < 0) { + throw new RangeError("Response log inspection limit must be a non-negative safe integer"); + } + const reader = source.getReader(); + let retained: Uint8Array = new Uint8Array(0); + let retainedBytes = 0; + let inspectionUnavailable = false; + let finalized = false; + let released = false; + + const discard = () => { + retained = new Uint8Array(0); + retainedBytes = 0; + }; + const release = () => { + if (released) return; + try { + reader.releaseLock(); + released = true; + } catch { + // Some runtimes defer settling a cancelled read; its continuation retries. + return; + } + }; + const retain = (chunk: Uint8Array) => { + if (inspectionUnavailable || chunk.byteLength === 0) return; + const remaining = limit - retainedBytes; + if (options.isJson && chunk.byteLength > remaining) { + inspectionUnavailable = true; + discard(); + return; + } + const count = Math.min(chunk.byteLength, remaining); + if (count === 0) return; + const required = retainedBytes + count; + if (required > retained.byteLength) { + // One geometrically grown allocation bounds both bytes and object count; + // retaining a separate slice per tiny chunk would still grow metadata. + const capacity = Math.min(limit, Math.max(required, retained.byteLength * 2, 4096)); + const grown = new Uint8Array(capacity); + grown.set(retained.subarray(0, retainedBytes)); + retained = grown; + } + retained.set(chunk.subarray(0, count), retainedBytes); + retainedBytes = required; + }; + const finalize = (outcome: ResponseLogBodyOutcome) => { + if (finalized) return; + finalized = true; + const inspect = !inspectionUnavailable && (!options.isJson || outcome === "eof"); + const bytes = retained.subarray(0, retainedBytes); + discard(); + let text: string | undefined; + try { + if (inspect) text = new TextDecoder().decode(bytes); + } catch { + // A diagnostic decoding failure must still finalize the request log. + text = undefined; + } + try { + options.onFinalize(outcome, text); + } catch { + // Optional logging must neither corrupt delivery nor prevent cancellation. + return; + } + }; + + return new ReadableStream({ + async pull(controller) { + try { + const { done, value } = await reader.read(); + if (finalized) { + release(); + return; + } + if (done) { + finalize("eof"); + release(); + controller.close(); + return; + } + try { + retain(value); + } catch { + // Inspection allocation is not a reason to drop client bytes. + inspectionUnavailable = true; + discard(); + } + controller.enqueue(value); + } catch (error) { + if (finalized) { + release(); + return; + } + finalize("error"); + release(); + controller.error(error); + } + }, + cancel(reason) { + if (finalized) return; + finalize("cancel"); + // Never await tee cancellation: it can wait for the other branch's EOF. + // Cancelling settles our pending read; no retained inspection bytes remain. + void reader.cancel(reason).catch(() => undefined); + release(); + }, + }, { highWaterMark: 0 }); +} diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 7f01dee197..357c114e2f 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -39,3 +39,26 @@ These optimizations do not add request queues, retry policies, or RSS-based admi Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. + +## Non-stream response-log inspection + +`src/server/relay.ts` delegates JSON and non-JSON error-body delivery to +`src/server/response-log-body.ts`. A single pull-driven stream forwards the original bytes; +logging neither waits for the whole body before delivery nor eagerly drains a second tee branch. +The inspection copy retains at most 32 MiB for JSON, or an 8 KiB byte prefix for other errors. +One geometrically grown allocation also bounds retained chunk metadata. These are per-response +inspection-copy limits, not a process-memory ceiling or an output-size limit; decoding and parsing +can allocate additional bounded objects. + +JSON is inspected only after clean EOF when the entire body fits. Crossing its budget discards +its inspection copy and skips parsing without truncating delivery, changing the HTTP status, +or inventing usage. Error and cancellation paths never parse a partial JSON document. Other +error bodies retain their bounded diagnostic prefix. The existing request-log parser and final +log writer remain responsible for redaction, usage provenance, and attempt accounting. + +EOF logs the original status, a failed body read logs 502, and downstream cancellation logs 499 +with `client_cancel`, exactly once. The same cancellation reason reaches the source reader; +reader cancellation is not awaited because a tee sibling can remain open. This limit does not +apply to SSE turn length: existing frame/output-item budgets and post-disconnect drain ownership +remain unchanged. `tests/server/consume-for-inspection-cancel.test.ts` registers the shared +body-lifecycle cases and covers the integration and a late SSE terminal beyond the JSON budget. diff --git a/tests/helpers/response-log-body-cases.ts b/tests/helpers/response-log-body-cases.ts new file mode 100644 index 0000000000..6b31e3db18 --- /dev/null +++ b/tests/helpers/response-log-body-cases.ts @@ -0,0 +1,146 @@ +import assert from 'node:assert/strict'; +import { relayResponseLogBody, MAX_RESPONSE_LOG_INSPECTION_BYTES, MAX_NON_JSON_ERROR_INSPECTION_BYTES } from '../../src/server/response-log-body'; +const enc = new TextEncoder(); +const tick = () => new Promise(r => setTimeout(r, 0)); +function controlled() { + let controller!: ReadableStreamDefaultController; + const reasons: unknown[] = []; + const stream = new ReadableStream({ start(c) { controller = c; }, cancel(r) { reasons.push(r); } }, { highWaterMark: 0 }); + return { stream, reasons, push: (v: string | Uint8Array) => controller.enqueue(typeof v === 'string' ? enc.encode(v) : v), close: () => controller.close(), error: (e: unknown) => controller.error(e) }; +} +function wrapped(source: ReadableStream, isJson: boolean, inspectionLimit?: number) { + const calls: [string, string | undefined][] = []; + return { body: relayResponseLogBody(source, { isJson, inspectionLimit, onFinalize: (...v) => calls.push(v) }), calls }; +} +export function registerResponseLogBodyCases(test: (name: string, run: () => Promise) => unknown): void { + for (const isJson of [false, true]) { + test(`first chunk before EOF; no early log (${isJson})`, async () => { + const s = controlled(), w = wrapped(s.stream, isJson); + const reader = w.body.getReader(), first = reader.read(); + s.push('{}'); + assert.deepEqual(await first, { value: enc.encode('{}'), done: false }); + assert.equal(w.calls.length, 0); + s.close(); + assert.equal((await reader.read()).done, true); + assert.deepEqual(w.calls, [['eof', '{}']]); + assert.equal(s.stream.locked, false); + }); + test(`slow readers do not cause prefetch (${isJson})`, async () => { + let pulls = 0; + const source = new ReadableStream({ pull(c) { ++pulls; c.enqueue(enc.encode('x')); } }, { highWaterMark: 0 }); + const w = wrapped(source, isJson), r = w.body.getReader(); + await tick(); assert.equal(pulls, 0); + await r.read(); await tick(); assert.equal(pulls, 1); + await r.cancel('stop'); assert.equal(w.calls.length, 1); + }); + test(`pending cancellation finalizes once and propagates reason (${isJson})`, async () => { + const s = controlled(), w = wrapped(s.stream, isJson), r = w.body.getReader(); + const pending = r.read(); await tick(); + const reason = new Error('client stop'); + await r.cancel(reason); await pending; await tick(); + assert.deepEqual(s.reasons, [reason]); + assert.deepEqual(w.calls, [['cancel', isJson ? undefined : '']]); + assert.equal(s.stream.locked, false); + }); + test(`source error preserves exception and prefix rules (${isJson})`, async () => { + const s = controlled(), w = wrapped(s.stream, isJson), r = w.body.getReader(); + const first = r.read(); s.push('{}'); await first; + const failure = new Error('upstream reset'); + const pending = r.read(); s.error(failure); + await assert.rejects(pending, (e: unknown) => e === failure); + assert.deepEqual(w.calls, [['error', isJson ? undefined : '{}']]); + assert.equal(s.stream.locked, false); + }); + } + test('binary and split UTF-8 forwarding is byte exact', async () => { + const chunks = [new Uint8Array([0xff, 0, 0xe3]), new Uint8Array([0x81, 0x82, 0xfe])]; + const source = new ReadableStream({ start(c) { chunks.forEach(v => c.enqueue(v)); c.close(); } }); + const w = wrapped(source, false); + assert.deepEqual(new Uint8Array(await new Response(w.body).arrayBuffer()), new Uint8Array(chunks.flatMap(v => [...v]))); + assert.equal(w.calls.length, 1); + }); + for (const [size, chunkSize, inspect] of [[8, 8, true], [9, 9, false], [9, 3, false], [8, 1, true]] as const) { + test(`JSON cap: ${size} bytes in ${chunkSize}-byte chunks`, async () => { + let sent = 0; + const source = new ReadableStream({ pull(c) { if (sent === size) return c.close(); const n = Math.min(chunkSize, size-sent); sent += n; c.enqueue(enc.encode('x'.repeat(n))); } }); + const w = wrapped(source, true, 8); + assert.equal(await new Response(w.body).text(), 'x'.repeat(size)); + assert.deepEqual(w.calls, [['eof', inspect ? 'x'.repeat(size) : undefined]]); + }); + } + test('actual oversized JSON budget still forwards every byte', async () => { + const size = MAX_RESPONSE_LOG_INSPECTION_BYTES + 1; + let sent = 0; + const block = new Uint8Array(65536).fill(120); + const source = new ReadableStream({ pull(c) { if (sent === size) return c.close(); const n = Math.min(block.length, size-sent); sent += n; c.enqueue(block.subarray(0,n)); } }); + const w = wrapped(source, true), r = w.body.getReader(); + let got = 0; + for (;;) { const { done, value } = await r.read(); if (done) break; got += value.length; assert.equal(value[0], 120); } + assert.equal(got, size); + assert.deepEqual(w.calls, [['eof', undefined]]); + }); + test('non-JSON prefix is byte bounded, not character bounded', async () => { + const text = '한'.repeat(10000); + const w = wrapped(new Response(text).body!, false); + assert.equal(await new Response(w.body).text(), text); + assert.deepEqual(w.calls, [['eof', new TextDecoder().decode(enc.encode(text).subarray(0, MAX_NON_JSON_ERROR_INSPECTION_BYTES))]]); + }); + test('cancel before any read never pulls', async () => { + let pulls = 0; let reason: unknown; + const s = new ReadableStream({ pull() { ++pulls; }, cancel(r) { reason = r; } }, { highWaterMark: 0 }); + const w = wrapped(s, true); + await w.body.cancel('before'); + assert.equal(pulls, 0); assert.equal(reason, 'before'); assert.equal(s.locked, false); + assert.deepEqual(w.calls, [['cancel', undefined]]); + }); + test('valid JSON prefix on cancellation is not treated as a complete response', async () => { + const s = controlled(), w = wrapped(s.stream, true), r = w.body.getReader(); + const pending = r.read(); s.push('{"usage":{"total_tokens":900}}'); await pending; + await r.cancel(); assert.deepEqual(w.calls, [['cancel', undefined]]); + }); + for (const outcome of ['eof', 'cancel', 'error']) { + test(`throwing logger does not disrupt ${outcome}`, async () => { + const s = controlled(); let calls = 0; + const body = relayResponseLogBody(s.stream, { isJson: false, onFinalize() { ++calls; throw new Error('logging failed'); } }); + const r = body.getReader(); const pending = r.read(); s.push('ok'); await pending; + if (outcome === 'eof') { s.close(); assert.equal((await r.read()).done, true); } + if (outcome === 'cancel') { await r.cancel('gone'); assert.deepEqual(s.reasons, ['gone']); } + if (outcome === 'error') { const e = new Error('reset'); s.error(e); await assert.rejects(r.read(), (v: unknown) => v === e); } + assert.equal(calls, 1); assert.equal(s.stream.locked, false); + }); + } + test('cancel rejection does not stall teardown', async () => { + const source = new ReadableStream({ cancel() { return Promise.reject(new Error('cancel failed')); } }); + const w = wrapped(source, false); await w.body.cancel(); await tick(); + assert.equal(w.calls.length, 1); assert.equal(source.locked, false); + }); + test('tee cancellation does not wait for sibling; sibling remains intact', async () => { + const s = controlled(), [left, right] = s.stream.tee(); + const w = wrapped(left, false), reader = w.body.getReader(), sibling = right.getReader(); + const a = reader.read(), b = sibling.read(); s.push('first'); await a; await b; + await reader.cancel('left stopped'); + assert.equal(w.calls.length, 1); assert.equal(s.reasons.length, 0); + const next = sibling.read(); s.push('second'); assert.equal(new TextDecoder().decode((await next).value), 'second'); + s.close(); assert.equal((await sibling.read()).done, true); + }); + test('cancel wins a racing source error without duplicate finalization', async () => { + const s = controlled(), w = wrapped(s.stream, true), r = w.body.getReader(); + const pending = r.read(); await tick(); + s.error(new Error('reset')); + await r.cancel('stop'); await pending; await tick(); + assert.deepEqual(w.calls, [['cancel', undefined]]); + }); + test('retained inspection is a copy, not a view of forwarded storage', async () => { + const s = controlled(), w = wrapped(s.stream, false), r = w.body.getReader(); + const buffer = enc.encode('original'); const read = r.read(); s.push(buffer); await read; buffer.fill(120); + s.close(); await r.read(); assert.deepEqual(w.calls, [['eof', 'original']]); + }); + test('empty JSON at zero cap and invalid limits', async () => { + const w = wrapped(new Response('').body!, true, 0); await new Response(w.body).text(); assert.deepEqual(w.calls, [['eof', '']]); + for (const limit of [-1, NaN, Infinity, 1.5]) { + const source = controlled().stream; + assert.throws(() => wrapped(source, true, limit), RangeError); assert.equal(source.locked, false); + } + }); + +} diff --git a/tests/server/consume-for-inspection-cancel.test.ts b/tests/server/consume-for-inspection-cancel.test.ts index 681b91a717..5d7e2081ea 100644 --- a/tests/server/consume-for-inspection-cancel.test.ts +++ b/tests/server/consume-for-inspection-cancel.test.ts @@ -3,10 +3,13 @@ import { consumeForInspection, consumeForResponseLogMetadata, getInspectionCounters, + responseWithDeferredRequestLog, resetInspectionCountersForTest, type SseInspector, } from "../../src/server/relay"; -import type { RequestLogContext } from "../../src/server/request-log"; +import { registerResponseLogBodyCases } from "../helpers/response-log-body-cases"; +import { MAX_RESPONSE_LOG_INSPECTION_BYTES } from "../../src/server/response-log-body"; +import type { RequestLogContext, RequestLogEntry } from "../../src/server/request-log"; // Regression for issue #44: native-passthrough turns are inspected on a teed background stream. // Codex disconnects the instant it finishes reading, so the inspection stream is frequently @@ -425,3 +428,80 @@ describe("consumeForInspection bare-error EOF finality", () => { expect(cancels).toBe(1); }); }); + +describe("bounded response log body lifecycle", () => registerResponseLogBodyCases(test)); + +for (const contentType of ["application/json", "text/plain"]) { + for (const outcome of ["eof", "error", "cancel"] as const) { + test(`deferred ${contentType} finalizes once on ${outcome}`, async () => { + const entries: RequestLogEntry[] = []; + let controller!: ReadableStreamDefaultController; + const reasons: unknown[] = []; + const source = new ReadableStream({ + start(c) { controller = c; }, + cancel(reason) { reasons.push(reason); }, + }); + const ctx: RequestLogContext = { + model: "fixture-model", provider: "custom", usageFromBridge: true, + usage: { inputTokens: 3, outputTokens: 2 }, + }; + const response = responseWithDeferredRequestLog(new Response(source, { + status: 503, statusText: "Unavailable", headers: { "content-type": contentType, "x-fixture": "retained" }, + }), "bounded-log-fixture", Date.now(), ctx, entry => entries.push(entry)); + const reader = response.body!.getReader(); + const first = reader.read(); + const bytes = encoder.encode(contentType === "application/json" ? '{"model":"wire-model"}' : "provider failed"); + controller.enqueue(bytes); + expect((await first).value).toEqual(bytes); + expect(entries).toHaveLength(0); + expect(response.status).toBe(503); + expect(response.statusText).toBe("Unavailable"); + expect(response.headers.get("x-fixture")).toBe("retained"); + if (outcome === "eof") { + controller.close(); + expect((await reader.read()).done).toBe(true); + } else if (outcome === "error") { + const failure = new Error("fixture reset"); + controller.error(failure); + await expect(reader.read()).rejects.toThrow("fixture reset"); + } else { + const pending = reader.read(); + await reader.cancel("fixture cancelled"); + await pending; + expect(reasons).toEqual(["fixture cancelled"]); + } + await tick(); + expect(entries).toHaveLength(1); + expect(entries[0]?.status).toBe(outcome === "cancel" ? 499 : outcome === "error" ? 502 : 503); + expect(entries[0]?.closeReason).toBe(outcome === "cancel" ? "client_cancel" : "non_stream"); + expect(ctx.usageFromBridge).toBe(true); + expect(ctx.usage).toEqual({ inputTokens: 3, outputTokens: 2 }); + expect(source.locked).toBe(false); + if (contentType === "application/json" && outcome !== "eof") { + expect(ctx.resolvedModel).toBeUndefined(); + } + }); + } +} + +test("JSON inspection budget does not detach a long SSE terminal observer", async () => { + const terminals: string[] = []; + const completed: unknown[] = []; + const padding = encoder.encode(`: ${"x".repeat(64 * 1024)}\n\n`); + let remaining = Math.ceil(MAX_RESPONSE_LOG_INSPECTION_BYTES / padding.byteLength) + 1; + const source = new ReadableStream({ + pull(controller) { + if (remaining-- > 0) controller.enqueue(padding); + else { + controller.enqueue(completedFrame("after-json-budget")); + controller.close(); + } + }, + }); + await new Promise(resolve => consumeForInspection( + source, status => terminals.push(status), undefined, resolve, + undefined, undefined, response => completed.push(response), + )); + expect(terminals).toEqual(["completed"]); + expect(completed).toHaveLength(1); +}, 10_000);