diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index 5f3fc99649..a5e9628bc8 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -45,7 +45,7 @@ After GUI registration or OAuth login, the confirmation dialog lets you open the | `upstreamFailoverThreshold?` | `number` | `3` | Consecutive transient failures before future new sessions fail over. Set `0` to disable. For regular Responses and native compact sends, proven pre-connection DNS/TCP reachability failures are tracked at the provider-host level: they never affect account health, account cooldowns, thread/session affinity, active-account selection, or Pool routing, and never count toward this threshold. | | `upstreamHostCircuitThreshold?` | `number` | `0` | Opt-in circuit threshold for proven pre-connection DNS/TCP failures on native OpenAI forward Responses and compact sends. `0` disables it; `1`–`20` opens a 30-second provider-origin cooldown after that many terminal logical requests. While open, requests receive `503` with `Retry-After` before account selection or upstream send; after cooldown, one half-open request is admitted. Timeouts and HTTP responses never count, and any HTTP response closes the circuit. Applies only to Codex Pool routing with no pinned account; it is inert for `codexAccountMode: "direct"` and account-qualified selectors. | | `maxUpstreamBodyBytes?` | `number` | `0` | Opt-in ceiling, in bytes, on a serialized native Responses **passthrough** body. `0` or omitted disables it — no limit is inferred for any destination. When set, a built body above the ceiling is refused locally before the send: streaming turns receive a terminal `response.failed` / `context_length_exceeded` so the client compacts instead of resending, and non-streaming turns receive a `413` naming the size, the number of embedded `input_image` items, and roughly how many megabytes of image data they represent. Checked at every build and rebuild point, including OAuth-refresh replay and alternate-account retry. Translated adapter paths are not covered. There is deliberately no default: the only measured ceiling here belongs to the WebSocket transport, which already falls back to HTTP for oversized turns, so a default would refuse requests that currently succeed. Set it when your gateway has a known request-size limit and you would rather see an actionable local error than an opaque upstream failure. | -| `maxInboundBodyBytes?` | `number` | `0` | Opt-in ceiling, in bytes, on a decompressed **inbound** data-plane request body — the mirror of `maxUpstreamBodyBytes` above. `0` or omitted keeps the built-in 256 MiB default. Raise it when a large-context session can no longer compact: Codex replays the whole history to the compaction model, so on the 922k-token opt-in window the compaction request is itself the one that crosses the limit, and the session is stuck at the only operation that would have shrunk it. Clamped to 1 MiB–512 MiB. The ceiling is not negotiable: the reader materializes the body several times over (wire bytes, decoded bytes, the decoded string, and the parsed object graph), so peak memory is a multiple of whatever is admitted, and an unbounded value would be a memory exhaustion lever. Applies to `/v1/responses`, `/v1/responses/compact`, `/v1/chat/completions`, and `/v1/messages`. The listener's accept size is fixed when the proxy binds, so a change takes effect on restart. A body above the limit is refused locally with HTTP 413 and `code: "inbound_body_too_large"`, which is deliberately distinct from the `context_length_exceeded` 413 a provider size refusal produces. | +| `maxInboundBodyBytes?` | `number` | `0` | Opt-in ceiling, in bytes, on a decompressed **inbound** data-plane request body — the mirror of `maxUpstreamBodyBytes` above. `0` or omitted keeps the built-in 256 MiB default. Raise it when a large-context session can no longer compact: Codex replays the whole history to the compaction model, so on the 922k-token opt-in window the compaction request is itself the one that crosses the limit, and the session is stuck at the only operation that would have shrunk it. Clamped to 1 MiB–512 MiB. The ceiling is not negotiable: the reader materializes the body several times over (wire bytes, decoded bytes, the decoded string, and the parsed object graph), so peak memory is a multiple of whatever is admitted, and an unbounded value would be a memory exhaustion lever. Applies to `/v1/responses`, `/v1/responses/compact`, `/v1/chat/completions`, and `/v1/messages`; image and search routes retain the built-in limit. Raised-limit readers share a process-wide 512 MiB admission budget, so concurrent excess is refused temporarily with HTTP 503 and `code: "server_busy"`. The listener's accept size is fixed when the proxy binds, so a change takes effect on restart. A body above the limit is refused locally with HTTP 413 and `code: "inbound_body_too_large"`, which is deliberately distinct from the `context_length_exceeded` 413 a provider size refusal produces. | | `modelCacheTtlMs?` | `number` | `300000` | Freshness window for the per-provider `/models` cache. | | `cacheRetention?` | `"none" \| "short" \| "long"` | `"short"` | Anthropic prompt-cache policy: disabled, 5-minute ephemeral, or 1-hour extended. | | `tokenGuardian?` | `OcxTokenGuardianConfig` | off | Optional proactive OAuth refresh and Codex-account warmup policy. | diff --git a/src/server/images.ts b/src/server/images.ts index 5642038474..02e56fcacf 100644 --- a/src/server/images.ts +++ b/src/server/images.ts @@ -29,7 +29,7 @@ import { sidecarEnter } from "../lib/sidecar-tracker"; import type { OcxConfig } from "../types"; import { resolveFirstUsableOpenAiSidecar, selectImagesProvider } from "../providers/openai-sidecar"; import { getProviderRegistryEntry } from "../providers/registry"; -import { readJsonRequestBody, resolveInboundBodyLimitBytes } from "./request-decompress"; +import { readJsonRequestBody } from "./request-decompress"; import { ForwardAdmissionCredentialError, validateForwardAdmissionCredential } from "./auth-cors"; import type { RequestLogContext } from "./request-log"; import { codexLogAccountId, decodeRequestErrorResponse } from "./responses"; @@ -602,7 +602,7 @@ export async function handleImages( ): Promise { let body: unknown; try { - body = await readJsonRequestBody(req, undefined, resolveInboundBodyLimitBytes(config.maxInboundBodyBytes)); + body = await readJsonRequestBody(req); } catch (err) { return decodeRequestErrorResponse(err, "images"); } diff --git a/src/server/request-decompress.ts b/src/server/request-decompress.ts index 1939479429..e2ed2f49c3 100644 --- a/src/server/request-decompress.ts +++ b/src/server/request-decompress.ts @@ -38,6 +38,37 @@ export const MAX_CONFIGURABLE_INBOUND_BODY_BYTES = 512 * 1024 * 1024; /** Floor for the opt-in. Below this an ordinary multi-image turn cannot be admitted at all. */ export const MIN_CONFIGURABLE_INBOUND_BODY_BYTES = 1024 * 1024; +/** + * Process-wide decoded-body admission. Readers reserve their full configured allowance before + * consuming a body, so increasing the per-request limit reduces the number of large bodies that + * can be decoded concurrently instead of multiplying the process's worst-case memory exposure. + */ +export const MAX_CONCURRENT_INBOUND_BODY_BYTES = MAX_CONFIGURABLE_INBOUND_BODY_BYTES; +let reservedInboundBodyBytes = 0; + +export class InboundBodyCapacityError extends Error { + constructor() { + super("inbound request body capacity is temporarily exhausted"); + this.name = "InboundBodyCapacityError"; + } +} + +function reserveInboundBodyCapacity(maxBytes: number): () => void { + // Preserve the established default-cap concurrency. The configurable range above that default + // is the additional high-amplification surface this gate exists to serialize. + if (maxBytes <= MAX_DECOMPRESSED_BODY_BYTES) return () => {}; + if (maxBytes > MAX_CONCURRENT_INBOUND_BODY_BYTES - reservedInboundBodyBytes) { + throw new InboundBodyCapacityError(); + } + reservedInboundBodyBytes += maxBytes; + let active = true; + return () => { + if (!active) return; + active = false; + reservedInboundBodyBytes -= maxBytes; + }; +} + /** * Resolve the configured inbound admission limit, clamped to the supported range. * @@ -298,43 +329,52 @@ export async function readBoundedJsonRequestBody( budget?: TranslatorBudget, options?: { emptyBodyFallback?: unknown; signal?: AbortSignal }, ): Promise { - const encoding = req.headers.get("content-encoding"); - const declaredLength = declaredBodyLength(req); - // Reject an honest oversized declaration before reading. Missing, malformed, - // and dishonest declarations remain bounded by the streaming reader below. - if (declaredLength !== null && declaredLength > maxBytes) { - const error = new DecompressedBodyTooLargeError(declaredLength, maxBytes, "declared_wire"); - cancelStreamWithoutWaiting(req.body, error); - throw error; - } - const releaseReservation = budget && declaredLength !== null && declaredLength > 0 - ? budget.observeAcceptedRequestCopy(declaredLength) - : undefined; - let raw: Uint8Array; + const releaseInboundCapacity = reserveInboundBodyCapacity(maxBytes); try { - raw = await readRequestBodyBytesCapped(req.body, maxBytes, options?.signal ?? req.signal); - } finally { - releaseReservation?.(); - } - assertBodySizeWithinLimit(raw, maxBytes, "observed_wire_lower_bound"); - const releaseRaw = budget?.observeAcceptedRequestCopy(raw.byteLength); - let releaseDecoded: (() => void) | undefined; - let releaseText: (() => void) | undefined; - try { - const decoded = decodeRequestBody(raw, encoding, maxBytes); - releaseDecoded = decoded === raw ? undefined : budget?.observeAcceptedRequestCopy(decoded.byteLength); - const text = new TextDecoder().decode(decoded); - releaseText = budget?.observeAcceptedRequestCopy(new TextEncoder().encode(text).byteLength); - if (options && "emptyBodyFallback" in options && text.trim() === "") { - return options.emptyBodyFallback; + const encoding = req.headers.get("content-encoding"); + const declaredLength = declaredBodyLength(req); + // Reject an honest oversized declaration before reading. Missing, malformed, + // and dishonest declarations remain bounded by the streaming reader below. + if (declaredLength !== null && declaredLength > maxBytes) { + const error = new DecompressedBodyTooLargeError(declaredLength, maxBytes, "declared_wire"); + cancelStreamWithoutWaiting(req.body, error); + throw error; + } + const releaseReservation = budget && declaredLength !== null && declaredLength > 0 + ? budget.observeAcceptedRequestCopy(declaredLength) + : undefined; + let raw: Uint8Array; + try { + raw = await readRequestBodyBytesCapped(req.body, maxBytes, options?.signal ?? req.signal); + } finally { + releaseReservation?.(); + } + assertBodySizeWithinLimit(raw, maxBytes, "observed_wire_lower_bound"); + const releaseRaw = budget?.observeAcceptedRequestCopy(raw.byteLength); + let releaseDecoded: (() => void) | undefined; + let releaseText: (() => void) | undefined; + try { + const decoded = decodeRequestBody(raw, encoding, maxBytes); + releaseDecoded = decoded === raw ? undefined : budget?.observeAcceptedRequestCopy(decoded.byteLength); + const text = new TextDecoder().decode(decoded); + // The source bytes are already the exact UTF-8 representation consumed by TextDecoder. + // Re-encoding a potentially huge string just to measure it creates another body-sized copy. + releaseText = budget?.observeAcceptedRequestCopy(decoded.byteLength); + if (options && "emptyBodyFallback" in options && text.trim() === "") { + return options.emptyBodyFallback; + } + const parsed = JSON.parse(text); + // Use the input size as the parsed graph's conservative accounting proxy. Serializing the + // graph again doubled the hottest allocation solely for observational telemetry. + budget?.observeAcceptedRequestCopy(decoded.byteLength); + return parsed; + } finally { + releaseText?.(); + releaseDecoded?.(); + releaseRaw?.(); } - const parsed = JSON.parse(text); - budget?.observeAcceptedRequestCopy(new TextEncoder().encode(JSON.stringify(parsed)).byteLength); - return parsed; } finally { - releaseText?.(); - releaseDecoded?.(); - releaseRaw?.(); + releaseInboundCapacity(); } } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index c1ce136ca4..b3bf6b8372 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -253,6 +253,7 @@ import { describeInboundBodyRefusal, resolveInboundBodyLimitBytes, DecompressedBodyTooLargeError, + InboundBodyCapacityError, UnsupportedContentEncodingError, } from "../request-decompress"; import { resolveAdapter, resolveWireProtocolOverride } from "../adapter-resolve"; @@ -1610,6 +1611,9 @@ export function codexForwardTerminalOutcomeRecorder( export function decodeRequestErrorResponse(err: unknown, label: string): Response { + if (err instanceof InboundBodyCapacityError) { + return formatErrorResponse(503, "server_error", err.message, { code: "server_busy" }); + } if (isTranslatorBudgetExceededError(err)) { return formatErrorResponse(413, "request_too_large", "request translation buffer exceeded the safe limit", { code: "translation_buffer_limit", diff --git a/src/server/search.ts b/src/server/search.ts index 808bcd86c6..e09d15fc25 100644 --- a/src/server/search.ts +++ b/src/server/search.ts @@ -33,7 +33,7 @@ import { type ExactOpenAiSidecarAccount, } from "../providers/openai-sidecar"; import { routeModel } from "../router"; -import { readJsonRequestBody, resolveInboundBodyLimitBytes } from "./request-decompress"; +import { readJsonRequestBody } from "./request-decompress"; import { ForwardAdmissionCredentialError, validateForwardAdmissionCredential } from "./auth-cors"; import type { RequestLogContext } from "./request-log"; import { codexLogAccountId, decodeRequestErrorResponse } from "./responses"; @@ -64,7 +64,7 @@ export async function handleSearch( } let body: unknown; try { - body = await readJsonRequestBody(req, undefined, resolveInboundBodyLimitBytes(config.maxInboundBodyBytes)); + body = await readJsonRequestBody(req); } catch (err) { return decodeRequestErrorResponse(err, "search"); } diff --git a/tests/usage/request-decompress.test.ts b/tests/usage/request-decompress.test.ts index cb88dd29e1..67b9f18858 100644 --- a/tests/usage/request-decompress.test.ts +++ b/tests/usage/request-decompress.test.ts @@ -6,6 +6,7 @@ import { describeInboundBodyRefusal, MAX_DECOMPRESSED_BODY_BYTES, MAX_CONFIGURABLE_INBOUND_BODY_BYTES, + InboundBodyCapacityError, MIN_CONFIGURABLE_INBOUND_BODY_BYTES, readBoundedJsonRequestBody, readJsonRequestBody, @@ -221,6 +222,31 @@ describe("decodeRequestBody", () => { }); describe("configurable inbound body limit (Issue #3573)", () => { + test("large readers share a process-wide decoded-byte admission budget", async () => { + const controller = new AbortController(); + const stalled = new Request("http://localhost/v1/responses", { + method: "POST", + body: new ReadableStream({ pull() {} }), + signal: controller.signal, + // Required by Node's Request typing for a streaming body; Bun accepts the same shape. + duplex: "half", + } as RequestInit); + const largeLimit = 300 * 1024 * 1024; + const first = readBoundedJsonRequestBody(stalled, largeLimit); + + await expect(readBoundedJsonRequestBody( + new Request("http://localhost/v1/responses", { method: "POST", body: "{}" }), + largeLimit, + )).rejects.toBeInstanceOf(InboundBodyCapacityError); + + controller.abort(new DOMException("test cleanup", "AbortError")); + await expect(first).rejects.toMatchObject({ name: "AbortError" }); + await expect(readBoundedJsonRequestBody( + new Request("http://localhost/v1/responses", { method: "POST", body: "{}" }), + largeLimit, + )).resolves.toEqual({}); + }); + test("an unconfigured proxy keeps the 256 MiB default", () => { expect(resolveInboundBodyLimitBytes(undefined)).toBe(MAX_DECOMPRESSED_BODY_BYTES); expect(resolveInboundBodyLimitBytes(0)).toBe(MAX_DECOMPRESSED_BODY_BYTES);