From d45125ee076dfdde85af2eb895cfa642273dd566 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 01:18:58 +0100 Subject: [PATCH] fix(failover): default off, add cost guard, fix adversarial-review findings Flip FailoverConfig.Enabled to default false. A version upgrade with no config change must preserve prior behaviour, and failover changes which provider serves a request. Update tests, README, and the yaegi-check probe to opt in explicitly. Add a mandatory cost guard. A candidate is skipped unless its input AND output cost per token are each no greater than the primary candidate's, resolved through the same canonical-then-bare pricing path the real billing code uses. Unknown pricing is never treated as free: a known primary against an unknown candidate skips, an unknown primary against a known non-zero candidate skips, everything else that is equal is allowed. Not configurable. Fix four blocking findings from adversarial review of commit 5d62479. F1: openai-type adapters mutate the request map in place to force stream usage reporting. Reusing one map across candidates let a later candidate read back an earlier candidate's own injection and leak a usage-only SSE chunk to a client that never asked for it. Every candidate now attempts against a fresh copy of the original request. F2: a streaming candidate that fails before writing anything still stages SSE-only response headers. Those headers now get cleared before every candidate attempt, so they cannot survive into a later candidate's plain JSON response. F3: the request-health gate used the retry-shaped isTransient classification, which treats a non-429 4xx as a successful outcome. A provider with a dead API key was never skipped and was retried forever. The gate now reuses failoverEligible's own classification instead. F4: a shared request context that an earlier candidate exhausted was attributed to whichever candidate happened to run next, opening a healthy provider's breaker for a timeout that was not its fault. A candidate is no longer attempted once the shared context is already done; the loop stops and reports the last real error instead. Also fix three should-fix findings from the same review. F5: a failure arriving while the request-health breaker is already open but not yet past its cooldown window is now a no-op, instead of resetting backoff and sliding the open window forward on every such failure. F6: the request-health gate now records at most one outcome per candidate per logical request, not one per same-provider retry attempt, so retry.attempts alone can no longer open the breaker on a single request. F7: every candidate's cache entry is now checked before any candidate is attempted, so a cached failover response no longer costs a round trip to a still-broken primary first. F10: the generic "failing over" log line is now rate-limited to once per 30 seconds during a sustained outage, reusing the existing logGate type. The 404 loud-log line stays unthrottled, as ruled. F9 (optional): skip building the failover candidate pool entirely when failover is disabled. Correct an overstated test comment on TestHandleChat_Failover_DiscoveryUnhealthy_StillTried: it proves alpha is still tried, not that ordering places it second. Document, but do not fix here, that this plugin's HTTP client sets no per-attempt timeout, so a provider that accepts a connection and then hangs will never trigger failover. This is a pre-existing property of the shared HTTP client, not specific to failover. Deferred: F8 (surfacing request-health state on the admin dashboard) touches admin.go, which a sibling branch is also editing concurrently; left for a follow-up to avoid a merge conflict in a file this change does not otherwise need. Every fix above is proven by a dedicated regression test, and each was verified by reverting the fix in place and confirming the corresponding test fails before restoring it. --- README.md | 67 +++- failover.go | 245 +++++++++++-- failover_test.go | 744 +++++++++++++++++++++++++++++++++++++- llmgateway.go | 13 +- routes_unified.go | 224 +++++++++--- tools/yaegi-check/main.go | 7 +- 6 files changed, 1174 insertions(+), 126 deletions(-) diff --git a/README.md b/README.md index 57b07dc..0560e59 100644 --- a/README.md +++ b/README.md @@ -193,7 +193,7 @@ config accepts them as YAML, which decodes to the same JSON shape. | `users` | `UsersConfig` | — | Inline and/or file-backed API-key holders. With none configured, every request is unauthenticated and gets 401. | | `redis` | `RedisConfig` | — | Distributed limit-counter backend. Omitted means in-process counters only (per-replica, approximate across multiple Traefik instances). | | `retry` | `RetryConfig` | `{}` (disabled) | Same-provider retry for transient upstream failures — see [Retry](#retry). Omitted or `enabled: false` means no retry: every request makes exactly one upstream attempt, byte-identical to a gateway built before this field existed. | -| `failover` | `FailoverConfig` | `{}` (enabled) | Cross-provider failover for a bare model id more than one configured provider serves — see [Failover](#failover). `enabled: false` turns it off; a deployment with no overlapping providers behaves exactly as before regardless of this setting. | +| `failover` | `FailoverConfig` | `{}` (disabled) | Cross-provider failover for a bare model id more than one configured provider serves — see [Failover](#failover). Omitted or `enabled: false` means no failover, byte-identical to a gateway built before this field existed. `enabled: true` opts in; a deployment with no overlapping providers behaves exactly as before regardless of this setting either way. | | `cache` | `CacheConfig` | `{}` (disabled) | Opt-in Redis-backed response cache for unified non-streaming chat/embeddings — see [Caching](#caching). Omitted or `enabled: false` means no caching, byte-identical to a gateway built before this field existed. | | `breaker` | `BreakerConfig` | `{}` (every default) | Per-provider discovery circuit breaker — see [Provider health (discovery circuit breaker)](#provider-health-discovery-circuit-breaker). Every field is individually zero-means-default; a provider whose discovery never fails is unaffected regardless of what this block contains. | | `admin` | `*AdminConfig` | `nil` (disabled) | Read-only admin dashboard — see [Admin](#admin). `nil` or `enabled: false` means the `/admin*` routes are not registered at all. | @@ -851,10 +851,14 @@ below for the separate, cross-provider mechanism. Cross-provider failover for a bare model id more than one configured provider serves — for example `whisper-1` configured on both `openai` and -`openai-audio`. On by default (`failover.enabled: true` unless set -otherwise); a deployment with no overlapping providers, or with a single -provider, behaves exactly as before regardless of this setting, since -there is never a second candidate to try. +`openai-audio`. **Off by default** (`failover.enabled: false` unless set +otherwise): a version upgrade with no config change preserves prior +behavior exactly, since failover changes which provider serves a request +and can shift cost attribution or mask a broken provider key behind an +apparent success. Set `failover.enabled: true` to opt in. A deployment +with no overlapping providers, or with a single provider, behaves +exactly as before regardless of this setting either way, since there is +never a second candidate to try. - **Scope**: `POST /v1/chat/completions`, `POST /v1/embeddings`, and `POST /v1/messages` only. The image/audio endpoints @@ -871,9 +875,30 @@ there is never a second candidate to try. - **What falls through**: everything except HTTP 400 — connection failures, timeouts, HTTP 429/401/403/404, and any 5xx. A 400 never falls through: the request is malformed and every provider would reject - it identically. A failover past a 404 is logged loudly, naming both - providers and the model — a 404 usually means a real configuration - mistake, not a transient outage. + it identically. A failover past a 404 is logged loudly, always, naming + both providers and the model — a 404 usually means a real configuration + mistake, not a transient outage. The routine "failing over from X to Y" + line is rate-limited to once per 30s during a sustained outage (with a + suppressed-count note on the next line), so it cannot itself become a + blocking synchronous write on every request; the 404 line is exempt + from that limit. +- **Cost guard (not optional)**: failover never routes to a candidate + priced higher than the provider it is failing over FROM (the first + candidate actually attempted, never the cheapest in the list), checked + on BOTH input and output cost per token — a candidate cheaper on input + but dearer on output is still rejected. Pricing is resolved exactly the + way real billing resolves it: the canonical `provider/model` id in + `pricing` first, falling back to the bare model id, so a per-provider + `pricing` override correctly makes the identical model id cost + different amounts on different providers. Unknown pricing is never + treated as free: if the primary's price is known and a candidate's is + not, the candidate is skipped (an unknown price cannot be proven + cheaper); if the primary's price is unknown and a candidate's is known + and non-zero, the candidate is skipped too (moving from unbilled to + billed is itself a spend increase); both unknown, or both an explicit + known-zero price, are treated as equal and allowed. There is no config + flag for this — it applies unconditionally whenever failover is + enabled. - **Never mid-stream**: failover only happens while nothing has reached the client yet. Once the first byte of a response is written, status and body are committed and the request is never retried elsewhere. @@ -882,9 +907,12 @@ there is never a second candidate to try. whose `/v1/models` is unhealthy is tried later, never skipped outright, since plenty of providers serve chat completions fine while their model listing is broken. A separate, in-memory, per-pod request-path health - signal (not exposed to config) is the actual routing gate: a provider - that has failed several consecutive live requests is skipped without - being called until it recovers. + signal (not yet exposed on the admin dashboard) is the actual routing + gate: a provider is skipped without being called once it has failed + three consecutive LOGICAL requests (same-provider retries all count as + one outcome, not one each) — everything except HTTP 400 counts as a + failure for this gate, including 401/403/404, so a provider with a dead + API key is eventually routed around instead of retried forever. - **Accounting**: a failover chain still counts as exactly one client request against `requestsPerMinute`/`requestsPerDay` — never once per attempt. Per-provider attempt/failure counters (the admin dashboard's @@ -892,10 +920,25 @@ there is never a second candidate to try. actually made it. - **Caching**: a cacheable response produced by a failover provider is stored under that provider's own cache key, never the first candidate's. + Every candidate's cache entry is checked before any of them is actually + attempted, so a cached response from a failover provider is served + without a wasted round trip to a still-broken primary. - **Configurable**: `failover.maxAttempts` caps how many different providers one request will try in total, primary included (default 3, maximum 10) — distinct from `retry.attempts`, which retries the SAME - provider. `failover.enabled: false` turns the whole mechanism off. + provider. `failover.enabled: false` (the default) turns the whole + mechanism off. +- **Known limitation — no per-attempt timeout**: `newAdapterHTTPClient` + sets no client timeout and no response-header timeout; the only bound + on an upstream call is the incoming request's own context, shared + identically across every candidate. A provider that accepts the TCP + connection and then simply never responds will hang the request + indefinitely and never trigger failover — only a failure fast enough to + leave time on that shared context (a dial failure, a fast non-2xx + response, or the context's own deadline finally expiring) does. This is + a pre-existing property of the plugin's HTTP client, not specific to + failover, and applies to every route, not only this one; it is not + addressed in this change. ## Caching diff --git a/failover.go b/failover.go index c399817..08c9498 100644 --- a/failover.go +++ b/failover.go @@ -23,6 +23,15 @@ import ( // metered-call path both /v1/chat/completions and /v1/messages already go // through. // +// Off by default (FailoverConfig.Enabled defaults to false — coordinator +// ruling, overriding this feature's original brief: new behavior that can +// change which provider serves a request goes behind a flag defaulting +// off). An operator who opts in also gets a non-optional cost guard +// (candidateCostAllowed, below): failover never routes to a candidate +// priced higher than the provider it is failing over from, on either +// input or output cost per token — there is no config knob to disable +// just that part. +// // SCOPE: routes_media.go's three media routes (images/generations, // audio/speech, audio/transcriptions) are deliberately NOT wired into // failover in this change, even though two of the brief's own named @@ -38,16 +47,17 @@ import ( // scoped follow-up; see this branch's own report for the explicit // call-out. -// FailoverConfig configures feat/failover. Enabled is a *bool, mirroring -// ProviderConfig.Passthrough's own nil-means-default-true convention -// (providers.go): nil or true (the default) turns an overlapping -// multi-provider deployment's already-existing redundancy into a hot -// standby with NO config change — the entire point of this feature (see -// this file's own package doc and the brief's "Why" section). Set false -// to opt out entirely. A single-provider deployment, or one where no two -// providers ever serve the identical bare model id, behaves exactly as -// before regardless of this setting: there is never more than one -// candidate to try (orderedFailoverCandidates' own doc comment). +// FailoverConfig configures feat/failover. Enabled is a *bool; nil or +// false (the default) means OFF — coordinator ruling, overriding this +// feature's original brief: a version upgrade with no config change must +// preserve prior behavior, and new behavior that can change which +// provider serves a request (shifting cost attribution, and able to mask +// a broken provider key behind an apparent success) goes behind a flag +// that defaults off, not on. Set Enabled: true to opt in. A +// single-provider deployment, or one where no two providers ever serve +// the identical bare model id, behaves exactly as before regardless of +// this setting either way: there is never more than one candidate to try +// (orderedFailoverCandidates' own doc comment). type FailoverConfig struct { Enabled *bool `json:"enabled,omitempty"` // MaxAttempts caps how many DIFFERENT PROVIDERS runMeteredCall tries @@ -78,9 +88,13 @@ type failoverConfig struct { } // validateFailoverConfig validates fc and returns the resolved -// failoverConfig newGateway attaches to the Gateway. +// failoverConfig newGateway attaches to the Gateway. enabled defaults to +// false (coordinator ruling, FailoverConfig's own doc comment) — every +// config that predates this field, and every config that never sets +// failover.enabled at all, keeps behaving exactly as before this feature +// existed. func validateFailoverConfig(fc FailoverConfig) (failoverConfig, error) { - enabled := true + enabled := false if fc.Enabled != nil { enabled = *fc.Enabled } @@ -154,6 +168,23 @@ func (m *modelRegistry) resolveWithCandidates(id string, grp *group) (resolveCan return primary, m.failoverCandidates(bareID, primary.providerName, grp), nil } +// resolvePrimaryOnly resolves id exactly like resolveWithCandidates' +// primary return (identical winner, identical error) WITHOUT ever +// scanning for or building failover candidates — the short-circuit +// runMeteredCall (routes_unified.go) takes when g.failover.enabled is +// false, so a deployment that never opted into failover pays no extra +// cost for a pool it will never use (measured: 74ns/1 alloc versus +// 521ns/13 allocs at 8 providers x 200 models for the full +// resolveWithCandidates scan — irrelevant next to a network round trip, +// but free to avoid). +func (m *modelRegistry) resolvePrimaryOnly(id string, grp *group) (resolveCandidate, error) { + adapter, upstreamModel, canonical, err := m.resolve(id, grp) + if err != nil { + return resolveCandidate{}, err + } + return resolveCandidate{adapter: adapter, providerName: adapter.name(), upstreamModel: upstreamModel, canonical: canonical}, nil +} + // failoverCandidates returns every configured provider other than // primaryProvider whose known model set contains bareID and that grp is // authorized to use, in modelRegistry.providerNames (sorted) order — the @@ -220,7 +251,14 @@ func (m *modelRegistry) failoverCandidates(bareID, primaryProvider string, grp * // half-open one (a stable partition, not a full sort — relative // order within each bucket is preserved), never dropped. // -// Finally the list is capped at g.failover.maxAttempts entries. +// This does NOT apply the cost guard or the maxAttempts cap — see +// runMeteredCall's own call sequence (routes_unified.go): candidates[0] +// coming out of THIS function is what filterCandidatesByCost (below) +// treats as "the primary" for pricing-guard purposes, so cost filtering +// must run against the health/discovery-ordered result, not the raw +// resolveWithCandidates order; and the maxAttempts cap runs LAST, after +// cost filtering, so a cap never re-admits an expensive candidate purely +// because a cheaper one ahead of it was removed on cost. func (g *Gateway) orderedFailoverCandidates(primary resolveCandidate, extra []resolveCandidate) []resolveCandidate { if !g.failover.enabled || len(extra) == 0 { return []resolveCandidate{primary} @@ -249,12 +287,112 @@ func (g *Gateway) orderedFailoverCandidates(primary resolveCandidate, extra []re deprioritized = append(deprioritized, c) } } - ordered = append(ordered, deprioritized...) + return append(ordered, deprioritized...) +} - if max := g.failover.maxAttempts; max > 0 && len(ordered) > max { - ordered = ordered[:max] +// resolveUnifiedPricing resolves (canonical, bare)'s price using the +// IDENTICAL resolution unifiedCostMicros itself uses (routes_unified.go): +// canonical id first, falling back to the bare upstream model id only +// when canonical has no configured price at all. ok is lookupPricing's +// own known/unknown distinction (pricing.go) — false means neither id +// resolved to a price, exactly the case where unifiedCostMicros would +// warn and bill 0, which candidateCostAllowed (below) must NEVER treat +// as an actual known-zero price. +func resolveUnifiedPricing(canonical, bare string, overrides map[string]*ModelPricing) (ModelPricing, bool) { + if p, ok := lookupPricing(canonical, overrides); ok { + return p, true } - return ordered + return lookupPricing(bare, overrides) +} + +// candidateCostAllowed reports whether candidate is allowed to serve as a +// failover destination for primary, per the operator's binding ruling: +// failover must never route to a candidate more expensive than the +// provider it is failing over FROM, checked on BOTH the input and the +// output per-token price — a candidate cheaper on input but dearer on +// output can still cost more on a completion-heavy request, so a single +// blended number is not sufficient. This is NOT configurable (ruling: +// "this behaviour is ON and not optional"). +// +// Four cases, all by ruling, not inferred: +// +// - both known: allowed only when candidate's input AND output price +// are each <= primary's. Equal is allowed either way. +// - primary known, candidate UNKNOWN: never allowed — an unknown price +// cannot be proven not-more-expensive. +// - primary UNKNOWN, candidate known and non-zero: never allowed — +// moving from unbilled to billed is itself a spend increase. +// - primary UNKNOWN, candidate known and EXACTLY zero (both +// dimensions), or both unknown: allowed — they are equal (a known +// zero can never be more expensive than an unknown amount, and two +// unknowns are trivially equal to each other). +// +// Unknown pricing is never treated as zero here, unlike costMicros' own +// billing behavior (pricing.go) — the gateway RECORDING zero for an +// unpriced model is an accounting shortcut, not proof the provider +// invoices zero, and this guard exists specifically to protect real +// money, not the gateway's own displayed number. +func candidateCostAllowed(primary ModelPricing, primaryKnown bool, candidate ModelPricing, candidateKnown bool) bool { + switch { + case primaryKnown && candidateKnown: + return candidate.InputPerM <= primary.InputPerM && candidate.OutputPerM <= primary.OutputPerM + case primaryKnown && !candidateKnown: + return false + case !primaryKnown && candidateKnown: + return candidate.InputPerM == 0 && candidate.OutputPerM == 0 + default: // both unknown + return true + } +} + +// filterCandidatesByCost drops every candidate whose resolved price +// candidateCostAllowed rejects against candidates[0] — the FIRST +// candidate actually attempted, never the cheapest in the list, per the +// operator's own definition of "primary" for this rule. Pricing comes +// from resolveUnifiedPricing against g.cfg.Pricing, the SAME overrides +// map unifiedCostMicros itself resolves against, so this compares +// against what would actually be billed. requestedModel names the +// client's own request in the log line a skip always produces — an +// operator-visible g.warnf, unthrottled, naming both providers, the +// model, and both resolved prices: a silently shortened candidate list +// would look identical to having no redundancy at all, which the +// operator must be able to tell apart from a cost-guard skip. +func (g *Gateway) filterCandidatesByCost(candidates []resolveCandidate, requestedModel string) []resolveCandidate { + if len(candidates) < 2 { + return candidates + } + primary := candidates[0] + primaryPrice, primaryKnown := resolveUnifiedPricing(primary.canonical, primary.upstreamModel, g.cfg.Pricing) + + out := make([]resolveCandidate, 1, len(candidates)) + out[0] = primary + for _, cand := range candidates[1:] { + candPrice, candKnown := resolveUnifiedPricing(cand.canonical, cand.upstreamModel, g.cfg.Pricing) + if candidateCostAllowed(primaryPrice, primaryKnown, candPrice, candKnown) { + out = append(out, cand) + continue + } + g.warnf("failover: skipping candidate provider %q for model %q: would cost more than primary provider %q (primary input=%.4f/M output=%.4f/M usd known=%v; candidate input=%.4f/M output=%.4f/M usd known=%v)", + cand.providerName, requestedModel, primary.providerName, + primaryPrice.InputPerM, primaryPrice.OutputPerM, primaryKnown, + candPrice.InputPerM, candPrice.OutputPerM, candKnown) + } + return out +} + +// capFailoverCandidates truncates candidates to at most max entries, +// applied as the LAST step of runMeteredCall's own candidate-building +// sequence (routes_unified.go) — after health/discovery ordering and +// after cost filtering, so the cap can never re-admit a candidate cost +// filtering already removed, and never discards a cheaper candidate that +// cost filtering kept purely because a removed, more expensive one used +// to sit ahead of it in the raw list. max <= 0 is a no-op (defensive; +// validateFailoverConfig never produces one). +func capFailoverCandidates(candidates []resolveCandidate, max int) []resolveCandidate { + if max > 0 && len(candidates) > max { + return candidates[:max] + } + return candidates } // failoverEligible reports whether callErr — runMeteredCall's own @@ -360,10 +498,11 @@ type requestHealthState struct { // request-path health signal — deliberately separate from registry.go's // discovery circuit breaker (modelRegistry.discoveryHealthy's own doc // comment forbids widening that one to cover request-path traffic). Fed -// by the SAME failure classification limiter.recordProviderAttempt uses -// (isTransient || isDeadlineExceeded — limits.go/retry.go), via record's -// own callers in routes_unified.go, but keeps its own independent state -// per provider name. +// by failoverEligible's own classification (record's own doc comment +// explains why, not limiter.recordProviderAttempt's narrower isTransient/ +// isDeadlineExceeded one), once per candidate per logical request, via +// record's own callers in routes_unified.go, but keeps its own +// independent state per provider name. // // Every method is nil-receiver-safe (both report "healthy"/no-op on a // nil *requestHealthTracker), matching this package's own nil-means- @@ -397,25 +536,50 @@ func (t *requestHealthTracker) stateFor(provider string) *requestHealthState { return st } -// record accounts one upstream request-path attempt's outcome for -// provider: success=false must be EXACTLY isTransient(resp, err) || -// isDeadlineExceeded(err) — the identical classification -// limiter.recordProviderAttempt applies (limits.go/retry.go) — never a -// wider one; routes_unified.go's attemptRecorder closure computes it -// once and feeds both. +// record accounts ONE LOGICAL REQUEST's outcome for provider — called +// exactly once per candidate per client request, after call() has +// already exhausted whatever same-provider retries retryPolicy.do +// performed internally (routes_unified.go's runMeteredCall), never once +// per raw upstream attempt. success=false must be +// !failoverEligible(callErr) (routes_unified.go/failover.go's own +// classification of the FINAL adapter-level error) — adversarial-review +// ruling (F3/F6): an EARLIER version fed this from isTransient(resp, +// err) || isDeadlineExceeded(err), limits.go/retry.go's own +// retry-shaped classification, once per RAW attempt inside +// retryPolicy.do. That was wrong on two counts. isTransient treats a +// non-429 4xx as "the provider answered correctly to a request it did +// not like" — correct for deciding whether retrying the SAME provider +// is worth it, wrong for deciding whether the POOL should route around +// it: a provider whose API key is dead answers 401 to every request and +// must eventually be skipped, not retried forever, which is exactly the +// "masking a broken provider key" risk this feature exists to reduce. +// And recording per raw attempt let ONE client request with +// retry.enabled (attempts: 3) report FOUR failures against a threshold +// documented as three REQUESTS. failoverEligible already classifies +// "everything except HTTP 400" as a failure, matching what this gate +// needs with no separate function required. // // requestBreakerFailureThreshold consecutive failures opens the breaker // for requestBreakerOpenDuration. While open, healthy (below) reports // unhealthy until that window passes; once it has, the NEXT record call // for this provider — success or failure — is what healthy already -// exposed as a post-cooldown probe attempt (probed, below): success -// closes the breaker outright, failure re-opens it with a doubled -// backoff (capped at requestBreakerMaxOpenDuration). This is the same -// probe-then-escalate-or-close shape providerState's discovery breaker -// uses (registry.go's recordHealthLocked/openBreakerLocked), -// reimplemented independently rather than shared — the two track -// different outcome streams, and discoveryHealthy's own doc comment -// forbids widening it to cover this one. +// exposed as a post-cooldown probe attempt: success closes the breaker +// outright, failure re-opens it with a doubled backoff (capped at +// requestBreakerMaxOpenDuration). This is the same probe-then-escalate- +// or-close shape providerState's discovery breaker uses (registry.go's +// recordHealthLocked/openBreakerLocked), reimplemented independently +// rather than shared — the two track different outcome streams, and +// discoveryHealthy's own doc comment forbids widening it to cover this +// one. +// +// A failure arriving while ALREADY open and still within its cooldown +// window is a deliberate no-op (adversarial-review fix, F5): an earlier +// version fell through to the closed-state consecutiveFailures++ branch +// for this case too, which reset backoff to the flat base duration and +// slid openUntil forward on every such failure instead of leaving the +// existing window alone — doubling never actually happened outside the +// narrow post-cooldown probe path the doc comment claimed was the only +// place it did. // // Deliberately no single-flight guard on the probe window (unlike // registry.go's tryBeginRefresh inFlight flag for its own, much lower- @@ -434,8 +598,6 @@ func (t *requestHealthTracker) record(provider string, success bool) { defer st.mu.Unlock() now := t.nowFn() - probed := st.open && !now.Before(st.openUntil) - if success { st.open = false st.consecutiveFailures = 0 @@ -444,7 +606,14 @@ func (t *requestHealthTracker) record(provider string, success bool) { return } - if probed { + // Already open, cooldown not yet elapsed: this failure changes + // nothing (F5 fix, above). + if st.open && now.Before(st.openUntil) { + return + } + + if st.open { + // Cooldown elapsed: this failure IS the probe outcome. Escalate. if st.backoff <= 0 { st.backoff = requestBreakerOpenDuration } else { diff --git a/failover_test.go b/failover_test.go index d74024b..cb1e25e 100644 --- a/failover_test.go +++ b/failover_test.go @@ -262,6 +262,48 @@ func TestRequestHealthTracker_FailedProbe_ReopensWithDoubledBackoff(t *testing.T assert.True(t, tr.healthy("p"), "must recover once the doubled window has fully elapsed") } +// TestRequestHealthTracker_FailureWhileOpenBeforeCooldown_IsANoOp is the +// direct regression test for adversarial-review finding F5: a failure +// arriving while the breaker is ALREADY open and its cooldown has not +// yet elapsed must change nothing — not the backoff, not openUntil. An +// earlier version fell through to the closed-state consecutiveFailures++ +// branch for this case, which reset backoff to the flat base duration +// and slid openUntil forward on every such failure, so doubling never +// actually happened outside the narrow post-cooldown probe path. +func TestRequestHealthTracker_FailureWhileOpenBeforeCooldown_IsANoOp(t *testing.T) { + t.Parallel() + tr := newRequestHealthTracker() + now := time.Now() + tr.nowFn = func() time.Time { return now } + + for i := 0; i < requestBreakerFailureThreshold; i++ { + tr.record("p", false) + } + st := tr.stateFor("p") + st.mu.Lock() + openUntilAfterTrip := st.openUntil + backoffAfterTrip := st.backoff + st.mu.Unlock() + require.Equal(t, requestBreakerOpenDuration, backoffAfterTrip) + + // Well before the cooldown elapses: several more failures arrive + // (e.g. other in-flight requests against the same dead provider). + now = now.Add(requestBreakerOpenDuration / 2) + for i := 0; i < 5; i++ { + require.False(t, tr.healthy("p")) + tr.record("p", false) + } + + st.mu.Lock() + openUntilAfterExtraFailures := st.openUntil + backoffAfterExtraFailures := st.backoff + failuresAfterExtra := st.consecutiveFailures + st.mu.Unlock() + assert.Equal(t, openUntilAfterTrip, openUntilAfterExtraFailures, "openUntil must not slide forward from failures arriving before the cooldown elapses") + assert.Equal(t, backoffAfterTrip, backoffAfterExtraFailures, "backoff must stay at the base duration, not reset by every failure while already open") + assert.Equal(t, requestBreakerFailureThreshold, failuresAfterExtra, "consecutiveFailures must not keep growing once the breaker is already open") +} + func TestRequestHealthTracker_BackoffCapsAtMax(t *testing.T) { t.Parallel() tr := newRequestHealthTracker() @@ -364,10 +406,18 @@ func TestValidateFailoverConfig_Defaults(t *testing.T) { t.Parallel() fc, err := validateFailoverConfig(FailoverConfig{}) require.NoError(t, err) - assert.True(t, fc.enabled, "nil Enabled must default to true — failover is a hot standby with no config change") + assert.False(t, fc.enabled, "nil Enabled must default to false — coordinator ruling: new behavior that can change which provider serves a request goes behind a flag defaulting off") assert.Equal(t, defaultFailoverMaxAttempts, fc.maxAttempts) } +func TestValidateFailoverConfig_ExplicitEnable(t *testing.T) { + t.Parallel() + yes := true + fc, err := validateFailoverConfig(FailoverConfig{Enabled: &yes}) + require.NoError(t, err) + assert.True(t, fc.enabled) +} + func TestValidateFailoverConfig_ExplicitDisable(t *testing.T) { t.Parallel() no := false @@ -402,9 +452,10 @@ func testGatewayForCandidates(t *testing.T, providerNames []string, fc failoverC adapters[name] = newFakeAdapter(name) providers[name] = &ProviderConfig{Models: []string{"shared"}} } - reg, err := newModelRegistry(adapters, &Config{Providers: providers}, func(string, ...any) {}) + cfg := &Config{Providers: providers} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) require.NoError(t, err) - return &Gateway{registry: reg, failoverHealth: newRequestHealthTracker(), failover: fc} + return &Gateway{registry: reg, failoverHealth: newRequestHealthTracker(), failover: fc, cfg: cfg, name: "test"} } func candNames(cands []resolveCandidate) []string { @@ -475,14 +526,127 @@ func TestOrderedFailoverCandidates_DiscoveryUnhealthy_DeprioritizedNotDropped(t assert.Equal(t, []string{"beta", "alpha"}, candNames(got), "discovery-unhealthy must be deprioritized (moved after), never dropped from the list") } -func TestOrderedFailoverCandidates_MaxAttemptsCapsList(t *testing.T) { +// TestOrderedFailoverCandidates_DoesNotCap proves orderedFailoverCandidates +// itself no longer applies maxAttempts — capFailoverCandidates does, as the +// LAST step of runMeteredCall's own sequence, after cost filtering. +func TestOrderedFailoverCandidates_DoesNotCap(t *testing.T) { t.Parallel() g := testGatewayForCandidates(t, []string{"alpha", "beta", "gamma"}, failoverConfig{enabled: true, maxAttempts: 2}) primary := resolveCandidate{providerName: "alpha"} extra := []resolveCandidate{{providerName: "beta"}, {providerName: "gamma"}} got := g.orderedFailoverCandidates(primary, extra) - assert.Len(t, got, 2, "maxAttempts must cap the total candidate list, primary included") - assert.Equal(t, []string{"alpha", "beta"}, candNames(got)) + assert.Equal(t, []string{"alpha", "beta", "gamma"}, candNames(got), "orderedFailoverCandidates must return the full health/discovery-ordered list uncapped") +} + +func TestCapFailoverCandidates(t *testing.T) { + t.Parallel() + all := []resolveCandidate{{providerName: "alpha"}, {providerName: "beta"}, {providerName: "gamma"}} + + got := capFailoverCandidates(all, 2) + assert.Equal(t, []string{"alpha", "beta"}, candNames(got), "maxAttempts must cap the total candidate list, primary included") + + got = capFailoverCandidates(all, 10) + assert.Len(t, got, 3, "a cap larger than the list must not truncate anything") + + got = capFailoverCandidates(all, 0) + assert.Len(t, got, 3, "max <= 0 must be a defensive no-op") +} + +// --- resolveUnifiedPricing / candidateCostAllowed / filterCandidatesByCost --- + +func TestResolveUnifiedPricing_CanonicalFirst_ThenBareFallback(t *testing.T) { + t.Parallel() + overrides := map[string]*ModelPricing{ + "beta/shared": {InputPerM: 1, OutputPerM: 2}, + "shared": {InputPerM: 5, OutputPerM: 6}, + } + p, ok := resolveUnifiedPricing("beta/shared", "shared", overrides) + require.True(t, ok) + assert.Equal(t, ModelPricing{InputPerM: 1, OutputPerM: 2}, p, "an override keyed on the canonical id must win over the bare fallback — the exact case unifiedCostMicros itself resolves this way") +} + +func TestResolveUnifiedPricing_FallsBackToBare(t *testing.T) { + t.Parallel() + overrides := map[string]*ModelPricing{"shared": {InputPerM: 5, OutputPerM: 6}} + p, ok := resolveUnifiedPricing("beta/shared", "shared", overrides) + require.True(t, ok) + assert.Equal(t, ModelPricing{InputPerM: 5, OutputPerM: 6}, p) +} + +func TestResolveUnifiedPricing_Unknown(t *testing.T) { + t.Parallel() + _, ok := resolveUnifiedPricing("beta/shared", "shared", nil) + assert.False(t, ok, "neither the canonical nor the bare id has a price — this must never be silently treated as zero") +} + +func TestResolveUnifiedPricing_BuiltinTable(t *testing.T) { + t.Parallel() + p, ok := resolveUnifiedPricing("openai/gpt-5", "gpt-5", nil) + require.True(t, ok, "the built-in pricing table must still resolve with no operator overrides configured") + assert.Equal(t, builtinPricing["gpt-5"], p) +} + +// TestCandidateCostAllowed is the direct regression test for the +// operator's binding ruling: table-driven across every named case, +// including the two-dimension requirement (cheap on one axis, dear on +// the other, must still be rejected). +func TestCandidateCostAllowed(t *testing.T) { + t.Parallel() + cheap := ModelPricing{InputPerM: 1, OutputPerM: 2} + dear := ModelPricing{InputPerM: 3, OutputPerM: 4} + mixedCheapInputDearOutput := ModelPricing{InputPerM: 0.5, OutputPerM: 10} + zero := ModelPricing{} + + tests := []struct { + name string + primary, candidate ModelPricing + primaryKnown, candidateKnown bool + want bool + }{ + {name: "both known, candidate strictly cheaper on both", primary: dear, primaryKnown: true, candidate: cheap, candidateKnown: true, want: true}, + {name: "both known, identical price", primary: cheap, primaryKnown: true, candidate: cheap, candidateKnown: true, want: true}, + {name: "both known, candidate strictly dearer on both", primary: cheap, primaryKnown: true, candidate: dear, candidateKnown: true, want: false}, + {name: "both known, cheaper input but dearer output must still reject", primary: cheap, primaryKnown: true, candidate: mixedCheapInputDearOutput, candidateKnown: true, want: false}, + {name: "both known, dearer input but cheaper output must still reject", primary: mixedCheapInputDearOutput, primaryKnown: true, candidate: cheap, candidateKnown: true, want: false}, + {name: "primary known, candidate unknown: never allowed", primary: dear, primaryKnown: true, candidate: ModelPricing{}, candidateKnown: false, want: false}, + {name: "primary known-zero, candidate unknown: never allowed", primary: zero, primaryKnown: true, candidate: ModelPricing{}, candidateKnown: false, want: false}, + {name: "primary unknown, candidate known non-zero: never allowed", primary: ModelPricing{}, primaryKnown: false, candidate: cheap, candidateKnown: true, want: false}, + {name: "primary unknown, candidate known but only input non-zero: never allowed", primary: ModelPricing{}, primaryKnown: false, candidate: ModelPricing{InputPerM: 0.01}, candidateKnown: true, want: false}, + {name: "primary unknown, candidate known-zero: allowed", primary: ModelPricing{}, primaryKnown: false, candidate: zero, candidateKnown: true, want: true}, + {name: "both unknown: allowed", primary: ModelPricing{}, primaryKnown: false, candidate: ModelPricing{}, candidateKnown: false, want: true}, + {name: "both known-zero: allowed", primary: zero, primaryKnown: true, candidate: zero, candidateKnown: true, want: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := candidateCostAllowed(tt.primary, tt.primaryKnown, tt.candidate, tt.candidateKnown) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestFilterCandidatesByCost_SkipsPricierKeepsCheaper(t *testing.T) { + t.Parallel() + g := testGatewayForCandidates(t, []string{"alpha", "beta", "gamma"}, failoverConfig{enabled: true, maxAttempts: 3}) + g.cfg.Pricing = map[string]*ModelPricing{ + "alpha/shared": {InputPerM: 2, OutputPerM: 4}, // primary + "beta/shared": {InputPerM: 3, OutputPerM: 4}, // dearer input -> rejected + "gamma/shared": {InputPerM: 2, OutputPerM: 4}, // equal -> allowed + } + candidates := []resolveCandidate{ + {providerName: "alpha", upstreamModel: "shared", canonical: "alpha/shared"}, + {providerName: "beta", upstreamModel: "shared", canonical: "beta/shared"}, + {providerName: "gamma", upstreamModel: "shared", canonical: "gamma/shared"}, + } + got := g.filterCandidatesByCost(candidates, "shared") + assert.Equal(t, []string{"alpha", "gamma"}, candNames(got)) +} + +func TestFilterCandidatesByCost_FewerThanTwoCandidates_NoOp(t *testing.T) { + t.Parallel() + g := testGatewayForCandidates(t, []string{"alpha"}, failoverConfig{enabled: true, maxAttempts: 3}) + candidates := []resolveCandidate{{providerName: "alpha", upstreamModel: "shared", canonical: "alpha/shared"}} + got := g.filterCandidatesByCost(candidates, "shared") + assert.Equal(t, candidates, got) } // ===================================================================== @@ -504,6 +668,13 @@ func twoProviderFailoverConfig(srvAlpha, srvBeta *httptest.Server) *Config { cfg.Users = &UsersConfig{Inline: []*UserConfig{ {Name: "alice", Group: "default", APIKey: "sk-alice", Limits: &LimitsConfig{}}, }} + // Explicitly opt in: FailoverConfig.Enabled now defaults to false + // (coordinator ruling) — every test in this file that exercises actual + // failover behavior must opt in on purpose, the same way an operator + // would, rather than relying on a default this feature no longer ships + // with. + yes := true + cfg.Failover = FailoverConfig{Enabled: &yes} return cfg } @@ -570,6 +741,8 @@ func TestHandleChat_Failover_FallsThroughAndServes(t *testing.T) { } cfg.Groups = map[string]*GroupConfig{"default": {}} cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice", Limits: &LimitsConfig{}}}} + yes := true + cfg.Failover = FailoverConfig{Enabled: &yes} next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") @@ -699,10 +872,17 @@ func TestHandleChat_Failover_RequestUnhealthy_SkippedWithoutBeingCalled(t *testi } // TestHandleChat_Failover_DiscoveryUnhealthy_StillTried proves discovery -// health is an ordering hint, never a hard skip: a provider whose -// DISCOVERY breaker is open must still be attempted (and can still serve -// the response) when it is the only one that actually works. Would FAIL -// if discoveryHealthy were (mis-)used as a skip gate. +// health is never a HARD SKIP: a provider whose DISCOVERY breaker is +// open must still be attempted, and can still serve the response, when +// beta (the only other candidate) fails. Would FAIL if discoveryHealthy +// were used to exclude a candidate outright, but does NOT independently +// prove ordering (deleting the deprioritization partition in +// orderedFailoverCandidates leaves this test passing too, since beta +// fails either way and alpha is tried regardless of position) — the +// ordering claim itself is covered only by +// TestOrderedFailoverCandidates_DiscoveryUnhealthy_DeprioritizedNotDropped +// (adversarial-review correction: this comment previously overstated +// what this test proves). func TestHandleChat_Failover_DiscoveryUnhealthy_StillTried(t *testing.T) { alphaSrv := jsonServer(http.StatusOK, successRespBody) // alpha's CHAT endpoint works fine defer alphaSrv.Close() @@ -727,10 +907,9 @@ func TestHandleChat_Failover_DiscoveryUnhealthy_StillTried(t *testing.T) { st.mu.Unlock() require.False(t, gw.registry.discoveryHealthy("alpha")) - // beta is healthy by every signal but its own chat endpoint 500s — - // ordering puts beta (discovery-healthy) FIRST, so this proves the - // request actually reaches alpha SECOND, not that alpha happened to - // be tried first anyway. + // beta is healthy by every signal but its own chat endpoint 500s, so + // the request can only succeed if alpha is tried at some point in + // the chain despite its open discovery breaker. body := map[string]any{"model": "shared", "messages": []any{}} req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) rec := httptest.NewRecorder() @@ -864,11 +1043,507 @@ func TestHandleChat_Failover_CachedUnderWinningProviderKey(t *testing.T) { assert.Equal(t, int64(1), atomic.LoadInt64(&betaCalls), "the second request must hit cache under beta's key, not call beta's backend again") } +// --- cost guard (operator ruling): never fail over to a pricier candidate --- + +// costGuardConfig builds a two-provider (alpha, beta) config exactly like +// twoProviderFailoverConfig but WITHOUT setting cfg.Pricing, so each test +// configures its own overrides for its own case. +func costGuardConfig(srvAlpha, srvBeta *httptest.Server) *Config { + cfg := twoProviderFailoverConfig(srvAlpha, srvBeta) + cfg.Providers["alpha"].Models = []string{"shared"} + cfg.Providers["beta"].Models = []string{"shared"} + return cfg +} + +// driveCostGuardRequest posts one chat-completion request for "shared" +// through h and returns the recorder and everything logged to stderr +// while it ran. +func driveCostGuardRequest(t *testing.T, h http.Handler) (*httptest.ResponseRecorder, string) { + t.Helper() + origStderr := captureStderrStart() + body := map[string]any{"model": "shared", "messages": []any{map[string]any{"role": "user", "content": "hi"}}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec, captureStderrStop(origStderr) +} + +// TestHandleChat_Failover_CostGuard_SkipsPricierCandidate is the direct +// regression test for the operator's requirement: a candidate priced +// higher than the primary must never serve. Would FAIL if the cost guard +// were reverted (mutation-tested in this branch's own history). +func TestHandleChat_Failover_CostGuard_SkipsPricierCandidate(t *testing.T) { + var betaCalls int64 + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := countingServer(http.StatusOK, successRespBody, &betaCalls) + defer betaSrv.Close() + + cfg := costGuardConfig(alphaSrv, betaSrv) + cfg.Pricing = map[string]*ModelPricing{ + "alpha/shared": {InputPerM: 1, OutputPerM: 2}, + "beta/shared": {InputPerM: 1, OutputPerM: 3}, // dearer on output only — still must be rejected + } + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + rec, logOutput := driveCostGuardRequest(t, h) + + assert.Equal(t, http.StatusInternalServerError, rec.Code, "beta is dearer, so alpha's own failure must be returned as-is, body=%s", rec.Body.String()) + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls), "beta must never be called once the cost guard rejects it") + assert.Contains(t, logOutput, "alpha") + assert.Contains(t, logOutput, "beta") + assert.Contains(t, logOutput, "shared") +} + +// TestHandleChat_Failover_CostGuard_AllowsCheaperOrEqualCandidate is the +// mirror positive case: a candidate priced at or below the primary must +// still serve normally. +func TestHandleChat_Failover_CostGuard_AllowsCheaperOrEqualCandidate(t *testing.T) { + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := jsonServer(http.StatusOK, successRespBody) + defer betaSrv.Close() + + cfg := costGuardConfig(alphaSrv, betaSrv) + cfg.Pricing = map[string]*ModelPricing{ + "alpha/shared": {InputPerM: 2, OutputPerM: 4}, + "beta/shared": {InputPerM: 2, OutputPerM: 4}, // exactly equal — allowed + } + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + rec, _ := driveCostGuardRequest(t, h) + assert.Equal(t, http.StatusOK, rec.Code, "beta is equally priced, so failover must still serve it, body=%s", rec.Body.String()) + assert.Equal(t, successRespBody, rec.Body.String()) +} + +// TestHandleChat_Failover_CostGuard_PrimaryKnown_CandidateUnknown_Skips +// covers the operator's first unknown-pricing ruling: primary priced, +// candidate unpriced — never allowed, since an unknown price cannot be +// proven not-more-expensive. +func TestHandleChat_Failover_CostGuard_PrimaryKnown_CandidateUnknown_Skips(t *testing.T) { + var betaCalls int64 + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := countingServer(http.StatusOK, successRespBody, &betaCalls) + defer betaSrv.Close() + + cfg := costGuardConfig(alphaSrv, betaSrv) + cfg.Pricing = map[string]*ModelPricing{ + "alpha/shared": {InputPerM: 1, OutputPerM: 2}, + // beta has no override, and "shared"/"beta/shared" is not in the + // built-in table either — its price is genuinely unknown. + } + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + rec, _ := driveCostGuardRequest(t, h) + assert.Equal(t, http.StatusInternalServerError, rec.Code) + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls)) +} + +// TestHandleChat_Failover_CostGuard_PrimaryUnknown_CandidateKnownNonZero_Skips +// covers the operator's second unknown-pricing ruling: moving from +// unbilled to billed is itself a spend increase. +func TestHandleChat_Failover_CostGuard_PrimaryUnknown_CandidateKnownNonZero_Skips(t *testing.T) { + var betaCalls int64 + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := countingServer(http.StatusOK, successRespBody, &betaCalls) + defer betaSrv.Close() + + cfg := costGuardConfig(alphaSrv, betaSrv) + cfg.Pricing = map[string]*ModelPricing{ + "beta/shared": {InputPerM: 1, OutputPerM: 2}, // alpha has no override at all: unknown + } + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + rec, _ := driveCostGuardRequest(t, h) + assert.Equal(t, http.StatusInternalServerError, rec.Code) + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls)) +} + +// TestHandleChat_Failover_CostGuard_PrimaryUnknown_CandidateKnownZero_Allows +// covers the ruling's own stated exception: a candidate that is known, +// explicit free can never be more expensive than an unknown amount. +func TestHandleChat_Failover_CostGuard_PrimaryUnknown_CandidateKnownZero_Allows(t *testing.T) { + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := jsonServer(http.StatusOK, successRespBody) + defer betaSrv.Close() + + cfg := costGuardConfig(alphaSrv, betaSrv) + cfg.Pricing = map[string]*ModelPricing{ + "beta/shared": {InputPerM: 0, OutputPerM: 0}, // explicit, known-zero override + } + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + rec, _ := driveCostGuardRequest(t, h) + assert.Equal(t, http.StatusOK, rec.Code, "a known-zero candidate can never be more expensive than an unknown primary, body=%s", rec.Body.String()) +} + +// TestHandleChat_Failover_CostGuard_BothUnknown_Allows and +// TestHandleChat_Failover_CostGuard_BothKnownZero_Allows cover the +// ruling's "equal, so allowed" cases. +func TestHandleChat_Failover_CostGuard_BothUnknown_Allows(t *testing.T) { + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := jsonServer(http.StatusOK, successRespBody) + defer betaSrv.Close() + + cfg := costGuardConfig(alphaSrv, betaSrv) // no cfg.Pricing at all + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + rec, _ := driveCostGuardRequest(t, h) + assert.Equal(t, http.StatusOK, rec.Code, "both unknown must be treated as equal, body=%s", rec.Body.String()) +} + +func TestHandleChat_Failover_CostGuard_BothKnownZero_Allows(t *testing.T) { + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := jsonServer(http.StatusOK, successRespBody) + defer betaSrv.Close() + + cfg := costGuardConfig(alphaSrv, betaSrv) + cfg.Pricing = map[string]*ModelPricing{ + "alpha/shared": {InputPerM: 0, OutputPerM: 0}, + "beta/shared": {InputPerM: 0, OutputPerM: 0}, + } + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + rec, _ := driveCostGuardRequest(t, h) + assert.Equal(t, http.StatusOK, rec.Code, "both known-zero must be treated as equal, body=%s", rec.Body.String()) +} + +// TestHandleChat_Failover_CostGuard_PerProviderPricingOverride_RealShape is +// the requested production-shape case: the SAME bare model id +// (a real, built-in-priced model — "gpt-5") served by two providers, +// where alpha has NO override (resolves via the built-in table) and beta +// has an operator-set per-provider override making it dearer on the +// canonical "beta/gpt-5" key. Proves the guard resolves pricing through +// the exact canonical-then-bare path unifiedCostMicros itself uses, not +// a parallel, simplified notion of price — the whole point of ruling 1. +func TestHandleChat_Failover_CostGuard_PerProviderPricingOverride_RealShape(t *testing.T) { + const realModel = "gpt-5" + var betaCalls int64 + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := countingServer(http.StatusOK, successRespBody, &betaCalls) + defer betaSrv.Close() + + cfg := CreateConfig() + cfg.Providers = map[string]*ProviderConfig{ + "alpha": {Type: "openai", BaseURL: alphaSrv.URL, APIKey: "k", Models: []string{realModel}}, + "beta": {Type: "openai", BaseURL: betaSrv.URL, APIKey: "k", Models: []string{realModel}}, + } + cfg.Groups = map[string]*GroupConfig{"default": {}} + cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice", Limits: &LimitsConfig{}}}} + yes := true + cfg.Failover = FailoverConfig{Enabled: &yes} + // Only beta gets a per-provider override — alpha resolves gpt-5's + // price from builtinPricing instead, exactly the real production + // shape a per-provider Pricing override produces. + builtin := builtinPricing[realModel] + cfg.Pricing = map[string]*ModelPricing{ + "beta/" + realModel: {InputPerM: builtin.InputPerM * 10, OutputPerM: builtin.OutputPerM * 10}, + } + + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + origStderr := captureStderrStart() + body := map[string]any{"model": realModel, "messages": []any{map[string]any{"role": "user", "content": "hi"}}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + logOutput := captureStderrStop(origStderr) + + assert.Equal(t, http.StatusInternalServerError, rec.Code, "beta's per-provider override makes it 10x alpha's built-in price, so it must be rejected, body=%s", rec.Body.String()) + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls)) + assert.Contains(t, logOutput, "alpha") + assert.Contains(t, logOutput, "beta") + assert.Contains(t, logOutput, realModel) +} + +// --- adversarial-review regressions (F1-F4, F6, F7) --- + +// TestHandleChat_Failover_F1_StreamOptionsNotLeakedAcrossCandidates is the +// direct regression test for adversarial-review finding F1: an +// openai-type adapter mutates req in place (chatCompletion forces +// stream_options.include_usage=true onto it BEFORE it ever calls the +// upstream, provider_openai.go), so reusing one shared req map across +// candidates let a later candidate read back an earlier candidate's own +// forced injection and misreport clientAskedUsage=true to a client that +// never asked for streamed usage — leaking the gateway's own +// deliberately-suppressed usage-only SSE chunk. alpha fails before ever +// reaching the network (its own mutation to its OWN req copy still +// happens first, exactly like the real bug), beta serves a real SSE +// stream containing a usage-only chunk the client never asked for. +func TestHandleChat_Failover_F1_StreamOptionsNotLeakedAcrossCandidates(t *testing.T) { + const contentChunk = "data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"delta\":{\"content\":\"Hi\"}}]}\n\n" + const usageOnlyChunk = "data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"choices\":[],\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":2}}\n\n" + const doneMarker = "data: [DONE]\n\n" + + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + fl, ok := w.(http.Flusher) + require.True(t, ok) + for _, chunk := range []string{contentChunk, usageOnlyChunk, doneMarker} { + _, _ = w.Write([]byte(chunk)) + fl.Flush() + } + })) + defer betaSrv.Close() + + cfg := twoProviderFailoverConfig(alphaSrv, betaSrv) + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + // The client explicitly does NOT set stream_options at all — it + // never asked for usage in the stream. + body := map[string]any{"model": "shared", "stream": true, "messages": []any{map[string]any{"role": "user", "content": "hi"}}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rw := newRecordingWriter() + h.ServeHTTP(rw, req) + + require.Equal(t, http.StatusOK, rw.status) + assert.Contains(t, rw.buf.String(), `"content":"Hi"`, "the real content chunk must still be forwarded") + assert.NotContains(t, rw.buf.String(), `"prompt_tokens":5`, "a client that never asked for stream usage must never receive it, even though ALPHA's own (failed, discarded) attempt forced include_usage=true on ITS OWN copy of the request") +} + +// TestHandleChat_Failover_F2_StaleSSEHeadersDoNotSurviveToPlainResponse is +// the direct regression test for adversarial-review finding F2: +// newSSEWriter (sse.go) stages Content-Type/Cache-Control/ +// X-Accel-Buffering on sw.Header() the moment a streaming attempt starts, +// before any byte is written. alpha enters that streaming path (a real +// 200 + text/event-stream response) but its connection is severed before +// any complete SSE event is dispatched — an error, satisfying THE HARD +// CONSTRAINT (nothing was actually WRITTEN to the client) — so failover +// proceeds to beta, which answers a plain, non-streaming JSON body. +func TestHandleChat_Failover_F2_StaleSSEHeadersDoNotSurviveToPlainResponse(t *testing.T) { + alphaSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + hj, ok := w.(http.Hijacker) + require.True(t, ok) + conn, _, hjErr := hj.Hijack() + if hjErr == nil { + // Closing a hijacked chunked-encoding connection with no + // terminating chunk produces a genuine "unexpected EOF" on + // the client's read side (verified empirically), not a + // clean end-of-stream — readSSE returns that as a real + // error without ever dispatching an event, so + // sw.writeData/sw.WriteHeader is never reached even though + // newSSEWriter already staged its three headers. + _ = conn.Close() + } + })) + defer alphaSrv.Close() + betaSrv := jsonServer(http.StatusOK, successRespBody) + defer betaSrv.Close() + + cfg := twoProviderFailoverConfig(alphaSrv, betaSrv) + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + body := map[string]any{"model": "shared", "stream": true, "messages": []any{map[string]any{"role": "user", "content": "hi"}}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + require.Equal(t, http.StatusOK, rec.Code, "body=%s", rec.Body.String()) + assert.Equal(t, successRespBody, rec.Body.String()) + assert.Equal(t, "application/json", rec.Header().Get("Content-Type"), "beta's own Content-Type must win, not alpha's stale text/event-stream") + assert.Empty(t, rec.Header().Get("Cache-Control"), "alpha's staged SSE header must never survive into beta's plain JSON response") + assert.Empty(t, rec.Header().Get("X-Accel-Buffering"), "alpha's staged SSE header must never survive into beta's plain JSON response") +} + +// TestHandleChat_Failover_F3_401IsAFailure_OpensRequestHealthBreaker is the +// direct regression test for adversarial-review finding F3: a provider +// answering 401 to every request (a dead API key — exactly the "masking +// a broken provider key" risk this feature exists to reduce) must +// eventually be skipped, not retried forever. Would FAIL under the +// reverted isTransient-based classification, since isTransient returns +// false for any non-429 4xx. +func TestHandleChat_Failover_F3_401IsAFailure_OpensRequestHealthBreaker(t *testing.T) { + var alphaCalls, betaCalls int64 + alphaSrv := countingServer(http.StatusUnauthorized, `{"error":{"message":"invalid api key","type":"authentication_error"}}`, &alphaCalls) + defer alphaSrv.Close() + betaSrv := countingServer(http.StatusOK, successRespBody, &betaCalls) + defer betaSrv.Close() + + cfg := twoProviderFailoverConfig(alphaSrv, betaSrv) + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + body := map[string]any{"model": "shared", "messages": []any{}} + for i := 0; i < requestBreakerFailureThreshold; i++ { + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + require.Equal(t, http.StatusOK, rec.Code, "every one of these must still fail over to beta and succeed") + } + require.Equal(t, int64(requestBreakerFailureThreshold), atomic.LoadInt64(&alphaCalls)) + + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusOK, rec.Code) + assert.Equal(t, int64(requestBreakerFailureThreshold), atomic.LoadInt64(&alphaCalls), "alpha must be skipped after 3 consecutive 401s, not called a 4th time") + assert.Equal(t, int64(requestBreakerFailureThreshold+1), atomic.LoadInt64(&betaCalls)) +} + +// TestHandleChat_Failover_F4_SharedDeadlineExhausted_NeverMisattributedToNextCandidate +// is the direct regression test for adversarial-review finding F4: alpha +// burns the ENTIRE shared request context sleeping past its deadline; +// beta must never be dialed at all (an attempt after the deadline is +// already spent would fail near-instantly for a reason that has nothing +// to do with beta's own health), never be recorded as having failed, and +// the client-facing error must name alpha, not beta. +func TestHandleChat_Failover_F4_SharedDeadlineExhausted_NeverMisattributedToNextCandidate(t *testing.T) { + alphaSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(300 * time.Millisecond) // well past the request's own 50ms deadline below + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(successRespBody)) + })) + defer alphaSrv.Close() + var betaCalls int64 + betaSrv := countingServer(http.StatusOK, successRespBody, &betaCalls) + defer betaSrv.Close() + + cfg := twoProviderFailoverConfig(alphaSrv, betaSrv) + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + gw, ok := h.(*Gateway) + require.True(t, ok) + gw.limiter.spawn = func(f func()) { f() } + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", + strings.NewReader(`{"model":"shared","messages":[{"role":"user","content":"hi"}]}`)).WithContext(ctx) + req.Header.Set("Authorization", "Bearer sk-alice") + req.Header.Set("Content-Type", "application/json") + + origStderr := captureStderrStart() + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + logOutput := captureStderrStop(origStderr) + + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls), "beta must never be dialed once the shared deadline is already spent") + assert.True(t, gw.failoverHealth.healthy("beta"), "beta must never be marked unhealthy for a timeout that was alpha's own fault") + + now := time.Now() + betaAttempts, _ := gw.limiter.getCounter(kindProvider, "beta", metricProvAttempt, windowDay, now) + assert.Equal(t, int64(0), betaAttempts, "beta's own dashboard attempt counter must not move either") + + assert.Contains(t, logOutput, "alpha", "the error must name the provider that actually failed") +} + +// TestHandleChat_Failover_F6_RetryAttemptsDoNotOpenBreakerOnOneRequest is +// the direct regression test for adversarial-review finding F6: with +// retry.enabled and attempts:3, ONE client request against a +// permanently-failing provider must count as ONE outcome for the +// request-health gate, not one per raw retry attempt — otherwise a +// single request alone could reach requestBreakerFailureThreshold (3) +// purely from ITS OWN internal retries. +func TestHandleChat_Failover_F6_RetryAttemptsDoNotOpenBreakerOnOneRequest(t *testing.T) { + var alphaCalls int64 + alphaSrv := countingServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`, &alphaCalls) + defer alphaSrv.Close() + betaSrv := jsonServer(http.StatusOK, successRespBody) + defer betaSrv.Close() + + cfg := twoProviderFailoverConfig(alphaSrv, betaSrv) + cfg.Retry = RetryConfig{Enabled: true, Attempts: 3, Backoff: "1ms"} + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + gw, ok := h.(*Gateway) + require.True(t, ok) + + body := map[string]any{"model": "shared", "messages": []any{}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + require.Equal(t, http.StatusOK, rec.Code, "must still fail over to beta after alpha's own retries are exhausted") + + // retry.attempts=3 means 4 raw HTTP attempts against alpha for this + // ONE client request (retryPolicy's own doc comment, retry.go). + require.Equal(t, int64(4), atomic.LoadInt64(&alphaCalls)) + // The request-health gate must still read this as exactly ONE + // failure, not four — well below requestBreakerFailureThreshold (3). + assert.True(t, gw.failoverHealth.healthy("alpha"), "one client request's internal retries must count as one outcome for the request-health gate, not one per raw attempt") +} + +// TestHandleChat_Failover_F7_CachedFailoverResponse_NeverHitsDeadPrimary is +// the direct regression test for adversarial-review finding F7: even +// while alpha stays permanently broken, a SECOND identical request must +// be served from beta's cache entry WITHOUT alpha's dead backend being +// hit again on that second request — proving cache lookups run for +// every candidate BEFORE any of them is attempted, not interleaved with +// each candidate's own turn. +func TestHandleChat_Failover_F7_CachedFailoverResponse_NeverHitsDeadPrimary(t *testing.T) { + var alphaCalls, betaCalls int64 + alphaSrv := countingServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`, &alphaCalls) + defer alphaSrv.Close() + betaSrv := countingServer(http.StatusOK, successRespBody, &betaCalls) + defer betaSrv.Close() + + redisLn := newBehavioralRedisServer(t) + cfg := twoProviderFailoverConfig(alphaSrv, betaSrv) + cfg.Redis = &RedisConfig{Address: redisLn.Addr().String()} + cfg.Cache = CacheConfig{Enabled: true, TTL: "1m", MaxBodyBytes: 1 << 20} + + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + reqBody := map[string]any{"model": "shared", "messages": []any{map[string]any{"role": "user", "content": "hi"}}} + req1 := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", reqBody) + rec1 := httptest.NewRecorder() + h.ServeHTTP(rec1, req1) + require.Equal(t, http.StatusOK, rec1.Code) + require.Equal(t, int64(1), atomic.LoadInt64(&alphaCalls)) + require.Equal(t, int64(1), atomic.LoadInt64(&betaCalls)) + + req2 := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", reqBody) + rec2 := httptest.NewRecorder() + h.ServeHTTP(rec2, req2) + + assert.Equal(t, http.StatusOK, rec2.Code) + assert.Equal(t, "hit", rec2.Header().Get("X-Llmgw-Cache")) + assert.Equal(t, int64(1), atomic.LoadInt64(&alphaCalls), "alpha's dead backend must NOT be hit again just to discover beta's cache entry") + assert.Equal(t, int64(1), atomic.LoadInt64(&betaCalls)) +} + // TestHandleChat_Failover_SingleProvider_UpstreamError_BehavesExactlyAsBefore // pins the "single-provider deployment behaves exactly as before" // requirement explicitly for this branch, with failover left at its -// (enabled) default: with no second candidate to try, an upstream error -// must be returned to the client unchanged, identical to v0.1 behavior. +// (now disabled) default: with no second candidate to try, an upstream +// error must be returned to the client unchanged, identical to v0.1 +// behavior, regardless of the enabled switch either way. func TestHandleChat_Failover_SingleProvider_UpstreamError_BehavesExactlyAsBefore(t *testing.T) { srv := jsonServer(http.StatusTooManyRequests, `{"error":{"message":"rate limited upstream","type":"rate_limit_error"}}`) defer srv.Close() @@ -898,6 +1573,43 @@ func TestHandleChat_Failover_SingleProvider_UpstreamError_BehavesExactlyAsBefore assert.Equal(t, "upstream_error", out.Error.Type) } +// TestHandleChat_Failover_DefaultConfig_ByteIdenticalToPreFailover is the +// regression test for the coordinator's default-off ruling: a Config that +// never mentions "failover" at all — the shape of every config written +// before this feature existed, and the shape most operators keep using +// after upgrading without opting in — must behave byte-identically to +// having no failover code at all. Would FAIL if validateFailoverConfig's +// default reverted to enabled. +func TestHandleChat_Failover_DefaultConfig_ByteIdenticalToPreFailover(t *testing.T) { + var betaCalls int64 + alphaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + defer alphaSrv.Close() + betaSrv := countingServer(http.StatusOK, successRespBody, &betaCalls) + defer betaSrv.Close() + + cfg := CreateConfig() + cfg.Providers = map[string]*ProviderConfig{ + "alpha": {Type: "openai", BaseURL: alphaSrv.URL, APIKey: "k", Models: []string{"shared"}}, + "beta": {Type: "openai", BaseURL: betaSrv.URL, APIKey: "k", Models: []string{"shared"}}, + } + cfg.Groups = map[string]*GroupConfig{"default": {}} + cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice", Limits: &LimitsConfig{}}}} + // cfg.Failover is deliberately left untouched — the zero value, what + // every config predating this feature has. + + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + require.NoError(t, err) + + body := map[string]any{"model": "shared", "messages": []any{}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusInternalServerError, rec.Code, "an unconfigured Failover block must leave alpha's own 500 unchanged") + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls), "beta must never be called when failover was never opted into") +} + // TestHandleChat_Failover_ExplicitlyDisabled_5xxNeverFallsThrough proves // the on/off switch actually gates the behavior end to end. func TestHandleChat_Failover_ExplicitlyDisabled_5xxNeverFallsThrough(t *testing.T) { diff --git a/llmgateway.go b/llmgateway.go index 6cd63ac..08d06d8 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -545,6 +545,14 @@ type Gateway struct { // rather than a nil-pointer panic. failoverHealth *requestHealthTracker name string + // failoverLogGate rate-limits runMeteredCall's generic "failing over" + // log line (routes_unified.go) to once per storeErrorLogEvery + // (adversarial-review fix, F10) — reuses auth.go's own logGate type, + // the identical technique shouldLogAuthFailure/logStoreError already + // apply elsewhere in this package. Zero-value-usable: the first call + // on a fresh Gateway always logs. Deliberately does NOT gate the 404 + // loud-log line (operator ruling: that one must always log). + failoverLogGate logGate // failover is Config.Failover, validated and resolved once by // newGateway (validateFailoverConfig, failover.go). failover failoverConfig @@ -657,8 +665,9 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str g := &Gateway{next: next, name: name, cfg: config, auth: auth} // feat/failover: validated once, here — see FailoverConfig's own doc - // comment (failover.go) for the default (enabled) and what "single- - // provider deployment behaves exactly as before" means in practice. + // comment (failover.go) for the default (disabled, coordinator + // ruling) and what "single-provider deployment behaves exactly as + // before" means in practice. failoverCfg, err := validateFailoverConfig(config.Failover) if err != nil { return nil, err diff --git a/routes_unified.go b/routes_unified.go index 655af16..bdbcbf4 100644 --- a/routes_unified.go +++ b/routes_unified.go @@ -5,10 +5,12 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "math" "net/http" "strconv" + "time" ) // maxRequestBytes caps a client request body decoded into a @@ -274,12 +276,31 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr // attempt" and "cache under the winning provider's key" actually mean // in code, not just in the brief's prose. func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scopes []limitScope, body []byte, req map[string]any, requestedModel string, grp *group, endpoint, logPrefix string, envelope envelopeWriter, writeUpstream providerUpstreamErrorWriter, call adapterCall) { - primary, extra, err := g.registry.resolveWithCandidates(requestedModel, grp) + // F9 (adversarial-review, optional perf fix): a disabled deployment + // never needs the candidate pool at all — resolvePrimaryOnly skips + // failoverCandidates' own provider scan entirely, so a config that + // never opted into failover pays nothing extra for a pool it will + // never use. + var primary resolveCandidate + var extra []resolveCandidate + var err error + if g.failover.enabled { + primary, extra, err = g.registry.resolveWithCandidates(requestedModel, grp) + } else { + primary, err = g.registry.resolvePrimaryOnly(requestedModel, grp) + } if err != nil { writeModelResolveErrorEnvelope(sw, err, envelope) return } + // Sequence matters (each step's own doc comment explains why): + // health/discovery ordering first, so candidates[0] below is the + // ACTUAL first-attempted candidate the cost guard compares everyone + // else against; cost filtering second; the maxAttempts cap last, so + // it never re-admits a candidate the cost guard already removed. candidates := g.orderedFailoverCandidates(primary, extra) + candidates = g.filterCandidatesByCost(candidates, requestedModel) + candidates = capFailoverCandidates(candidates, g.failover.maxAttempts) // cacheable/streaming gate every cache step below on spec §2's scope, // exactly as before this feature: a non-streaming request, with a @@ -291,44 +312,40 @@ func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scop streaming, _ := req["stream"].(bool) cacheable := !streaming && g.cache != nil && groupCacheEnabled(grp) - for i, cand := range candidates { - if i > 0 { - // checkAndCount above (admitRequest, this file) already - // counted this request exactly once — never double-count: a - // failover attempt is a NEW upstream attempt, not a new - // logical request, so nothing here calls checkAndCount - // again. Per-provider ATTEMPT counters, below, still record - // every attempt independently via the freshly-bound - // recorder. - g.warnf("%s: failing over from provider %q to %q for model %q", logPrefix, candidates[i-1].providerName, cand.providerName, requestedModel) - } + // pristineReq is the client's own decoded request, snapshotted ONCE, + // BEFORE any candidate's adapter ever touches it, with "model" + // already rewritten to the upstream id every candidate in this + // failover chain shares by construction (failoverCandidates, + // failover.go, only ever adds providers serving the IDENTICAL bare + // id resolve's primary already resolved to). Every attempt below + // builds its OWN fresh top-level copy from this snapshot rather than + // reusing or mutating one shared map (adversarial-review fix, F1): + // openai-type adapters mutate req in place — chatCompletion forces + // stream_options.include_usage=true onto it (provider_openai.go) — + // and reusing the same map across attempts let a LATER candidate + // read back an EARLIER candidate's own forced injection and + // misreport clientAskedUsage=true to a client that never asked for + // it, leaking the gateway's own suppressed usage-only SSE chunk. + // Proven against a single-provider control; this is what fixes it + // for openai, openai-audio, and copilot alike. + pristineReq := make(map[string]any, len(req)+1) + for k, v := range req { + pristineReq[k] = v + } + if len(candidates) > 0 { + pristineReq["model"] = candidates[0].upstreamModel + } - // req["model"] is reset on EVERY iteration: cand.upstreamModel is - // the identical string across every candidate in a failover chain - // by construction (failoverCandidates, failover.go, only ever - // adds providers sharing the SAME bare id resolve's primary - // already resolved to), so this is a no-op rewrite on the common - // single-candidate path, not new cost. - req["model"] = cand.upstreamModel - - // cacheKey is computed fresh per candidate — cache correctness - // (brief): a response produced by a failover provider must be - // stored under THAT provider's own key, never the first - // candidate's. Called here, BEFORE gatewayAliasKey is injected - // below, preserving the ordering cacheKey's own doc comment - // (cache.go) requires of every caller: gatewayAliasKey is a - // purely internal echo-back mechanism cacheKey does not itself - // strip, so computing the key after injecting it would hash an - // extra, noisy field. - var cacheKeyStr string - if cacheable { - cacheKeyStr = cacheKey(cand.providerName, cand.upstreamModel, requestedModel, endpoint, req) - if cached, hit := g.cache.lookup(cacheKeyStr); hit { - // A cache hit accounts the request only — checkAndCount - // above already counted it — never token/cost, and never - // runs the zero-usage estimate branch below: the client - // never reached the upstream provider its body size would - // be estimating against. + // F7 fix (adversarial-review): check EVERY candidate's cache entry + // before attempting ANY of them, so a cached failover response never + // costs a real round trip to a dead primary just because normal + // ordering tries that primary first — the lookup used to sit inside + // each candidate's own turn, after every EARLIER candidate's real + // upstream call had already failed. + if cacheable { + for _, cand := range candidates { + key := cacheKey(cand.providerName, cand.upstreamModel, requestedModel, endpoint, pristineReq) + if cached, hit := g.cache.lookup(key); hit { sw.Header().Set("X-Llmgw-Cache", "hit") if cached.ContentType != "" { sw.Header().Set("Content-Type", cached.ContentType) @@ -337,6 +354,95 @@ func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scop _, _ = sw.Write(cached.Body) return } + } + } + + var lastErr error + var lastProviderName string + for i, cand := range candidates { + if i > 0 { + // F4 fix (adversarial-review): a shared request context an + // EARLIER candidate already exhausted must never be + // misattributed as THIS candidate's own failure. There is no + // per-candidate timeout budget in this architecture (every + // candidate shares r's SAME context — see README's own + // documented limitation), so once an earlier candidate burns + // the deadline, attempting the next one would fail near- + // instantly for a reason that has nothing to do with its own + // health. Stop here with the LAST REAL error instead of + // chasing further doomed candidates and wrongly opening a + // perfectly healthy provider's request-health breaker. + if r.Context().Err() != nil { + g.handleAdapterErrorEnvelope(sw, lastErr, lastProviderName, logPrefix, envelope, writeUpstream) + return + } + + // checkAndCount above (admitRequest, this file) already + // counted this request exactly once — never double-count: a + // failover attempt is a NEW upstream attempt, not a new + // logical request, so nothing here calls checkAndCount + // again. Per-provider ATTEMPT counters, below, still record + // every attempt independently via the freshly-bound + // recorder. + // + // Rate-limited (F10 fix, adversarial-review): during a + // sustained outage this line would otherwise be one + // blocking synchronous stderr write per request on shared + // Traefik ingress — the exact failure shouldLogAuthFailure/ + // logStoreError (auth.go/limits.go) already exist to + // prevent elsewhere in this package. The 404 line below is + // deliberately NOT gated by this (operator ruling: it must + // always log). + if log, suppressed := g.failoverLogGate.shouldLog(time.Now()); log { + suffix := "" + if suppressed > 0 { + suffix = fmt.Sprintf(" (%d more failovers suppressed since last log)", suppressed) + } + g.warnf("%s: failing over from provider %q to %q for model %q%s", logPrefix, candidates[i-1].providerName, cand.providerName, requestedModel, suffix) + } + } + + // F2 fix (adversarial-review): clear every header a PRIOR + // candidate's own writer may have staged on sw.Header() without + // committing it — newSSEWriter (sse.go) sets Content-Type: text/ + // event-stream, Cache-Control: no-cache, and X-Accel-Buffering: + // no the moment a streaming attempt starts, before any byte is + // written. If that candidate then fails before writing (still + // safe to fail over, THE HARD CONSTRAINT below), those staged + // headers must not survive into a plain-JSON response from the + // next candidate. Safe to clear unconditionally: this loop only + // ever reaches a new iteration while sw.wroteHeader is false, so + // nothing committed is ever at risk of being wiped. + hdr := sw.Header() + for k := range hdr { + hdr.Del(k) + } + + // attemptReq is a FRESH top-level copy of pristineReq for every + // candidate (F1 fix, this function's own doc comment above). + // gatewayAliasKey is (re-)injected here: every providerAdapter + // implementation unconditionally deletes it before it ever + // touches the network (its own doc comment, providers.go), so + // each fresh copy needs it set again to echo the client's own + // requested alias correctly (anthropic/gemini-type adapters read + // it to build their own response envelope). + attemptReq := make(map[string]any, len(pristineReq)+1) + for k, v := range pristineReq { + attemptReq[k] = v + } + attemptReq[gatewayAliasKey] = requestedModel + + // cacheKeyStr is still computed per candidate — cache + // correctness (brief): a response produced by a failover + // provider must be stored under THAT provider's own key, never + // the first candidate's. Computed from pristineReq (matching the + // pre-pass above exactly): gatewayAliasKey is a purely internal + // echo-back mechanism cacheKey does not itself strip (its own + // doc comment, cache.go), so hashing attemptReq instead would + // add noise no two requests would otherwise disagree on. + var cacheKeyStr string + if cacheable { + cacheKeyStr = cacheKey(cand.providerName, cand.upstreamModel, requestedModel, endpoint, pristineReq) // Set before call() below writes anything — headers must // precede the body a miss is about to produce (spec §2). A // later iteration's own miss overwrites this harmlessly: it @@ -347,16 +453,6 @@ func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scop sw.Header().Set("X-Llmgw-Cache", "miss") } - // gatewayAliasKey is (re-)injected on every iteration, AFTER the - // cacheKey computation above: every providerAdapter implementation - // unconditionally deletes it before it ever touches the network - // (its own doc comment, providers.go) — a failed first attempt - // leaves req without it, and the next candidate needs it set - // again to echo the client's own requested alias correctly - // (anthropic/gemini-type adapters read it to build their own - // response envelope). - req[gatewayAliasKey] = requestedModel - // respWriter is sw, wrapped in a capture tee only when this // request is cacheable — cacheCaptureWriter buffers everything // written so a 200 non-stream response can be stored after @@ -378,19 +474,28 @@ func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scop // misattribute a retried attempt to the first provider" // requirement) — every upstream attempt call() makes, via // upstreamJSON's retryPolicy.do (providers.go), reports through - // here. The same closure also feeds g.failoverHealth (failover.go), - // feat/failover's own request-path health signal, with the - // IDENTICAL isTransient||isDeadlineExceeded classification - // limiter.recordProviderAttempt already applies — see - // requestHealthTracker.record's own doc comment for why this must - // stay the same classification, not a wider one. + // here. limiter.recordProviderAttempt keeps its own existing + // per-raw-attempt classification unchanged (limits.go); feat/ + // failover's OWN request-path health gate is recorded separately + // below, once per candidate rather than once per raw attempt — + // see requestHealthTracker.record's own doc comment (failover.go) + // for why (F3/F6 fix, adversarial-review). providerName := cand.providerName upstreamModel := cand.upstreamModel ctx := withAttemptRecorder(r.Context(), func(resp *http.Response, attemptErr error) { g.limiter.recordProviderAttempt(providerName, upstreamModel, resp, attemptErr) - g.failoverHealth.record(providerName, !isTransient(resp, attemptErr) && !isDeadlineExceeded(attemptErr)) }) - result, callErr := call(cand.adapter, ctx, respWriter, req) + result, callErr := call(cand.adapter, ctx, respWriter, attemptReq) + + // Request-path health, recorded ONCE per candidate per logical + // request (F3/F6 fix, adversarial-review) — see + // requestHealthTracker.record's own doc comment (failover.go) + // for the full ruling: failoverEligible's classification + // (everything except HTTP 400 is a failure) is reused directly, + // not limiter's narrower isTransient/isDeadlineExceeded one, so + // a provider answering 401/403/404 to every request is + // eventually routed around instead of retried forever. + g.failoverHealth.record(providerName, !failoverEligible(callErr)) // Usage is accounted before the error branch below runs, not // after: every adapter that can fail mid-stream (forwardStream @@ -457,6 +562,9 @@ func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scop return } + lastErr = callErr + lastProviderName = providerName + hasNext := i < len(candidates)-1 && failoverEligible(callErr) if hasNext && isProviderNotFoundError(callErr) { // Loud, non-suppressed log line (brief: "Failover after a 404 @@ -465,7 +573,9 @@ func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scop // would hide it") — deliberately g.errorf, not g.logf, and // deliberately its own line rather than folded into the // generic "failing over" line above the next iteration - // already prints. + // already prints, and deliberately NOT gated by + // failoverLogGate (operator ruling: this one must always + // log). g.errorf("%s: provider %q returned 404 for model %q; failing over to %q — this usually means a real provider/model configuration mistake, not a transient outage", logPrefix, providerName, requestedModel, candidates[i+1].providerName) } if !hasNext { diff --git a/tools/yaegi-check/main.go b/tools/yaegi-check/main.go index f9b5efd..935046e 100644 --- a/tools/yaegi-check/main.go +++ b/tools/yaegi-check/main.go @@ -371,7 +371,12 @@ func run() error { })) defer failoverBUpstream.Close() - attemptAccountingOverride := `{"breaker":{"failureThreshold":2,"openDuration":"3s","maxOpenDuration":"6s"},` + + // failover.enabled must be set explicitly here: FailoverConfig.Enabled + // defaults to false (coordinator ruling — a version upgrade with no + // config change must preserve prior behavior), so without this the + // exerciseFailover probe below would find failover-a's 500 returned + // to the client unchanged, never reaching failover-b at all. + attemptAccountingOverride := `{"failover":{"enabled":true},"breaker":{"failureThreshold":2,"openDuration":"3s","maxOpenDuration":"6s"},` + `"providers":{"openai":{"type":"openai","baseUrl":"` + upstream.URL + `","apiKey":"sk-up","models":["` + testDataWantModel + `","` + builtinLookupModelID + `"]},` + `"brk":{"type":"openai","baseUrl":"` + brkUpstream.URL + `","apiKey":"sk-up","discovery":true,"discoveryInterval":"1ms"},"` + slowProviderName + `":{"type":"openai","baseUrl":"` + slowUpstream.URL + `","apiKey":"sk-up","models":["` + slowProviderModel + `"]},` +