[WRONG BRANCH] fix(server): bound raised inbound body concurrency

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. |

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<Response> {
   let body: unknown;
   try {
-    body = await readJsonRequestBody(req, undefined, resolveInboundBodyLimitBytes(config.maxInboundBodyBytes));
+    body = await readJsonRequestBody(req);
   } catch (err) {
     return decodeRequestErrorResponse(err, "images");
   }

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<unknown> {
-  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();
   }
 }
 

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",

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");
   }

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<Uint8Array>({ 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);