diff --git a/changelog.d/fixes/7847-bound-client-raw-request.md b/changelog.d/fixes/7847-bound-client-raw-request.md new file mode 100644 index 0000000000..80ef4870d8 --- /dev/null +++ b/changelog.d/fixes/7847-bound-client-raw-request.md @@ -0,0 +1 @@ +- fix(backend): stop `buildClientRawRequest` deep-cloning the whole request body on every chat request (#7847) — every consumer of `clientRawRequest.body` is observability and keeps at most a bounded copy, so the unbounded clone retained ~41x more than anything used it (3.19 MiB vs 0.08 MiB on a 3.05 MiB / 729-message agent request). Also makes `cloneBoundedForLog` idempotent: arrays, objects and strings all exceeded their own bounds once the truncation marker was added, so re-bounding an already bounded payload silently dropped a further item and misreported the original length diff --git a/open-sse/utils/requestLogger.ts b/open-sse/utils/requestLogger.ts index 4c8dc3f660..f2ef74e84e 100644 --- a/open-sse/utils/requestLogger.ts +++ b/open-sse/utils/requestLogger.ts @@ -102,9 +102,26 @@ function createEmptyStreamChunks() { }; } +const TRUNCATED_ARRAY_MARKER = "_omniroute_truncated_array"; +const TRUNCATED_KEYS_MARKER = "_omniroute_truncated_keys"; + +function isTruncatedArrayMarker(value: unknown): boolean { + return ( + typeof value === "object" && + value !== null && + !Array.isArray(value) && + (value as JsonRecord)[TRUNCATED_ARRAY_MARKER] === true + ); +} + function truncateLogString(value: string, maxLength = MAX_LOG_STRING_LENGTH): string { if (value.length <= maxLength) return value; - return `${value.slice(0, Math.floor(maxLength / 2))}\n[...truncated ${value.length - maxLength} chars...]\n${value.slice(-Math.ceil(maxLength / 2))}`; + // The marker has to fit INSIDE the budget (#7847): keeping maxLength characters and then + // adding the marker produced a result longer than maxLength, so re-bounding an already + // bounded string truncated it a second time and the function was not idempotent. + const marker = `\n[...truncated ${value.length - maxLength} chars...]\n`; + const keep = Math.max(0, maxLength - marker.length); + return `${value.slice(0, Math.floor(keep / 2))}${marker}${value.slice(-Math.ceil(keep / 2))}`; } /** @@ -134,6 +151,13 @@ export function cloneBoundedForLog(value: unknown, depth = 0, key: string | null if (depth >= 6) return "[MaxDepth]"; if (Array.isArray(value)) { + // Idempotence (#7847): an already-bounded array is [marker, ...tail] — MAX_LOG_ARRAY_ITEMS + 1 + // entries, which is over the limit. Re-truncating it would drop the marker plus one real + // item and rewrite originalLength with the truncated length (25 instead of the true 800), so + // the log would misreport how much was cut. Keep the original marker, re-bound only the tail. + if (isTruncatedArrayMarker(value[0])) { + return [value[0], ...value.slice(1).map((item) => cloneBoundedForLog(item, depth + 1))]; + } const exempt = key === "tools"; const shouldTruncate = !exempt && value.length > MAX_LOG_ARRAY_ITEMS; const source = shouldTruncate ? value.slice(-MAX_LOG_ARRAY_ITEMS) : value; @@ -141,7 +165,7 @@ export function cloneBoundedForLog(value: unknown, depth = 0, key: string | null if (shouldTruncate) { return [ { - _omniroute_truncated_array: true, + [TRUNCATED_ARRAY_MARKER]: true, originalLength: value.length, retainedTailItems: MAX_LOG_ARRAY_ITEMS, }, @@ -152,12 +176,19 @@ export function cloneBoundedForLog(value: unknown, depth = 0, key: string | null } const result: JsonRecord = {}; - const entries = Object.entries(value as JsonRecord); + // Idempotence (#7847): our own marker key must not be counted as payload, or a re-bounded + // object would push a real key out to make room for it and report `1` dropped instead of 20. + const carriedDropped = (value as JsonRecord)[TRUNCATED_KEYS_MARKER]; + const carried = typeof carriedDropped === "number" ? carriedDropped : 0; + const entries = Object.entries(value as JsonRecord).filter( + ([k]) => !(carried > 0 && k === TRUNCATED_KEYS_MARKER) + ); for (const [k, item] of entries.slice(0, MAX_LOG_OBJECT_KEYS)) { result[k] = cloneBoundedForLog(item, depth + 1, k); } - if (entries.length > MAX_LOG_OBJECT_KEYS) { - result._omniroute_truncated_keys = entries.length - MAX_LOG_OBJECT_KEYS; + const dropped = Math.max(0, entries.length - MAX_LOG_OBJECT_KEYS) + carried; + if (dropped > 0) { + result[TRUNCATED_KEYS_MARKER] = dropped; } return result; } diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index e4125beda7..e684730d80 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -102,7 +102,7 @@ import { generateRequestId } from "../../shared/utils/requestId"; import { logAuditEvent } from "../../lib/compliance/index"; import { enforceApiKeyPolicy } from "../../shared/utils/apiKeyPolicy"; import { hasProviderQuotaBypassScope } from "../../shared/constants/apiKeyPolicyScopes"; -import { cloneLogPayload } from "@/lib/logPayloads"; +import { cloneBoundedForLog } from "@omniroute/open-sse/utils/requestLogger.ts"; import { handleInternalUsageCommand } from "@/lib/usage/internalUsageCommand"; import { applyTaskAwareRouting, @@ -958,42 +958,10 @@ export async function handleChat( return withCorrelationId(withSessionHeader(response, sessionId), reqId); } -export function buildClientRawRequest(request: Request, body: unknown) { - const url = new URL(request.url); - return { - endpoint: url.pathname, - body: cloneLogPayload(body), - headers: Object.fromEntries(request.headers.entries()), - signal: request.signal ?? null, - }; -} - -/** - * #7360 follow-up: chatCore.ts's createStreamController (and, downstream, - * withRateLimit/acquireAccountSemaphore) only ever watches - * clientRawRequest.signal — the ORIGINAL client's request signal, which stays - * open for as long as the overall combo keeps retrying elsewhere. A target - * abandoned by comboTargetTimeoutMs (open-sse/services/combo/targetTimeoutRunner.ts) - * never learns it was abandoned, and hangs forever (leaking a permanent - * "pending" dashboard entry — trackPendingRequest(false) never runs; live - * incident, log id 1784418258231-14961a). Merges the per-target - * modelAbortSignal (when present) into clientRawRequest.signal so an - * abandoned dispatch can actually observe its own abort and reach its - * cleanup path — returns clientRawRequest unchanged when there's no - * modelAbortSignal to merge in (the non-combo / non-timed-out common case). - */ -export function resolveDispatchClientRawRequest( - clientRawRequest: { signal?: AbortSignal | null } | null | undefined, - modelAbortSignal: AbortSignal | null | undefined -): typeof clientRawRequest { - if (!modelAbortSignal) return clientRawRequest; - return { - ...clientRawRequest, - signal: clientRawRequest?.signal - ? mergeAbortSignals(clientRawRequest.signal, modelAbortSignal) - : modelAbortSignal, - }; -} +// The clientRawRequest envelope lives in ./chat/clientRawRequest.ts. Imported for local use +// below and re-exported for the historical public surface. +import { buildClientRawRequest, resolveDispatchClientRawRequest } from "./chat/clientRawRequest.ts"; +export { buildClientRawRequest, resolveDispatchClientRawRequest }; /** * Handle single model chat request diff --git a/src/sse/handlers/chat/clientRawRequest.ts b/src/sse/handlers/chat/clientRawRequest.ts new file mode 100644 index 0000000000..0fd80cd9a8 --- /dev/null +++ b/src/sse/handlers/chat/clientRawRequest.ts @@ -0,0 +1,57 @@ +/** + * The clientRawRequest envelope — the observability snapshot of a chat request. + * + * Extracted from chat.ts (#7847). Both helpers are about the same object: one builds it, the + * other merges a per-target abort signal into it before dispatch. Neither belongs in the + * request handler proper, and chat.ts sits against a frozen file-size ratchet. + * + * chat.ts re-exports both, so the public surface and tests/unit/chat-build-client-raw-request + * are unchanged. + */ +import { mergeAbortSignals } from "@omniroute/open-sse/executors/base.ts"; +import { cloneBoundedForLog } from "@omniroute/open-sse/utils/requestLogger.ts"; + +export function buildClientRawRequest(request: Request, body: unknown) { + const url = new URL(request.url); + return { + endpoint: url.pathname, + // #7847: bounded, not a full deep clone. Every consumer of clientRawRequest.body is + // observability — reqLogger.logClientRawRequest (which re-bounds it anyway, or drops it + // entirely when the logger is disabled), trackPendingRequest's `clientRequest`, and + // recordRejectedRequestUsage's `requestBody`. None feeds dispatch, translation or the + // upstream call, so cloning the whole payload retained ~41x more than anything kept: + // 3.19 MiB vs 0.08 MiB on the incident's 3.05 MiB / 729-message request. + // Still a clone, not an alias — `body` is rewritten downstream (plugin onRequest hook, + // compression), and this has to stay a snapshot of what the client actually sent. + body: cloneBoundedForLog(body), + headers: Object.fromEntries(request.headers.entries()), + signal: request.signal ?? null, + }; +} + +/** + * #7360 follow-up: chatCore.ts's createStreamController (and, downstream, + * withRateLimit/acquireAccountSemaphore) only ever watches + * clientRawRequest.signal — the ORIGINAL client's request signal, which stays + * open for as long as the overall combo keeps retrying elsewhere. A target + * abandoned by comboTargetTimeoutMs (open-sse/services/combo/targetTimeoutRunner.ts) + * never learns it was abandoned, and hangs forever (leaking a permanent + * "pending" dashboard entry — trackPendingRequest(false) never runs; live + * incident, log id 1784418258231-14961a). Merges the per-target + * modelAbortSignal (when present) into clientRawRequest.signal so an + * abandoned dispatch can actually observe its own abort and reach its + * cleanup path — returns clientRawRequest unchanged when there's no + * modelAbortSignal to merge in (the non-combo / non-timed-out common case). + */ +export function resolveDispatchClientRawRequest( + clientRawRequest: { signal?: AbortSignal | null } | null | undefined, + modelAbortSignal: AbortSignal | null | undefined +): typeof clientRawRequest { + if (!modelAbortSignal) return clientRawRequest; + return { + ...clientRawRequest, + signal: clientRawRequest?.signal + ? mergeAbortSignals(clientRawRequest.signal, modelAbortSignal) + : modelAbortSignal, + }; +} diff --git a/tests/unit/repro-7847-bound-client-raw-request.test.ts b/tests/unit/repro-7847-bound-client-raw-request.test.ts new file mode 100644 index 0000000000..165c87ae68 --- /dev/null +++ b/tests/unit/repro-7847-bound-client-raw-request.test.ts @@ -0,0 +1,116 @@ +// Repro + regression guard for #7847: buildClientRawRequest deep-clones the ENTIRE request body +// on every chat request, unbounded, even though every consumer of clientRawRequest.body is +// observability that either discards it or re-clones it *bounded*. +// +// Consumers traced at the time of writing — none feeds dispatch, translation or the upstream call: +// 1. chatCore.ts -> reqLogger.logClientRawRequest(...) (no-op when the logger is off, +// otherwise re-clones via cloneBoundedForLog) +// 2. chatCore.ts -> trackPendingRequest({ clientRequest }) -> /api/logs/[id] +// 3. chat.ts -> recordRejectedRequestUsage({ requestBody }) +// +// The incident: a 3.05 MiB request (729 messages / 86 tools) reached ~12,282 MiB of V8 heap. +import { test } from "node:test"; +import assert from "node:assert/strict"; + +const { buildClientRawRequest } = await import("../../src/sse/handlers/chat.ts"); +const { cloneBoundedForLog, MAX_LOG_ARRAY_ITEMS } = await import( + "../../open-sse/utils/requestLogger.ts" +); + +const MESSAGES = 800; + +function makeRequest(): Request { + return new Request("http://localhost:20128/v1/chat/completions", { + method: "POST", + headers: { "content-type": "application/json" }, + }); +} + +function longHistoryBody() { + return { + model: "claude-opus-5", + messages: Array.from({ length: MESSAGES }, (_, i) => ({ + role: i % 2 === 0 ? "user" : "assistant", + content: `message ${i} `.repeat(200), + })), + }; +} + +test("#7847: buildClientRawRequest must not retain an unbounded copy of the message history", () => { + const captured = buildClientRawRequest(makeRequest(), longHistoryBody()) as { + body: { messages: unknown[] }; + }; + + // The bounded clone keeps a truncation marker plus the tail, never the whole history. + assert.ok( + captured.body.messages.length <= MAX_LOG_ARRAY_ITEMS + 1, + `clientRawRequest.body retained ${captured.body.messages.length} of ${MESSAGES} messages — ` + + `every consumer is observability and keeps at most ${MAX_LOG_ARRAY_ITEMS}` + ); +}); + +test("#7847: the retained snapshot does not grow with the message count", () => { + // The amplification in #7847 is that retention scaled with history length. Ten times the + // history must not cost ten times the retained snapshot. + const size = (v: unknown) => JSON.stringify(v).length; + const at = (messages: number) => + size( + ( + buildClientRawRequest(makeRequest(), { + model: "claude-opus-5", + messages: Array.from({ length: messages }, (_, i) => ({ + role: i % 2 === 0 ? "user" : "assistant", + content: `message ${i} `.repeat(200), + })), + }) as { body: unknown } + ).body + ); + + const small = at(MESSAGES); + const tenfold = at(MESSAGES * 10); + // Not exactly equal: only the tail is retained, and the tail of the longer history carries + // wider index numbers ("message 7999" vs "message 799"). What matters is that the growth is + // a rounding error rather than the 10x a proportional retention would cost. + assert.ok( + tenfold < small * 1.5, + `retained ${tenfold} bytes for ${MESSAGES * 10} messages vs ${small} for ${MESSAGES} — ` + + `retention must be bounded; proportional retention would be ~${small * 10}` + ); +}); + +test("bounding at the entry does not change what the request logger ultimately stores", () => { + // chatCore hands clientRawRequest.body to reqLogger.logClientRawRequest, which applies + // cloneBoundedForLog itself. Bounding earlier may only be safe if that second pass is a + // no-op — otherwise the persisted log payload would change shape. + const body = longHistoryBody(); + const once = cloneBoundedForLog(body); + const twice = cloneBoundedForLog(once); + assert.deepEqual(twice, once, "cloneBoundedForLog must be idempotent for the entry clone to be safe"); +}); + +test("buildClientRawRequest still carries endpoint, headers and signal", () => { + const req = new Request("http://localhost:20128/v1/chat/completions?x=1", { + method: "POST", + headers: { "content-type": "application/json", "x-omniroute-session-id": "sess-1" }, + }); + const out = buildClientRawRequest(req, { model: "m", messages: [] }) as { + endpoint: string; + headers: Record; + signal: unknown; + }; + assert.equal(out.endpoint, "/v1/chat/completions"); + assert.equal(out.headers["x-omniroute-session-id"], "sess-1"); + assert.equal(out.headers["content-type"], "application/json"); + assert.ok("signal" in out); +}); + +test("the captured body is a snapshot — later mutation of the request body must not leak in", () => { + // chatCore rewrites `body` downstream (plugin onRequest hook, compression). The captured + // snapshot must not alias it, or the log would show post-mutation content. + const body = { model: "m", messages: [{ role: "user", content: "original" }] }; + const captured = buildClientRawRequest(makeRequest(), body) as { + body: { messages: { content: string }[] }; + }; + body.messages[0].content = "mutated by a downstream plugin"; + assert.equal(captured.body.messages[0].content, "original"); +}); diff --git a/tests/unit/request-logger-bounded-idempotence.test.ts b/tests/unit/request-logger-bounded-idempotence.test.ts new file mode 100644 index 0000000000..e8aa54169a --- /dev/null +++ b/tests/unit/request-logger-bounded-idempotence.test.ts @@ -0,0 +1,95 @@ +// cloneBoundedForLog must be idempotent (#7847). +// +// Since buildClientRawRequest now bounds the body at the entry point, the request logger applies +// cloneBoundedForLog to an ALREADY bounded value. If the second pass were not a no-op, the +// persisted log payload would change shape versus before the fix — and it did not used to be: +// +// arrays : [marker, ...24 items] is 25 entries, over the 24 limit, so a second pass dropped +// the marker plus one real item and rewrote originalLength as 25 instead of 800. +// objects : 80 keys + _omniroute_truncated_keys is 81, so a second pass evicted a real key to +// make room and reported 1 dropped instead of the true count. +// strings : the truncation marker was appended AFTER slicing to maxLength, so the "bounded" +// string was longer than the bound and got truncated again. +import { test } from "node:test"; +import assert from "node:assert/strict"; + +const { cloneBoundedForLog, MAX_LOG_ARRAY_ITEMS } = await import( + "../../open-sse/utils/requestLogger.ts" +); + +const MAX_KEYS = 80; +const MAX_STRING = 64 * 1024; + +test("arrays: second pass preserves the marker, the tail, and the true originalLength", () => { + const input = { messages: Array.from({ length: 800 }, (_, i) => ({ role: "user", n: i })) }; + const once = cloneBoundedForLog(input) as { messages: Record[] }; + const twice = cloneBoundedForLog(once) as { messages: Record[] }; + + assert.deepEqual(twice, once, "re-bounding must be a no-op"); + assert.equal(twice.messages.length, MAX_LOG_ARRAY_ITEMS + 1, "marker plus the retained tail"); + assert.equal( + twice.messages[0].originalLength, + 800, + "originalLength must keep describing the ORIGINAL array, not the bounded one" + ); + // The tail must still be the last items of the real history, not shifted by the marker. + assert.equal((twice.messages.at(-1) as { n: number }).n, 799); +}); + +test("objects: second pass keeps the real keys and the true dropped count", () => { + const input = Object.fromEntries(Array.from({ length: 100 }, (_, i) => [`k${i}`, i])); + const once = cloneBoundedForLog(input) as Record; + const twice = cloneBoundedForLog(once) as Record; + + assert.deepEqual(twice, once, "re-bounding must be a no-op"); + assert.equal(once._omniroute_truncated_keys, 20, "100 keys minus the 80 retained"); + assert.equal(twice._omniroute_truncated_keys, 20, "the dropped count must not be recomputed"); + assert.equal( + Object.keys(twice).filter((k) => k !== "_omniroute_truncated_keys").length, + MAX_KEYS, + "a real key must not be evicted to make room for the marker" + ); +}); + +test("strings: the bounded result respects the bound, so a second pass is a no-op", () => { + const once = cloneBoundedForLog("x".repeat(200_000)) as string; + const twice = cloneBoundedForLog(once) as string; + + assert.equal(twice, once, "re-bounding must be a no-op"); + assert.ok( + once.length <= MAX_STRING, + `bounded string is ${once.length} chars, over the ${MAX_STRING} bound — the marker must fit inside the budget` + ); + assert.match(once, /\[\.\.\.truncated \d+ chars\.\.\.\]/); +}); + +test("values already within the bounds are returned unchanged", () => { + const input = { model: "m", messages: [{ role: "user", content: "hi" }], n: 1, ok: true }; + assert.deepEqual(cloneBoundedForLog(input), input); + assert.deepEqual(cloneBoundedForLog(cloneBoundedForLog(input)), input); +}); + +test("the tools exemption survives re-bounding", () => { + // tools are deliberately exempt from array truncation (debug-critical inventory); a second + // pass must not start truncating them. + const input = { tools: Array.from({ length: 200 }, (_, i) => ({ name: `tool_${i}` })) }; + const once = cloneBoundedForLog(input) as { tools: unknown[] }; + const twice = cloneBoundedForLog(once) as { tools: unknown[] }; + + assert.equal(once.tools.length, 200, "tools must not be truncated"); + assert.equal(twice.tools.length, 200, "and must stay untruncated on a second pass"); + assert.deepEqual(twice, once); +}); + +test("nested structures stay stable across repeated bounding", () => { + const input = { + messages: Array.from({ length: 50 }, (_, i) => ({ + role: "user", + content: "y".repeat(100_000), + meta: Object.fromEntries(Array.from({ length: 90 }, (_, k) => [`m${k}`, `${i}-${k}`])), + })), + }; + const once = cloneBoundedForLog(input); + assert.deepEqual(cloneBoundedForLog(once), once); + assert.deepEqual(cloneBoundedForLog(cloneBoundedForLog(once)), once, "stable under repetition"); +});