From f00eaae5c3447355de736cc7850133010266bb00 Mon Sep 17 00:00:00 2001 From: luvs01 Date: Sat, 8 Aug 2026 23:59:03 +0900 Subject: [PATCH] fix(logs): bound response metadata inspection --- src/server/relay.ts | 76 +++++++++++++++++---- tests/consume-for-inspection-cancel.test.ts | 20 ++++++ tests/request-log.test.ts | 18 +++++ 3 files changed, 101 insertions(+), 13 deletions(-) diff --git a/src/server/relay.ts b/src/server/relay.ts index 3b5aae2366..1c9ff06df1 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -17,6 +17,7 @@ const nativePassthroughSseResponses = new WeakSet(); const eagerRelaySseResponses = new WeakSet(); export const MAX_INSPECTION_SSE_FRAME_BYTES = 4 * 1024 * 1024; +export const MAX_RESPONSE_LOG_INSPECTION_BYTES = 32 * 1024 * 1024; export const MAX_COMPLETED_OUTPUT_ITEMS = 256; export const MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES = 8 * 1024 * 1024; export const MAX_TAIL_ERROR_MESSAGE_CHARS = 512; @@ -428,26 +429,59 @@ 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)); + const reader = response.body.getReader(); + const isJson = contentType.includes("application/json"); + const inspectionLimit = isJson ? MAX_RESPONSE_LOG_INSPECTION_BYTES : 8192; + const inspectedChunks: Uint8Array[] = []; + let inspectedBytes = 0; + let inspectionOverflowed = false; + let finalized = false; + const finalizeJsonLog = () => { + if (finalized) return; + finalized = true; + if (!inspectionOverflowed || !isJson) { + inspectResponseLogJson(logCtx, new TextDecoder().decode(joinedBytes(inspectedChunks, inspectedBytes))); + } addFinalRequestLog(requestId, start, logCtx, response.status, { closeReason: "non_stream" }, addLog); - return text; }; const body = new ReadableStream({ - async start(controller) { + async pull(controller) { try { - controller.enqueue(new TextEncoder().encode(await finalizeJsonLog())); - controller.close(); + const { done, value } = await reader.read(); + if (done) { + finalizeJsonLog(); + controller.close(); + return; + } + if (!inspectionOverflowed) { + if (inspectedBytes + value.byteLength <= inspectionLimit) { + inspectedChunks.push(value.slice()); + inspectedBytes += value.byteLength; + } else { + inspectionOverflowed = true; + if (isJson) { + inspectedChunks.length = 0; + inspectedBytes = 0; + } else { + const remainingBytes = inspectionLimit - inspectedBytes; + if (remainingBytes > 0) inspectedChunks.push(value.slice(0, remainingBytes)); + inspectedBytes = inspectionLimit; + } + } + } + controller.enqueue(value); } catch (err) { - addFinalRequestLog(requestId, start, logCtx, 502, { closeReason: "non_stream" }, addLog); + if (!finalized) { + finalized = true; + addFinalRequestLog(requestId, start, logCtx, 502, { closeReason: "non_stream" }, addLog); + } try { controller.error(err); } catch { /* already torn down */ } } }, + cancel(reason) { + finalizeJsonLog(); + reader.cancel(reason).catch(() => {}); + }, }); return new Response(body, { status: response.status, @@ -929,6 +963,8 @@ export type InspectionConsumerOptions = { drainBounds?: Partial; upstream?: AbortController; now?: () => number; + /** Maximum total bytes inspected before detaching from the provider stream. */ + maxInspectionBytes?: number; /** Test seam for proving both public consumers dispose their owned inspector. */ inspectorFactory?: (handlers: SseInspectorHandlers) => SseInspector; }; @@ -951,11 +987,13 @@ function startBoundedInspectionPump(options: InspectionPumpOptions): void { let clientGone = false; let clientGoneReason: unknown; let drainedBytes = 0; + let inspectedBytes = 0; let drainDeadline = Number.POSITIVE_INFINITY; let drainTimer: ReturnType | undefined; let drainStopped = false; const drainMs = options.drainBounds?.ms ?? DEFAULT_INSPECTION_DRAIN_MS; const drainBytes = options.drainBounds?.bytes ?? DEFAULT_INSPECTION_DRAIN_BYTES; + const maxInspectionBytes = options.maxInspectionBytes ?? MAX_RESPONSE_LOG_INSPECTION_BYTES; const now = options.now ?? Date.now; let cancelFired = false; const fireCancel = () => { @@ -1025,7 +1063,19 @@ function startBoundedInspectionPump(options: InspectionPumpOptions): void { break; } if (!clientGone) { - inspector.feed(value); + const remainingBytes = Math.max(0, maxInspectionBytes - inspectedBytes); + const inspectedValue = value.byteLength > remainingBytes + ? value.subarray(0, remainingBytes) + : value; + if (inspectedValue.byteLength > 0) inspector.feed(inspectedValue); + inspectedBytes += inspectedValue.byteLength; + if (value.byteLength > remainingBytes || inspectedBytes >= maxInspectionBytes) { + // This reader normally owns one branch of a tee. Cancelling only the + // inspection branch leaves delivery to the client intact and restores + // client backpressure instead of greedily draining an untrusted source. + await reader.cancel("response log inspection byte limit reached"); + break; + } continue; } if (now() >= drainDeadline) { diff --git a/tests/consume-for-inspection-cancel.test.ts b/tests/consume-for-inspection-cancel.test.ts index 96632374e5..f4ebbe3415 100644 --- a/tests/consume-for-inspection-cancel.test.ts +++ b/tests/consume-for-inspection-cancel.test.ts @@ -112,6 +112,26 @@ describe("consumeForInspection cancel finalization (#44)", () => { }); describe("bounded post-disconnect inspection drain", () => { + test("metadata inspection detaches at its total byte limit", async () => { + const source = controlledStream(); + const done = new Promise(resolve => { + consumeForResponseLogMetadata( + source.stream, + {} as RequestLogContext, + undefined, + resolve, + undefined, + undefined, + { maxInspectionBytes: 8 }, + ); + }); + + source.push(encoder.encode("12345678")); + await done; + + expect(source.cancelReasons).toEqual(["response log inspection byte limit reached"]); + }); + test("consumeForInspection stops at the injected byte bound, cancels the reader, and aborts upstream", async () => { const source = controlledStream(); const clientGone = new AbortController(); diff --git a/tests/request-log.test.ts b/tests/request-log.test.ts index 50042794d7..834da6ac7c 100644 --- a/tests/request-log.test.ts +++ b/tests/request-log.test.ts @@ -23,6 +23,7 @@ import { sealRequestAttemptIdentity, type RequestLogContext, } from "../src/server/request-log"; +import { MAX_RESPONSE_LOG_INSPECTION_BYTES } from "../src/server/relay"; import { bridgeToResponsesSSE } from "../src/bridge"; import type { AdapterEvent, OcxUsage } from "../src/types"; import { @@ -655,6 +656,23 @@ describe("request log metadata", () => { }); }); + test("deferred JSON logging forwards oversized bodies without retaining them for inspection", async () => { + const entries: RequestLogEntry[] = []; + const prefix = JSON.stringify({ model: "must-not-be-inspected", padding: "" }).slice(0, -2); + const text = `${prefix}${"x".repeat(MAX_RESPONSE_LOG_INSPECTION_BYTES)}"}`; + const response = responseWithDeferredRequestLog( + new Response(text, { headers: { "content-type": "application/json" } }), + "ocx-test-json-bounded", + Date.now(), + { model: "requested-model", provider: "openai" }, + entry => entries.push(entry), + ); + + expect(await response.text()).toBe(text); + expect(entries).toHaveLength(1); + expect(entries[0]?.resolvedModel).toBeUndefined(); + }); + test("deferred JSON logging accepts ChatCompletions-shape usage", async () => { const entries: RequestLogEntry[] = []; const response = responseWithDeferredRequestLog(