From c0291f8fbb43340aba574a5c31c42388609586ce Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Sat, 20 Jun 2026 11:20:57 -0300 Subject: [PATCH] refactor(combo): de-dup upstream-error exhaustion classification across both dispatchers (#4366) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Segundo incremento da de-dup dos 2 dispatchers (handleTargetError). Após cada erro de target, ambos rodavam um bloco quase-idêntico que marca o provider exhausted (#1731), a conexão connection-errored (#1731v2) ou o provider transiently rate-limited. - combo/targetExhaustion.ts: applyComboTargetExhaustion(target, opts) — atualiza os 3 Sets de exhaustion e retorna providerExhausted. As MUTAÇÕES de Set (que dirigem o skip de targets, lidas por getExhaustedTargetSkipReason) são BYTE-IDÊNTICAS nos dois; as diferenças reais viram parâmetros: tag, allAccountsRateLimited (termo extra do RR, false no handleComboChat), exhaustedLogLevel (info no handleComboChat, debug no RR). Connection-level extraído p/ markConnectionLevelExhaustion (privado, <15 complexity). - combo.ts: −73 linhas; 4 imports órfãos removidos. - 7 testes de caracterização travam as mutações + o return. ÚNICA mudança de comportamento: o WORDING das mensagens de log do RR ganha o sufixo 'on remaining targets' (cosmético; mesmo #code, mesmas mutações, mesmos níveis de log). 376/376 combo (caracterização preservada), integração sse 5/5, typecheck 0, complexity neutro (1895), file-size encolhe. --- open-sse/services/combo.ts | 131 ++++-------------- open-sse/services/combo/targetExhaustion.ts | 127 +++++++++++++++++ .../combo/combo-target-exhaustion.test.ts | 131 ++++++++++++++++++ 3 files changed, 287 insertions(+), 102 deletions(-) create mode 100644 open-sse/services/combo/targetExhaustion.ts create mode 100644 tests/unit/combo/combo-target-exhaustion.test.ts diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 640b798135..d3535fe05b 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -7,7 +7,6 @@ import { checkFallbackError, - classifyErrorText, classifyLockoutReason, decayModelFailureCount, formatRetryAfter, @@ -15,8 +14,6 @@ import { isModelLocked, recordModelLockoutFailure, recordProviderFailure, - isProviderExhaustedReason, - hasPerModelQuota, } from "./accountFallback.ts"; import { RateLimitReason } from "../config/constants.ts"; import { errorResponse, unavailableResponse } from "../utils/error.ts"; @@ -110,7 +107,6 @@ import { MAX_FALLBACK_WAIT_MS, MAX_GLOBAL_ATTEMPTS, isAllAccountsRateLimitedResponse, - isProviderCircuitOpenResult, clampComboDepth, shouldSkipForPredictedTtft, shouldRecordProviderBreakerFailure, @@ -121,6 +117,7 @@ import { toRecordedTarget, getExhaustedTargetSkipReason, } from "./combo/comboPredicates.ts"; +import { applyComboTargetExhaustion } from "./combo/targetExhaustion.ts"; import { dedupeTargetsByExecutionKey, isRecord } from "./combo/comboData.ts"; import { resolveShadowTargets, scheduleShadowRouting } from "./combo/shadowRouting.ts"; import { @@ -1626,57 +1623,20 @@ export async function handleComboChat({ ); const { cooldownMs } = fallbackResult; - // #1731: If the entire provider quota is exhausted, mark it so subsequent - // same-provider targets are skipped immediately. API-key 429s still use - // the short resilience cooldown, but explicit quota text should stop the - // combo from trying another target for the same provider in this request. - // Passthrough/per-model-quota providers multiplex independent upstream - // models behind one provider connection; a quota 429 for one model must - // not skip fallback targets for another model on the same provider. - const providerExhausted = - Boolean(provider && provider !== "unknown") && - !hasPerModelQuota(provider, rawModel) && - (isProviderExhaustedReason(fallbackResult) || - classifyErrorText(errorText) === RateLimitReason.QUOTA_EXHAUSTED); - if (providerExhausted) { - exhaustedProviders.add(provider); - log.info( - "COMBO", - `Provider ${provider} quota exhausted — marking for skip on remaining targets (#1731)` - ); - } else if ( - result.status === 429 && - !isTokenLimitBreach && - provider && - provider !== "unknown" - ) { - transientRateLimitedProviders.add(provider); - } - // #1731: Connection-level errors (502/503/504) suggest the provider itself is having - // issues (e.g. upstream unreachable, proxy error). Skip remaining same-provider - // targets in this request to avoid hammering a known-bad connection. - if ( - !providerExhausted && - provider && - provider !== "unknown" && - [408, 500, 502, 503, 504, 524].includes(result.status) && - !isProviderCircuitOpenResult(result, errorText) - ) { - const connId = target.connectionId as string | undefined; - if (connId) { - exhaustedConnections.add(`${provider}:${connId}`); - log.info( - "COMBO", - `Provider ${provider} connection ${connId} error (${result.status}) — marking for skip on remaining targets (#1731v2)` - ); - } else { - exhaustedProviders.add(provider); - log.info( - "COMBO", - `Provider ${provider} connection error (${result.status}) — marking for skip on remaining targets (#1731)` - ); - } - } + // #1731 / #1731v2: classify the upstream error and update the exhaustion sets + // (shared with handleRoundRobinCombo). Returns whether the provider is fully exhausted. + const providerExhausted = applyComboTargetExhaustion(target, { + result, + fallbackResult, + errorText, + rawModel, + isTokenLimitBreach, + allAccountsRateLimited: false, + sets: { exhaustedProviders, exhaustedConnections, transientRateLimitedProviders }, + log, + tag: "COMBO", + exhaustedLogLevel: "info", + }); // #2101: Prevent infinite fallback loops with 400 Bad Request errors that indicate // request-body-specific issues (context overflow, malformed request, model access denied). @@ -2377,53 +2337,20 @@ async function handleRoundRobinCombo({ // same-provider targets are skipped immediately. API-key 429s still use // the short resilience cooldown, but explicit quota text should stop the // combo from trying another target for the same provider in this request. - // Passthrough/per-model-quota providers multiplex independent upstream - // models behind one provider connection; a quota 429 for one model must - // not skip fallback targets for another model on the same provider. - const providerExhausted = - Boolean(provider && provider !== "unknown") && - !hasPerModelQuota(provider, parseModel(modelStr).model || modelStr) && - (isProviderExhaustedReason(fallbackResult) || - classifyErrorText(errorText) === RateLimitReason.QUOTA_EXHAUSTED || - isAllAccountsRateLimited); - if (providerExhausted) { - exhaustedProviders.add(provider); - log.debug?.( - "COMBO-RR", - `Provider ${provider} quota exhausted — marking for skip (#1731)` - ); - } else if ( - result.status === 429 && - !isTokenLimitBreach && - provider && - provider !== "unknown" - ) { - transientRateLimitedProviders.add(provider); - } - - // #1731v2: Connection-level errors (502/503/504) — skip remaining same-connection targets - if ( - !providerExhausted && - provider && - provider !== "unknown" && - [408, 500, 502, 503, 504, 524].includes(result.status) && - !isProviderCircuitOpenResult(result, errorText) - ) { - const connId = target.connectionId as string | undefined; - if (connId) { - exhaustedConnections.add(`${provider}:${connId}`); - log.info( - "COMBO-RR", - `Provider ${provider} connection ${connId} error (${result.status}) — marking for skip (#1731v2)` - ); - } else { - exhaustedProviders.add(provider); - log.info( - "COMBO-RR", - `Provider ${provider} connection error (${result.status}) — marking for skip (#1731)` - ); - } - } + // #1731 / #1731v2: classify the upstream error and update the exhaustion sets + // (shared with handleComboChat). Returns whether the provider is fully exhausted. + const providerExhausted = applyComboTargetExhaustion(target, { + result, + fallbackResult, + errorText, + rawModel: parseModel(modelStr).model || modelStr, + isTokenLimitBreach, + allAccountsRateLimited: isAllAccountsRateLimited, + sets: { exhaustedProviders, exhaustedConnections, transientRateLimitedProviders }, + log, + tag: "COMBO-RR", + exhaustedLogLevel: "debug", + }); // Transient errors → mark in semaphore so round-robin stops stampeding this target. if ( diff --git a/open-sse/services/combo/targetExhaustion.ts b/open-sse/services/combo/targetExhaustion.ts new file mode 100644 index 0000000000..fb4a371b3c --- /dev/null +++ b/open-sse/services/combo/targetExhaustion.ts @@ -0,0 +1,127 @@ +/** + * Shared upstream-error → exhaustion-set classification for the combo dispatchers + * (Quality Gate v2 / Fase 9 — combo god-file decomposition, dispatcher de-dup fase 2b). + * + * Both dispatchers (handleComboChat's speculative loop + handleRoundRobinCombo's rotation) + * ran a near-identical block after each target's upstream error: mark the provider fully + * exhausted (#1731), the provider:connection pair connection-errored (#1731v2), or the + * provider transiently rate-limited — driving same-request target skipping (read back by + * getExhaustedTargetSkipReason). The SET mutations are byte-identical to the previous inline + * code in BOTH dispatchers; the only differences (preserved here as parameters) were: + * - the log tag ("COMBO" / "COMBO-RR"); + * - the round-robin's extra `|| isAllAccountsRateLimited` term in the exhaustion test + * (`allAccountsRateLimited`, false for handleComboChat); + * - the quota-exhausted log LEVEL ("info" for handleComboChat, "debug" for round-robin). + * The only standardization is the log MESSAGE wording (round-robin previously dropped the + * "on remaining targets" suffix) — diagnostic text only, same #code + provider info. + */ +import { classifyErrorText, hasPerModelQuota, isProviderExhaustedReason } from "../accountFallback.ts"; +import { RateLimitReason } from "../../config/constants.ts"; +import { isProviderCircuitOpenResult } from "./comboPredicates.ts"; +import type { ComboLogger, ResolvedComboTarget } from "./types.ts"; + +// Connection-level failure statuses: the provider connection itself is likely bad (upstream +// unreachable, proxy/gateway error), so remaining same-connection targets are skipped. +const CONNECTION_LEVEL_ERROR_STATUSES = [408, 500, 502, 503, 504, 524]; + +export type ComboExhaustionSets = { + exhaustedProviders: Set; + exhaustedConnections: Set; + transientRateLimitedProviders: Set; +}; + +export type ApplyComboTargetExhaustionOptions = { + result: { status: number; headers?: Headers | null }; + fallbackResult: Parameters[0]; + errorText: string; + rawModel: string; + isTokenLimitBreach: boolean; + allAccountsRateLimited: boolean; + sets: ComboExhaustionSets; + log: ComboLogger; + tag: string; + exhaustedLogLevel: "info" | "debug"; +}; + +/** + * Update the per-request exhaustion sets from a target's upstream error. + * @returns providerExhausted — callers gate the connection-level branch and the same-provider + * retry decision on this (was a `const providerExhausted` local in both dispatchers). + */ +export function applyComboTargetExhaustion( + target: ResolvedComboTarget, + opts: ApplyComboTargetExhaustionOptions +): boolean { + const { + result, + fallbackResult, + errorText, + rawModel, + isTokenLimitBreach, + allAccountsRateLimited, + sets, + log, + tag, + exhaustedLogLevel, + } = opts; + const { exhaustedProviders, exhaustedConnections, transientRateLimitedProviders } = sets; + const provider = target.provider; + + // #1731: full provider quota exhausted → skip remaining same-provider targets this request. + // Passthrough/per-model-quota providers multiplex models behind one connection, so a quota + // 429 for one model must NOT skip fallback targets for another model on the same provider. + const providerExhausted = + Boolean(provider && provider !== "unknown") && + !hasPerModelQuota(provider, rawModel) && + (isProviderExhaustedReason(fallbackResult) || + classifyErrorText(errorText) === RateLimitReason.QUOTA_EXHAUSTED || + allAccountsRateLimited); + if (providerExhausted) { + exhaustedProviders.add(provider); + const emit = exhaustedLogLevel === "debug" ? log.debug : log.info; + emit?.(tag, `Provider ${provider} quota exhausted — marking for skip on remaining targets (#1731)`); + } else { + if (result.status === 429 && !isTokenLimitBreach && provider && provider !== "unknown") { + transientRateLimitedProviders.add(provider); + } + markConnectionLevelExhaustion(target, { result, errorText, sets, log, tag }); + } + + return providerExhausted; +} + +/** + * #1731v2: connection-level errors (408/5xx, excluding the OmniRoute circuit-open signal) suggest + * the provider connection itself is bad → skip remaining same-connection (or same-provider, when + * no connectionId) targets this request. Only runs when the provider was NOT already marked fully + * exhausted above. Split out to keep applyComboTargetExhaustion under the complexity ceiling. + */ +function markConnectionLevelExhaustion( + target: ResolvedComboTarget, + opts: Pick +): void { + const { result, errorText, sets, log, tag } = opts; + const provider = target.provider; + if ( + !provider || + provider === "unknown" || + !CONNECTION_LEVEL_ERROR_STATUSES.includes(result.status) || + isProviderCircuitOpenResult(result, errorText) + ) { + return; + } + const connId = target.connectionId ?? undefined; + if (connId) { + sets.exhaustedConnections.add(`${provider}:${connId}`); + log.info( + tag, + `Provider ${provider} connection ${connId} error (${result.status}) — marking for skip on remaining targets (#1731v2)` + ); + } else { + sets.exhaustedProviders.add(provider); + log.info( + tag, + `Provider ${provider} connection error (${result.status}) — marking for skip on remaining targets (#1731)` + ); + } +} diff --git a/tests/unit/combo/combo-target-exhaustion.test.ts b/tests/unit/combo/combo-target-exhaustion.test.ts new file mode 100644 index 0000000000..b5dcce432c --- /dev/null +++ b/tests/unit/combo/combo-target-exhaustion.test.ts @@ -0,0 +1,131 @@ +// tests/unit/combo/combo-target-exhaustion.test.ts +// Characterization of applyComboTargetExhaustion — the de-duplicated #1731/#1731v2 upstream-error +// → exhaustion-set classification shared by both combo dispatchers. Locks the SET mutations +// (which drive same-request target skipping) and the providerExhausted return. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { + applyComboTargetExhaustion, + type ComboExhaustionSets, +} from "../../../open-sse/services/combo/targetExhaustion.ts"; + +const log = { info() {}, warn() {}, error() {}, debug() {} }; + +function sets(): ComboExhaustionSets { + return { + exhaustedProviders: new Set(), + exhaustedConnections: new Set(), + transientRateLimitedProviders: new Set(), + }; +} + +function target(overrides: Record = {}) { + return { + kind: "model", + executionKey: "ek", + modelStr: "test-dedup-provider/m1", + provider: "test-dedup-provider", + providerId: null, + connectionId: "conn-1", + ...overrides, + } as Parameters[0]; +} + +const baseOpts = { + errorText: "plain upstream error", + rawModel: "m1", + isTokenLimitBreach: false, + allAccountsRateLimited: false, + log, + tag: "COMBO", + exhaustedLogLevel: "info" as const, +}; + +test("marks provider exhausted when the fallback result signals quota exhaustion", () => { + const s = sets(); + const exhausted = applyComboTargetExhaustion(target(), { + ...baseOpts, + result: { status: 429 }, + fallbackResult: { creditsExhausted: true }, + sets: s, + }); + assert.equal(exhausted, true); + assert.ok(s.exhaustedProviders.has("test-dedup-provider")); + assert.equal(s.transientRateLimitedProviders.size, 0); +}); + +test("round-robin's allAccountsRateLimited term also marks the provider exhausted", () => { + const s = sets(); + const exhausted = applyComboTargetExhaustion(target(), { + ...baseOpts, + result: { status: 503 }, + fallbackResult: {}, + allAccountsRateLimited: true, + sets: s, + }); + assert.equal(exhausted, true); + assert.ok(s.exhaustedProviders.has("test-dedup-provider")); +}); + +test("a transient 429 (not exhausted) marks the provider rate-limited, not exhausted", () => { + const s = sets(); + const exhausted = applyComboTargetExhaustion(target(), { + ...baseOpts, + result: { status: 429 }, + fallbackResult: {}, + sets: s, + }); + assert.equal(exhausted, false); + assert.ok(s.transientRateLimitedProviders.has("test-dedup-provider")); + assert.equal(s.exhaustedProviders.size, 0); +}); + +test("connection-level 5xx with a connectionId poisons exhaustedConnections (#1731v2)", () => { + const s = sets(); + const exhausted = applyComboTargetExhaustion(target(), { + ...baseOpts, + result: { status: 502, headers: null }, + fallbackResult: {}, + sets: s, + }); + assert.equal(exhausted, false); + assert.ok(s.exhaustedConnections.has("test-dedup-provider:conn-1")); + assert.equal(s.exhaustedProviders.size, 0); +}); + +test("connection-level 5xx without a connectionId poisons exhaustedProviders (#1731)", () => { + const s = sets(); + applyComboTargetExhaustion(target({ connectionId: null }), { + ...baseOpts, + result: { status: 503, headers: null }, + fallbackResult: {}, + sets: s, + }); + assert.ok(s.exhaustedProviders.has("test-dedup-provider")); + assert.equal(s.exhaustedConnections.size, 0); +}); + +test("an unknown provider is never marked (guard)", () => { + const s = sets(); + applyComboTargetExhaustion(target({ provider: "unknown" }), { + ...baseOpts, + result: { status: 502, headers: null }, + fallbackResult: { creditsExhausted: true }, + allAccountsRateLimited: true, + sets: s, + }); + assert.equal(s.exhaustedProviders.size, 0); + assert.equal(s.exhaustedConnections.size, 0); +}); + +test("a 200/benign status with no exhaustion mutates nothing and returns false", () => { + const s = sets(); + const exhausted = applyComboTargetExhaustion(target(), { + ...baseOpts, + result: { status: 200 }, + fallbackResult: {}, + sets: s, + }); + assert.equal(exhausted, false); + assert.equal(s.exhaustedProviders.size + s.exhaustedConnections.size + s.transientRateLimitedProviders.size, 0); +});