Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions src/shared/utils/circuitBreaker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
5 changes: 4 additions & 1 deletion src/sse/handlers/chat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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
Expand Down
9 changes: 8 additions & 1 deletion src/sse/handlers/chatHelpers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down Expand Up @@ -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
Expand Down
79 changes: 79 additions & 0 deletions tests/unit/circuit-breaker-stream-controller-4602.test.ts
Original file line number Diff line number Diff line change
@@ -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();
});
Loading