diff --git a/src/shared/utils/circuitBreaker.ts b/src/shared/utils/circuitBreaker.ts index 00dbf6c28f..ee898e319b 100644 --- a/src/shared/utils/circuitBreaker.ts +++ b/src/shared/utils/circuitBreaker.ts @@ -25,6 +25,28 @@ import { } from "../../lib/db/domainState"; import type { FailureKind } from "./classify429"; +/** + * #4602 — Detect a LOCAL stream-lifecycle error that must NOT count as a + * whole-provider failure. The Codex WebSocket→SSE bridge can throw a bare + * `Invalid state: Controller is already closed` (an enqueue-after-close on our + * own ReadableStream controller). It carries no `statusCode`, so it defaults to + * HTTP 502 and would otherwise trip the provider circuit breaker — blacklisting + * the entire Codex provider for a bug that lives in our bridge, not upstream. + * Use this with the breaker's `isFailure` option so the bridge error is ignored + * by the provider breaker while genuine upstream 5xx failures still count. + */ +export function isLocalStreamLifecycleError(error: unknown): boolean { + if (!error) return false; + const message = + typeof error === "string" + ? error + : typeof (error as { message?: unknown }).message === "string" + ? ((error as { message: string }).message as string) + : ""; + if (!message) return false; + return /controller is already closed/i.test(message); +} + export const STATE = { CLOSED: "CLOSED", DEGRADED: "DEGRADED", diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index 1d2c17d365..3a5cbedc69 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -75,7 +75,7 @@ import { connectionHasExtraKeys } from "@omniroute/open-sse/services/apiKeyRotat // Pipeline integration — wired modules import { classify429FromError, type FailureKind } from "@/shared/utils/classify429"; import { resolveUseUpstream429BreakerHints } from "@/shared/utils/providerHints"; -import { getCircuitBreaker } from "../../shared/utils/circuitBreaker"; +import { getCircuitBreaker, isLocalStreamLifecycleError } from "../../shared/utils/circuitBreaker"; import { markAccountExhaustedFrom429 } from "../../domain/quotaCache"; import { RequestTelemetry, recordTelemetry } from "../../shared/utils/requestTelemetry"; import { generateRequestId } from "../../shared/utils/requestId"; @@ -937,6 +937,9 @@ async function handleSingleModelChat( const breaker = getCircuitBreaker(provider, { failureThreshold: providerProfile.failureThreshold, resetTimeout: providerProfile.resetTimeoutMs, + // #4602: a local WS-bridge "Controller is already closed" throw is not an + // upstream outage — keep it from tripping the whole-provider breaker. + isFailure: (e) => !isLocalStreamLifecycleError(e), onStateChange: (name: string, from: string, to: string) => log.info("CIRCUIT", `${name}: ${from} → ${to}`), ...(useHints429 diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index e8abf6533e..daad871601 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -26,7 +26,11 @@ import { isTlsFingerprintActive, } from "@omniroute/open-sse/utils/proxyFetch.ts"; import { resolveProxyForConnection } from "@/lib/localDb"; -import { CircuitBreakerOpenError, getCircuitBreaker } from "../../shared/utils/circuitBreaker"; +import { + CircuitBreakerOpenError, + getCircuitBreaker, + isLocalStreamLifecycleError, +} from "../../shared/utils/circuitBreaker"; import { classify429FromError, type FailureKind } from "../../shared/utils/classify429"; import { resolveUseUpstream429BreakerHints } from "../../shared/utils/providerHints"; @@ -324,6 +328,9 @@ export async function checkPipelineGates( failureThreshold: providerProfile.failureThreshold ?? providerProfile.circuitBreakerThreshold, degradationThreshold: providerProfile.degradationThreshold, resetTimeout: providerProfile.resetTimeoutMs ?? providerProfile.circuitBreakerReset, + // #4602: a local WS-bridge "Controller is already closed" throw is not an + // upstream outage — keep it from tripping the whole-provider breaker. + isFailure: (e) => !isLocalStreamLifecycleError(e), onStateChange: (name: string, from: string, to: string) => log.info("CIRCUIT", `${name}: ${from} → ${to}`), ...(useHints429 diff --git a/tests/unit/circuit-breaker-stream-controller-4602.test.ts b/tests/unit/circuit-breaker-stream-controller-4602.test.ts new file mode 100644 index 0000000000..941d39d997 --- /dev/null +++ b/tests/unit/circuit-breaker-stream-controller-4602.test.ts @@ -0,0 +1,79 @@ +/** + * #4602 — Codex WebSocket bridge failures must not trip the whole-provider + * circuit breaker. + * + * When `codexTransport = websocket`, a `/v1/responses` request can fail in + * ~300ms with a bare `Invalid state: Controller is already closed` throw (the + * WS→SSE bridge enqueues after the response controller is closed). That error + * carries no `statusCode`, so it defaults to HTTP 502, and 502 is a + * provider-failure code — a burst trips the OAuth provider breaker (threshold + * 3) and every subsequent Codex request fails with `503 ... circuit breaker is + * open`, even on the healthy HTTP/SSE path. A local stream-lifecycle error is + * NOT an upstream provider outage and must be excluded from the breaker. + */ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { + CircuitBreaker, + isLocalStreamLifecycleError, +} from "../../src/shared/utils/circuitBreaker.ts"; + +const uniqueName = (suffix: string) => + `cb-test-#4602-${suffix}-${Date.now()}-${Math.floor(Math.random() * 1e6)}`; + +test("#4602 isLocalStreamLifecycleError flags the WS controller-closed error and nothing else", () => { + assert.equal( + isLocalStreamLifecycleError(new Error("Invalid state: Controller is already closed")), + true + ); + assert.equal( + isLocalStreamLifecycleError({ message: "Controller is already closed" }), + true + ); + // Real upstream failures must still count. + assert.equal(isLocalStreamLifecycleError(new Error("502 Bad Gateway")), false); + assert.equal(isLocalStreamLifecycleError(new Error("upstream timed out")), false); + assert.equal(isLocalStreamLifecycleError(undefined), false); + assert.equal(isLocalStreamLifecycleError(null), false); +}); + +test("#4602 breaker stays CLOSED when only the WS controller-closed error is thrown", async () => { + const cb = new CircuitBreaker(uniqueName("ws-closed"), { + failureThreshold: 3, + resetTimeout: 30_000, + isFailure: (e) => !isLocalStreamLifecycleError(e), + }); + + for (let i = 0; i < 5; i++) { + await assert.rejects( + cb.execute(async () => { + throw new Error("Invalid state: Controller is already closed"); + }), + /Controller is already closed/ + ); + } + + // 5 bridge errors past a threshold of 3 — the provider breaker must NOT open. + assert.equal(cb.state, "CLOSED"); + assert.equal(cb.failureCount, 0); + cb.reset(); +}); + +test("#4602 a genuine upstream failure still trips the breaker with the same isFailure guard", async () => { + const cb = new CircuitBreaker(uniqueName("real-failure"), { + failureThreshold: 3, + resetTimeout: 30_000, + isFailure: (e) => !isLocalStreamLifecycleError(e), + }); + + for (let i = 0; i < 3; i++) { + await assert.rejects( + cb.execute(async () => { + throw new Error("502 Bad Gateway from upstream"); + }) + ); + } + + assert.equal(cb.state, "OPEN"); + cb.reset(); +});