fix(resilience): derive the chat_admission_busy Retry-After from observed lease occupancy

changelog.d/fixes/12395-heavy-admission-retry-after.md

@@ -0,0 +1 @@
+- **fix(chat-admission):** derive the `chat_admission_busy` 503 `Retry-After` from observed heavyweight-lease occupancy — the larger of the exhausted `OMNIROUTE_CHAT_ADMISSION_QUEUE_MS` window and the time since capacity last turned over, capped at 60 s — instead of a fixed 1 s (structural) / 2 s (byte-stage) hint that invited Codex/agent fan-out clients to re-send ~1 MiB `/v1/responses` bodies every second into a gate held for the whole SSE lifetime; an idle gate keeps the historical floors ([#12135](https://github.com/diegosouzapw/OmniRoute/issues/12135)) (#12395 — thanks @pacocartones)

docs/guides/TROUBLESHOOTING.md

@@ -538,8 +538,12 @@ When many concurrent requests hit a rate-limited provider, OmniRoute uses mutex
 
 - The chat completions endpoint returns a retryable `503` response whose error code is
   `chat_admission_busy`.
-- The response includes `Retry-After`; the byte-based path uses 2 seconds, while the
-  structure-based path uses 1 second and includes `reason: "structure_limit"`.
+- The response includes `Retry-After`. Since #12135 the value is derived from observed
+  occupancy — the larger of the `OMNIROUTE_CHAT_ADMISSION_QUEUE_MS` window the request already
+  waited and the time the current heavyweight leases have been held — rounded up to whole
+  seconds and capped at 60. On an idle gate it keeps the historical floors: 2 seconds on the
+  byte-based path, 1 second on the structure-based path (which also includes
+  `reason: "structure_limit"`).
 - This can happen while another heavyweight chat or long-running streaming response is still
   in flight.
 

src/shared/middleware/chatAdmissionResponses.ts

@@ -4,10 +4,34 @@ import { CORS_HEADERS } from "../utils/cors";
 
 const JSON_HEADERS = { ...CORS_HEADERS, "Content-Type": "application/json" };
 
-export function chatAdmissionRejectionResponse(status: 413 | 503, hardMaxBytes: number): Response {
+/**
+ * `Retry-After` floors for the retryable 503s — the pre-#12135 fixed values. A caller
+ * passes an occupancy-derived hint (`ChatAdmissionController#retryAfterSeconds`) and the
+ * header carries whichever is larger, so an idle gate still answers exactly as before
+ * while a gate whose leases have been busy for a whole SSE stream stops inviting a
+ * 1-second retry storm.
+ */
+const BYTE_STAGE_RETRY_AFTER_FLOOR_SECONDS = 2;
+const STRUCTURAL_RETRY_AFTER_FLOOR_SECONDS = 1;
+
+function retryAfterHeader(floorSeconds: number, hintSeconds: number | undefined): string {
+  const hint = Number.isFinite(hintSeconds) ? Math.ceil(hintSeconds as number) : 0;
+  return String(Math.max(floorSeconds, hint));
+}
+
+export function chatAdmissionRejectionResponse(
+  status: 413 | 503,
+  hardMaxBytes: number,
+  retryAfterSeconds?: number
+): Response {
   const isPayload = status === 413;
   const headers: Record<string, string> = { ...JSON_HEADERS };
-  if (!isPayload) headers["Retry-After"] = "2";
+  if (!isPayload) {
+    headers["Retry-After"] = retryAfterHeader(
+      BYTE_STAGE_RETRY_AFTER_FLOOR_SECONDS,
+      retryAfterSeconds
+    );
+  }
   const message = isPayload
     ? `Request body too large for chat completions (max ${Math.floor(
         hardMaxBytes / (1024 * 1024)
@@ -53,10 +77,19 @@ export function resourcePressureRejectionResponse(): Response {
   );
 }
 
-export function structuralRejectionResponse(status: 413 | 503, maxMessages: number): Response {
+export function structuralRejectionResponse(
+  status: 413 | 503,
+  maxMessages: number,
+  retryAfterSeconds?: number
+): Response {
   const historyLimit = status === 413;
   const headers: Record<string, string> = { ...JSON_HEADERS };
-  if (!historyLimit) headers["Retry-After"] = "1";
+  if (!historyLimit) {
+    headers["Retry-After"] = retryAfterHeader(
+      STRUCTURAL_RETRY_AFTER_FLOOR_SECONDS,
+      retryAfterSeconds
+    );
+  }
   const body = buildErrorBody(
     status,
     historyLimit

src/shared/middleware/chatBodyAdmission.ts

@@ -92,6 +92,15 @@ export const CHAT_ADMISSION_MAX_QUEUED_BYTES = parsePositiveInt(
   4 * 1024 * 1024
 );
 
+/**
+ * Ceiling for the occupancy-derived `Retry-After` on a capacity 503 (#12135). A
+ * heavyweight lease is held for the whole SSE lifetime, so the hint is derived from how
+ * long capacity has demonstrably been busy (`ChatAdmissionController#retryAfterSeconds`);
+ * this cap keeps a multi-minute stream from telling a client to sleep for minutes when
+ * another slot may free far sooner.
+ */
+export const CHAT_ADMISSION_RETRY_AFTER_MAX_SECONDS = 60;
+
 export const CHAT_HEAVY_MESSAGE_COUNT = parsePositiveInt(
   process.env.OMNIROUTE_CHAT_HEAVY_MESSAGE_COUNT,
   200
@@ -260,6 +269,9 @@ export class ChatAdmissionController {
    * `CHAT_MAX_HEAVY_IN_FLIGHT` bound, but still a real, finite ceiling instead of
    * the unconditional bypass this replaces. */
   #activeHealthy = 0;
+  /** #12135: acquisition time of every live heavy lease, keyed by an opaque token, so the
+   * capacity 503 can advertise a `Retry-After` derived from observed occupancy. */
+  #heavyLeaseStartedAt = new Map<symbol, number>();
   /** Per-key FIFOs. A key groups one client's waiters so they are served
    * round-robin against the shared budget instead of monopolizing a strict
    * FIFO (see #dispatchFair). */
@@ -397,6 +409,8 @@ export class ChatAdmissionController {
   tryAcquireHeavy(): ChatAdmissionLease | null {
     if (this.#activeHeavy >= this.maxHeavyInFlight) return null;
     this.#activeHeavy += 1;
+    const token = Symbol("heavy-lease");
+    this.#heavyLeaseStartedAt.set(token, Date.now());
     const done = trackRequest();
     let released = false;
     return {
@@ -407,12 +421,39 @@ export class ChatAdmissionController {
         if (released) return;
         released = true;
         this.#activeHeavy = Math.max(0, this.#activeHeavy - 1);
+        this.#heavyLeaseStartedAt.delete(token);
         done();
         this.#dispatchFair();
       },
     };
   }
 
+  /**
+   * `Retry-After` (whole seconds) for a capacity 503, derived from live occupancy instead
+   * of a fixed constant (#12135). A heavyweight lease is held for the ENTIRE SSE lifetime
+   * (tens of seconds to minutes), so a fixed 1–2 s hint invited clients to re-send the
+   * same ~1 MiB body every second into a gate that could not possibly have cleared. The
+   * hint is the larger of:
+   *  - `queueMs`, the bounded wait the caller already exhausted — the server itself needed
+   *    longer than that, so advertising less is dishonest; and
+   *  - the age of the YOUNGEST live heavy lease: the time since heavyweight capacity last
+   *    turned over. Every slot has been continuously held at least that long, so it is the
+   *    observed floor on how long "busy" has lasted (the oldest lease would be a pessimist
+   *    with N slots in flight).
+   * Rounded up and capped at `CHAT_ADMISSION_RETRY_AFTER_MAX_SECONDS`. The response
+   * builders floor the result at their historical value (1 s structural, 2 s byte-stage),
+   * so an idle gate answers exactly as before.
+   */
+  retryAfterSeconds(queueMs: number, now = Date.now()): number {
+    let youngestAgeMs = Number.POSITIVE_INFINITY;
+    for (const startedAt of this.#heavyLeaseStartedAt.values()) {
+      youngestAgeMs = Math.min(youngestAgeMs, now - startedAt);
+    }
+    const occupancyMs = Number.isFinite(youngestAgeMs) ? youngestAgeMs : 0;
+    const hintSeconds = Math.ceil(Math.max(0, queueMs, occupancyMs) / 1000);
+    return Math.min(CHAT_ADMISSION_RETRY_AFTER_MAX_SECONDS, Math.max(1, hintSeconds));
+  }
+
   /**
    * Wait up to `timeoutMs` for heavyweight capacity, retrying atomically on each
    * release. Resolves `null` when the deadline expires with no capacity freed, in
@@ -843,26 +884,41 @@ export async function admitChatStructure(
   // Structural-only waits happen on byte-light bodies (a byte-heavy body already
   // holds the byte-stage lease), so the conservative 256KB weight bounds the
   // parsed JSON the waiter keeps resident while parked.
+  const queueMs = options.queueMs ?? 0;
   const acquiredCount = await controller.acquireHeavyWithin(
-    options.queueMs ?? 0,
+    queueMs,
     options.signal,
     CHAT_LARGE_BODY_BYTES,
     options.sessionId
   );
   if (!acquiredCount) {
-    return { admit: false, response: structuralRejectionResponse(503, maxMessages) };
+    return {
+      admit: false,
+      response: structuralRejectionResponse(
+        503,
+        maxMessages,
+        controller.retryAfterSeconds(queueMs)
+      ),
+    };
   }
 
   // #503-fanout: same composed count+budget gate as the fast path above.
   const acquiredBudget = await controller.acquireBudgetWithin(
     CHAT_LARGE_BODY_BYTES,
-    options.queueMs ?? 0,
+    queueMs,
     options.signal,
     options.sessionId
   );
   if (acquiredBudget.status !== "acquired") {
     acquiredCount.release();
-    return { admit: false, response: structuralRejectionResponse(503, maxMessages) };
+    return {
+      admit: false,
+      response: structuralRejectionResponse(
+        503,
+        maxMessages,
+        controller.retryAfterSeconds(queueMs)
+      ),
+    };
   }
   return {
     admit: true,
@@ -1006,14 +1062,18 @@ export async function admitChatRequest(
     return true;
   };
 
+  // #12135: the capacity 503 advertises an occupancy-derived Retry-After.
+  const busyResponse = () =>
+    chatAdmissionRejectionResponse(503, hardMaxBytes, controller.retryAfterSeconds(queueMs));
+
   // A known-large declaration can reserve before ingestion. Unknown lengths are boundedly
   // sniffed below; this avoids consuming scarce heavyweight capacity for small chunked bodies.
   if (
     contentLength !== null &&
     contentLength >= largeBodyBytes &&
     !(await reserve(Math.min(contentLength, hardMaxBytes)))
   ) {
-    return { admit: false, response: chatAdmissionRejectionResponse(503, hardMaxBytes) };
+    return { admit: false, response: busyResponse() };
   }
 
   const reader = request.body?.getReader();
@@ -1039,7 +1099,7 @@ export async function admitChatRequest(
       }
       if (totalBytes >= largeBodyBytes && !(await reserve(totalBytes))) {
         await reader.cancel("chat admission capacity unavailable").catch(() => undefined);
-        return { admit: false, response: chatAdmissionRejectionResponse(503, hardMaxBytes) };
+        return { admit: false, response: busyResponse() };
       }
       chunks.push(value);
     }