fix(resilience): stop combo fallback on local process pressure

changelog.d/fixes/19124-terminal-local-process-pressure.md

@@ -0,0 +1 @@
+- **fix(resilience):** Stop combo retries and fallback on process-wide `resource_pressure` / `heap_pressure` 503s, preserving the original retry hint and healthy account affinity. Cancel in-flight hedges and release their delay timers. (#19124 — thanks @aldoeliacim)

open-sse/services/combo/comboAttemptLoop.ts

@@ -213,6 +213,10 @@ export type DispatchWithCooldownRetryExtra = {
   compatRejectedTargets: ResolvedComboTarget[];
 };
 
+/**
+ * Dispatches ordered targets with bounded set and cooldown retries, returning the first
+ * success or terminal response, or an aggregate failure when eligible targets are exhausted.
+ */
 export async function dispatchWithCooldownRetry(opts: {
   state: AttemptLoopState;
   deps: AttemptLoopDeps;
@@ -468,10 +472,15 @@ export async function dispatchWithCooldownRetry(opts: {
           dispatchPosition + 1 < dispatchIndices.length
         ) {
           const hedgeDelay = resolveDelayMs(deps.config.hedgeDelayMs, 500);
+          let hedgeTimer: ReturnType<typeof setTimeout> | undefined;
           const timeoutPromise = new Promise<void>((r) => {
-            setTimeout(r, hedgeDelay);
+            hedgeTimer = setTimeout(r, hedgeDelay);
           });
-          await Promise.race([task, globalPromise, timeoutPromise, loopSafetyPromise]);
+          try {
+            await Promise.race([task, globalPromise, timeoutPromise, loopSafetyPromise]);
+          } finally {
+            clearTimeout(hedgeTimer);
+          }
         } else {
           await Promise.race([task, globalPromise, loopSafetyPromise]);
         }
@@ -711,6 +720,7 @@ export async function dispatchWithCooldownRetry(opts: {
               }
             );
             if (compatFallbackResult) {
+              if (!compatFallbackResult.ok) return compatFallbackResult;
               recordComboRequest(deps.combo.name, null, {
                 success: true,
                 latencyMs: Date.now() - deps.startTime,

open-sse/services/combo/comboCompatFallback.ts

@@ -1,3 +1,5 @@
+import { readAttemptFailure } from "./attemptFailure.ts";
+import { isLocalProcessPressureResponse } from "./comboPredicates.ts";
 import type {
   ComboLogger,
   HandleSingleModel,
@@ -37,6 +39,7 @@ export interface CompatFallbackContext {
  * upstream `Response`. Targets that are unavailable, in cooldown, or return a non-ok
  * response are skipped. Returns `null` when no rejected target yields a success, so the
  * caller can fall through to its existing error/503 path with an unchanged error surface.
+ * Process-wide pressure instead returns its original terminal response, never a success.
  */
 export async function attemptCompatRejectedFallback(
   rejectedTargets: ResolvedComboTarget[],
@@ -75,6 +78,11 @@ export async function attemptCompatRejectedFallback(
       effectiveComboStrategy: ctx.strategy,
       fallbackAttempts: i,
     });
+    if (
+      result.status === 503 &&
+      isLocalProcessPressureResponse(result, (await readAttemptFailure(result)).errorBody)
+    )
+      return result;
     if (result.ok) {
       ctx.log.info("COMBO", `Last-resort compat fallback succeeded via ${target.modelStr}`);
       return result;

open-sse/services/combo/comboPredicates.ts

@@ -6,6 +6,7 @@
  * predicates are re-exported from combo.ts for backward compatibility.
  */
 
+import { isMarkedLocalProcessPressureResponse } from "../../utils/localPressureResponse.ts";
 import {
   EXECUTOR_CONTRACT_VIOLATION_CODE,
   LOCAL_CONFIGURATION_ERROR_CODES,
@@ -400,6 +401,17 @@ export function isAccountSemaphoreCapacityFailure(error?: {
   );
 }
 
+/** Process-wide shedding cannot recover by retrying another combo target. Only a shed this
+ * process created is terminal: an upstream can return the same 503 body and keeps fallback. */
+export function isLocalProcessPressureResponse(response: Response, body: unknown): boolean {
+  if (response.status !== 503 || !isMarkedLocalProcessPressureResponse(response)) return false;
+  if (!body || typeof body !== "object") return false;
+  const error = (body as Record<string, unknown>).error;
+  if (!error || typeof error !== "object") return false;
+  const code = (error as Record<string, unknown>).code;
+  return code === "resource_pressure" || code === "heap_pressure";
+}
+
 export function isComboTargetTimeoutFailure(error?: {
   code?: string | null;
   type?: string | null;

open-sse/services/combo/dispatchPrelude.ts

@@ -24,7 +24,13 @@ import { errorResponse, errorResponseWithComboDiagnostics } from "../../utils/er
 import { parseModel } from "../model.ts";
 import { handlePipelineChat, type PipelineStep } from "../pipeline.ts";
 import type { resolveComboSetupConfig } from "../comboConfig.ts";
-import { clampComboDepth, clampGlobalAttempts, resolveDelayMs } from "./comboPredicates.ts";
+import {
+  clampComboDepth,
+  clampGlobalAttempts,
+  resolveDelayMs,
+  isLocalProcessPressureResponse,
+} from "./comboPredicates.ts";
+import { readAttemptFailure } from "./attemptFailure.ts";
 import {
   deriveRequestCompatibilityRequirements,
   isVisionIncompatibleTarget,
@@ -285,6 +291,11 @@ async function evaluatePinnedResponse(args: {
     );
     return null;
   }
+  if (
+    pinnedResult.status === 503 &&
+    isLocalProcessPressureResponse(pinnedResult, (await readAttemptFailure(pinnedResult)).errorBody)
+  )
+    return pinnedResult;
   const pinnedStatus = pinnedResult.status || 500;
   if (![401, 408, 429, 500, 502, 503, 504].includes(pinnedStatus)) {
     return pinnedResult;

open-sse/services/combo/executeTargetAttempt.ts

@@ -64,6 +64,7 @@ import {
   isHedgeCancelledSentinel,
   isAccountSemaphoreCapacityFailure,
   isLocalQueueCapacityErrorBody,
+  isLocalProcessPressureResponse,
   toRecordedTarget,
   resolveDelayMs,
   isModelScoped400,
@@ -123,6 +124,10 @@ import {
 import { isEmptyStreamSiblingRecoveryEnabled } from "../streamRecovery.ts";
 import { waitForCooldownAwareRetry } from "../../../src/sse/services/cooldownAwareRetry.ts";
 
+/**
+ * Executes an admitted target with configured retries and response validation. Returns a
+ * success or terminal response, or a nonterminal result when the loop may try another target.
+ */
 export async function executeTargetAttempt(
   opts: ExecuteTargetAttemptOptions
 ): Promise<ExecuteTargetResult> {
@@ -779,6 +784,11 @@ export async function executeTargetAttempt(
 
     const failure = await readAttemptFailure(result, { log: deps.log, modelStr });
     const { errorText, errorBody, structuredError } = failure;
+    // Return the original verdict before retry hints, health or affinity mutations.
+    // The attempt loop treats this as fatal and cancels already-running hedges.
+    if (isLocalProcessPressureResponse(result, errorBody)) {
+      return { ok: false, response: result };
+    }
     const retryAfter = resolveAttemptRetryAfter(failure);
 
     // Track earliest retryAfter

open-sse/services/combo/roundRobinCombo.ts

@@ -111,6 +111,7 @@ import {
   isLocalKeyPolicyBreachErrorBody,
   isAccountSemaphoreCapacityFailure,
   isLocalQueueCapacityErrorBody,
+  isLocalProcessPressureResponse,
   isProviderCircuitOpenResult,
   toRecordedTarget,
   getExhaustedTargetSkipReason,
@@ -1112,6 +1113,16 @@ export async function handleRoundRobinCombo({
           } catch {
             logRetryHintUnreadable(log, "COMBO-RR", modelStr, result.status, "clone failed");
           }
+          // Process-wide shedding is terminal, not a target failure. Preserve the
+          // original body/headers and leave health, semaphore cooldown and pins alone.
+          if (isLocalProcessPressureResponse(result, errorBody)) {
+            rrEvents.failed(
+              errorText || `HTTP ${result.status}`,
+              Date.now() - startTime,
+              selectedConnectionIdOf(result)
+            );
+            return result;
+          }
           retryAfter ||= readProseRetryAfter(bodyText); // #13672 opt-in prose hints
 
           if (result.status === 499) {
@@ -1453,6 +1464,7 @@ export async function handleRoundRobinCombo({
       strategy: "round-robin",
     });
     if (compatFallbackResult) {
+      if (!compatFallbackResult.ok) return compatFallbackResult;
       recordComboRequest(combo.name, null, {
         success: true,
         latencyMs: Date.now() - startTime,

open-sse/services/combo/runtimeUnits.ts

@@ -12,7 +12,12 @@ import { waitForCooldownAwareRetry } from "../../../src/sse/services/cooldownAwa
 import { buildContextBudgetResponse, readBudgetFailure } from "./budgetExhaustion.ts";
 import type { ComboDiagnostics } from "../../utils/error.ts";
 import { recordComboRequest } from "../comboMetrics.ts";
-import { resolveDelayMs, requestScopedReplayKey } from "./comboPredicates.ts";
+import {
+  resolveDelayMs,
+  requestScopedReplayKey,
+  isLocalProcessPressureResponse,
+} from "./comboPredicates.ts";
+import { readAttemptFailure } from "./attemptFailure.ts";
 import {
   isProviderSafetyStreamFailure,
   qualityValidationFailure,
@@ -193,6 +198,10 @@ function orderUnitsForStrategy(strategy: string, units: ResolvedComboUnit[]): Re
   return units;
 }
 
+/**
+ * Runs strategy-ordered model and nested-combo units within the shared attempt budget.
+ * Returns the serving or terminal unit and response, or a final failure with no unit.
+ */
 export async function executeRuntimeUnitCombo(args: {
   body: Record<string, unknown>;
   combo: ComboLike;
@@ -359,6 +368,13 @@ export async function executeRuntimeUnitCombo(args: {
         });
         return { response, unit };
       }
+      if (
+        response.status === 503 &&
+        isLocalProcessPressureResponse(response, (await readAttemptFailure(response)).errorBody)
+      ) {
+        // Do not replay a shed child or replace its response with an aggregate.
+        return { response, unit };
+      }
       lastResponse = response;
       if (response.ok) {
         if (unit.kind === "combo-ref") {

open-sse/utils/heapPressure.ts

@@ -1,3 +1,4 @@
+import { markLocalProcessPressureResponse } from "./localPressureResponse.ts";
 import v8 from "node:v8";
 
 /**
@@ -71,11 +72,13 @@ export function checkHeapPressureGuard(
     success: false,
     status: 503,
     error: HEAP_PRESSURE_MESSAGE,
-    response: new Response(
-      JSON.stringify({
-        error: { message: HEAP_PRESSURE_MESSAGE, type: "server_error", code: "heap_pressure" },
-      }),
-      { status: 503, headers: { "Content-Type": "application/json", "Retry-After": "5" } }
+    response: markLocalProcessPressureResponse(
+      new Response(
+        JSON.stringify({
+          error: { message: HEAP_PRESSURE_MESSAGE, type: "server_error", code: "heap_pressure" },
+        }),
+        { status: 503, headers: { "Content-Type": "application/json", "Retry-After": "5" } }
+      )
     ),
   };
 }

open-sse/utils/localPressureResponse.ts

@@ -0,0 +1,27 @@
+/**
+ * Provenance of process-pressure sheds.
+ *
+ * A `resource_pressure` / `heap_pressure` 503 means this process is shedding load from its own
+ * V8/cgroup state, so retrying another combo target cannot help. The same status and body can
+ * also arrive from an upstream (a chained gateway), and that one must keep ordinary fallback.
+ * Only a response this process built is branded; the WeakSet identity cannot be forged by an
+ * upstream body or header. Internal wrappers that must allocate a new Response inherit the brand.
+ */
+const localPressureResponses = new WeakSet<Response>();
+
+/** Brand a 503 created by one of this process's pressure guards. */
+export function markLocalProcessPressureResponse(response: Response): Response {
+  localPressureResponses.add(response);
+  return response;
+}
+
+/** True only for a pressure shed this process created (or an internal wrapper of one). */
+export function isMarkedLocalProcessPressureResponse(response: Response): boolean {
+  return localPressureResponses.has(response);
+}
+
+/** Preserve the brand when an internal response wrapper must allocate. */
+export function inheritLocalProcessPressureResponse(source: Response, target: Response): Response {
+  if (localPressureResponses.has(source)) localPressureResponses.add(target);
+  return target;
+}

open-sse/utils/resourcePressure.ts

@@ -1,3 +1,4 @@
+import { markLocalProcessPressureResponse } from "./localPressureResponse.ts";
 import { checkHeapPressureGuard, HEAP_PRESSURE_THRESHOLD_MB } from "./heapPressure.ts";
 import { buildErrorBody } from "./error.ts";
 import {
@@ -195,6 +196,10 @@ function describeCachedPressure(params: {
   };
 }
 
+/**
+ * Logs critical local pressure and returns a branded 503 guard result with a retry hint
+ * for the supplied pressure reason and optional diagnostic details.
+ */
 function buildCriticalGuard(
   reason: PressureReason,
   detail: Record<string, number | string | null | undefined> = {}
@@ -207,17 +212,19 @@ function buildCriticalGuard(
     success: false,
     status: 503,
     error: PRESSURE_MESSAGE,
-    response: new Response(
-      JSON.stringify(
-        buildErrorBody(503, PRESSURE_MESSAGE, undefined, {
-          type: "server_error",
-          code: "resource_pressure",
-        })
-      ),
-      {
-        status: 503,
-        headers: { "Content-Type": "application/json", "Retry-After": RETRY_AFTER_SECONDS },
-      }
+    response: markLocalProcessPressureResponse(
+      new Response(
+        JSON.stringify(
+          buildErrorBody(503, PRESSURE_MESSAGE, undefined, {
+            type: "server_error",
+            code: "resource_pressure",
+          })
+        ),
+        {
+          status: 503,
+          headers: { "Content-Type": "application/json", "Retry-After": RETRY_AFTER_SECONDS },
+        }
+      )
     ),
   };
 }

src/sse/handlers/chat.ts

@@ -57,6 +57,7 @@ import type { CompressionExclusions } from "@omniroute/open-sse/services/compres
 import { resolveComboConfig } from "@omniroute/open-sse/services/comboConfig.ts";
 import { injectHandoffIntoBody } from "@omniroute/open-sse/services/contextHandoff.ts";
 import { runWithTransientBackendRetry } from "@omniroute/open-sse/services/transientBackendRetry.ts";
+import { isMarkedLocalProcessPressureResponse } from "@omniroute/open-sse/utils/localPressureResponse.ts";
 import {
   HTTP_STATUS,
   ANTIGRAVITY_PRE_RESPONSE_TIMEOUT_CODE,
@@ -1308,10 +1309,12 @@ async function handleChatImplementation(
     comboPreselectedCredentials.clear();
 
     // ── Global Fallback Provider (#689) ────────────────────────────────────
-    // If combo exhausted all models, try the global fallback before giving up.
+    // If combo exhausted all models, try the global fallback before giving up. A local pressure
+    // shed did not exhaust anything: the fallback would only be shed again, with retries.
     if (
       !response.ok &&
       [502, 503].includes(response.status) &&
+      !isMarkedLocalProcessPressureResponse(response) &&
       typeof (settings as any)?.globalFallbackModel === "string" &&
       (settings as any).globalFallbackModel.trim()
     ) {

src/sse/handlers/chat/selectedConnectionHeader.ts

@@ -1,6 +1,11 @@
+import { inheritLocalProcessPressureResponse } from "@omniroute/open-sse/utils/localPressureResponse.ts";
 import { inheritTrustedLocalRateLimitResponse } from "@omniroute/open-sse/services/rateLimitManager/errors.ts";
 import { inheritResponsePolicies } from "./responsePolicies.ts";
 
+/**
+ * Returns the response with its selected connection ID, or unchanged when no ID is given.
+ * Immutable headers require a wrapper that preserves trusted markers and response policies.
+ */
 export function withSelectedConnectionHeader(
   response: Response,
   connectionId: string | null | undefined
@@ -20,7 +25,10 @@ export function withSelectedConnectionHeader(
     // A cloned response must keep the trusted local rate-limit marker plus the per-Response
     // routing policies of the original (#15594 provider-probe marker, #16072 empty-turn
     // execution record), or the probe fence / empty-turn guard loses track of it.
-    const trusted = inheritTrustedLocalRateLimitResponse(response, cloned);
+    const trusted = inheritLocalProcessPressureResponse(
+      response,
+      inheritTrustedLocalRateLimitResponse(response, cloned)
+    );
     return inheritResponsePolicies(response, trusted);
   }
 }