diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 5ab9d1d79b35..f0dc5efc3c8e 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -158,6 +158,7 @@ import { isStreamReadinessFailureErrorBody, isStreamEarlyEofErrorBody, isTokenLimitBreachErrorBody, + isLocalQueueCapacityErrorBody, toRecordedTarget, getExhaustedTargetSkipReason, clampPercent, @@ -1657,6 +1658,7 @@ export async function handleComboChat({ // FIX 5: a local per-API-key token-limit 429 must not cool shared accounts. const isTokenLimitBreach = result.status === 429 && isTokenLimitBreachErrorBody(errorBody); + const isLocalQueueCapacity = isLocalQueueCapacityErrorBody(errorBody); // Fix #1681: Status 499 means client disconnected — stop combo loop immediately. // There is no point trying fallback models when nobody is listening. @@ -1675,6 +1677,22 @@ export async function handleComboChat({ // so the combo would wrongly fall through to the next model after a 499. return { ok: false, response: result }; } + if (isLocalQueueCapacity) { + log.info( + "COMBO", + `Local rate-limit queue capacity reached for ${modelStr} — returning without upstream fallback` + ); + recordComboRequest(combo.name, modelStr, { + success: false, + latencyMs: Date.now() - startTime, + fallbackCount, + strategy, + target: toRecordedTarget(target), + }); + recordedAttempts++; + if (i > 0) fallbackCount++; + return { ok: false, response: result }; + } // Combo fallback is target-level orchestration: a non-ok target response is // treated as local to that target and the combo continues to the next target. @@ -2989,6 +3007,23 @@ async function handleRoundRobinCombo({ // FIX 5: a local per-API-key token-limit 429 must not cool shared accounts. const isTokenLimitBreach = result.status === 429 && isTokenLimitBreachErrorBody(errorBody); + const isLocalQueueCapacity = isLocalQueueCapacityErrorBody(errorBody); + + if (isLocalQueueCapacity) { + log.info( + "COMBO-RR", + `Local rate-limit queue capacity reached for ${modelStr} — returning without upstream fallback` + ); + recordComboRequest(combo.name, modelStr, { + success: false, + latencyMs: Date.now() - startTime, + fallbackCount, + strategy: "round-robin", + target: toRecordedTarget(target), + }); + recordedAttempts++; + return result; + } // Round-robin uses the same target-level fallback rule as other combo // strategies: non-ok target responses fall through to the next target. diff --git a/open-sse/services/combo/comboPredicates.ts b/open-sse/services/combo/comboPredicates.ts index 93f1568c96dc..6a424d7439e3 100644 --- a/open-sse/services/combo/comboPredicates.ts +++ b/open-sse/services/combo/comboPredicates.ts @@ -197,6 +197,10 @@ const REQUEST_SCOPED_UPSTREAM_ERROR_CODES: Record = { upstream_response_failed: true, // Local combo per-target timer (targetTimeoutRunner) — not a connection health signal. combo_target_timeout: true, + // Local limiter queue-capacity codes — not a provider/connection health signal. + rate_limit_queue_timeout: true, + rate_limit_queue_full: true, + rate_limit_queue_wedged: true, }; /** Request/model-specific failures must not poison provider-wide resilience state. */ @@ -206,7 +210,11 @@ export function isRequestScopedUpstreamFailure(error?: { }): boolean { const code = typeof error?.code === "string" ? error.code.toLowerCase() : ""; const type = typeof error?.type === "string" ? error.type.toLowerCase() : ""; - return REQUEST_SCOPED_UPSTREAM_ERROR_CODES[code] === true || type === "context_length_exceeded"; + return ( + REQUEST_SCOPED_UPSTREAM_ERROR_CODES[code] === true || + type === "context_length_exceeded" || + type === "local_queue_capacity" + ); } /** Request-scoped classification that also has access to the HTTP body. */ @@ -358,6 +366,21 @@ export function isTokenLimitBreachErrorBody(errorBody: unknown): boolean { return (error as Record).code === "TOKEN_LIMIT_EXCEEDED"; } +/** Local limiter capacity is not an upstream/provider failure and must not cascade. */ +export function isLocalQueueCapacityErrorBody(errorBody: unknown): boolean { + if (!errorBody || typeof errorBody !== "object") return false; + const error = (errorBody as Record).error; + if (!error || typeof error !== "object") return false; + const code = String((error as Record).code || "").toUpperCase(); + const type = String((error as Record).type || "").toLowerCase(); + return ( + code === "RATE_LIMIT_QUEUE_TIMEOUT" || + code === "RATE_LIMIT_QUEUE_FULL" || + code === "RATE_LIMIT_QUEUE_WEDGED" || + type === "local_queue_capacity" + ); +} + export function toRecordedTarget(target: ResolvedComboTarget) { return { executionKey: target.executionKey, diff --git a/tests/unit/rate-limit-local-capacity-classification.test.ts b/tests/unit/rate-limit-local-capacity-classification.test.ts new file mode 100644 index 000000000000..1ed86fddc620 --- /dev/null +++ b/tests/unit/rate-limit-local-capacity-classification.test.ts @@ -0,0 +1,79 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +const { isLocalQueueCapacityErrorBody, isRequestScopedUpstreamFailure, shouldSkipConnDisable } = + await import("../../open-sse/services/combo/comboPredicates.ts"); +const { handleComboChat } = await import("../../open-sse/services/combo.ts"); + +function localQueueResponse() { + return new Response( + JSON.stringify({ + error: { + message: "Local rate-limit queue wait exceeded", + code: "RATE_LIMIT_QUEUE_TIMEOUT", + type: "local_queue_capacity", + }, + }), + { status: 429, headers: { "content-type": "application/json" } } + ); +} + +function createLog() { + return { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} }; +} + +test("local queue capacity is request-scoped and never disables a healthy connection", () => { + assert.equal( + isRequestScopedUpstreamFailure({ + code: "RATE_LIMIT_QUEUE_TIMEOUT", + type: "local_queue_capacity", + }), + true + ); + assert.equal( + shouldSkipConnDisable( + { + status: 429, + errorCode: "RATE_LIMIT_QUEUE_TIMEOUT", + errorType: "local_queue_capacity", + }, + false, + false, + "nvidia" + ), + true + ); + assert.equal( + isLocalQueueCapacityErrorBody({ + error: { code: "RATE_LIMIT_QUEUE_TIMEOUT", type: "local_queue_capacity" }, + }), + true + ); +}); + +test("combo returns a local queue capacity response without retrying or rotating targets", async () => { + let calls = 0; + const response = await handleComboChat({ + body: { model: "openai/gpt-4" }, + combo: { + name: `local-queue-capacity-${Math.random().toString(16).slice(2)}`, + strategy: "priority", + models: ["openai/gpt-4", "openai/gpt-4o-mini"], + config: { maxRetries: 2, maxSetRetries: 1, retryDelayMs: 0, fallbackDelayMs: 0 }, + }, + handleSingleModel: async () => { + calls += 1; + return localQueueResponse(); + }, + isModelAvailable: async () => true, + log: createLog() as never, + settings: {}, + allCombos: null, + }); + + assert.equal(response.status, 429); + assert.equal(calls, 1, "a local queue limit must not amplify into retries or fallback calls"); + const body = await response.json(); + assert.equal(body.error.code, "RATE_LIMIT_QUEUE_TIMEOUT"); + assert.equal(body.error.type, "local_queue_capacity"); +});