From aa32d2ed77ca41f75a5aa0b9ecfe2b9126777c9d Mon Sep 17 00:00:00 2001 From: Prudhvi Vuda <53619858+Prudhvivuda@users.noreply.github.com> Date: Thu, 20 Aug 2026 05:30:39 -0400 Subject: [PATCH] fix(resilience): retry Codex pre-output transport failures on the same account (#9708) (#10792) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merged via merge-train (release/v3.8.50, batch1 2026-08-20) — static gates (typecheck/file-size/complexity/cognitive/changelog) green on the combined tree; test:unit reds observed in the boarded run were verified pre-existing on the pure release tip (unrelated flake), not caused by this PR. Thanks for the contribution! --- .../fixes/9708-codex-same-account-retry.md | 1 + src/sse/handlers/chat.ts | 43 +++- src/sse/services/auth.ts | 34 ++- src/sse/services/sameAccountTransportRetry.ts | 108 ++++++++ ...-same-account-transport-retry-9708.test.ts | 243 ++++++++++++++++++ 5 files changed, 426 insertions(+), 3 deletions(-) create mode 100644 changelog.d/fixes/9708-codex-same-account-retry.md create mode 100644 src/sse/services/sameAccountTransportRetry.ts create mode 100644 tests/unit/codex-same-account-transport-retry-9708.test.ts diff --git a/changelog.d/fixes/9708-codex-same-account-retry.md b/changelog.d/fixes/9708-codex-same-account-retry.md new file mode 100644 index 0000000000..2ccb7fb97f --- /dev/null +++ b/changelog.d/fixes/9708-codex-same-account-retry.md @@ -0,0 +1 @@ +- **fix(resilience):** retry a retryable Codex pre-output 502/503/504/507 once on the same account (2–3s jitter) before cooling the connection, and stop translating that mixed pool into an all-accounts quota `429` ([#9708](https://github.com/diegosouzapw/OmniRoute/issues/9708)) diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index b8aff4cd2a..25e8caccc4 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -176,6 +176,10 @@ import { resolveCooldownAwareRetrySettings, waitForCooldownAwareRetry, } from "../services/cooldownAwareRetry"; +import { + shouldRetrySameAccountTransport, + sameAccountTransportRetryDelayMs, +} from "../services/sameAccountTransportRetry"; import { constrainConnectionsToQuota, resolveQuotaKeyScope } from "../../lib/quota/quotaKey"; import { checkConnectionCapacity } from "../utils/backpressure"; import { @@ -333,7 +337,11 @@ function isManagedComboUnsupported( const managedComboRejection = () => buildManagedLeaseErrorResponse( - new LeaseContextError(409, "LEASE_UNSUPPORTED_ROUTE", "Managed leases do not support this route") + new LeaseContextError( + 409, + "LEASE_UNSUPPORTED_ROUTE", + "Managed leases do not support this route" + ) ); const comboPromoteDeps = { updateCombo, info: log.info, warn: log.warn }; @@ -1534,6 +1542,7 @@ async function handleSingleModelChat( // re-attempt to exactly one for the whole request. Declared outside both retry // loops so it can never reset and loop. let streamEarlyEofRetries = 0; + const sameAccountTransportRetries = new Map(); const occupancySessionKey = runtimeOptions.sessionAffinityKey ?? runtimeOptions.sessionId ?? `request:${randomUUID()}`; let initialPreselectedCredentials = runtimeOptions.preselectedCredentials; @@ -1818,7 +1827,8 @@ async function handleSingleModelChat( comboStrategy, isCombo, comboStepId: runtimeOptions.comboStepId ?? null, - comboExecutionKey: runtimeOptions.comboExecutionKey ?? runtimeOptions.comboStepId ?? null, + comboExecutionKey: + runtimeOptions.comboExecutionKey ?? runtimeOptions.comboStepId ?? null, extendedContext, modelApiFormat: apiFormat, modelTargetFormat: targetFormat, @@ -2201,6 +2211,35 @@ async function handleSingleModelChat( } } + // #9708: retry a retryable pre-output transport failure once on the same + // account (jittered 2-3s) before cooling the connection. A first 503/507 + // must not rotate away from a still-healthy Codex prompt-cache partition. + const transportAttempts = sameAccountTransportRetries.get(credentials.connectionId) || 0; + if ( + shouldRetrySameAccountTransport({ + status: result.status, + errorText: errorStr, + errorCode: result.errorCode, + errorType: result.errorType, + attempt: transportAttempts, + hasForcedConnection, + }) + ) { + sameAccountTransportRetries.set(credentials.connectionId, transportAttempts + 1); + const waitMs = sameAccountTransportRetryDelayMs(); + log.warn( + "RETRY", + `${provider}/${model} retryable pre-output ${result.status} — retrying same account once after ${waitMs}ms` + ); + const completed = await waitForCooldownAwareRetry(waitMs, requestSignal); + if (!completed) { + releaseOAuthSession(); + return errorResponse(499, "Request aborted"); + } + preselectedCredentials = credentials; + continue; + } + // 8. Fallback to next account // A3 guard: if 401 and connection has extra keys, skip connection-level disable // (key-level failure already recorded in chatCore.ts via T07) diff --git a/src/sse/services/auth.ts b/src/sse/services/auth.ts index bf9076df92..2a4d572b01 100644 --- a/src/sse/services/auth.ts +++ b/src/sse/services/auth.ts @@ -61,6 +61,10 @@ import { } from "@omniroute/open-sse/services/quotaPreflight.ts"; import { resolveResilienceSettings } from "@/lib/resilience/settings"; import { resolveModelLockoutSettings } from "@/lib/resilience/modelLockoutSettings"; +import { + buildMixedAvailabilityError, + isTransportCooldownErrorCode, +} from "../services/sameAccountTransportRetry"; import { syncHealthFromDB, type KeyHealth } from "@omniroute/open-sse/services/apiKeyRotator.ts"; import { classifyProviderError, @@ -1049,7 +1053,9 @@ async function getProviderSearchPool(provider: string): Promise { if (!nodeId) continue; if ( nodePrefix && - (nodePrefix === provider || nodePrefix === canonicalProvider || nodePrefix === canonicalAlias) + (nodePrefix === provider || + nodePrefix === canonicalProvider || + nodePrefix === canonicalAlias) ) { searchPool.add(nodeId); } @@ -1686,6 +1692,32 @@ export async function getProviderCredentials( } if (policyEligibleConnections.length === 0 && availableConnections.length > 0) { + const transportUnavailable = connections.filter( + (connection) => + connectionFilterStatus.get(connection.id) === "rateLimited" && + isTransportCooldownErrorCode(connection.errorCode) + ); + if (transportUnavailable.length > 0) { + const mixed = buildMixedAvailabilityError({ + provider, + quotaFilteredCount: blockedByPolicy.length, + transportUnavailableCount: transportUnavailable.length, + transportStatus: Number(transportUnavailable[0]?.errorCode) || 503, + }); + const retryAfter = + getEarliestFutureDate( + transportUnavailable.map((connection) => connection.rateLimitedUntil || null) + ) || new Date(Date.now() + 3000).toISOString(); + invalidateManagedLease(options, "HEALTH_OR_COOLDOWN"); + return { + allRateLimited: true, + retryAfter, + retryAfterHuman: formatRetryAfter(retryAfter), + lastError: mixed.lastError, + lastErrorCode: mixed.lastErrorCode, + }; + } + const earliestResetAt = getEarliestFutureDate(blockedByPolicy.map((entry) => entry.resetAt)); const earliestResetMs = parseFutureDateMs(earliestResetAt); diff --git a/src/sse/services/sameAccountTransportRetry.ts b/src/sse/services/sameAccountTransportRetry.ts new file mode 100644 index 0000000000..4c82d94765 --- /dev/null +++ b/src/sse/services/sameAccountTransportRetry.ts @@ -0,0 +1,108 @@ +/** + * Same-account retry for retryable pre-output transport failures (#9708). + * + * A 503/507 (connection reset, retry-buffer overflow, early EOF before useful + * output) must not immediately cool the account and rotate. One jittered + * same-account retry absorbs brief proxy blips and keeps Codex prompt-cache + * affinity. A second failure then takes a short cooldown and may rotate. + */ + +export const SAME_ACCOUNT_TRANSPORT_RETRY_MAX = 1; +export const SAME_ACCOUNT_TRANSPORT_RETRY_MIN_DELAY_MS = 2000; +export const SAME_ACCOUNT_TRANSPORT_RETRY_JITTER_MS = 1000; + +const RETRYABLE_TRANSPORT_STATUSES = new Set([502, 503, 504, 507]); + +const RETRYABLE_TRANSPORT_TEXT = [ + /upstream connect error/i, + /disconnect\/reset before headers/i, + /remote connection failure/i, + /connection reset/i, + /exceeded request buffer limit/i, + /early eof/i, + /econnreset/i, + /socket hang up/i, + /und_err_socket/i, +]; + +const NON_RETRYABLE_ERROR_TYPES = new Set(["lease_error", "account_semaphore_capacity"]); + +export function isRetryableTransportStatus(status: unknown): boolean { + const numeric = Number(status); + return Number.isFinite(numeric) && RETRYABLE_TRANSPORT_STATUSES.has(numeric); +} + +export function isRetryablePreOutputTransportError( + status: unknown, + errorText: string | null | undefined, + errorCode?: string | null, + errorType?: string | null +): boolean { + if (errorType && NON_RETRYABLE_ERROR_TYPES.has(errorType)) return false; + if (errorCode && String(errorCode).startsWith("LEASE_")) return false; + + const text = String(errorText || ""); + const numericStatus = Number(status); + if (numericStatus === 429 || numericStatus === 401 || numericStatus === 400) return false; + if (/quota (threshold|exhausted)|credits exhausted/i.test(text)) return false; + if (/invalid_request|prompt is too long|context.?length|unsupported model/i.test(text)) { + return false; + } + + const statusRetryable = isRetryableTransportStatus(status); + const textRetryable = RETRYABLE_TRANSPORT_TEXT.some((pattern) => pattern.test(text)); + const codeRetryable = + errorCode === "STREAM_EARLY_EOF" || + errorCode === "proxy_unreachable" || + errorCode === "PROXY_UNREACHABLE"; + + return statusRetryable || textRetryable || codeRetryable; +} + +export function sameAccountTransportRetryDelayMs(random: () => number = Math.random): number { + const draw = random(); + const unit = Number.isFinite(draw) ? Math.min(Math.max(draw, 0), 1) : 0; + return Math.round( + SAME_ACCOUNT_TRANSPORT_RETRY_MIN_DELAY_MS + SAME_ACCOUNT_TRANSPORT_RETRY_JITTER_MS * unit + ); +} + +export function shouldRetrySameAccountTransport(options: { + status: unknown; + errorText?: string | null; + errorCode?: string | null; + errorType?: string | null; + attempt: number; + hasForcedConnection?: boolean; + hasEmittedOutput?: boolean; +}): boolean { + if (options.hasForcedConnection) return false; + if (options.hasEmittedOutput) return false; + if (options.attempt >= SAME_ACCOUNT_TRANSPORT_RETRY_MAX) return false; + return isRetryablePreOutputTransportError( + options.status, + options.errorText, + options.errorCode, + options.errorType + ); +} + +export function isTransportCooldownErrorCode(errorCode: unknown): boolean { + return isRetryableTransportStatus(errorCode); +} + +export function buildMixedAvailabilityError(options: { + provider: string; + quotaFilteredCount: number; + transportUnavailableCount: number; + transportStatus?: number | null; +}): { status: number; lastError: string; lastErrorCode: number } { + const quota = Math.max(0, options.quotaFilteredCount); + const transport = Math.max(0, options.transportUnavailableCount); + const upstreamStatus = options.transportStatus || 503; + return { + status: 503, + lastErrorCode: 503, + lastError: `No ${options.provider} accounts currently available: ${quota} quota-filtered, ${transport} temporarily unavailable after upstream ${upstreamStatus}`, + }; +} diff --git a/tests/unit/codex-same-account-transport-retry-9708.test.ts b/tests/unit/codex-same-account-transport-retry-9708.test.ts new file mode 100644 index 0000000000..e436df0080 --- /dev/null +++ b/tests/unit/codex-same-account-transport-retry-9708.test.ts @@ -0,0 +1,243 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const { + isRetryablePreOutputTransportError, + shouldRetrySameAccountTransport, + sameAccountTransportRetryDelayMs, + isTransportCooldownErrorCode, + buildMixedAvailabilityError, + SAME_ACCOUNT_TRANSPORT_RETRY_MAX, +} = await import("../../src/sse/services/sameAccountTransportRetry.ts"); + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-9708-codex-retry-")); +process.env.DATA_DIR = TEST_DATA_DIR; +process.env.API_KEY_SECRET ||= "codex-9708-test-secret"; + +const core = await import("../../src/lib/db/core.ts"); +const providersDb = await import("../../src/lib/db/providers.ts"); +const quotaCache = await import("../../src/domain/quotaCache.ts"); +const auth = await import("../../src/sse/services/auth.ts"); + +async function resetStorage() { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); +} + +function futureIso(ms = 60_000) { + return new Date(Date.now() + ms).toISOString(); +} + +async function seedConnection(provider: string, overrides: Record = {}) { + return providersDb.createProviderConnection({ + provider, + authType: overrides.authType || "oauth", + name: overrides.name || `${provider}-${Math.random().toString(16).slice(2, 8)}`, + accessToken: overrides.accessToken || `tok-${Math.random().toString(16).slice(2, 10)}`, + isActive: overrides.isActive ?? true, + testStatus: overrides.testStatus || "active", + priority: overrides.priority, + rateLimitedUntil: overrides.rateLimitedUntil, + lastError: overrides.lastError, + lastErrorType: overrides.lastErrorType, + errorCode: overrides.errorCode, + providerSpecificData: overrides.providerSpecificData || {}, + }); +} + +test.beforeEach(async () => { + await resetStorage(); +}); + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +test("#9708: 503 connection-reset and 507 buffer errors are retryable pre-output transport", () => { + assert.equal( + isRetryablePreOutputTransportError( + 503, + "upstream connect error or disconnect/reset before headers reset reason: remote connection failure" + ), + true + ); + assert.equal( + isRetryablePreOutputTransportError( + 507, + "exceeded request buffer limit while retrying upstream" + ), + true + ); + assert.equal(isRetryablePreOutputTransportError(504, "gateway timeout"), true); + assert.equal(isRetryablePreOutputTransportError(502, "Bad Gateway"), true); +}); + +test("#9708: quota, auth, and deterministic 400s never enter the same-account retry path", () => { + assert.equal( + isRetryablePreOutputTransportError( + 429, + "All codex accounts reached configured quota threshold" + ), + false + ); + assert.equal(isRetryablePreOutputTransportError(401, "unauthorized"), false); + assert.equal(isRetryablePreOutputTransportError(400, "prompt is too long"), false); + assert.equal( + shouldRetrySameAccountTransport({ + status: 503, + errorText: "remote connection failure", + attempt: 0, + hasForcedConnection: true, + }), + false + ); + assert.equal( + shouldRetrySameAccountTransport({ + status: 503, + errorText: "remote connection failure", + attempt: 0, + hasEmittedOutput: true, + }), + false + ); +}); + +test("#9708: same-account retry is bounded to exactly one attempt", () => { + assert.equal( + shouldRetrySameAccountTransport({ + status: 503, + errorText: "remote connection failure", + attempt: 0, + }), + true + ); + assert.equal( + shouldRetrySameAccountTransport({ + status: 503, + errorText: "remote connection failure", + attempt: SAME_ACCOUNT_TRANSPORT_RETRY_MAX, + }), + false + ); +}); + +test("#9708: retry delay stays in the 2-3s jitter window", () => { + assert.equal( + sameAccountTransportRetryDelayMs(() => 0), + 2000 + ); + assert.equal( + sameAccountTransportRetryDelayMs(() => 1), + 3000 + ); + assert.equal( + sameAccountTransportRetryDelayMs(() => 0.5), + 2500 + ); +}); + +test("#9708: mixed-cause pool error is 503, not all-accounts-quota 429", () => { + const mixed = buildMixedAvailabilityError({ + provider: "codex", + quotaFilteredCount: 2, + transportUnavailableCount: 1, + transportStatus: 507, + }); + assert.equal(mixed.status, 503); + assert.equal(mixed.lastErrorCode, 503); + assert.match(mixed.lastError, /2 quota-filtered/); + assert.match(mixed.lastError, /1 temporarily unavailable after upstream 507/); + assert.equal(mixed.lastError.includes("quota threshold"), false); +}); + +test("#9708: simulate first 503 then success on the same account; second failure rotates", () => { + function simulate(results: Array<{ status: number; error?: string; success?: boolean }>) { + let attempt = 0; + let markUnavailable = 0; + let i = 0; + let connectionId = "acct-a"; + while (true) { + const result = results[Math.min(i, results.length - 1)]; + if (result.success) { + return { outcome: "success", attempt, markUnavailable, connectionId }; + } + if ( + shouldRetrySameAccountTransport({ + status: result.status, + errorText: result.error, + attempt, + }) + ) { + attempt += 1; + i += 1; + continue; + } + markUnavailable += 1; + connectionId = "acct-b"; + return { outcome: "fallback", attempt, markUnavailable, connectionId }; + } + } + + const recovered = simulate([ + { status: 503, error: "remote connection failure" }, + { success: true, status: 200 }, + ]); + assert.equal(recovered.outcome, "success"); + assert.equal(recovered.attempt, 1); + assert.equal(recovered.markUnavailable, 0); + assert.equal(recovered.connectionId, "acct-a"); + + const rotated = simulate([ + { status: 507, error: "exceeded request buffer limit while retrying upstream" }, + { status: 507, error: "exceeded request buffer limit while retrying upstream" }, + ]); + assert.equal(rotated.outcome, "fallback"); + assert.equal(rotated.attempt, 1); + assert.equal(rotated.markUnavailable, 1); + assert.equal(rotated.connectionId, "acct-b"); +}); + +test("#9708: getProviderCredentials does not report all-quota 429 when a sibling is only transport-cooled", async () => { + const resetAt = futureIso(120_000); + const quotaA = await seedConnection("codex", { + name: "codex-quota-a", + priority: 1, + providerSpecificData: { + limitPolicy: { enabled: true, thresholdPercent: 75, windows: ["session"] }, + }, + }); + const quotaB = await seedConnection("codex", { + name: "codex-quota-b", + priority: 2, + providerSpecificData: { + limitPolicy: { enabled: true, thresholdPercent: 75, windows: ["session"] }, + }, + }); + await seedConnection("codex", { + name: "codex-transport-blip", + priority: 3, + rateLimitedUntil: futureIso(8_000), + errorCode: 507, + lastError: "exceeded request buffer limit while retrying upstream", + lastErrorType: "server_error", + }); + + quotaCache.setQuotaCache(quotaA.id, "codex", { + session: { remainingPercentage: 0, resetAt }, + }); + quotaCache.setQuotaCache(quotaB.id, "codex", { + session: { remainingPercentage: 0, resetAt }, + }); + + const result = await auth.getProviderCredentials("codex"); + assert.equal(result.allRateLimited, true); + assert.notEqual(result.lastErrorCode, 429); + assert.equal(result.lastErrorCode, 503); + assert.match(String(result.lastError), /temporarily unavailable after upstream 507/i); + assert.equal(isTransportCooldownErrorCode(507), true); +});