diff --git a/src/config.ts b/src/config.ts index 1308b3a64b..a5c3345471 100644 --- a/src/config.ts +++ b/src/config.ts @@ -445,6 +445,18 @@ const retryOn429PolicySchema = z.object({ respectRetryAfter: z.boolean().optional(), }).strict(); +/** + * Bounds for the opt-in transient-5xx retry policy. Delays intentionally stay + * non-configurable in the first version: the runtime uses the shared safe + * constants in lib/upstream-retry.ts (400ms base, 5s cap, Retry-After honored). + * Strict, so an unknown key is rejected at every validation boundary instead of + * being silently ignored. + */ +const transientRetryOn5xxPolicySchema = z.object({ + enabled: z.boolean().optional(), + attempts: z.number().int().min(1).max(10).optional(), +}).strict(); + const requestPacingRuleSchema = z.object({ // Keep the RPM-derived timer within the same one-hour bound as minIntervalMs. requestsPerMinute: z.number().min(1 / 60).max(60_000).optional(), @@ -517,6 +529,7 @@ const providerConfigSchema = z.object({ .transform(normalizeNonBlankStringArray) .optional(), retryOn429: retryOn429PolicySchema.optional(), + transientRetryOn5xx: transientRetryOn5xxPolicySchema.optional(), codexAccountMode: z.enum(["pool", "direct"]).optional(), // Validated rather than passed through: this schema ends in `.passthrough()`, so an // undeclared key survives verbatim. A misspelled `codexToolMode` therefore used to be @@ -1391,6 +1404,59 @@ function sanitizeRetryOn429ForLoad(parsed: unknown): void { } } +/** + * Load-time degradation for `transientRetryOn5xx` (loadConfig only), mirroring + * {@link sanitizeRetryOn429ForLoad}: one hand-edited invalid optional field must not + * trip the whole provider schema. Invalid fields are dropped with a warning; an + * explicitly present but invalid master switch drops the whole policy so a malformed + * disable-oriented edit stays disabled. + */ +function sanitizeTransientRetryOn5xxForLoad(parsed: unknown): void { + if (!parsed || typeof parsed !== "object") return; + const root = parsed as Record; + const providers = root.providers; + if (!providers || typeof providers !== "object" || Array.isArray(providers)) return; + for (const [name, provider] of Object.entries(providers as Record)) { + const safeProviderName = JSON.stringify(redactSecretString(name)); + if (!provider || typeof provider !== "object" || Array.isArray(provider)) continue; + const p = provider as Record; + const policy = p.transientRetryOn5xx; + if (policy === undefined) continue; + if (!policy || typeof policy !== "object" || Array.isArray(policy)) { + delete p.transientRetryOn5xx; + console.warn(`⚠️ config.json providers.${safeProviderName}.transientRetryOn5xx (${typeof policy}) is invalid — ignoring the policy`); + continue; + } + const policyRecord = policy as Record; + if ("enabled" in policyRecord && typeof policyRecord.enabled !== "boolean") { + delete p.transientRetryOn5xx; + console.warn(`⚠️ config.json providers.${safeProviderName}.transientRetryOn5xx.enabled (${typeof policyRecord.enabled}) is invalid — ignoring the whole policy`); + continue; + } + const policyShape = transientRetryOn5xxPolicySchema.shape; + const hadPolicyEntries = Object.keys(policyRecord).length > 0; + const cleaned: Record = {}; + for (const [key, fieldSchema] of Object.entries(policyShape)) { + const value = policyRecord[key]; + if (value === undefined) continue; + if (fieldSchema.safeParse(value).success) cleaned[key] = value; + else console.warn(`⚠️ config.json providers.${safeProviderName}.transientRetryOn5xx.${key} (${typeof value}) is invalid — ignoring the field`); + } + const knownKeys = new Set(Object.keys(policyShape)); + for (const key of Object.keys(policyRecord)) { + if (!knownKeys.has(key)) { + console.warn(`⚠️ config.json providers.${safeProviderName}.transientRetryOn5xx.${JSON.stringify(redactSecretString(key))} is not a recognized field — ignoring it`); + } + } + if (hadPolicyEntries && Object.keys(cleaned).length === 0) { + delete p.transientRetryOn5xx; + console.warn(`⚠️ config.json providers.${safeProviderName}.transientRetryOn5xx has no valid fields left — removing the policy (an empty policy would enable retries with defaults)`); + } else { + p.transientRetryOn5xx = cleaned; + } + } +} + /** * Management write-boundary validation for `retryOn429` (fail closed). Unlike the * lenient load-time sanitizer, invalid values and unknown keys are rejected outright so @@ -1413,6 +1479,26 @@ export function retryOn429PolicyConfigError(policy: unknown): string | null { return `retryOn429.${field} is invalid (${first.message})`; } +/** + * Management write-boundary validation for `transientRetryOn5xx` (fail closed), + * mirroring {@link retryOn429PolicyConfigError}. Reuses the shared policy schema. + * Never echoes values, and secret-shaped unknown field names are redacted. + */ +export function transientRetryOn5xxPolicyConfigError(policy: unknown): string | null { + if (policy === undefined) return null; + const result = transientRetryOn5xxPolicySchema.safeParse(policy); + if (result.success) return null; + const first = result.error.issues[0]; + if (!first) return "transientRetryOn5xx is invalid"; + if (first.code === "unrecognized_keys") { + const names = first.keys.map(key => JSON.stringify(redactSecretString(key))).join(", "); + return `transientRetryOn5xx has unrecognized field${first.keys.length > 1 ? "s" : ""}: ${names}`; + } + if (first.path.length === 0) return `transientRetryOn5xx is invalid (${first.message})`; + const field = String(first.path[first.path.length - 1]); + return `transientRetryOn5xx.${field} is invalid (${first.message})`; +} + /** * Load-time degradation for `providers..modelCosts`, mirroring * {@link sanitizeRetryOn429ForLoad}. A hand-edited malformed display-price row @@ -1816,6 +1902,7 @@ export function loadConfig(): OcxConfig { const parsed = JSON.parse(raw); sanitizeAliasesForLoad(parsed); sanitizeRetryOn429ForLoad(parsed); + sanitizeTransientRetryOn5xxForLoad(parsed); sanitizeModelCostsForLoad(parsed); const result = configSchema.safeParse(parsed); if (result.success) { @@ -2199,6 +2286,7 @@ function configDiagnosticsFromRaw(raw: string): ConfigDiagnostics { // schema and send the caller a default-config fallback (the config command could then // persist that fallback over the user's providers/keys). sanitizeRetryOn429ForLoad(parsed); + sanitizeTransientRetryOn5xxForLoad(parsed); sanitizeModelCostsForLoad(parsed); const result = configSchema.safeParse(parsed); if (result.success) { diff --git a/src/providers/key-failover.ts b/src/providers/key-failover.ts index 7a2d2d330d..ecd5965d1b 100644 --- a/src/providers/key-failover.ts +++ b/src/providers/key-failover.ts @@ -9,7 +9,7 @@ * Modelled after src/codex/routing.ts cooldown logic but scoped to plain API-key pools. */ import { saveConfigPreservingClaudeCode } from "../config"; -import type { OcxConfig, OcxProviderConfig, RateLimitRetryPolicy } from "../types"; +import type { OcxConfig, OcxProviderConfig, RateLimitRetryPolicy, TransientRetryPolicy } from "../types"; import { resolveProviderTransport, type OcxProviderTransport } from "./xai-transport"; import { sweepExpiredOnWrite } from "../lib/state-store-sweeper"; @@ -34,6 +34,33 @@ const DEFAULT_RATE_LIMIT_RETRY = { respectRetryAfter: true, } as const satisfies Required; +/** + * Default transient-5xx retry policy used when a provider opts in via a bare + * `transientRetryOn5xx: {}` (presence = opt-in with these defaults). + */ +const DEFAULT_TRANSIENT_RETRY = { + enabled: true, + attempts: 3, +} as const satisfies Required; + +/** + * Normalize a provider's `transientRetryOn5xx` policy, or return null when the knob is absent, + * explicitly disabled, or the provider is not key-auth. OAuth/forward credentials are + * never replayed on the same token, and local runtimes have no remote key to preserve. + * The returned policy is fully defaulted so callers never re-check fields. + */ +export function transientRetryPolicyFor( + provider: Pick, +): Required | null { + const policy = provider.transientRetryOn5xx; + if (!policy || policy.enabled === false) return null; + if (provider.authMode !== undefined && provider.authMode !== "key") return null; + return { + enabled: policy.enabled ?? DEFAULT_TRANSIENT_RETRY.enabled, + attempts: policy.attempts ?? DEFAULT_TRANSIENT_RETRY.attempts, + }; +} + /** Map<`${providerName}\0${keyId}`, KeyCooldown> */ const keyCooldowns = new Map(); diff --git a/src/server/chat-native.ts b/src/server/chat-native.ts index 2e84dc68d3..69ce39af15 100644 --- a/src/server/chat-native.ts +++ b/src/server/chat-native.ts @@ -21,6 +21,7 @@ import { isModelTextOnly } from "../vision"; import { applyUpstreamRecoveryInit, fetchWithResetRetry, + fetchWithTransientRetry, prepareSameTarget429Wait, type UpstreamSendRecovery, } from "../lib/upstream-retry"; @@ -33,6 +34,7 @@ import { rateLimitRetryDelayMs, rateLimitRetryPolicyFor, rotateProviderTransportOn429, + transientRetryPolicyFor, } from "../providers/key-failover"; import { fastPolicyForModel } from "../providers/service-tier"; import type { RouteResult } from "../router"; @@ -204,7 +206,9 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio const send = async (request: AdapterRequest, recovery?: "rate-limit-429" | "key-429"): Promise => { try { - return await fetchWithResetRetry( + const transientPolicy = transientRetryPolicyFor(activeProvider); + const doFetchWithRetry = transientPolicy ? fetchWithTransientRetry : fetchWithResetRetry; + return await doFetchWithRetry( (transportRecovery?: UpstreamSendRecovery) => { noteAttemptSend(attempt, logCtx.usageLogInputTokens, transportRecovery ?? recovery); return fetchWithHeaderTimeout( @@ -223,7 +227,11 @@ export async function handleNativeChatCompletions(options: HandleNativeChatOptio }), ); }, - { abortSignal: upstream.signal, label: safeHostLabel(request.url) }, + { + abortSignal: upstream.signal, + label: safeHostLabel(request.url), + ...(transientPolicy ? { attempts: transientPolicy.attempts } : {}), + }, ); } finally { request.releaseBodyObservation?.(); diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 691d75ada4..0844604873 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -218,6 +218,7 @@ import { rateLimitRetryDelayMs, rateLimitRetryPolicyFor, rotateProviderTransportOn429, + transientRetryPolicyFor, } from "../../providers/key-failover"; import { shouldAttemptImageTierRetry } from "../image-retry"; import { isXaiResponsesDestination, resolveProviderTransport } from "../../providers/xai-transport"; @@ -5278,11 +5279,12 @@ async function handleResponsesInner( }), }); } else { - // #1851 scope guard: transient-5xx retry on this generic adapter path is opt-in for - // direct Google AI Studio only (Vertex/Antigravity use fetchResponse above). Other - // adapters keep reset-only retry so combo failover still hops on the first 5xx + // #1851 scope guard: transient-5xx retry on this generic adapter path is opt-in via + // provider config (transientRetryOn5xx) or the legacy Google adapter exception. + // Other adapters keep reset-only retry so combo failover still hops on the first 5xx // instead of burning ~1.2s of same-target retries per hop. - const fetchWithRetryPolicy = route.provider.adapter === "google" ? fetchWithTransientRetry : fetchWithResetRetry; + const transientPolicy = transientRetryPolicyFor(route.provider); + const fetchWithRetryPolicy = transientPolicy || route.provider.adapter === "google" ? fetchWithTransientRetry : fetchWithResetRetry; upstreamResponse = await fetchWithRetryPolicy( recovery => { noteAttemptSend(logCtx.activeAttempt, inputTokenEstimate, recovery); @@ -5296,7 +5298,11 @@ async function handleResponsesInner( modelId: route.modelId, })); }, - { abortSignal: upstream.signal, label: safeHostLabel(builtInitialRequest.url) }, + { + abortSignal: upstream.signal, + label: safeHostLabel(builtInitialRequest.url), + ...(transientPolicy ? { attempts: transientPolicy.attempts } : {}), + }, ); } } catch (err) { @@ -5797,9 +5803,10 @@ async function handleResponsesInner( }), }); } - // Same #1851 scope guard as the initial send: transient-5xx retry only for direct - // Google AI Studio; every other adapter keeps reset-only semantics here. - const fetchContinuationWithRetryPolicy = route.provider.adapter === "google" ? fetchWithTransientRetry : fetchWithResetRetry; + // Same #1851 scope guard as the initial send: transient-5xx retry via + // provider config (transientRetryOn5xx) or the legacy Google adapter exception. + const transientPolicy = transientRetryPolicyFor(route.provider); + const fetchContinuationWithRetryPolicy = transientPolicy || route.provider.adapter === "google" ? fetchWithTransientRetry : fetchWithResetRetry; return await fetchContinuationWithRetryPolicy( recovery => { noteAttemptSend(logCtx.activeAttempt, continuationEstimate, recovery ?? replayKind); @@ -5819,8 +5826,12 @@ async function handleResponsesInner( }), ); }, - { abortSignal: upstream.signal, label: safeHostLabel(builtContinuationRequest.url) }, - ); + { + abortSignal: upstream.signal, + label: safeHostLabel(builtContinuationRequest.url), + ...(transientPolicy ? { attempts: transientPolicy.attempts } : {}), + }, + ); } finally { builtContinuationRequest.releaseBodyObservation?.(); } diff --git a/src/types.ts b/src/types.ts index 68f3cd4f7c..cec4306367 100644 --- a/src/types.ts +++ b/src/types.ts @@ -90,6 +90,7 @@ export type { OpenRouterProviderRouting, ResponsesItemIdRepairConfig, RateLimitRetryPolicy, + TransientRetryPolicy, ProviderCostOverlay, RequestPacingRule, ProviderRequestPacingConfig, diff --git a/src/types/provider.ts b/src/types/provider.ts index b7ba042506..38f7dbd414 100644 --- a/src/types/provider.ts +++ b/src/types/provider.ts @@ -51,6 +51,21 @@ export interface RateLimitRetryPolicy { respectRetryAfter?: boolean; } +/** + * Transient-5xx retry policy (providers..transientRetryOn5xx). When present and not + * explicitly disabled, the proxy retries the initial request with exponential backoff on + * transient upstream statuses (500/502/503/504/520/521/522) before surfacing the error. + * All fields optional; the runtime applies the shared defaults (attempts=3, + * enabled=true) with the fixed backoff constants in lib/upstream-retry.ts + * (400ms base, 5s cap, Retry-After honored). Only honored for key-auth providers. + */ +export interface TransientRetryPolicy { + /** Master switch. The presence of the object also enables the policy (default true). */ + enabled?: boolean; + /** Extra retry attempts after the first transient failure (1..10, default 3). */ + attempts?: number; +} + /** * User-configured display price for one model (USD per 1M tokens). * Mirrors the `Cost4` shape used by the usage cost estimator; structurally @@ -537,6 +552,15 @@ export interface OcxProviderConfig { * before any response bytes are relayed, so the replay is lossless. */ retryOn429?: RateLimitRetryPolicy; + /** + * Opt-in transient-5xx retry policy. When present, the proxy retries the initial + * request with exponential backoff on transient upstream statuses (500/502/503/504/520/521/522) + * before surfacing the error. Pre-stream only: a transient error arrives before any + * response bytes are relayed, so the replay is lossless. This extends the same protection + * that the Google adapter already receives via fetchWithTransientRetry to any key-auth + * provider that opts in. Disabled by default. + */ + transientRetryOn5xx?: TransientRetryPolicy; /** * Model ids whose OpenAI-compatible chat endpoint accepts `reasoning_split: true` and returns * thinking separately in `reasoning_content` / `reasoning_details` instead of visible content. diff --git a/tests/transient-retry-policy.test.ts b/tests/transient-retry-policy.test.ts new file mode 100644 index 0000000000..a9c8257d08 --- /dev/null +++ b/tests/transient-retry-policy.test.ts @@ -0,0 +1,203 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { transientRetryPolicyFor } from "../src/providers/key-failover"; +import { handleResponses } from "../src/server/responses"; +import { transientRetryOn5xxPolicyConfigError } from "../src/config"; +import type { OcxConfig, OcxProviderConfig } from "../src/types"; + +describe("transientRetryPolicyFor", () => { + test("null when absent or explicitly disabled", () => { + expect(transientRetryPolicyFor({} as OcxProviderConfig)).toBeNull(); + expect(transientRetryPolicyFor({ transientRetryOn5xx: { enabled: false } } as OcxProviderConfig)).toBeNull(); + }); + + test("null for OAuth, forward, local, and unknown auth modes (fail closed)", () => { + expect(transientRetryPolicyFor({ + authMode: "oauth", + transientRetryOn5xx: {}, + } as OcxProviderConfig)).toBeNull(); + expect(transientRetryPolicyFor({ + authMode: "forward", + transientRetryOn5xx: {}, + } as OcxProviderConfig)).toBeNull(); + expect(transientRetryPolicyFor({ + authMode: "local", + transientRetryOn5xx: {}, + } as OcxProviderConfig)).toBeNull(); + expect(transientRetryPolicyFor({ + authMode: "custom-unknown", + transientRetryOn5xx: {}, + } as OcxProviderConfig)).toBeNull(); + expect(transientRetryPolicyFor({ + authMode: "key", + transientRetryOn5xx: {}, + } as OcxProviderConfig)).not.toBeNull(); + }); + + test("applies defaults when the object is present", () => { + expect(transientRetryPolicyFor({ transientRetryOn5xx: {} } as OcxProviderConfig)).toEqual({ + enabled: true, + attempts: 3, + }); + }); + + test("honors explicit attempts", () => { + expect(transientRetryPolicyFor({ + transientRetryOn5xx: { attempts: 5 }, + } as OcxProviderConfig)).toEqual({ + enabled: true, + attempts: 5, + }); + }); +}); + +describe("transientRetryOn5xxPolicyConfigError", () => { + test("accepts undefined and valid policies", () => { + expect(transientRetryOn5xxPolicyConfigError(undefined)).toBeNull(); + expect(transientRetryOn5xxPolicyConfigError({})).toBeNull(); + expect(transientRetryOn5xxPolicyConfigError({ enabled: true, attempts: 3 })).toBeNull(); + }); + + test("rejects invalid field values", () => { + expect(transientRetryOn5xxPolicyConfigError({ enabled: "yes" })).toContain("enabled"); + expect(transientRetryOn5xxPolicyConfigError({ attempts: 0 })).toContain("attempts"); + expect(transientRetryOn5xxPolicyConfigError({ attempts: 99 })).toContain("attempts"); + }); + + test("rejects unknown fields without echoing secret-shaped names", () => { + const err = transientRetryOn5xxPolicyConfigError({ intervalMs: 1000 }); + expect(err).toContain("unrecognized"); + expect(err).toContain("intervalMs"); + }); +}); + +describe("transient-5xx fetch path", () => { + const originalFetch = globalThis.fetch; + + afterEach(() => { + globalThis.fetch = originalFetch; + }); + + function makeConfig(transientRetryOn5xx?: unknown): OcxConfig { + return { + port: 0, + defaultProvider: "corp", + providers: { + corp: { + adapter: "openai-chat", + baseUrl: "https://gateway.corp.example", + authMode: "key", + apiKey: "key-alpha-000111222333", + ...(transientRetryOn5xx !== undefined ? { transientRetryOn5xx } : {}), + }, + }, + } as OcxConfig; + } + + function postResponses(config: OcxConfig): Promise { + return handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "corp/some-model", input: "hello", stream: false }), + }), config, { model: "corp/some-model", provider: "corp" }); + } + + function okPayload(): Response { + return new Response(JSON.stringify({ + id: "chatcmpl-mock", + object: "chat.completion", + created: 1, + model: "some-model", + choices: [{ index: 0, message: { role: "assistant", content: "ok" }, finish_reason: "stop" }], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }), { status: 200, headers: { "content-type": "application/json" } }); + } + + test("503 then success: retries transparently and returns 200", async () => { + let sends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (url.startsWith("https://gateway.corp.example/")) { + sends += 1; + if (sends < 3) { + return new Response(JSON.stringify({ error: { message: "server_is_overloaded" } }), + { status: 503, headers: { "content-type": "application/json" } }); + } + return okPayload(); + } + return originalFetch(input, init); + }) as typeof fetch; + + const response = await postResponses(makeConfig({ attempts: 3 })); + expect(response.status).toBe(200); + expect(sends).toBe(3); + }); + + test("retries exhausted: final 503 is surfaced to the client", async () => { + let sends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (url.startsWith("https://gateway.corp.example/")) { + sends += 1; + return new Response(JSON.stringify({ error: { message: "server_is_overloaded" } }), + { status: 503, headers: { "content-type": "application/json" } }); + } + return originalFetch(input, init); + }) as typeof fetch; + + const response = await postResponses(makeConfig({ attempts: 2 })); + expect(response.status).toBe(503); + expect(sends).toBe(2); + }); + + test("no opt-in: 503 is surfaced immediately with a single send", async () => { + let sends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (url.startsWith("https://gateway.corp.example/")) { + sends += 1; + return new Response(JSON.stringify({ error: { message: "server_is_overloaded" } }), + { status: 503, headers: { "content-type": "application/json" } }); + } + return originalFetch(input, init); + }) as typeof fetch; + + const response = await postResponses(makeConfig()); + expect(response.status).toBe(503); + expect(sends).toBe(1); + }); + + test("explicitly disabled: 503 is surfaced immediately", async () => { + let sends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (url.startsWith("https://gateway.corp.example/")) { + sends += 1; + return new Response(JSON.stringify({ error: { message: "server_is_overloaded" } }), + { status: 503, headers: { "content-type": "application/json" } }); + } + return originalFetch(input, init); + }) as typeof fetch; + + const response = await postResponses(makeConfig({ enabled: false })); + expect(response.status).toBe(503); + expect(sends).toBe(1); + }); + + test("400 is never retried even with the policy enabled", async () => { + let sends = 0; + globalThis.fetch = (async (input, init) => { + const url = input instanceof Request ? input.url : String(input); + if (url.startsWith("https://gateway.corp.example/")) { + sends += 1; + return new Response(JSON.stringify({ error: { message: "bad request" } }), + { status: 400, headers: { "content-type": "application/json" } }); + } + return originalFetch(input, init); + }) as typeof fetch; + + const response = await postResponses(makeConfig({ attempts: 3 })); + expect(response.status).toBe(400); + expect(sends).toBe(1); + }); +}); +