diff --git a/changelog.d/fixes/19124-terminal-local-process-pressure.md b/changelog.d/fixes/19124-terminal-local-process-pressure.md new file mode 100644 index 000000000000..ba1dba85e437 --- /dev/null +++ b/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) diff --git a/open-sse/services/combo/comboAttemptLoop.ts b/open-sse/services/combo/comboAttemptLoop.ts index 0137876b6715..6fcb9d590c94 100644 --- a/open-sse/services/combo/comboAttemptLoop.ts +++ b/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 | undefined; const timeoutPromise = new Promise((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, diff --git a/open-sse/services/combo/comboCompatFallback.ts b/open-sse/services/combo/comboCompatFallback.ts index a614572b6a10..62df6a570f4c 100644 --- a/open-sse/services/combo/comboCompatFallback.ts +++ b/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; diff --git a/open-sse/services/combo/comboPredicates.ts b/open-sse/services/combo/comboPredicates.ts index b602305b4dd5..b2875de0968d 100644 --- a/open-sse/services/combo/comboPredicates.ts +++ b/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).error; + if (!error || typeof error !== "object") return false; + const code = (error as Record).code; + return code === "resource_pressure" || code === "heap_pressure"; +} + export function isComboTargetTimeoutFailure(error?: { code?: string | null; type?: string | null; diff --git a/open-sse/services/combo/dispatchPrelude.ts b/open-sse/services/combo/dispatchPrelude.ts index c657f0cc619e..0e0ff34785b2 100644 --- a/open-sse/services/combo/dispatchPrelude.ts +++ b/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; diff --git a/open-sse/services/combo/executeTargetAttempt.ts b/open-sse/services/combo/executeTargetAttempt.ts index 0a691ffe639a..63a955494e13 100644 --- a/open-sse/services/combo/executeTargetAttempt.ts +++ b/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 { @@ -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 diff --git a/open-sse/services/combo/roundRobinCombo.ts b/open-sse/services/combo/roundRobinCombo.ts index 063b4c055732..2768840c7e3e 100644 --- a/open-sse/services/combo/roundRobinCombo.ts +++ b/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, diff --git a/open-sse/services/combo/runtimeUnits.ts b/open-sse/services/combo/runtimeUnits.ts index 304ba99124b8..d1c54626e579 100644 --- a/open-sse/services/combo/runtimeUnits.ts +++ b/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; 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") { diff --git a/open-sse/utils/heapPressure.ts b/open-sse/utils/heapPressure.ts index 659853356d06..566a55af0a7e 100644 --- a/open-sse/utils/heapPressure.ts +++ b/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" } } + ) ), }; } diff --git a/open-sse/utils/localPressureResponse.ts b/open-sse/utils/localPressureResponse.ts new file mode 100644 index 000000000000..82f5dfda7a12 --- /dev/null +++ b/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(); + +/** 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; +} diff --git a/open-sse/utils/resourcePressure.ts b/open-sse/utils/resourcePressure.ts index 9bc4edaffadf..024b3cd728eb 100644 --- a/open-sse/utils/resourcePressure.ts +++ b/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 = {} @@ -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 }, + } + ) ), }; } diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index bcea3f95a40d..2d14057e9135 100644 --- a/src/sse/handlers/chat.ts +++ b/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() ) { diff --git a/src/sse/handlers/chat/selectedConnectionHeader.ts b/src/sse/handlers/chat/selectedConnectionHeader.ts index aa6f575bff00..4fe9086c43a4 100644 --- a/src/sse/handlers/chat/selectedConnectionHeader.ts +++ b/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); } } diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index 3a5ca9fb2a07..3caa53715b49 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -1,3 +1,4 @@ +import { inheritLocalProcessPressureResponse } from "@omniroute/open-sse/utils/localPressureResponse.ts"; import { credentialCooldownResponse } from "./credentialCooldownResponse"; import { getModelInfo, @@ -1191,6 +1192,10 @@ export async function safeLogEvents({ } catch {} } +/** + * Returns the response with a session ID header, or unchanged when no session ID is given. + * Immutable headers require a wrapper preserving trusted rate-limit and pressure markers. + */ export function withSessionHeader(response: Response, sessionId: string | null): Response { if (!response || !sessionId) return response; @@ -1204,10 +1209,17 @@ export function withSessionHeader(response: Response, sessionId: string | null): headers: response.headers, }); cloned.headers.set("X-OmniRoute-Session-Id", sessionId); - return inheritTrustedLocalRateLimitResponse(response, cloned); + return inheritLocalProcessPressureResponse( + response, + inheritTrustedLocalRateLimitResponse(response, cloned) + ); } } +/** + * Returns the response with a correlation ID header, or unchanged when no ID is given. + * Immutable headers require a wrapper preserving trusted rate-limit and pressure markers. + */ export function withCorrelationId(response: Response, correlationId: string | null): Response { if (!response || !correlationId) return response; @@ -1221,7 +1233,10 @@ export function withCorrelationId(response: Response, correlationId: string | nu headers: response.headers, }); cloned.headers.set("X-Correlation-Id", correlationId); - return inheritTrustedLocalRateLimitResponse(response, cloned); + return inheritLocalProcessPressureResponse( + response, + inheritTrustedLocalRateLimitResponse(response, cloned) + ); } } diff --git a/src/sse/handlers/emergencyFallbackHeader.ts b/src/sse/handlers/emergencyFallbackHeader.ts index 08004cad3b1f..168b31f4d3f3 100644 --- a/src/sse/handlers/emergencyFallbackHeader.ts +++ b/src/sse/handlers/emergencyFallbackHeader.ts @@ -1,3 +1,4 @@ +import { inheritLocalProcessPressureResponse } from "@omniroute/open-sse/utils/localPressureResponse.ts"; import { inheritTrustedLocalRateLimitResponse } from "@omniroute/open-sse/services/rateLimitManager/errors.ts"; import { EMERGENCY_FALLBACK_CONFIG, @@ -43,7 +44,10 @@ export function withEmergencyFallbackHeader( headers: response.headers, }); cloned.headers.set(OMNIROUTE_RESPONSE_HEADERS.emergencyFallback, value); - return inheritTrustedLocalRateLimitResponse(response, cloned); + return inheritLocalProcessPressureResponse( + response, + inheritTrustedLocalRateLimitResponse(response, cloned) + ); } } diff --git a/stryker.conf.json b/stryker.conf.json index b437ed2983e2..57afc0f8987d 100644 --- a/stryker.conf.json +++ b/stryker.conf.json @@ -332,6 +332,7 @@ "tests/unit/combo-identical-error-streak.test.ts", "tests/unit/combo-priority-compat-fallback.test.ts", "tests/unit/combo-priority-quota-exhaustion-cutoff-5923.test.ts", + "tests/unit/combo-process-pressure-terminal.test.ts", "tests/unit/combo-provider-cooldown-sibling.test.ts", "tests/unit/combo-provider-cooldown.test.ts", "tests/unit/combo-provider-diversity-wiring.test.ts", diff --git a/tests/unit/chat-global-fallback-local-pressure-19124.test.ts b/tests/unit/chat-global-fallback-local-pressure-19124.test.ts new file mode 100644 index 000000000000..2b071fcc0e7f --- /dev/null +++ b/tests/unit/chat-global-fallback-local-pressure-19124.test.ts @@ -0,0 +1,155 @@ +/** + * #19124: a shed this process created must end the combo at the chat handler too. The Global + * Fallback Model runs after handleComboChat and retries 503s, so it has to skip the local shed + * while an upstream 503 with the very same body keeps its fallback. + */ +import test from "node:test"; +import assert from "node:assert/strict"; +import { createChatPipelineHarness } from "../integration/_chatPipelineHarness.ts"; + +const harness = await createChatPipelineHarness("chat-global-fallback-local-pressure"); +const { + BaseExecutor, + buildClaudeResponse, + buildRequest, + combosDb, + handleChat, + resetStorage, + seedConnection, + settingsDb, + toPlainHeaders, +} = harness; +const { reloadResourcePressureRuntime } = await import("../../open-sse/utils/resourcePressure.ts"); + +const originalFetch = globalThis.fetch; +const MiB = 1024 ** 2; + +/** + * Installs a pressure runtime that counts each guard check. The first `healthyChecks` read a + * healthy heap, so the request passes admission and pressure rises before the provider dispatch. + */ +function reloadPressure(healthyChecks = Infinity) { + const guard = { checks: 0 }; + reloadResourcePressureRuntime({ + heapThresholdMb: 100, + immediateHeapUsedMb: () => { + guard.checks += 1; + return guard.checks <= healthyChecks ? 1 : 500; + }, + sample: async () => ({ + observedAtMs: Date.now(), + v8: { heapUsedBytes: MiB, heapLimitBytes: 10_000 * MiB }, + process: { + rssBytes: MiB, + externalBytes: 0, + arrayBuffersBytes: 0, + availableBytes: null, + constrainedBytes: null, + }, + cgroup: { + currentBytes: null, + maxBytes: null, + highBytes: null, + fileBytes: null, + events: null, + }, + psi: null, + }), + }); + return guard; +} + +/** Sends one non-streaming request to the named combo through the full chat handler. */ +function requestCombo(model: string) { + return handleChat( + buildRequest({ + body: { model, stream: false, messages: [{ role: "user", content: "pressure fallback" }] }, + }) + ); +} + +test.beforeEach(async () => { + BaseExecutor.RETRY_CONFIG.delayMs = 0; + await resetStorage(); + reloadPressure(); +}); + +test.afterEach(async () => { + globalThis.fetch = originalFetch; + reloadPressure(); + await resetStorage(); +}); + +test.after(async () => { + await harness.cleanup(); +}); + +test("a local pressure shed skips the Global Fallback Model and keeps the original 503", async () => { + await seedConnection("openai", { apiKey: "sk-openai-pressure-shed" }); + await seedConnection("claude", { apiKey: "sk-claude-pressure-shed" }); + await combosDb.createCombo({ + name: "router-pressure-shed", + strategy: "priority", + config: { maxRetries: 0, retryDelayMs: 0 }, + models: ["openai/gpt-4.1"], + }); + let fetchCalls = 0; + globalThis.fetch = async () => { + fetchCalls += 1; + return new Response("provider must not run", { status: 500 }); + }; + + const withoutFallback = reloadPressure(1); + const control = await requestCombo("router-pressure-shed"); + assert.equal(control.status, 503); + assert.ok(withoutFallback.checks > 1, "the request must get past admission to the combo target"); + + await settingsDb.updateSettings({ globalFallbackModel: "claude/claude-sonnet-4.6" }); + const withFallback = reloadPressure(1); + const response = await requestCombo("router-pressure-shed"); + + assert.equal(response.status, 503); + assert.equal(response.headers.get("Retry-After"), "5"); + assert.equal((await response.json()).error.code, "resource_pressure"); + assert.equal(fetchCalls, 0); + assert.equal( + withFallback.checks, + withoutFallback.checks, + "the Global Fallback Model must add no dispatch to a request the process already shed" + ); +}); + +test("an upstream 503 with a pressure body still reaches the Global Fallback Model", async () => { + await seedConnection("openai", { apiKey: "sk-openai-upstream-pressure" }); + await seedConnection("claude", { apiKey: "sk-claude-upstream-pressure" }); + await combosDb.createCombo({ + name: "router-upstream-pressure", + strategy: "priority", + config: { maxRetries: 0, retryDelayMs: 0 }, + models: ["openai/gpt-4.1"], + }); + await settingsDb.updateSettings({ globalFallbackModel: "claude/claude-sonnet-4.6" }); + + const attemptedKeys: string[] = []; + globalThis.fetch = async (_url, init) => { + const headers = toPlainHeaders(init?.headers); + const key = headers["x-api-key"] ?? headers.Authorization ?? headers.authorization ?? ""; + attemptedKeys.push(key); + if (key === "sk-claude-upstream-pressure") return buildClaudeResponse("Fallback answered"); + return new Response( + JSON.stringify({ + error: { code: "resource_pressure", type: "server_error", message: "Upstream pressure" }, + }), + { status: 503, headers: { "Content-Type": "application/json", "Retry-After": "7" } } + ); + }; + + const response = await requestCombo("router-upstream-pressure"); + + assert.equal(response.status, 200); + assert.deepEqual(attemptedKeys, [ + "Bearer sk-openai-upstream-pressure", + "sk-claude-upstream-pressure", + ]); + assert.equal((await response.json()).choices[0].message.content, "Fallback answered"); +}); diff --git a/tests/unit/combo-process-pressure-terminal.test.ts b/tests/unit/combo-process-pressure-terminal.test.ts new file mode 100644 index 000000000000..e338a8b5eaab --- /dev/null +++ b/tests/unit/combo-process-pressure-terminal.test.ts @@ -0,0 +1,473 @@ +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import test from "node:test"; +import { setTimeout as nativeSet, clearTimeout as nativeClear } from "node:timers"; +import type { ComboLike, HandleSingleModel } from "../../open-sse/services/combo/types.ts"; + +const originalDataDir = process.env.DATA_DIR; +const originalDisableBackup = process.env.DISABLE_SQLITE_AUTO_BACKUP; +const originalFetch = globalThis.fetch; +globalThis.fetch = async () => { + throw new Error("offline pressure fixture forbids network access"); +}; +const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), "combo-process-pressure-")); +process.env.DATA_DIR = dataDir; +process.env.DISABLE_SQLITE_AUTO_BACKUP = "true"; +const { handleComboChat, resolveComboTargets } = await import("../../open-sse/services/combo.ts"); +const { resetDbInstance } = await import("../../src/lib/db/core.ts"); +const { recordSessionModelUsage, getLastSessionModel } = + await import("../../src/lib/db/contextHandoffs.ts"); +const { createProviderConnection } = await import("../../src/lib/db/providers.ts"); +const { saveModelsDevCapabilities } = await import("../../src/lib/modelsDevSync.ts"); +const { getLKGP, setLKGP } = await import("../../src/lib/db/settings/lkgp.ts"); +const { applySessionStickiness, recordStickyBinding, peekStickyConnectionId } = + await import("../../open-sse/services/combo/sessionStickiness.ts"); +const { checkHeapPressureGuard } = await import("../../open-sse/utils/heapPressure.ts"); +const { markLocalProcessPressureResponse } = + await import("../../open-sse/utils/localPressureResponse.ts"); +const { resetAllCircuitBreakers, getCircuitBreaker } = + await import("../../src/shared/utils/circuitBreaker.ts"); +const { getAllModelLockouts, clearAllModelLockouts } = + await import("../../open-sse/services/accountFallback.ts"); +const { clearCooldownState, isProviderInCooldown } = + await import("../../open-sse/services/providerCooldownTracker.ts"); +const semaphore = await import("../../open-sse/services/accountSemaphore.ts"); +const eventBus = await import("../../src/lib/events/eventBus.ts"); + +/** Provides a no-op logger callback for offline routing fixtures. */ +const noop = () => {}; +const log = { info: noop, warn: noop, debug: noop, error: noop }; +const body = { messages: [{ role: "user", content: "offline pressure regression" }] }; +const models = ["providerA/first", "providerB/second"]; +let sequence = 0; +const config = { + maxRetries: 0, + maxSetRetries: 0, + retryDelayMs: 0, + setRetryDelayMs: 0, + fallbackDelayMs: 0, +}; +/** Returns a successful JSON chat response for healthy fallback controls. */ +const ok = () => Response.json({ choices: [{ message: { content: "healthy" } }] }); +/** Returns a branded local 503 for the requested pressure code with a custody receipt. */ +function pressure(code: string): Response { + const response = + code === "heap_pressure" + ? checkHeapPressureGuard(2, 1)!.response + : markLocalProcessPressureResponse( + Response.json( + { error: { code, type: "server_error", message: "Local process pressure" } }, + { status: 503, headers: { "Retry-After": "7" } } + ) + ); + response.headers.set("X-Pressure-Receipt", "original-verdict"); + return response; +} +/** Returns a uniquely named two-target combo with the strategy and configuration overrides. */ +function combo(strategy: string, overrides: Record = {}): ComboLike { + return { + name: `process-pressure-${sequence++}`, + strategy, + models, + config: { ...config, ...overrides }, + }; +} +/** Runs the selected combo with the offline request, dispatcher, and optional nested combos. */ +function run( + selected: ComboLike, + handleSingleModel: HandleSingleModel, + allCombos: ComboLike[] = [] +) { + return handleComboChat({ + body, + combo: selected, + handleSingleModel, + allCombos, + log, + settings: {}, + }); +} +/** Asserts original response identity, pressure headers, and unread body bytes, then consumes it. */ +async function assertOriginal(response: Response, original: Response, bytes: string) { + assert.equal(response, original, "terminal pressure retains response identity and custody"); + assert.equal(response.status, 503); + const expectedRetryAfter = JSON.parse(bytes).error.code === "heap_pressure" ? "5" : "7"; + assert.equal(response.headers.get("Retry-After"), expectedRetryAfter); + assert.equal(response.headers.get("X-Pressure-Receipt"), "original-verdict"); + assert.equal(response.bodyUsed, false); + assert.equal(await response.text(), bytes); +} + +test.beforeEach(() => { + resetAllCircuitBreakers(); + clearAllModelLockouts(); + clearCooldownState(); + semaphore.resetAll(); +}); +test.after(() => { + resetDbInstance(); + globalThis.fetch = originalFetch; + if (originalDisableBackup === undefined) delete process.env.DISABLE_SQLITE_AUTO_BACKUP; + else process.env.DISABLE_SQLITE_AUTO_BACKUP = originalDisableBackup; + if (originalDataDir === undefined) delete process.env.DATA_DIR; + else process.env.DATA_DIR = originalDataDir; + // Keep the private fixture on disk for native closure/FD inspection. +}); + +for (const code of ["resource_pressure", "heap_pressure"]) { + test(`${code} stops pinned-context redispatch`, async () => { + const selected = combo("priority"); + selected.context_cache_protection = true; + await createProviderConnection({ + provider: "providerA", + authType: "apikey", + name: "offline pin", + isActive: true, + }); + const sessionId = `pressure-session-${sequence}`; + recordSessionModelUsage(sessionId, selected.name, models[0], "providerA"); + const original = pressure(code); + const bytes = await original.clone().text(); + const calls: string[] = []; + const response = await handleComboChat({ + body, + combo: selected, + settings: {}, + allCombos: [selected], + log, + relayOptions: { sessionId }, + handleSingleModel: async (_body, model) => { + calls.push(model); + return calls.length === 1 ? original : ok(); + }, + }); + assert.deepEqual(calls, [models[0]]); + await assertOriginal(response, original, bytes); + assert.equal(getLastSessionModel(sessionId, selected.name), models[0]); + }); + for (const strategy of ["priority", "round-robin"]) { + test(`${code} stops ${strategy} compatibility last-resort dispatch`, async () => { + const selected = combo(strategy); + selected.models = [ + "openai/pressure-kept", + "openai/pressure-rejected1", + "openai/pressure-rejected2", + ]; + saveModelsDevCapabilities({ + openai: Object.fromEntries( + ["pressure-kept", "pressure-rejected1", "pressure-rejected2"].map((name) => [ + name, + { + tool_call: name === "pressure-kept", + modalities_input: JSON.stringify(["text"]), + modalities_output: JSON.stringify(["text"]), + limit_context: 128000, + limit_output: 4096, + reasoning: false, + attachment: false, + structured_output: true, + temperature: true, + knowledge_cutoff: null, + release_date: null, + last_updated: null, + status: null, + family: null, + open_weights: false, + limit_input: 128000, + interleaved_field: null, + }, + ]) + ), + }); + const original = pressure(code); + const bytes = await original.clone().text(); + const calls: string[] = []; + const response = await handleComboChat({ + body: { ...body, tools: [{ type: "function", function: { name: "offline" } }] }, + combo: selected, + settings: {}, + allCombos: [], + log, + isModelAvailable: async (model) => !model.endsWith("pressure-kept"), + handleSingleModel: async (_body, model) => { + calls.push(model); + return calls.length === 1 ? original : ok(); + }, + }); + assert.deepEqual(calls, ["openai/pressure-rejected1"]); + await assertOriginal(response, original, bytes); + }); + } + test(`${code} releases a native hedge-delay timer on immediate terminal return`, async (t) => { + const timers: ReturnType[] = []; + const cleared = new Set>(); + + t.mock.method(globalThis, "setTimeout", (fn: () => void, ms: number, ...args: unknown[]) => { + const timer = nativeSet(fn, ms, ...args); + if (ms === 180000) timers.push(timer); + return timer; + }); + t.mock.method(globalThis, "clearTimeout", (timer: ReturnType) => { + cleared.add(timer); + return nativeClear(timer); + }); + try { + const original = pressure(code); + const calls: string[] = []; + const response = await run( + combo("priority", { + hedging: true, + zeroLatencyOptimizationsEnabled: true, + hedgeDelayMs: 180000, + }), + async (_body, model) => { + calls.push(model); + return original; + } + ); + assert.deepEqual(calls, [models[0]]); + assert.equal(response, original); + assert.equal(timers.length, 1, "test must observe the actual speculative timer"); + assert.ok(cleared.has(timers[0]), "terminal return must clear its losing hedge timer"); + } finally { + for (const timer of timers) nativeClear(timer); + } + }); + for (const strategy of ["priority", "round-robin", "fill-first"]) { + for (const retries of [0, 2]) { + test(`replay_139070680f ${code}: ${strategy} stops retries=${retries}`, async () => { + const calls: string[] = []; + const original = pressure(code); + const bytes = await original.clone().text(); + const selected = combo(strategy, { maxRetries: retries, maxSetRetries: retries }); + const response = await run(selected, async (_body, model) => { + calls.push(model); + return calls.length === 1 ? original : ok(); + }); + assert.deepEqual( + calls, + [models[0]], + "process-wide shedding cannot dispatch another target" + ); + await assertOriginal(response, original, bytes); + assert.deepEqual(getAllModelLockouts(), []); + assert.equal(isProviderInCooldown("providerA"), false); + assert.equal(getCircuitBreaker("providerA").getStatus().failureCount, 0); + // A shed invocation must release its round-robin slot, not strand capacity. + assert.equal((await run(selected, async () => ok())).status, 200); + }); + } + } + for (const strategy of ["priority", "round-robin"]) { + test(`${code} preserves ${strategy} sticky and persisted LKGP bindings`, async () => { + const connection = await createProviderConnection({ + provider: "providerA", + authType: "apikey", + name: "offline affinity", + isActive: true, + }); + const selected = combo(strategy); + assert.ok(typeof connection.id === "string"); + const connectionId = connection.id; + selected.models = [{ kind: "model", model: models[0], connectionId }, models[1]]; + const targets = resolveComboTargets(selected, []); + const sticky = await applySessionStickiness(targets, body.messages, selected.name); + assert.ok(sticky.messageHash); + recordStickyBinding(sticky.messageHash, connectionId, selected.name); + await setLKGP(selected.name, selected.name, "providerA", connectionId); + await setLKGP(selected.name, targets[0].executionKey, "providerA", connectionId); + const before = await Promise.all([ + getLKGP(selected.name, selected.name), + getLKGP(selected.name, targets[0].executionKey), + ]); + const original = pressure(code); + const calls: string[] = []; + const response = await run(selected, async (_body, model) => { + calls.push(model); + return original; + }); + assert.equal(response, original); + assert.deepEqual(calls, [models[0]]); + assert.equal(peekStickyConnectionId(sticky.messageHash), connectionId); + assert.deepEqual( + await Promise.all([ + getLKGP(selected.name, selected.name), + getLKGP(selected.name, targets[0].executionKey), + ]), + before + ); + assert.deepEqual(getAllModelLockouts(), []); + assert.equal(isProviderInCooldown("providerA", connectionId), false); + }); + } + for (const childStrategy of ["priority", "round-robin"]) { + test(`${code} propagates from nested ${childStrategy} without parent retries`, async () => { + const child = combo(childStrategy); + const middle = combo("priority", { nestedComboMode: "execute", maxRetries: 2 }); + middle.models = [{ kind: "combo-ref", comboName: child.name }, "providerC/middle"]; + const parent = combo("priority", { nestedComboMode: "execute", maxRetries: 2 }); + parent.models = [{ kind: "combo-ref", comboName: middle.name }, "providerD/outer"]; + const calls: string[] = []; + const original = pressure(code); + const bytes = await original.clone().text(); + const response = await run( + parent, + async (_body, model) => { + calls.push(model); + return calls.length === 1 ? original : ok(); + }, + [parent, middle, child] + ); + assert.deepEqual(calls, [models[0]]); + await assertOriginal(response, original, bytes); + }); + } + test(`${code} stops a direct execute-mode model unit`, async () => { + const selected = combo("priority", { nestedComboMode: "execute", maxRetries: 2 }); + const unusedChild = combo("priority"); + selected.models = [models[0], { kind: "combo-ref", comboName: unusedChild.name }]; + const calls: string[] = []; + const original = pressure(code); + const bytes = await original.clone().text(); + const response = await run( + selected, + async (_body, model) => { + calls.push(model); + return calls.length === 1 ? original : ok(); + }, + [selected, unusedChild] + ); + assert.deepEqual(calls, [models[0]]); + await assertOriginal(response, original, bytes); + }); + test( + `${code} aborts an already-running hedge and closes its owned FD`, + { timeout: 5000 }, + async () => { + const selected = combo("priority", { + hedging: true, + zeroLatencyOptimizationsEnabled: true, + hedgeDelayMs: 0, + targetTimeoutMs: 4000, + }); + selected.models = [...models, "providerC/must-not-launch"]; + const original = pressure(code); + const bytes = await original.clone().text(); + const calls: string[] = []; + let fdClosed = false; + let aborted = false; + let settle!: () => void; + const settled = new Promise((resolve) => { + settle = resolve; + }); + const response = await run(selected, async (_body, model, target) => { + calls.push(model); + if (model !== models[0]) return original; + const fd = fs.openSync(path.join(dataDir, `owned-hedge-${sequence}`), "w+"); + const signal = target?.modelAbortSignal; + assert.ok(signal, "real combo runner must deliver cancellation to the provider leg"); + return new Promise((resolve) => { + signal.addEventListener( + "abort", + () => { + aborted = true; + fs.closeSync(fd); + assert.throws(() => fs.fstatSync(fd), { code: "EBADF" }); + fdClosed = true; + resolve(new Response(null, { status: 499 })); + settle(); + }, + { once: true } + ); + }); + }); + await settled; + assert.deepEqual(calls, models); + assert.equal(aborted, true); + assert.equal(fdClosed, true); + await assertOriginal(response, original, bytes); + assert.deepEqual(getAllModelLockouts(), []); + } + ); +} + +for (const strategy of ["priority", "round-robin"]) { + for (const failure of [ + () => Response.json({ error: { message: "Real upstream unavailable" } }, { status: 503 }), + () => + Response.json( + { error: { code: "other_503", message: "resource_pressure in prose" } }, + { status: 503 } + ), + () => Response.json({ error: { code: "resource_pressure" } }, { status: 502 }), + () => new Response("heap_pressure", { status: 503 }), + // A chained upstream (another gateway) can return the very same pressure bodies; only a + // response this process shed itself is terminal. + () => + Response.json( + { + error: { code: "resource_pressure", type: "server_error", message: "Upstream pressure" }, + }, + { status: 503, headers: { "Retry-After": "7" } } + ), + () => + Response.json( + { error: { code: "heap_pressure", type: "server_error", message: "Upstream pressure" } }, + { status: 503, headers: { "Retry-After": "5" } } + ), + ]) { + test(`${strategy} preserves ordinary fallback control ${failure.toString()}`, async () => { + const calls: string[] = []; + const result = await run(combo(strategy), async (_body, model) => { + calls.push(model); + return calls.length === 1 ? failure() : ok(); + }); + assert.equal(result.status, 200); + assert.deepEqual(calls, models); + }); + } +} + +test("round-robin closes the live attempt it opened before a terminal pressure shed", async () => { + const seen: string[] = []; + const unsubscribers = (["combo.target.attempt", "combo.target.failed"] as const).map((event) => + eventBus.on(event, () => seen.push(event)) + ); + try { + const original = pressure("resource_pressure"); + const response = await run(combo("round-robin"), async () => original); + assert.equal(response, original); + assert.deepEqual(seen, ["combo.target.attempt", "combo.target.failed"]); + } finally { + for (const unsubscribe of unsubscribers) unsubscribe(); + } +}); + +test("internal header wrappers keep the local pressure brand when they must clone", async () => { + const { withSelectedConnectionHeader } = + await import("../../src/sse/handlers/chat/selectedConnectionHeader.ts"); + const { isMarkedLocalProcessPressureResponse } = + await import("../../open-sse/utils/localPressureResponse.ts"); + const shed = pressure("resource_pressure"); + // Immutable headers force the wrapper down its clone path. + const frozen = new Response(shed.body, { status: 503, headers: shed.headers }); + Object.defineProperty(frozen, "headers", { + value: new Proxy(new Headers(shed.headers), { + /** Rejects header mutations while binding all other header methods to their target. */ + get(target, prop) { + if (prop === "set") + return () => { + throw new TypeError("immutable"); + }; + const value = Reflect.get(target, prop); + return typeof value === "function" ? value.bind(target) : value; + }, + }), + }); + markLocalProcessPressureResponse(frozen); + const wrapped = withSelectedConnectionHeader(frozen, "conn-1"); + assert.notEqual(wrapped, frozen); + assert.equal(isMarkedLocalProcessPressureResponse(wrapped), true); +});