From 5a34111125e45950d2abd21f3bd05913ca8024a6 Mon Sep 17 00:00:00 2001 From: Paco Cartones <253313177+pacocartones@users.noreply.github.com> Date: Wed, 2 Sep 2026 08:14:54 +0200 Subject: [PATCH] fix(resilience): count resolved 5xx results against the provider breaker (#12360) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CircuitBreaker.execute() treated every resolved promise as a success, but handleChatCore() reports most upstream failures by resolving with { success: false, status: 5xx }. On the chat path that spurious _onSuccess() decayed failureCount right before the call site's _onFailure() for the same attempt, so a provider answering 503s indefinitely stayed CLOSED at failureCount: 1 and kept receiving traffic — the breaker was structurally unable to open. Combo dispatches hit the same cancellation through the shared per-provider breaker. execute() now takes an optional per-call classifyResult; without it the resolved-means-success contract every throw-based caller relies on is unchanged. executeChatWithBreaker() passes ignore and the chat path accounts for the outcome exactly once where the request context lives, so a combo success is no longer counted twice. Validated in a combined worktree with all 25 PRs of this batch boarded together: typecheck:core clean, 443/443 node-runner tests plus 14/14 vitest across every test file the batch touches, and check-changelog-integrity, check:cycles (418 files), check:provider-consistency (272 REGISTRY entries, 355 canonical providers), check:docs-counts, check:docs-sync (42 locales) and check-file-size all green. Thanks @pacocartones. --- .../12360-circuit-breaker-resolved-5xx.md | 1 + src/shared/utils/circuitBreaker.ts | 46 ++++- src/sse/handlers/chat.ts | 8 +- src/sse/handlers/chatHelpers.ts | 16 +- src/sse/handlers/chatPredicates.ts | 32 +++ stryker.conf.json | 1 + .../unit/breaker-network-error-guard.test.ts | 51 ++++- ...circuit-breaker-resolved-5xx-12254.test.ts | 192 ++++++++++++++++++ 8 files changed, 337 insertions(+), 10 deletions(-) create mode 100644 changelog.d/fixes/12360-circuit-breaker-resolved-5xx.md create mode 100644 tests/unit/circuit-breaker-resolved-5xx-12254.test.ts diff --git a/changelog.d/fixes/12360-circuit-breaker-resolved-5xx.md b/changelog.d/fixes/12360-circuit-breaker-resolved-5xx.md new file mode 100644 index 0000000000..419d2ff76f --- /dev/null +++ b/changelog.d/fixes/12360-circuit-breaker-resolved-5xx.md @@ -0,0 +1 @@ +- **fix(resilience):** count resolved upstream 5xx results against the provider circuit breaker on the chat path — `CircuitBreaker.execute()` no longer reads a resolved `{ success: false, status: 5xx }` as a success that cancels the call-site failure, so a provider answering 503s now trips its breaker instead of staying `CLOSED` at `failureCount: 1`; single-model and combo dispatches are each accounted exactly once ([#12254](https://github.com/diegosouzapw/OmniRoute/issues/12254)) diff --git a/src/shared/utils/circuitBreaker.ts b/src/shared/utils/circuitBreaker.ts index 773d757aa2..02e4e67811 100644 --- a/src/shared/utils/circuitBreaker.ts +++ b/src/shared/utils/circuitBreaker.ts @@ -150,6 +150,25 @@ interface CircuitBreakerOptions { backoffEscalationCount?: number; } +/** + * How a RESOLVED `execute()` result is accounted (#12254). Callers such as + * `handleChatCore()` report most upstream failures by resolving with + * `{ success: false, status: 5xx }` instead of throwing, so a breaker that reads every + * resolution as a success never trips on that path. + */ +export type CircuitBreakerResultOutcome = "success" | "failure" | "ignore"; + +export interface CircuitBreakerExecuteOptions { + /** + * Classify a resolved result. Omitted: every resolution is a success (the + * throw-based contract every other caller relies on). Return "ignore" when the + * call site accounts for the outcome itself with request context the breaker + * does not have — the chat path does (`classifyProviderBreakerResult()` in + * chat.ts, `recordProviderFailure()`/`recordProviderSuccess()` in combo.ts). + */ + classifyResult?: (result: T) => CircuitBreakerResultOutcome; +} + export interface TransitionRecord { from: string; to: string; @@ -300,7 +319,7 @@ export class CircuitBreaker { ); } - async execute(fn: () => Promise): Promise { + async execute(fn: () => Promise, options?: CircuitBreakerExecuteOptions): Promise { this._refreshOpenState(); if (this.state === STATE.OPEN) { @@ -325,7 +344,7 @@ export class CircuitBreaker { try { const result = await fn(); - this._onSuccess(); + this._recordResolvedResult(result, options?.classifyResult); return result; } catch (error) { if (this.isFailure(error)) { @@ -387,6 +406,29 @@ export class CircuitBreaker { // ─── Internal ───────────────────────────────── + /** + * Account a resolved `execute()` result exactly once. A classifier that throws + * falls back to the legacy "resolved = success" reading, mirroring `classifyError`. + */ + _recordResolvedResult( + result: T, + classifyResult?: (result: T) => CircuitBreakerResultOutcome + ): void { + let outcome: CircuitBreakerResultOutcome = "success"; + if (classifyResult) { + try { + outcome = classifyResult(result); + } catch { + outcome = "success"; + } + } + if (outcome === "failure") { + this._onFailure(); + } else if (outcome === "success") { + this._onSuccess(); + } + } + _onSuccess() { if (this.state === STATE.OPEN) { this._transition(STATE.CLOSED, "success-recovery"); diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index 26e58825f5..9d8bc608e1 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -105,10 +105,10 @@ import { import { buildModalityBridgeHeader } from "@/lib/guardrails/modalityBridge/bridgeStats"; import { resolveConversationId } from "@omniroute/open-sse/services/conversationTracker.ts"; import { + classifyProviderBreakerResult, isAntigravityMissingProjectError, isProviderBreakerFailureStatus, resolveStreamReadinessClassificationError, - shouldTripProviderBreakerForResult, } from "./chatPredicates"; import { markAntigravityMissingCloudCodeProject } from "@omniroute/open-sse/services/antigravityProjectPersistence.ts"; import { connectionHasExtraKeys } from "@omniroute/open-sse/services/apiKeyRotator.ts"; @@ -1923,7 +1923,9 @@ async function handleSingleModelChat( if (result.success) { clearModelLock(provider, credentials.connectionId, model); - if (!forceLiveComboTest) { + // #12254: exactly-once breaker accounting — combo successes are recorded by + // combo.ts (recordProviderSuccess); live combo tests never touch the breaker. + if (classifyProviderBreakerResult(result, isCombo, forceLiveComboTest) === "success") { breaker._onSuccess(); } if (injectedHandoff && runtimeOptions.sessionId && comboName) { @@ -2370,7 +2372,7 @@ async function handleSingleModelChat( // breaker for real traffic (#9817). if ( !(await shouldIsolateProbeFailures()) && - shouldTripProviderBreakerForResult(result, isCombo, forceLiveComboTest) + classifyProviderBreakerResult(result, isCombo, forceLiveComboTest) === "failure" ) { breaker._onFailure(); } diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index 7bf4eef08f..b736203b75 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -399,6 +399,13 @@ export function checkResourcePressureBeforeProviderWork(): ResourcePressureGuard } } +// #12254: handleChatCore resolves `{ success: false, status: 5xx }` for most upstream +// failures, so execute() must not read a resolution as a success (it used to, and that +// spurious _onSuccess() cancelled the call site's _onFailure() for the same attempt). +// The chat path accounts for the outcome exactly once where the request context lives: +// chat.ts via classifyProviderBreakerResult(), combo.ts via recordProviderFailure/Success. +const chatPathOwnsBreakerAccounting = () => "ignore" as const; + export async function executeChatWithBreaker({ bypassCircuitBreaker, breaker, @@ -592,13 +599,16 @@ export async function executeChatWithBreaker({ } if (tlsFingerprintActive) { - const tracked = await breaker.execute(async () => - runWithTlsTracking(tlsTrackingIdentity, chatFn) + const tracked = await breaker.execute( + async () => runWithTlsTracking(tlsTrackingIdentity, chatFn), + { classifyResult: chatPathOwnsBreakerAccounting } ); return { result: tracked.result, tlsFingerprintUsed: tracked.tlsFingerprintUsed }; } - const result = await breaker.execute(chatFn); + const result = await breaker.execute(chatFn, { + classifyResult: chatPathOwnsBreakerAccounting, + }); return { result, tlsFingerprintUsed: false }; } catch (cbErr: any) { if (cbErr instanceof CircuitBreakerOpenError) { diff --git a/src/sse/handlers/chatPredicates.ts b/src/sse/handlers/chatPredicates.ts index db9abd3da2..f1ccbbaab2 100644 --- a/src/sse/handlers/chatPredicates.ts +++ b/src/sse/handlers/chatPredicates.ts @@ -43,6 +43,38 @@ export function shouldTripProviderBreakerForResult( ); } +export type ProviderBreakerResultOutcome = "success" | "failure" | "ignore"; + +/** + * #12254: single source of truth for how a resolved dispatch result is accounted + * against the per-provider breaker. `handleChatCore()` resolves with + * `{ success: false, status: 5xx }` for most upstream failures, so `breaker.execute()` + * cannot classify it — the call site does, exactly once: + * - combo dispatches and live combo tests are "ignore": the combo target loop owns the + * accounting (`recordProviderFailure()` / `recordProviderSuccess()`), which also knows + * about same-provider-next and `skipProviderBreaker`; + * - a successful single-model dispatch is a "success"; + * - a failed one is a "failure" only when `shouldTripProviderBreakerForResult()` agrees. + */ +export function classifyProviderBreakerResult( + result: { + success?: boolean; + status: number; + response?: Response; + errorCode?: string | null; + errorType?: string | null; + error?: unknown; + }, + isCombo: boolean, + forceLiveComboTest: boolean +): ProviderBreakerResultOutcome { + if (forceLiveComboTest || isCombo) return "ignore"; + if (result.success) return "success"; + return shouldTripProviderBreakerForResult(result, isCombo, forceLiveComboTest) + ? "failure" + : "ignore"; +} + export function isAntigravityMissingProjectError( provider: string, result: { status?: number; errorCode?: string; errorType?: string } diff --git a/stryker.conf.json b/stryker.conf.json index 8345b15d89..adfd35dcdf 100644 --- a/stryker.conf.json +++ b/stryker.conf.json @@ -147,6 +147,7 @@ "tests/unit/circuit-breaker-failure-kind.test.ts", "tests/unit/circuit-breaker-local-execution.test.ts", "tests/unit/circuit-breaker-registry-cap.test.ts", + "tests/unit/circuit-breaker-resolved-5xx-12254.test.ts", "tests/unit/circuit-breaker-stream-controller-4602.test.ts", "tests/unit/claude-code-parity.test.ts", "tests/unit/claude-effort-suffix-strip.test.ts", diff --git a/tests/unit/breaker-network-error-guard.test.ts b/tests/unit/breaker-network-error-guard.test.ts index c03c25e561..8d53edc080 100644 --- a/tests/unit/breaker-network-error-guard.test.ts +++ b/tests/unit/breaker-network-error-guard.test.ts @@ -1,6 +1,9 @@ import { test } from "node:test"; import assert from "node:assert/strict"; -import { shouldTripProviderBreakerForResult } from "../../src/sse/handlers/chatPredicates.ts"; +import { + classifyProviderBreakerResult, + shouldTripProviderBreakerForResult, +} from "../../src/sse/handlers/chatPredicates.ts"; import { recordProviderFailure, clearProviderFailure, @@ -70,6 +73,50 @@ test("forceLiveComboTest=true prevents breaker trip (combo will try next target) assert.equal(result, false); }); +// #12254: the single-model call site accounts for a RESOLVED dispatch result exactly +// once through this classifier — `breaker.execute()` no longer reads a resolved +// `{ success: false, status: 5xx }` as a success. +test("classifyProviderBreakerResult: a resolved 503 on the single-model path is a failure", () => { + const outcome = classifyProviderBreakerResult( + { success: false, status: 503, errorCode: null, errorType: null, error: "overloaded" }, + false, + false + ); + assert.equal(outcome, "failure"); +}); +test("classifyProviderBreakerResult: a successful single-model dispatch is a success", () => { + const outcome = classifyProviderBreakerResult({ success: true, status: 200 }, false, false); + assert.equal(outcome, "success"); +}); +test("classifyProviderBreakerResult: excluded failures are ignored, not counted as successes", () => { + const outcome = classifyProviderBreakerResult( + { success: false, status: 502, errorCode: "proxy_unreachable", errorType: null }, + false, + false + ); + assert.equal(outcome, "ignore"); +}); +test("classifyProviderBreakerResult: combo dispatches leave accounting to combo.ts (success and failure)", () => { + assert.equal( + classifyProviderBreakerResult({ success: true, status: 200 }, true, false), + "ignore" + ); + assert.equal( + classifyProviderBreakerResult({ success: false, status: 503, errorCode: null }, true, false), + "ignore" + ); +}); +test("classifyProviderBreakerResult: live combo tests never touch the breaker", () => { + assert.equal( + classifyProviderBreakerResult({ success: true, status: 200 }, false, true), + "ignore" + ); + assert.equal( + classifyProviderBreakerResult({ success: false, status: 503, errorCode: null }, false, true), + "ignore" + ); +}); + test("queue-timeout recordProviderFailure never opens the provider breaker", () => { // Control first: that many real failures WOULD open the breaker — proving the // isQueueTimeout flag, not an inert provider, is what keeps it closed. @@ -136,4 +183,4 @@ test("persistent dead proxy across windows still opens the breaker", () => { } finally { Date.now = originalNow; } -}); \ No newline at end of file +}); diff --git a/tests/unit/circuit-breaker-resolved-5xx-12254.test.ts b/tests/unit/circuit-breaker-resolved-5xx-12254.test.ts new file mode 100644 index 0000000000..613f1d846d --- /dev/null +++ b/tests/unit/circuit-breaker-resolved-5xx-12254.test.ts @@ -0,0 +1,192 @@ +/** + * #12254: `handleChatCore()` reports most upstream failures by RESOLVING with + * `{ success: false, status: 5xx }` rather than throwing. `CircuitBreaker.execute()` + * used to treat every resolved promise as a success, so a provider could return 5xx + * indefinitely while its breaker stayed CLOSED — the spurious `_onSuccess()` decayed + * the counter by one and cancelled the very next call-site `_onFailure()` for the same + * attempt, pinning `failureCount` at 1. + * + * The first test drives the real single-model pipeline + * (chat.ts → executeChatWithBreaker → breaker.execute) against an upstream that always + * answers 503. The remaining tests pin the `execute()` result-classification contract. + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { createChatPipelineHarness } from "../integration/_chatPipelineHarness.ts"; + +const harness = await createChatPipelineHarness("circuit-breaker-resolved-5xx-12254"); +const { BaseExecutor, buildRequest, handleChat, resetStorage, seedConnection, settingsDb } = + harness; +const { CircuitBreaker, getCircuitBreaker, STATE } = + await import("../../src/shared/utils/circuitBreaker.ts"); + +const originalFetch = globalThis.fetch; +const originalRetryConfig = { + maxAttempts: BaseExecutor.RETRY_CONFIG.maxAttempts, + delayMs: BaseExecutor.RETRY_CONFIG.delayMs, +}; + +const uniqueName = (s: string) => `cb-12254-${s}-${Date.now()}-${Math.floor(Math.random() * 1e6)}`; + +test.beforeEach(async () => { + BaseExecutor.RETRY_CONFIG.maxAttempts = 1; + BaseExecutor.RETRY_CONFIG.delayMs = 0; + await resetStorage(); +}); + +test.afterEach(async () => { + globalThis.fetch = originalFetch; + BaseExecutor.RETRY_CONFIG.maxAttempts = originalRetryConfig.maxAttempts; + BaseExecutor.RETRY_CONFIG.delayMs = originalRetryConfig.delayMs; + await resetStorage(); +}); + +test.after(async () => { + await harness.cleanup(); +}); + +test("#12254: consecutive resolved 503s open the provider breaker on the single-model path", async () => { + const failureThreshold = 3; + await settingsDb.updateSettings({ + requestRetry: 0, + maxRetryIntervalSec: 0, + resilienceSettings: { + providerBreaker: { + apikey: { failureThreshold, degradationThreshold: 2, resetTimeoutMs: 60_000 }, + }, + }, + }); + + let upstreamCalls = 0; + globalThis.fetch = async () => { + upstreamCalls += 1; + return new Response(JSON.stringify({ error: { message: "Service temporarily overloaded" } }), { + status: 503, + headers: { "Content-Type": "application/json" }, + }); + }; + + const breaker = getCircuitBreaker("openai"); + const trace: string[] = []; + for (let i = 0; i < failureThreshold; i++) { + // A 503 puts the dispatched connection into cooldown; seed a fresh active one so + // every request really reaches the upstream and flows through breaker.execute(). + await seedConnection("openai", { apiKey: `sk-openai-resolved-503-${i}` }); + const upstreamCallsBefore = upstreamCalls; + const response = await handleChat( + buildRequest({ + body: { + model: "openai/o3-mini", + stream: false, + messages: [{ role: "user", content: `resolved 503 attempt ${i}` }], + }, + }) + ); + trace.push( + `req${i}: http=${response.status} upstreamCalls=${upstreamCalls} failureCount=${breaker.failureCount} state=${breaker.state}` + ); + assert.equal(response.status, 503, trace.join("\n")); + assert.ok(upstreamCalls > upstreamCallsBefore, `request ${i} must reach the upstream`); + assert.equal( + breaker.failureCount, + i + 1, + `each resolved 503 must count exactly once\n${trace.join("\n")}` + ); + } + + assert.equal(breaker.state, STATE.OPEN, trace.join("\n")); + + // The breaker now protects the chat path: the next request is short-circuited + // before any upstream dispatch. + await seedConnection("openai", { apiKey: "sk-openai-resolved-503-after-open" }); + const upstreamCallsBeforeOpen = upstreamCalls; + const rejected = await handleChat( + buildRequest({ + body: { + model: "openai/o3-mini", + stream: false, + messages: [{ role: "user", content: "breaker is open" }], + }, + }) + ); + assert.equal(rejected.status, 503); + assert.equal(upstreamCalls, upstreamCallsBeforeOpen, "an OPEN breaker must not dispatch"); + assert.match(await rejected.text(), /circuit breaker/i); +}); + +test("#12254: execute() counts a resolved failure payload when the classifier says so", async () => { + const cb = new CircuitBreaker(uniqueName("resolved-failure"), { + failureThreshold: 3, + resetTimeout: 30_000, + }); + const chatFn = async () => ({ success: false, status: 503, error: "overloaded" }); + const classifyResult = (result: { success: boolean }) => + result.success ? ("success" as const) : ("failure" as const); + + for (let i = 0; i < 3; i++) { + await cb.execute(chatFn, { classifyResult }); + } + + assert.equal(cb.failureCount, 3); + assert.equal(cb.state, STATE.OPEN); + cb.reset(); +}); + +test("#12254: execute() leaves accounting to the caller when the classifier returns ignore", async () => { + const cb = new CircuitBreaker(uniqueName("ignore"), { + failureThreshold: 3, + resetTimeout: 30_000, + }); + cb._onFailure(); + cb._onFailure(); + assert.equal(cb.failureCount, 2); + const stateBefore = cb.state; + + // Neither a resolved failure nor a resolved success may move the counter or the + // state: the caller records the outcome exactly once itself. + await cb.execute(async () => ({ success: false, status: 503 }), { + classifyResult: () => "ignore", + }); + await cb.execute(async () => ({ success: true, status: 200 }), { + classifyResult: () => "ignore", + }); + + assert.equal(cb.failureCount, 2); + assert.equal(cb.state, stateBefore); + cb.reset(); +}); + +test("#12254: execute() without a classifier keeps the resolved-is-success contract", async () => { + const cb = new CircuitBreaker(uniqueName("default"), { + failureThreshold: 3, + resetTimeout: 30_000, + }); + cb._onFailure(); + assert.equal(cb.failureCount, 1); + + await cb.execute(async () => ({ success: false, status: 503 })); + + // Gradual recovery on success: the legacy behaviour every throw-based caller relies on. + assert.equal(cb.failureCount, 0); + assert.equal(cb.state, STATE.CLOSED); + cb.reset(); +}); + +test("#12254: a throwing classifier never wedges the breaker", async () => { + const cb = new CircuitBreaker(uniqueName("throwing"), { + failureThreshold: 3, + resetTimeout: 30_000, + }); + + const result = await cb.execute(async () => "ok", { + classifyResult: () => { + throw new Error("classifier bug"); + }, + }); + + assert.equal(result, "ok"); + assert.equal(cb.state, STATE.CLOSED); + assert.equal(cb.failureCount, 0); + cb.reset(); +});