From 5d624791fabba30bcdcdd68e9951dae7524890db Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 00:32:12 +0100 Subject: [PATCH] feat(failover): add cross-provider failover for overlapping models Add failover.go with the full candidate pipeline. resolveWithCandidates resolves a request exactly like resolve does, then adds every other provider sharing the identical bare model id. orderedFailoverCandidates applies a hard skip for request-path-unhealthy providers, deprioritizes discovery-unhealthy providers without dropping them, fails open when every candidate reads unhealthy, and caps the list at maxAttempts. Add requestHealthTracker, a separate per-pod request-path health signal fed by the same isTransient/isDeadlineExceeded classification recordProviderAttempt already uses. It never widens the existing discovery circuit breaker, per that breaker's own documented scope. Rewrite runMeteredCall (routes_unified.go) to loop over candidates instead of resolving once. Each attempt rebinds the attempt recorder to its own provider, recomputes the cache key before the response is cached, and re-injects the alias echo key an adapter may have deleted on a prior failed attempt. Failover never runs once any byte has reached the client, since statusTrackingWriter already tracks that state. A failure falls through on everything except HTTP 400, matching the brief exactly. A 404 failover logs loudly, naming both providers and the model, since it usually signals a real configuration mistake. Add FailoverConfig (Config.Failover): an Enabled pointer defaulting to true, and a MaxAttempts bound between 1 and 10, defaulting to 3. A single-provider deployment, or one with no overlapping bare model ids, behaves exactly as before regardless of this setting. Add a yaegi-check probe (tools/yaegi-check/main.go) driving two providers under the real interpreter, one failing and one serving, proving the candidate loop runs correctly interpreted. Update README.md with a Failover section and correct the Retry section's stale claim that no cross-provider failover exists. Media routes (images, audio speech, audio transcriptions) do not get failover in this change. Each has its own request-shape handling that needs its own scoped follow-up. --- README.md | 59 ++- failover.go | 487 ++++++++++++++++++ failover_test.go | 1009 +++++++++++++++++++++++++++++++++++++ llmgateway.go | 48 +- routes_unified.go | 306 +++++++---- tools/yaegi-check/main.go | 76 ++- 6 files changed, 1866 insertions(+), 119 deletions(-) create mode 100644 failover.go create mode 100644 failover_test.go diff --git a/README.md b/README.md index a2f0871..57b07dc 100644 --- a/README.md +++ b/README.md @@ -193,6 +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. | | `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. | @@ -807,10 +808,10 @@ released, so this never becomes a ceiling on total concurrent requests. ## Retry -Same-provider retry for a transient upstream failure — no cross-provider -failover. Off by default (`retry.enabled: false`); every config that -predates this field keeps making exactly one upstream attempt per -request. +Same-provider retry for a transient upstream failure. Off by default +(`retry.enabled: false`); every config that predates this field keeps +making exactly one upstream attempt per request. See [Failover](#failover) +below for the separate, cross-provider mechanism. - **Scope**: every adapter upstream call — chat, embeddings, model listing, and the three media endpoints (`images/generations`, @@ -846,6 +847,56 @@ request. is never double-counted. Request counters still increment once per client request, unchanged. +## Failover + +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. + +- **Scope**: `POST /v1/chat/completions`, `POST /v1/embeddings`, and + `POST /v1/messages` only. The image/audio endpoints + (`images/generations`, `audio/speech`, `audio/transcriptions`) are not + wired into failover yet — each has its own request-shape handling + (multipart rebuild, binary/streaming responses) that needs its own + scoped follow-up. +- **Eligible candidates**: a bare model id resolved via the collision + winner (the same mechanism `GET /v1/models` documents for overlapping + catalogs) offers every OTHER configured provider that also serves that + id and that the caller's group is authorized to use. A `provider/model` + request, or an alias whose target is `provider/model`, is an explicit, + exact choice and is never eligible for substitution. +- **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. +- **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. +- **Health signals, kept separate**: the discovery-endpoint health + surfaced on the admin dashboard is an ordering hint only — a provider + 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. +- **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 + success-rate badge) attribute every attempt to the provider that + 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. +- **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. + ## Caching An opt-in, Redis-backed response cache for the unified, non-streaming diff --git a/failover.go b/failover.go new file mode 100644 index 0000000..c399817 --- /dev/null +++ b/failover.go @@ -0,0 +1,487 @@ +package traefikllmgateway + +import ( + "context" + "errors" + "fmt" + "net/http" + "sync" + "time" +) + +// This file implements feat/failover: cross-provider failover for a bare +// model id more than one configured provider serves (registry.go's +// bareWinner collision case — production logs show whisper-1, tts-1, +// tts-1-hd, text-embedding-3-small, text-embedding-3-large, and +// text-embedding-ada-002 served by two or more of openai/openai-audio/ +// copilot today). Before this feature, registry.go's resolve returns +// exactly one provider adapter and retry.go's retryPolicy only ever +// retries the SAME provider — so a down provider fails every request for +// its models even when another configured provider serves the identical +// model. This file turns that already-existing redundancy into a hot +// standby, hooked into routes_unified.go's runMeteredCall, the ONE shared +// metered-call path both /v1/chat/completions and /v1/messages already go +// through. +// +// 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 +// overlapping models (whisper-1, tts-1/tts-1-hd) are media-route +// traffic. Each of those three handlers has its own request-shape +// preprocessing (multipart rebuild for transcriptions, a JSON +// re-marshal for speech, a binary/streaming response for speech) that +// would need its own careful "nothing written yet" analysis and its own +// test coverage, rather than reusing runMeteredCall's shared shape — the +// brief's own hook-point instruction ("do not fork a second +// implementation") is about not duplicating the chat/messages pipeline, +// not about extending every possible call site in one pass. Left as a +// 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). +type FailoverConfig struct { + Enabled *bool `json:"enabled,omitempty"` + // MaxAttempts caps how many DIFFERENT PROVIDERS runMeteredCall tries + // in total for one logical request, the primary included — distinct + // from RetryConfig.Attempts (retry.go), which retries the SAME + // provider. 0 (the default) uses defaultFailoverMaxAttempts. A value + // outside 1..maxFailoverMaxAttempts is a construction error. + MaxAttempts int `json:"maxAttempts,omitempty"` +} + +// defaultFailoverMaxAttempts is applied when FailoverConfig.MaxAttempts is +// left at 0. maxFailoverMaxAttempts bounds an explicit override, the same +// "cheap sanity ceiling on an operator value" precedent +// maxBreakerFailureThreshold/maxRetryAttempts already set (registry.go/ +// retry.go). +const ( + defaultFailoverMaxAttempts = 3 + maxFailoverMaxAttempts = 10 +) + +// failoverConfig is FailoverConfig, validated and defaulted once at +// construction (newGateway, llmgateway.go) — mirrors breakerConfig/ +// retryPolicy's own resolved-value-not-raw-config shape (registry.go/ +// retry.go). +type failoverConfig struct { + maxAttempts int + enabled bool +} + +// validateFailoverConfig validates fc and returns the resolved +// failoverConfig newGateway attaches to the Gateway. +func validateFailoverConfig(fc FailoverConfig) (failoverConfig, error) { + enabled := true + if fc.Enabled != nil { + enabled = *fc.Enabled + } + + maxAttempts := fc.MaxAttempts + if maxAttempts == 0 { + maxAttempts = defaultFailoverMaxAttempts + } + if maxAttempts < 1 || maxAttempts > maxFailoverMaxAttempts { + return failoverConfig{}, fmt.Errorf("llmgateway: failover.maxAttempts must be between 1 and %d, got %d", maxFailoverMaxAttempts, maxAttempts) + } + + return failoverConfig{enabled: enabled, maxAttempts: maxAttempts}, nil +} + +// resolveCandidate is one provider that can serve a client-requested model +// id — modelRegistry.resolve's own (adapter, upstreamModel, canonical) +// triple, tagged with the provider's own name so a failover attempt can +// rebind its cache key, attempt recorder, and error log to the ACTUAL +// provider that served or failed the request, never the first one resolve +// happened to pick. +type resolveCandidate struct { + adapter providerAdapter + providerName string + upstreamModel string + canonical string +} + +// resolveWithCandidates resolves id exactly as modelRegistry.resolve +// (registry.go) does — identical winner, identical error — and +// additionally reports every OTHER configured provider that also serves +// the same bare upstream model id and remains authorized for grp, as +// feat/failover's candidate pool. +// +// primary/err are byte-identical to what resolve(id, grp) alone would +// return: this method calls resolve internally and never second-guesses +// its winner or its error — feat/failover only ADDS candidates after an +// already-successful resolution, it never changes which model or error a +// request resolves to (in particular, authorization denial/unknown-model +// behavior for the primary winner is completely unchanged by this +// branch). extra is nil whenever there is nothing to add: +// +// - id resolved via an explicit "provider/model" prefix, or an alias +// whose target is itself provider-prefixed: the requester's own +// explicit provider choice is never a candidate for pool +// substitution — ruling, this feature's own design: "provider/model" +// addressing is exact, not eligible for failover. +// - id resolved via a bare-id collision winner (bareWinner), or an +// alias whose target is itself a bare-id collision winner: extra is +// every OTHER configured provider whose known model set also +// contains that identical bare id string, in +// modelRegistry.providerNames (sorted) order, and authorized for grp +// on that provider. +func (m *modelRegistry) resolveWithCandidates(id string, grp *group) (resolveCandidate, []resolveCandidate, error) { + adapter, upstreamModel, canonical, err := m.resolve(id, grp) + if err != nil { + return resolveCandidate{}, nil, err + } + primary := resolveCandidate{adapter: adapter, providerName: adapter.name(), upstreamModel: upstreamModel, canonical: canonical} + + bareID := id + if target, isAlias := m.aliases[id]; isAlias { + if _, _, ok := m.splitConfiguredProvider(target); ok { + return primary, nil, nil + } + bareID = target + } else if _, _, ok := m.splitConfiguredProvider(id); ok { + return primary, nil, nil + } + + return primary, m.failoverCandidates(bareID, primary.providerName, grp), 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 +// raw candidate pool orderedFailoverCandidates (below) narrows and +// reorders further by health. +// +// Only grp.allowsProvider is re-checked per candidate: grp.allowsModel is +// already known true from the primary winner's own successful resolve +// call, and — for the identical bare id string every candidate here +// shares by construction — allowsModel's result does not vary by +// provider (auth.go's allowsModel takes no provider argument), so +// re-checking it per candidate would be redundant work, not a +// correctness gap. +func (m *modelRegistry) failoverCandidates(bareID, primaryProvider string, grp *group) []resolveCandidate { + var extra []resolveCandidate + for _, name := range m.providerNames { + if name == primaryProvider { + continue + } + if !m.states[name].hasModel(bareID) || !grp.allowsProvider(name) { + continue + } + // canonical assigned to a local before the struct literal, matching + // resolveAgainst's own documented Yaegi precedent (registry.go): a + // defensive, cheap precaution against the same class of interpreter + // trap, not a proven necessity for a struct-literal field (only a + // multi-value RETURN tuple is the proven-bad shape there). + canonical := name + "/" + bareID + extra = append(extra, resolveCandidate{ + adapter: m.adapters[name], + providerName: name, + upstreamModel: bareID, + canonical: canonical, + }) + } + return extra +} + +// orderedFailoverCandidates builds runMeteredCall's actual try list from +// resolveWithCandidates' own (primary, extra) split. +// +// Failover is a no-op (returns exactly [primary]) whenever it is disabled +// (g.failover.enabled false) or there is nothing to add (len(extra) == +// 0) — the second case is what makes "single-provider deployment behaves +// exactly as before" hold unconditionally, regardless of the enabled +// switch: there is never more than one candidate to try when no other +// configured provider shares this bare model id. +// +// Otherwise, two health signals apply, deliberately kept separate +// (modelRegistry.discoveryHealthy's own doc comment, registry.go, is +// binding on this): +// +// - REQUEST-PATH health (g.failoverHealth, this file) is a HARD SKIP +// gate: a candidate it reports unhealthy is excluded outright, +// never attempted. If that filtering would remove every candidate +// (this local, per-pod signal reporting the whole fleet down +// simultaneously is more likely stale than true), it is discarded +// and every original candidate is tried anyway — failing open +// against the gateway's own imperfect signal rather than refusing +// to route a request at all. +// - DISCOVERY health (g.registry.discoveryHealthy) is an ORDERING HINT +// ONLY, exactly as its own doc comment requires: a discovery- +// unhealthy candidate is moved after every discovery-healthy/ +// 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. +func (g *Gateway) orderedFailoverCandidates(primary resolveCandidate, extra []resolveCandidate) []resolveCandidate { + if !g.failover.enabled || len(extra) == 0 { + return []resolveCandidate{primary} + } + + all := make([]resolveCandidate, 0, len(extra)+1) + all = append(all, primary) + all = append(all, extra...) + + healthy := make([]resolveCandidate, 0, len(all)) + for _, c := range all { + if g.failoverHealth.healthy(c.providerName) { + healthy = append(healthy, c) + } + } + if len(healthy) == 0 { + healthy = all + } + + ordered := make([]resolveCandidate, 0, len(healthy)) + var deprioritized []resolveCandidate + for _, c := range healthy { + if g.registry.discoveryHealthy(c.providerName) { + ordered = append(ordered, c) + } else { + deprioritized = append(deprioritized, c) + } + } + ordered = append(ordered, deprioritized...) + + if max := g.failover.maxAttempts; max > 0 && len(ordered) > max { + ordered = ordered[:max] + } + return ordered +} + +// failoverEligible reports whether callErr — runMeteredCall's own +// adapterCall return, evaluated only once nothing has reached the client +// yet (the hard constraint this whole feature is built around: once an +// adapter has written anything, status and body are committed and can +// never be retried elsewhere) — should try the next candidate instead of +// answering the client with callErr. +// +// Ruling (brief, "Which failures fall through" — decided, not +// relitigated here): every failure except HTTP 400 falls through. 5xx, +// connection failures, timeouts, 429, 401, 403, and 404 all do; 400 +// never does, because the request itself is malformed and every provider +// would reject it identically, so retrying just multiplies latency and +// burns budget. +// +// Two failure shapes are excluded for the SAME "retrying changes +// nothing" reasoning 400 itself is excluded for, even though neither is +// literally an HTTP 400: context.Canceled (the client walked away — +// failing over would only waste an attempt against a connection nobody +// is waiting on) and errRequestBuildFailed (upstreamBytes, providers.go +// — the identical malformed request shape reaches every candidate the +// same way, differing only in which operator-configured base URL it is +// built against). A *translateError or *responseTranslationError is this +// gateway's OWN translation code failing on an already-decoded request — +// not a signal about the PROVIDER's availability at all — so neither is +// eligible either. +func failoverEligible(callErr error) bool { + if callErr == nil { + return false + } + if errors.Is(callErr, context.Canceled) || errors.Is(callErr, errRequestBuildFailed) { + return false + } + // Plain type assertions, not errors.As — matching this package's + // established Yaegi-safe convention for these exact two pointer error + // types (providerHTTPError's own doc comment, providers.go; + // handleAdapterErrorEnvelope, routes_unified.go, does the identical + // assertion for the identical reason). + if perr, ok := callErr.(*providerHTTPError); ok { + return perr.status != http.StatusBadRequest + } + if _, ok := callErr.(*translateError); ok { + return false + } + if _, ok := callErr.(*responseTranslationError); ok { + return false + } + return true +} + +// isProviderNotFoundError reports whether err is a *providerHTTPError +// carrying HTTP 404 — routes_unified.go's runMeteredCall uses this to +// decide when a failover needs its own loud, non-suppressed log line +// (brief: "Failover after a 404 must log loudly ... a 404 usually means +// a real configuration mistake, and silently succeeding elsewhere would +// hide it"). +func isProviderNotFoundError(err error) bool { + perr, ok := err.(*providerHTTPError) + return ok && perr.status == http.StatusNotFound +} + +// requestBreakerFailureThreshold/requestBreakerOpenDuration/ +// requestBreakerMaxOpenDuration size requestHealthTracker's own breaker — +// deliberately separate constants from registry.go's discovery breaker +// (defaultBreakerFailureThreshold/defaultBreakerOpenDuration/ +// defaultBreakerMaxOpenDuration): a request-path outage needs to be +// noticed and routed around within seconds, not the discovery breaker's +// minutes-to-hours cadence, and modelRegistry.discoveryHealthy's own doc +// comment forbids widening that breaker to cover this outcome stream at +// all. Fixed constants, not configurable — unlike FailoverConfig above, +// the brief only asks for max-attempts and an on/off switch to be +// operator-configurable; this breaker's thresholds are an internal +// implementation detail of the routing-gate signal, not a documented +// tuning surface. +const ( + requestBreakerFailureThreshold = 3 + requestBreakerOpenDuration = 30 * time.Second + requestBreakerMaxOpenDuration = 5 * time.Minute +) + +// requestHealthState is one provider's REQUEST-PATH health (feat/ +// failover), mu-guarded — mirrors providerState's per-provider, +// individually-mutexed shape (registry.go) at a much smaller scale: this +// tracks exactly one outcome stream (request-path attempts), never +// discovery. The zero value is closed/healthy, so a provider that never +// fails never touches anything below open/backoff/openUntil. +// +// Field order below is fieldalignment-verified (golangci-lint's govet +// enable-all, run with -fix against a scratch copy to derive the exact +// zero-waste sequence, then hand-applied here so this doc comment +// survives — the same convention providerState's own doc comment +// explains, registry.go). +type requestHealthState struct { + openUntil time.Time + backoff time.Duration + consecutiveFailures int + mu sync.Mutex + open bool +} + +// requestHealthTracker is feat/failover's own per-pod, in-memory, +// 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. +// +// Every method is nil-receiver-safe (both report "healthy"/no-op on a +// nil *requestHealthTracker), matching this package's own nil-means- +// disabled convention for an optional subsystem (responseCache's own doc +// comment, llmgateway.go's Gateway.cache field) — a Gateway assembled +// directly in a test without going through newGateway, or any future +// caller that does the same, degrades to "failover never skips a +// candidate for request health" rather than a nil-pointer panic. +type requestHealthTracker struct { + states map[string]*requestHealthState + nowFn func() time.Time + mu sync.Mutex +} + +// newRequestHealthTracker returns an empty tracker — every provider name +// reads healthy until its first recorded failure. +func newRequestHealthTracker() *requestHealthTracker { + return &requestHealthTracker{states: make(map[string]*requestHealthState), nowFn: time.Now} +} + +// stateFor returns provider's requestHealthState, creating it on first +// use. +func (t *requestHealthTracker) stateFor(provider string) *requestHealthState { + t.mu.Lock() + defer t.mu.Unlock() + st, ok := t.states[provider] + if !ok { + st = &requestHealthState{} + t.states[provider] = st + } + 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. +// +// 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. +// +// Deliberately no single-flight guard on the probe window (unlike +// registry.go's tryBeginRefresh inFlight flag for its own, much lower- +// volume, background-goroutine probe): this signal sits on the live +// request path, where many concurrent requests can observe the same +// post-cooldown window at once. Serializing them would need its own +// lock-and-CAS dance for a benefit — one broken provider getting at most +// one real probe instead of a short burst of them right as its backoff +// expires — this per-pod, best-effort signal does not need to pay for. +func (t *requestHealthTracker) record(provider string, success bool) { + if t == nil || provider == "" { + return + } + st := t.stateFor(provider) + st.mu.Lock() + defer st.mu.Unlock() + now := t.nowFn() + + probed := st.open && !now.Before(st.openUntil) + + if success { + st.open = false + st.consecutiveFailures = 0 + st.backoff = 0 + st.openUntil = time.Time{} + return + } + + if probed { + if st.backoff <= 0 { + st.backoff = requestBreakerOpenDuration + } else { + st.backoff *= 2 + } + if st.backoff > requestBreakerMaxOpenDuration { + st.backoff = requestBreakerMaxOpenDuration + } + st.openUntil = now.Add(st.backoff) + return + } + + st.consecutiveFailures++ + if st.consecutiveFailures >= requestBreakerFailureThreshold { + st.open = true + st.backoff = requestBreakerOpenDuration + st.openUntil = now.Add(st.backoff) + } +} + +// healthy reports whether provider's request-path breaker currently +// permits routing to it — feat/failover's hard SKIP gate +// (orderedFailoverCandidates, above), distinct from +// modelRegistry.discoveryHealthy's deprioritize-only contract. An +// unrecorded provider (never failed, or never attempted at all) +// reads healthy — the same "never seen a failure" default the discovery +// breaker's own zero value already applies (providerState's doc +// comment, registry.go). +func (t *requestHealthTracker) healthy(provider string) bool { + if t == nil || provider == "" { + return true + } + st := t.stateFor(provider) + st.mu.Lock() + defer st.mu.Unlock() + if !st.open { + return true + } + return !t.nowFn().Before(st.openUntil) +} diff --git a/failover_test.go b/failover_test.go new file mode 100644 index 0000000..d74024b --- /dev/null +++ b/failover_test.go @@ -0,0 +1,1009 @@ +package traefikllmgateway + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --- modelRegistry.resolveWithCandidates / failoverCandidates --- + +func TestModelRegistry_ResolveWithCandidates_BareCollision_ReturnsOrderedExtras(t *testing.T) { + t.Parallel() + adapters := map[string]providerAdapter{ + "beta": newFakeAdapter("beta"), + "alpha": newFakeAdapter("alpha"), + "gamma": newFakeAdapter("gamma"), + } + cfg := &Config{Providers: map[string]*ProviderConfig{ + "beta": {Models: []string{"shared"}}, + "alpha": {Models: []string{"shared"}}, + "gamma": {Models: []string{"shared"}}, + }} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + require.NoError(t, err) + + primary, extra, err := reg.resolveWithCandidates("shared", allowAllGroup()) + require.NoError(t, err) + assert.Equal(t, "alpha", primary.providerName, "primary must match resolve's own sorted-first winner") + + names := make([]string, len(extra)) + for i, c := range extra { + names[i] = c.providerName + } + assert.Equal(t, []string{"beta", "gamma"}, names, "extras must be every OTHER owning provider, in providerNames sorted order, primary excluded") + for _, c := range extra { + assert.Equal(t, "shared", c.upstreamModel) + assert.Equal(t, c.providerName+"/shared", c.canonical) + } +} + +func TestModelRegistry_ResolveWithCandidates_ProviderPrefixed_NoExtras(t *testing.T) { + t.Parallel() + adapters := map[string]providerAdapter{ + "beta": newFakeAdapter("beta"), + "alpha": newFakeAdapter("alpha"), + } + cfg := &Config{Providers: map[string]*ProviderConfig{ + "beta": {Models: []string{"shared"}}, + "alpha": {Models: []string{"shared"}}, + }} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + require.NoError(t, err) + + primary, extra, err := reg.resolveWithCandidates("beta/shared", allowAllGroup()) + require.NoError(t, err) + assert.Equal(t, "beta", primary.providerName) + assert.Nil(t, extra, "an explicit provider/model request must never gain failover candidates") +} + +func TestModelRegistry_ResolveWithCandidates_AliasToProviderPrefixedTarget_NoExtras(t *testing.T) { + t.Parallel() + adapters := map[string]providerAdapter{ + "beta": newFakeAdapter("beta"), + "alpha": newFakeAdapter("alpha"), + } + cfg := &Config{ + Providers: map[string]*ProviderConfig{ + "beta": {Models: []string{"shared"}}, + "alpha": {Models: []string{"shared"}}, + }, + ModelAliases: map[string]string{"aliased": "beta/shared"}, + } + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + require.NoError(t, err) + + primary, extra, err := reg.resolveWithCandidates("aliased", allowAllGroup()) + require.NoError(t, err) + assert.Equal(t, "beta", primary.providerName) + assert.Nil(t, extra, "an alias targeting an explicit provider/model must never gain failover candidates") +} + +func TestModelRegistry_ResolveWithCandidates_AliasToBareTarget_ReturnsExtras(t *testing.T) { + t.Parallel() + adapters := map[string]providerAdapter{ + "beta": newFakeAdapter("beta"), + "alpha": newFakeAdapter("alpha"), + } + cfg := &Config{ + Providers: map[string]*ProviderConfig{ + "beta": {Models: []string{"shared"}}, + "alpha": {Models: []string{"shared"}}, + }, + ModelAliases: map[string]string{"aliased": "shared"}, + } + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + require.NoError(t, err) + + primary, extra, err := reg.resolveWithCandidates("aliased", allowAllGroup()) + require.NoError(t, err) + assert.Equal(t, "alpha", primary.providerName, "an alias to a bare-collision target resolves via the same sorted-first winner") + require.Len(t, extra, 1) + assert.Equal(t, "beta", extra[0].providerName) +} + +// TestModelRegistry_ResolveWithCandidates_ExtraFiltersByProviderAuthorization +// proves a group restricted away from a would-be failover provider never +// gets it offered as a candidate, even though that provider genuinely +// serves the bare id. +func TestModelRegistry_ResolveWithCandidates_ExtraFiltersByProviderAuthorization(t *testing.T) { + t.Parallel() + adapters := map[string]providerAdapter{ + "beta": newFakeAdapter("beta"), + "alpha": newFakeAdapter("alpha"), + } + cfg := &Config{Providers: map[string]*ProviderConfig{ + "beta": {Models: []string{"shared"}}, + "alpha": {Models: []string{"shared"}}, + }} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + require.NoError(t, err) + + grp := &group{name: "restricted", providers: []string{"alpha"}} + primary, extra, err := reg.resolveWithCandidates("shared", grp) + require.NoError(t, err) + assert.Equal(t, "alpha", primary.providerName) + assert.Empty(t, extra, "beta is denied to this group, so it must never appear as a failover candidate") +} + +// TestModelRegistry_ResolveWithCandidates_ResolveError_PropagatesUnchanged +// proves feat/failover never changes resolve's own error/status contract: +// an unknown or denied model still resolves to the identical error, with +// no candidates offered. +func TestModelRegistry_ResolveWithCandidates_ResolveError_PropagatesUnchanged(t *testing.T) { + t.Parallel() + adapters := map[string]providerAdapter{"alpha": newFakeAdapter("alpha")} + cfg := &Config{Providers: map[string]*ProviderConfig{"alpha": {Models: []string{"gpt-test"}}}} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + require.NoError(t, err) + + _, extra, err := reg.resolveWithCandidates("no-such-model", allowAllGroup()) + assert.ErrorIs(t, err, errModelUnknown) + assert.Nil(t, extra) + + grp := &group{name: "denied", models: []string{"something-else"}} + _, extra, err = reg.resolveWithCandidates("gpt-test", grp) + assert.ErrorIs(t, err, errModelDenied) + assert.Nil(t, extra) +} + +// --- requestHealthTracker --- + +func TestRequestHealthTracker_HealthyByDefault(t *testing.T) { + t.Parallel() + tr := newRequestHealthTracker() + assert.True(t, tr.healthy("never-seen")) +} + +func TestRequestHealthTracker_NilSafe(t *testing.T) { + t.Parallel() + var tr *requestHealthTracker + assert.True(t, tr.healthy("anything"), "a nil tracker must report healthy, never panic") + assert.NotPanics(t, func() { tr.record("anything", false) }) +} + +// TestRequestHealthTracker_OpensAfterThreshold_ThenSkips is the direct +// regression test for "a provider marked unhealthy by REQUEST-path health +// is skipped without being called": would FAIL if record() never opened +// the breaker (a reverted feature would leave healthy() returning true +// forever). +func TestRequestHealthTracker_OpensAfterThreshold_ThenSkips(t *testing.T) { + t.Parallel() + tr := newRequestHealthTracker() + now := time.Now() + tr.nowFn = func() time.Time { return now } + + for i := 0; i < requestBreakerFailureThreshold-1; i++ { + tr.record("p", false) + assert.True(t, tr.healthy("p"), "must stay healthy before the threshold is reached") + } + tr.record("p", false) + assert.False(t, tr.healthy("p"), "must open exactly at the failure threshold") +} + +func TestRequestHealthTracker_SuccessBeforeThreshold_ResetsCounter(t *testing.T) { + t.Parallel() + tr := newRequestHealthTracker() + for i := 0; i < requestBreakerFailureThreshold-1; i++ { + tr.record("p", false) + } + tr.record("p", true) + // The counter must have reset to zero, not merely "not yet open": one + // more failure alone must not open it. + tr.record("p", false) + assert.True(t, tr.healthy("p"), "a success below threshold must reset the failure count, not just pause it") +} + +// TestRequestHealthTracker_RecoversAfterCooldown_ThenCloses proves the +// breaker admits exactly one probe once its cooldown elapses, and a +// successful probe closes it. +func TestRequestHealthTracker_RecoversAfterCooldown_ThenCloses(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) + } + require.False(t, tr.healthy("p")) + + now = now.Add(requestBreakerOpenDuration - time.Millisecond) + assert.False(t, tr.healthy("p"), "must stay unhealthy until the cooldown fully elapses") + + now = now.Add(2 * time.Millisecond) + assert.True(t, tr.healthy("p"), "must admit exactly one probe once the cooldown elapses") + + tr.record("p", true) + assert.True(t, tr.healthy("p")) + // A single further failure alone must not reopen it — closeBreaker + // resets the failure counter to zero, mirroring registry.go's own + // discovery breaker. + tr.record("p", false) + assert.True(t, tr.healthy("p")) +} + +// TestRequestHealthTracker_FailedProbe_ReopensWithDoubledBackoff proves a +// failed post-cooldown probe re-opens the breaker with a LONGER backoff, +// not the same one — would FAIL if reverted to a fixed-duration reopen. +func TestRequestHealthTracker_FailedProbe_ReopensWithDoubledBackoff(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) + } + now = now.Add(requestBreakerOpenDuration + time.Millisecond) + require.True(t, tr.healthy("p"), "cooldown elapsed, must admit the probe") + tr.record("p", false) // the probe itself fails + + assert.False(t, tr.healthy("p"), "a failed probe must reopen the breaker") + // The new backoff is requestBreakerOpenDuration*2: advancing only + // past the ORIGINAL window must still read unhealthy. + now = now.Add(requestBreakerOpenDuration + time.Millisecond) + assert.False(t, tr.healthy("p"), "a failed probe must double the backoff, not repeat the original window") + + now = now.Add(requestBreakerOpenDuration + 2*time.Millisecond) + assert.True(t, tr.healthy("p"), "must recover once the doubled window has fully elapsed") +} + +func TestRequestHealthTracker_BackoffCapsAtMax(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) + } + // Fail every probe repeatedly — backoff must never exceed + // requestBreakerMaxOpenDuration, so this loop must always terminate + // within a bounded number of doublings. + for i := 0; i < 20; i++ { + now = now.Add(requestBreakerMaxOpenDuration + time.Millisecond) + require.True(t, tr.healthy("p")) + tr.record("p", false) + } + st := tr.stateFor("p") + st.mu.Lock() + backoff := st.backoff + st.mu.Unlock() + assert.LessOrEqual(t, backoff, requestBreakerMaxOpenDuration) +} + +// TestRequestHealthTracker_IndependentPerProvider proves one provider's +// failures never affect another's health — the per-pod signal is keyed by +// provider name, not shared global state. +func TestRequestHealthTracker_IndependentPerProvider(t *testing.T) { + t.Parallel() + tr := newRequestHealthTracker() + for i := 0; i < requestBreakerFailureThreshold; i++ { + tr.record("broken", false) + } + assert.False(t, tr.healthy("broken")) + assert.True(t, tr.healthy("fine"), "an unrelated provider's health must be untouched") +} + +// TestRequestHealthTracker_ConcurrentRecord_NoRace exercises record/healthy +// from many goroutines at once — the whole point of this signal is to sit +// on the live request path. +func TestRequestHealthTracker_ConcurrentRecord_NoRace(t *testing.T) { + t.Parallel() + tr := newRequestHealthTracker() + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + tr.record("p", i%3 != 0) + tr.healthy("p") + }(i) + } + wg.Wait() +} + +// --- failoverEligible / isProviderNotFoundError --- + +func TestFailoverEligible(t *testing.T) { + t.Parallel() + deadlineExceededErr := fmt.Errorf("%w: %w", errUpstream, fmt.Errorf("dial tcp: %w", context.DeadlineExceeded)) + + tests := []struct { + err error + name string + want bool + }{ + {name: "nil error", err: nil, want: false}, + {name: "400 never falls through", err: &providerHTTPError{status: http.StatusBadRequest}, want: false}, + {name: "500 falls through", err: &providerHTTPError{status: http.StatusInternalServerError}, want: true}, + {name: "502 falls through", err: &providerHTTPError{status: http.StatusBadGateway}, want: true}, + {name: "429 falls through", err: &providerHTTPError{status: http.StatusTooManyRequests}, want: true}, + {name: "401 falls through", err: &providerHTTPError{status: http.StatusUnauthorized}, want: true}, + {name: "403 falls through", err: &providerHTTPError{status: http.StatusForbidden}, want: true}, + {name: "404 falls through", err: &providerHTTPError{status: http.StatusNotFound}, want: true}, + {name: "connection failure falls through", err: fmt.Errorf("%w: dial tcp: connection refused", errUpstream), want: true}, + {name: "deadline exceeded (timeout) falls through", err: deadlineExceededErr, want: true}, + {name: "client canceled does not fall through", err: fmt.Errorf("%w: %w", errUpstream, context.Canceled), want: false}, + {name: "request build failure does not fall through", err: fmt.Errorf("%w: %w: build request", errUpstream, errRequestBuildFailed), want: false}, + {name: "translateError does not fall through", err: &translateError{msg: "bad field"}, want: false}, + {name: "responseTranslationError does not fall through", err: &responseTranslationError{err: errors.New("boom")}, want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, failoverEligible(tt.err)) + }) + } +} + +func TestIsProviderNotFoundError(t *testing.T) { + t.Parallel() + assert.True(t, isProviderNotFoundError(&providerHTTPError{status: http.StatusNotFound})) + assert.False(t, isProviderNotFoundError(&providerHTTPError{status: http.StatusInternalServerError})) + assert.False(t, isProviderNotFoundError(errors.New("plain error"))) + assert.False(t, isProviderNotFoundError(nil)) +} + +// --- validateFailoverConfig --- + +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.Equal(t, defaultFailoverMaxAttempts, fc.maxAttempts) +} + +func TestValidateFailoverConfig_ExplicitDisable(t *testing.T) { + t.Parallel() + no := false + fc, err := validateFailoverConfig(FailoverConfig{Enabled: &no}) + require.NoError(t, err) + assert.False(t, fc.enabled) +} + +func TestValidateFailoverConfig_MaxAttemptsBounds(t *testing.T) { + t.Parallel() + _, err := validateFailoverConfig(FailoverConfig{MaxAttempts: -1}) + assert.Error(t, err) + + _, err = validateFailoverConfig(FailoverConfig{MaxAttempts: maxFailoverMaxAttempts + 1}) + assert.Error(t, err) + + fc, err := validateFailoverConfig(FailoverConfig{MaxAttempts: maxFailoverMaxAttempts}) + require.NoError(t, err) + assert.Equal(t, maxFailoverMaxAttempts, fc.maxAttempts) +} + +// --- Gateway.orderedFailoverCandidates --- + +// testGatewayForCandidates builds a minimal *Gateway sufficient for +// orderedFailoverCandidates: a real modelRegistry (for discoveryHealthy) +// and a fresh requestHealthTracker, without going through newGateway/New. +func testGatewayForCandidates(t *testing.T, providerNames []string, fc failoverConfig) *Gateway { + t.Helper() + adapters := make(map[string]providerAdapter, len(providerNames)) + providers := make(map[string]*ProviderConfig, len(providerNames)) + for _, name := range providerNames { + adapters[name] = newFakeAdapter(name) + providers[name] = &ProviderConfig{Models: []string{"shared"}} + } + reg, err := newModelRegistry(adapters, &Config{Providers: providers}, func(string, ...any) {}) + require.NoError(t, err) + return &Gateway{registry: reg, failoverHealth: newRequestHealthTracker(), failover: fc} +} + +func candNames(cands []resolveCandidate) []string { + out := make([]string, len(cands)) + for i, c := range cands { + out[i] = c.providerName + } + return out +} + +func TestOrderedFailoverCandidates_Disabled_ReturnsPrimaryOnly(t *testing.T) { + t.Parallel() + g := testGatewayForCandidates(t, []string{"alpha", "beta"}, failoverConfig{enabled: false, maxAttempts: 3}) + primary := resolveCandidate{providerName: "alpha"} + extra := []resolveCandidate{{providerName: "beta"}} + got := g.orderedFailoverCandidates(primary, extra) + assert.Equal(t, []string{"alpha"}, candNames(got), "disabled failover must never try a second provider") +} + +func TestOrderedFailoverCandidates_NoExtras_ReturnsPrimaryOnly_RegardlessOfEnabled(t *testing.T) { + t.Parallel() + g := testGatewayForCandidates(t, []string{"alpha"}, failoverConfig{enabled: true, maxAttempts: 3}) + primary := resolveCandidate{providerName: "alpha"} + got := g.orderedFailoverCandidates(primary, nil) + assert.Equal(t, []string{"alpha"}, candNames(got), "single-provider deployment must behave exactly as before regardless of the enabled switch") +} + +func TestOrderedFailoverCandidates_RequestUnhealthy_HardSkipped(t *testing.T) { + t.Parallel() + g := testGatewayForCandidates(t, []string{"alpha", "beta"}, failoverConfig{enabled: true, maxAttempts: 3}) + for i := 0; i < requestBreakerFailureThreshold; i++ { + g.failoverHealth.record("alpha", false) + } + primary := resolveCandidate{providerName: "alpha"} + extra := []resolveCandidate{{providerName: "beta"}} + got := g.orderedFailoverCandidates(primary, extra) + assert.Equal(t, []string{"beta"}, candNames(got), "a request-unhealthy provider must be excluded outright, never merely deprioritized") +} + +func TestOrderedFailoverCandidates_AllRequestUnhealthy_FailsOpenToFullList(t *testing.T) { + t.Parallel() + g := testGatewayForCandidates(t, []string{"alpha", "beta"}, failoverConfig{enabled: true, maxAttempts: 3}) + for _, name := range []string{"alpha", "beta"} { + for i := 0; i < requestBreakerFailureThreshold; i++ { + g.failoverHealth.record(name, false) + } + } + primary := resolveCandidate{providerName: "alpha"} + extra := []resolveCandidate{{providerName: "beta"}} + got := g.orderedFailoverCandidates(primary, extra) + assert.ElementsMatch(t, []string{"alpha", "beta"}, candNames(got), "every candidate unhealthy must fail OPEN to the full list, not refuse to route at all") +} + +func TestOrderedFailoverCandidates_DiscoveryUnhealthy_DeprioritizedNotDropped(t *testing.T) { + t.Parallel() + g := testGatewayForCandidates(t, []string{"alpha", "beta"}, failoverConfig{enabled: true, maxAttempts: 3}) + // Force alpha's discovery breaker open directly. + st := g.registry.states["alpha"] + st.mu.Lock() + st.health = breakerOpen + st.openUntil = time.Now().Add(time.Hour) + st.mu.Unlock() + require.False(t, g.registry.discoveryHealthy("alpha")) + + primary := resolveCandidate{providerName: "alpha"} + extra := []resolveCandidate{{providerName: "beta"}} + got := g.orderedFailoverCandidates(primary, extra) + 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) { + 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)) +} + +// ===================================================================== +// End-to-end HTTP tests +// ===================================================================== + +// twoProviderFailoverConfig builds a Config with two openai-type providers, +// "alpha" and "beta" (alpha sorted first — the bareWinner/primary +// candidate), both configured with the identical bare model id "shared", +// pointed at srvAlpha/srvBeta respectively, one group with no +// restrictions, and one user "alice" with unlimited limits. +func twoProviderFailoverConfig(srvAlpha, srvBeta *httptest.Server) *Config { + cfg := CreateConfig() + cfg.Providers = map[string]*ProviderConfig{ + "alpha": {Type: "openai", BaseURL: srvAlpha.URL, APIKey: "k", Models: []string{"shared"}}, + "beta": {Type: "openai", BaseURL: srvBeta.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{}}, + }} + return cfg +} + +const successRespBody = `{"id":"chatcmpl-ok","object":"chat.completion","model":"shared","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":2}}` + +func jsonServer(status int, body string) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write([]byte(body)) + })) +} + +func countingServer(status int, body string, calls *int64) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt64(calls, 1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write([]byte(body)) + })) +} + +// TestHandleChat_Failover_FallsThroughAndServes is table-driven across +// every failure shape the brief names as a fallthrough case (5xx, +// connection failure, 429, 401, 403, 404) — proving alpha's failure is +// classified eligible AND that beta's real response reaches the client +// each time. Would FAIL if failoverEligible or the candidate loop's +// hasNext wiring were reverted (alpha's error would just be returned to +// the client instead of falling through). +func TestHandleChat_Failover_FallsThroughAndServes(t *testing.T) { + tests := []struct { + name string + alphaStatus int + unreachable bool + }{ + {name: "5xx", alphaStatus: http.StatusInternalServerError}, + {name: "connection_failure", unreachable: true}, + {name: "429", alphaStatus: http.StatusTooManyRequests}, + {name: "401", alphaStatus: http.StatusUnauthorized}, + {name: "403", alphaStatus: http.StatusForbidden}, + {name: "404", alphaStatus: http.StatusNotFound}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var alphaURL string + if tt.unreachable { + // A closed listener: connections are refused immediately, + // deterministic and fast, unlike a black-holed address. + dead := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + alphaURL = dead.URL + dead.Close() + } else { + alphaSrv := jsonServer(tt.alphaStatus, `{"error":{"message":"alpha down","type":"server_error"}}`) + defer alphaSrv.Close() + alphaURL = alphaSrv.URL + } + betaSrv := jsonServer(http.StatusOK, successRespBody) + defer betaSrv.Close() + + cfg := CreateConfig() + cfg.Providers = map[string]*ProviderConfig{ + "alpha": {Type: "openai", BaseURL: alphaURL, 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{}}}} + + 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": "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) + logOutput := captureStderrStop(origStderr) + + require.Equal(t, http.StatusOK, rec.Code, "must fall through to beta and serve its response, body=%s", rec.Body.String()) + assert.Equal(t, successRespBody, rec.Body.String()) + + if tt.alphaStatus == http.StatusNotFound { + assert.Contains(t, logOutput, "404", "a 404 failover must log loudly") + assert.Contains(t, logOutput, "alpha") + assert.Contains(t, logOutput, "beta") + } + }) + } +} + +// TestHandleChat_Failover_400_DoesNotFallThrough is the regression test +// for the brief's one hard exception: a malformed request must never +// retry against a second provider. Would FAIL if failoverEligible treated +// 400 the same as every other status. +func TestHandleChat_Failover_400_DoesNotFallThrough(t *testing.T) { + var betaCalls int64 + alphaSrv := jsonServer(http.StatusBadRequest, `{"error":{"message":"bad request","type":"invalid_request_error"}}`) + 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{}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusBadRequest, rec.Code, "a 400 must be returned as-is, never retried") + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls), "beta must never be called after alpha's 400") +} + +// TestHandleChat_Failover_MidStream_DoesNotFailOver_NoCorruption is the +// regression test for THE HARD CONSTRAINT: once anything has reached the +// client, failover must never happen and the response must never be +// corrupted with a second envelope. Would FAIL if the loop checked +// failoverEligible before sw.wroteHeader. +func TestHandleChat_Failover_MidStream_DoesNotFailOver_NoCorruption(t *testing.T) { + const firstChunk = "data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"delta\":{\"content\":\"Hi\"}}]}\n\n" + + var betaCalls int64 + alphaSrv := 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) + _, _ = w.Write([]byte(firstChunk)) + fl.Flush() + hj, ok := w.(http.Hijacker) + require.True(t, ok) + conn, _, hjErr := hj.Hijack() + if hjErr == nil { + _ = conn.Close() + } + })) + 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", "stream": true, "messages": []any{}} + 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, "headers already committed before the drop") + assert.Equal(t, firstChunk, rec.Body.String(), "no trailing envelope, no failover content appended") + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls), "beta must never be called once alpha has started writing") +} + +// TestHandleChat_Failover_RequestUnhealthy_SkippedWithoutBeingCalled is the +// end-to-end regression test for the request-path health hard-skip gate: +// after enough consecutive failures, alpha must be excluded from routing +// entirely — its upstream server must receive no further hits — while beta +// still serves. +func TestHandleChat_Failover_RequestUnhealthy_SkippedWithoutBeingCalled(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() + + 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)) + + // alpha's request-path breaker must now be open — one more request + // must skip it entirely. + 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 NOT receive a further hit once request-unhealthy") + assert.Equal(t, int64(requestBreakerFailureThreshold+1), atomic.LoadInt64(&betaCalls)) +} + +// 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. +func TestHandleChat_Failover_DiscoveryUnhealthy_StillTried(t *testing.T) { + alphaSrv := jsonServer(http.StatusOK, successRespBody) // alpha's CHAT endpoint works fine + defer alphaSrv.Close() + betaSrv := jsonServer(http.StatusInternalServerError, `{"error":{"message":"down","type":"server_error"}}`) + 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) + + // Force alpha's DISCOVERY breaker open directly (feat/provider-health's + // own signal) — its chat-completions endpoint is untouched and stays + // healthy; only discovery is marked broken, per discoveryHealthy's own + // documented scope. + st := gw.registry.states["alpha"] + st.mu.Lock() + st.health = breakerOpen + st.openUntil = time.Now().Add(time.Hour) + 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. + 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, "alpha must still be tried and must serve, body=%s", rec.Body.String()) + assert.Equal(t, successRespBody, rec.Body.String()) +} + +// TestHandleChat_Failover_TenantRequestCount_IncrementsOnceAcrossFailover +// is the regression test for "never double-count the request": a +// tenant's own req/min counter must move by exactly 1 across a whole +// failover chain, never once per attempt. +func TestHandleChat_Failover_TenantRequestCount_IncrementsOnceAcrossFailover(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 := 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) + + 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) + + now := time.Now() + reqCount, ok := gw.limiter.getCounter("user", "alice", metricReq, windowMin, now) + require.True(t, ok) + assert.Equal(t, int64(1), reqCount, "one client-facing request must count as exactly one, regardless of how many providers it took") + + totalReq, ok := gw.limiter.getCounter(totalScopeKind, totalScopeID, metricReq, windowMin, now) + require.True(t, ok) + assert.Equal(t, int64(1), totalReq) +} + +// TestHandleChat_Failover_PerProviderAttemptCounters_AttributeCorrectly is +// the regression test for "rebind the attempt recorder per attempt": alpha +// must show exactly one attempt/one failure, and beta exactly one +// attempt/zero failures — never both attributed to alpha (the closure's +// original, fixed-at-construction bug this feature explicitly fixes). +func TestHandleChat_Failover_PerProviderAttemptCounters_AttributeCorrectly(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 := 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) + // Deterministic: run the fire-and-forget counter write synchronously + // (mirrors TestHandleChat_ContextDeadlineExceeded_RecordsProviderFailure). + gw.limiter.spawn = func(f func()) { f() } + + 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) + + now := time.Now() + alphaAttempts, _ := gw.limiter.getCounter(kindProvider, "alpha", metricProvAttempt, windowDay, now) + alphaFails, _ := gw.limiter.getCounter(kindProvider, "alpha", metricProvFail, windowDay, now) + betaAttempts, _ := gw.limiter.getCounter(kindProvider, "beta", metricProvAttempt, windowDay, now) + betaFails, _ := gw.limiter.getCounter(kindProvider, "beta", metricProvFail, windowDay, now) + + assert.Equal(t, int64(1), alphaAttempts, "alpha attempts") + assert.Equal(t, int64(1), alphaFails, "alpha fails — a reverted 'rebind per attempt' bug would misattribute this to whichever provider the closure was first bound to") + assert.Equal(t, int64(1), betaAttempts, "beta attempts") + assert.Equal(t, int64(0), betaFails, "beta fails") +} + +// TestHandleChat_Failover_CachedUnderWinningProviderKey is the regression +// test for cache-key correctness: a response produced by the FAILOVER +// provider (beta) must be retrievable via a cache key built with BETA's +// own name, not alpha's — would FAIL if cacheKeyStr were still computed +// once, before the loop, from the primary candidate alone. +func TestHandleChat_Failover_CachedUnderWinningProviderKey(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() + + 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) + gw, ok := h.(*Gateway) + require.True(t, ok) + + reqBody := map[string]any{"model": "shared", "messages": []any{map[string]any{"role": "user", "content": "hi"}}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", reqBody) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + require.Equal(t, http.StatusOK, rec.Code) + require.Equal(t, "miss", rec.Header().Get("X-Llmgw-Cache")) + require.Equal(t, int64(1), atomic.LoadInt64(&betaCalls)) + + // Reconstruct the exact key runMeteredCall would compute for BETA's + // own candidate — upstreamModel is "shared" (the bare id every + // candidate shares in this fixture), requestedModel is the client's + // own "shared" string, endpoint is cacheEndpointChat, and req is the + // decoded body with "model" rewritten (exactly what the gateway + // hashes) — see cacheKey's own doc comment, cache.go. + hashed := map[string]any{"model": "shared", "messages": reqBody["messages"]} + wantKey := cacheKey("beta", "shared", "shared", cacheEndpointChat, hashed) + cached, hit := gw.cache.lookup(wantKey) + require.True(t, hit, "the response must be stored under beta's own cache key") + assert.Contains(t, string(cached.Body), `"id":"chatcmpl-ok"`) + + // A second, identical request must now be served from cache without + // beta's real backend being hit again. + 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, int64(1), atomic.LoadInt64(&betaCalls), "the second request must hit cache under beta's key, not call beta's backend again") +} + +// 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. +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() + + cfg := CreateConfig() + cfg.Providers = map[string]*ProviderConfig{ + "openai": {Type: "openai", BaseURL: srv.URL, APIKey: "k", Models: []string{"gpt-test"}}, + } + cfg.Groups = map[string]*GroupConfig{"default": {}} + cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice"}}} + 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": "gpt-test", "messages": []any{}} + req := newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusTooManyRequests, rec.Code, "a single-provider deployment must still surface the upstream error unchanged — there is nothing to fail over to") + var out struct { + Error struct { + Type string `json:"type"` + } `json:"error"` + } + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &out)) + assert.Equal(t, "upstream_error", out.Error.Type) +} + +// TestHandleChat_Failover_ExplicitlyDisabled_5xxNeverFallsThrough proves +// the on/off switch actually gates the behavior end to end. +func TestHandleChat_Failover_ExplicitlyDisabled_5xxNeverFallsThrough(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 := twoProviderFailoverConfig(alphaSrv, betaSrv) + no := false + cfg.Failover = FailoverConfig{Enabled: &no} + 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, "with failover disabled, alpha's own 500 must be returned as-is, wrapped in the upstream-error envelope") + assert.Equal(t, int64(0), atomic.LoadInt64(&betaCalls)) +} + +// TestHandleChat_Failover_ContextDeadlineExceeded_ClassifiedEligible_ +// AndRecordsHealthFailure proves timeout classification end to end +// through the SAME mechanism recordProviderAttempt already uses +// (isDeadlineExceeded) — see this test file's package-level note on why a +// full "alpha times out, beta still serves" HTTP scenario cannot be +// demonstrated: every candidate in one logical request shares ONE +// request context, so once that context's own deadline is what makes +// alpha's failure a genuine timeout, zero budget remains for any +// subsequent candidate. This proves the half that IS achievable and +// correct: alpha's attempt is recorded as a request-health failure +// exactly like a plain connection error would be. +func TestHandleChat_Failover_ContextDeadlineExceeded_RecordsRequestHealthFailure(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(300 * time.Millisecond) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(successRespBody)) + })) + defer srv.Close() + + cfg := CreateConfig() + cfg.Providers = map[string]*ProviderConfig{ + "alpha": {Type: "openai", BaseURL: srv.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{}}}} + 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) + + 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") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + assert.True(t, gw.failoverHealth.healthy("alpha"), "one deadline-exceeded attempt alone must not yet open the breaker (below threshold)") + + st := gw.failoverHealth.stateFor("alpha") + st.mu.Lock() + failures := st.consecutiveFailures + st.mu.Unlock() + assert.Equal(t, 1, failures, "the deadline-exceeded attempt must be classified as a request-health failure, not silently ignored") +} + +// capturedStderr holds the state captureStderrStart needs to hand back to +// captureStderrStop — the same os.Pipe swap +// TestHandleChat_MidStreamUpstreamDrop_NoTrailingEnvelope already uses in +// routes_unified_test.go, factored out here so the table-driven fallthrough +// test above can capture the 404 loud-log assertion per sub-test. +type capturedStderr struct { + orig *os.File + pr *os.File + pw *os.File +} + +// captureStderrStart redirects os.Stderr to an in-memory pipe. Callers +// must not run this concurrently with another goroutine writing to +// os.Stderr in the SAME process (the same constraint the existing +// mid-stream-drop test already accepts) — every caller in this file runs +// its captured request synchronously, sequentially, never inside +// t.Parallel(). +func captureStderrStart() capturedStderr { + orig := os.Stderr + pr, pw, err := os.Pipe() + if err != nil { + panic(err) // os.Pipe failing is not a condition any test here can meaningfully recover from + } + os.Stderr = pw + return capturedStderr{orig: orig, pr: pr, pw: pw} +} + +// captureStderrStop restores os.Stderr and returns everything written to +// it since the matching captureStderrStart call. +func captureStderrStop(c capturedStderr) string { + os.Stderr = c.orig + _ = c.pw.Close() + out, _ := io.ReadAll(c.pr) + return string(out) +} diff --git a/llmgateway.go b/llmgateway.go index ed15ff0..6cd63ac 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -70,7 +70,13 @@ type Config struct { // at all) resolves to every documented default and behaves exactly // like a deployment with no circuit breaker at all for a provider // that never fails — see validateBreakerConfig's own doc comment. - Breaker BreakerConfig `json:"breaker"` + // Failover configures feat/failover — cross-provider failover for a + // bare model id more than one configured provider serves. See + // FailoverConfig's own doc comment (failover.go) for the full + // contract; a struct value, not a pointer, for the same + // zero-means-default reason as Breaker/Retry/Cache below. + Failover FailoverConfig `json:"failover"` + Breaker BreakerConfig `json:"breaker"` // Retry is a struct value, not a pointer, because its own Enabled // field is the on/off signal (unlike Redis/Users, where the block's // mere presence is the signal) — so its tag omits "omitempty": @@ -493,18 +499,24 @@ func defaultBodyAdmissionCap() int { const maxExplicitBodyAdmissionCap = 10_000 // Gateway is the Traefik middleware handler. +// +// Field order below is fieldalignment-verified (golangci-lint's govet +// enable-all, run with -fix against a scratch copy to derive the exact +// zero-waste sequence, then hand-applied here so every field keeps its +// original doc comment) — see providerState's own doc comment +// (registry.go) for the general convention. type Gateway struct { - next http.Handler - cfg *Config - auth *authStore - limiter *limiter - registry *modelRegistry - adapters map[string]providerAdapter + next http.Handler // targetClient is the shared, connection-pooled *http.Client the // MCP/A2A target proxy (mcp_a2a.go) issues every upstream request // through — built once via newAdapterHTTPClient, the same constructor // each provider adapter uses for its own client. targetClient *http.Client + auth *authStore + limiter *limiter + registry *modelRegistry + adapters map[string]providerAdapter + cfg *Config // cache is nil whenever response caching is not configured or not // usable (cfg.Cache.Enabled is false, or true with no config.Redis — // see buildResponseCache, cache.go). Every call site checks for nil @@ -525,7 +537,17 @@ type Gateway struct { // resized afterward, matching a Go channel's own fixed-capacity // contract. bodyAdmission chan struct{} - name string + // failoverHealth is feat/failover's own per-pod, in-memory + // request-path health signal (failover.go), built once, here, by + // newGateway. Every method on it is nil-receiver-safe, so a Gateway + // assembled directly (bypassing newGateway, as a few older tests do) + // degrades to "failover never skips a candidate for request health" + // rather than a nil-pointer panic. + failoverHealth *requestHealthTracker + name string + // failover is Config.Failover, validated and resolved once by + // newGateway (validateFailoverConfig, failover.go). + failover failoverConfig } // telemetryStartupOnce keeps the anonymous "plugin loaded" ping to one per @@ -634,6 +656,16 @@ 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. + failoverCfg, err := validateFailoverConfig(config.Failover) + if err != nil { + return nil, err + } + g.failover = failoverCfg + g.failoverHealth = newRequestHealthTracker() + // bodyAdmissionCap: an explicit MaxInFlightBodyRequests always wins // over the self-tuned default (house rule: explicit override always // wins) — see Config.MaxInFlightBodyRequests' own doc comment. 0 (the diff --git a/routes_unified.go b/routes_unified.go index 143404e..655af16 100644 --- a/routes_unified.go +++ b/routes_unified.go @@ -256,123 +256,223 @@ func (g *Gateway) runUnified(w http.ResponseWriter, r *http.Request, u *user, gr // envelope shape they answer in. logPrefix names the route in every log // line this function writes, so a log reader can always tell which route // produced a given line without the classification logic existing twice. +// +// feat/failover: this function now tries a LIST of candidate providers +// (g.registry.resolveWithCandidates/g.orderedFailoverCandidates, +// failover.go), not just the one resolve alone would pick, and is the +// ONE hook point both /v1/chat/completions (runUnified, above) and +// /v1/messages (handleMessages, routes_messages.go) share — per the +// brief's own instruction, failover is wired here once rather than +// forked into a second implementation. THE HARD CONSTRAINT this loop is +// built around: call(...) writes directly to respWriter, which tees +// straight through to sw; the very first byte written commits the +// response, so a candidate is only ever retried while sw.wroteHeader is +// still false. Every step that was a one-shot computation before this +// feature (the cache key, the attempt recorder's bound provider name, +// the usage/canonical id billed) now happens fresh INSIDE the loop, once +// per candidate — this is what "rebind the attempt recorder per +// 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) { - adapter, upstreamModel, canonical, err := g.registry.resolve(requestedModel, grp) + primary, extra, err := g.registry.resolveWithCandidates(requestedModel, grp) if err != nil { writeModelResolveErrorEnvelope(sw, err, envelope) return } + candidates := g.orderedFailoverCandidates(primary, extra) - req["model"] = upstreamModel - - // cacheable gates every cache step below on spec §2's scope: a - // non-streaming request, with a working cache (g.cache is nil - // whenever caching is off or Redis was absent — buildResponseCache, - // cache.go), for a group that has not opted out (groupCacheEnabled). - // cacheKey is computed here — after "model" is rewritten to - // upstreamModel but deliberately BEFORE gatewayAliasKey is injected - // below — so the hashed request body reflects exactly what goes - // upstream. requestedModel (the client's own, un-rewritten alias - // string) and endpoint are passed as separate cacheKey arguments, - // not left in req: a translating adapter bakes the echoed alias - // straight into the cached response body's own "model" field, so two - // clients requesting the same upstream model under different alias - // forms (e.g. "claude-x" vs "anthropic/claude-x") must never collide - // into one cache entry (cacheKey's own doc comment, cache.go). + // cacheable/streaming gate every cache step below on spec §2's scope, + // exactly as before this feature: a non-streaming request, with a + // working cache (g.cache is nil whenever caching is off or Redis was + // absent — buildResponseCache, cache.go), for a group that has not + // opted out (groupCacheEnabled). Computed once, outside the loop: it + // depends only on req["stream"]/grp, neither of which a failover + // attempt changes. streaming, _ := req["stream"].(bool) cacheable := !streaming && g.cache != nil && groupCacheEnabled(grp) - var cacheKeyStr string - if cacheable { - cacheKeyStr = cacheKey(adapter.name(), 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. - sw.Header().Set("X-Llmgw-Cache", "hit") - if cached.ContentType != "" { - sw.Header().Set("Content-Type", cached.ContentType) + + 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) + } + + // 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. + sw.Header().Set("X-Llmgw-Cache", "hit") + if cached.ContentType != "" { + sw.Header().Set("Content-Type", cached.ContentType) + } + sw.WriteHeader(cached.Status) + _, _ = sw.Write(cached.Body) + return } - sw.WriteHeader(cached.Status) - _, _ = sw.Write(cached.Body) + // 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 + // is never observable until SOME candidate actually writes, + // by which point the loop has already committed to that + // candidate's outcome (THE HARD CONSTRAINT, this function's + // own doc comment). + 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 + // call() returns, without any adapter knowing caching exists + // (adapterCall's contract is unchanged either way). A fresh + // capture per candidate: reusing one across attempts would let a + // candidate that never wrote anything (the common failover case) + // leave a stale buffer behind — capture.status stays 0 either + // way, so this is defensive clarity, not a proven bug fix. + var respWriter http.ResponseWriter = sw + var capture *cacheCaptureWriter + if cacheable { + capture = newCacheCaptureWriter(sw, g.cache.maxBodyBytes) + respWriter = capture + } + + // Feature A (v0.22) attempt recorder, REBOUND every iteration to + // THIS candidate's own provider name (feat/failover's own "never + // 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. + 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) + + // Usage is accounted before the error branch below runs, not + // after: every adapter that can fail mid-stream (forwardStream + // in each of the three provider files) still returns whatever + // usage it had already captured alongside the error — a usage + // chunk that arrived just before a dropped connection must + // still be billed, or a client that aborts right after that + // frame arrives could repeat the trick to dodge every + // token/cost budget. A non-streaming failure (providerHTTPError, + // translateError, or a connection failure before any write) + // always carries zero usage by contract, so accounting it here + // is a no-op — account skips every write once both total tokens + // and cost are zero. routes_messages.go's + // callTranslatedMessages/callAnthropicMessagesPassthrough follow + // the identical contract for their own error paths (item 1 fix, + // 2026-08-22 review). This still runs exactly once per + // CANDIDATE, including a candidate that goes on to fail over — + // its own (zero, by contract) usage is billed as a no-op, never + // skipped, so a later successful candidate's real usage is never + // silently doubled with a phantom first entry either. + if callErr == nil && result.total() == 0 { + if streaming { + g.logf("%s: zero usage reported for a streaming response from model %q; accounting the request only", logPrefix, cand.canonical) + } else { + result.prompt = int64(math.Ceil(float64(len(body)) / 4)) + result.estimated = true + } + } + g.limiter.account(scopes, result, unifiedCostMicros(cand.canonical, cand.upstreamModel, result, g.cfg.Pricing)) + if result.estimated { + g.logf("%s: usage for model %q logged as estimated (%d prompt tokens derived from request body size, not the provider's reported usage)", logPrefix, cand.canonical, result.prompt) + } + + // A miss stores the response after everything above has already + // run — accounting must never be skipped or delayed waiting on a + // cache write. Only a genuine upstream 200 is stored (spec §2's + // "on 200 non-stream, tee and SET"): a non-2xx status never + // reaches here as capture.status (forwardJSON/translate error + // paths return *providerHTTPError/*translateError instead of + // writing through respWriter, so callErr is non-nil and + // capture.status stays 0), and callErr == nil is checked + // directly regardless. capture.oversize excludes a response + // cacheCaptureWriter stopped buffering past maxBodyBytes: + // store's own maxBodyBytes check would reject it too, but only + // after being handed a silently truncated body — skip the call + // outright instead of ever constructing a corrupt cache entry. + if cacheable && callErr == nil && capture.status == http.StatusOK && !capture.oversize { + g.cache.store(cacheKeyStr, capture.status, capture.contentType, capture.buf.Bytes(), effectiveTTL(g.cache, grp)) + } + + if callErr == nil { return } - // Set before call() below writes anything — headers must precede - // the body a miss is about to produce (spec §2). - sw.Header().Set("X-Llmgw-Cache", "miss") - } - 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 call() returns, without - // any adapter knowing caching exists (adapterCall's contract is - // unchanged either way). - var respWriter http.ResponseWriter = sw - var capture *cacheCaptureWriter - if cacheable { - capture = newCacheCaptureWriter(sw, g.cache.maxBodyBytes) - respWriter = capture - } - - // Feature A (v0.22): every upstream attempt call() makes — via - // upstreamJSON's retryPolicy.do, providers.go — reports through to - // this recorder, closing over the adapter/model this request already - // resolved to. See attemptRecorder's own doc comment (providers.go). - ctx := withAttemptRecorder(r.Context(), func(resp *http.Response, attemptErr error) { - g.limiter.recordProviderAttempt(adapter.name(), upstreamModel, resp, attemptErr) - }) - result, callErr := call(adapter, ctx, respWriter, req) - - // Usage is accounted before the error branch below runs, not after: - // every adapter that can fail mid-stream (forwardStream in each of the - // three provider files) still returns whatever usage it had already - // captured alongside the error — a usage chunk that arrived just before - // a dropped connection must still be billed, or a client that aborts - // right after that frame arrives could repeat the trick to dodge every - // token/cost budget. A non-streaming failure (providerHTTPError, - // translateError, or a connection failure before any write) always - // carries zero usage by contract, so accounting it here is a no-op — - // account skips every write once both total tokens and cost are zero. - // routes_messages.go's callTranslatedMessages/ - // callAnthropicMessagesPassthrough follow the identical contract for - // their own error paths (item 1 fix, 2026-08-22 review: a translation - // failure AFTER a successful, billed upstream call must still return - // that call's real usage, not usage{}). - if callErr == nil && result.total() == 0 { - if streaming { - g.logf("%s: zero usage reported for a streaming response from model %q; accounting the request only", logPrefix, canonical) - } else { - result.prompt = int64(math.Ceil(float64(len(body)) / 4)) - result.estimated = true + // THE HARD CONSTRAINT: once sw has written anything, status and + // body are committed and can never be retried elsewhere — a + // mid-stream failure (headers, or a partial SSE body, already + // reached the client) always falls straight through to the + // existing, unchanged error classification, exactly as it did + // before this feature existed, regardless of how many candidates + // remain. + if sw.wroteHeader { + g.handleAdapterErrorEnvelope(sw, callErr, providerName, logPrefix, envelope, writeUpstream) + return } - } - g.limiter.account(scopes, result, unifiedCostMicros(canonical, upstreamModel, result, g.cfg.Pricing)) - if result.estimated { - g.logf("%s: usage for model %q logged as estimated (%d prompt tokens derived from request body size, not the provider's reported usage)", logPrefix, canonical, result.prompt) - } - // A miss stores the response after everything above has already run — - // accounting must never be skipped or delayed waiting on a cache - // write. Only a genuine upstream 200 is stored (spec §2's "on 200 - // non-stream, tee and SET"): a non-2xx status never reaches here as - // capture.status (forwardJSON/translate error paths return - // *providerHTTPError/*translateError instead of writing through - // respWriter, so callErr is non-nil and capture.status stays 0), and - // callErr == nil is checked directly regardless. capture.oversize - // excludes a response cacheCaptureWriter stopped buffering past - // maxBodyBytes: store's own maxBodyBytes check would reject it too, - // but only after being handed a silently truncated body — skip the - // call outright instead of ever constructing a corrupt cache entry. - if cacheable && callErr == nil && capture.status == http.StatusOK && !capture.oversize { - g.cache.store(cacheKeyStr, capture.status, capture.contentType, capture.buf.Bytes(), effectiveTTL(g.cache, grp)) - } - - if callErr != nil { - g.handleAdapterErrorEnvelope(sw, callErr, adapter.name(), logPrefix, envelope, writeUpstream) - return + hasNext := i < len(candidates)-1 && failoverEligible(callErr) + if hasNext && isProviderNotFoundError(callErr) { + // Loud, non-suppressed log line (brief: "Failover after a 404 + // must log loudly ... a 404 usually means a real + // configuration mistake, and silently succeeding elsewhere + // 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. + 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 { + g.handleAdapterErrorEnvelope(sw, callErr, providerName, logPrefix, envelope, writeUpstream) + return + } + // Falls through to the next candidate. } } diff --git a/tools/yaegi-check/main.go b/tools/yaegi-check/main.go index af663a2..f9b5efd 100644 --- a/tools/yaegi-check/main.go +++ b/tools/yaegi-check/main.go @@ -346,6 +346,31 @@ func run() error { })) defer brkUpstream.Close() + // failoverAUpstream/failoverBUpstream back exerciseFailover (feat/ + // failover, below): two providers configured with the IDENTICAL bare + // model id (failoverModelID) — the exact shape registry.go's bareWinner + // collision handles, and the shape production logs show for whisper-1/ + // tts-1/tts-1-hd/the embeddings models across openai/openai-audio/ + // copilot. A always answers 500; B always answers 200. This proves the + // failover candidate loop (runMeteredCall, routes_unified.go) falls + // through and serves correctly under the REAL interpreter, not merely + // compiled — this package's own doc comment is exactly why that + // distinction matters on this codebase. + failoverAUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(`{"error":{"message":"failover-a is down","type":"server_error"}}`)) + })) + defer failoverAUpstream.Close() + failoverBHits := new(int64) + failoverBUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt64(failoverBHits, 1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"id":"c-failover","object":"chat.completion","model":"` + failoverModelID + `","choices":[{"index":0,"message":{"role":"assistant","content":"served by failover-b"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`)) + })) + defer failoverBUpstream.Close() + attemptAccountingOverride := `{"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"},"` + @@ -354,7 +379,9 @@ func run() error { // (exerciseMessagesRoute, below): a real anthropic-type provider, // interpreted end to end through // callAnthropicMessagesPassthrough. - `"anthropic":{"type":"anthropic","baseUrl":"` + anthropicProbeUpstream.URL + `","apiKey":"sk-anth","models":["claude-test"]}},` + // #nosec G101 -- test fixture literal, not a real credential + `"anthropic":{"type":"anthropic","baseUrl":"` + anthropicProbeUpstream.URL + `","apiKey":"sk-anth","models":["claude-test"]},` + // #nosec G101 -- test fixture literal, not a real credential + `"failover-a":{"type":"openai","baseUrl":"` + failoverAUpstream.URL + `","apiKey":"sk-up","models":["` + failoverModelID + `"]},` + + `"failover-b":{"type":"openai","baseUrl":"` + failoverBUpstream.URL + `","apiKey":"sk-up","models":["` + failoverModelID + `"]}},` + // modelMeta (feature v0.23): a config-override entry for // testDataWantModel, so exerciseHandler's GET /v1/models // assertion below proves resolveModelMeta's config-override @@ -405,7 +432,7 @@ func run() error { return fmt.Errorf("%s.New's returned value does not implement http.Handler", pkgName) } - if err := exerciseHandler(handler, builtinLookupContextTokens); err != nil { + if err := exerciseHandler(handler, builtinLookupContextTokens, failoverBHits); err != nil { return err } // exerciseBreaker (feat/provider-health, adversarial-review round 2): @@ -429,6 +456,11 @@ func run() error { // the config-override path testDataWantModel already exercises. const builtinLookupModelID = "gpt-4o" +// failoverModelID is the bare model id both "failover-a" and "failover-b" +// (run(), above) are configured with — the run()-owned upstream servers +// exerciseFailover (below) drives. +const failoverModelID = "failover-shared" + // mcpProbeServerName is the federated MCP server run() configures against // mcpProbeUpstream, and mcpProbeToolName is a tool id carrying its // "_" prefix so resolveFederatedTool routes POST /mcp's @@ -634,8 +666,8 @@ func exerciseBreaker(handler http.Handler, hits, healthy *int64) error { // for real under Yaegi, not merely that the package's imports resolve. // builtinContextTokens is builtinLookupModelID's own real, extracted // builtinModelMetaTable value (SHOULD-6) — see that constant's own doc -// comment. -func exerciseHandler(handler http.Handler, builtinContextTokens int) error { +// comment. failoverBHits backs exerciseFailover's own call, below. +func exerciseHandler(handler http.Handler, builtinContextTokens int, failoverBHits *int64) error { authedReq := httptest.NewRequest(http.MethodGet, "/v1/models", nil) authedReq.Header.Set("Authorization", "Bearer "+testDataUserAPIKey) authedRec := httptest.NewRecorder() @@ -702,6 +734,10 @@ func exerciseHandler(handler http.Handler, builtinContextTokens int) error { return err } + if err := exerciseFailover(handler, failoverBHits); err != nil { + return err + } + if err := exerciseFederatedTooLarge(handler); err != nil { return err } @@ -1013,6 +1049,38 @@ func exerciseFederatedTooLarge(handler http.Handler) error { return nil } +// exerciseFailover drives one real POST /v1/chat/completions against the +// interpreted handler for failoverModelID — a bare model id both +// "failover-a" (always 500s) and "failover-b" (always 200s) are +// configured with — proving the failover candidate loop (runMeteredCall, +// routes_unified.go/failover.go) falls through a genuine provider +// failure and serves the second provider's response under the REAL +// interpreter. Compiled tests already prove this (failover_test.go); this +// harness exists because compiled tests have repeatedly passed on this +// codebase while the interpreted shape failed (this package's own doc +// comment). +func exerciseFailover(handler http.Handler, bHits *int64) error { + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader( + `{"model":"`+failoverModelID+`","messages":[{"role":"user","content":"hi"}]}`, + )) + req.Header.Set("Authorization", "Bearer "+testDataUserAPIKey) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + return fmt.Errorf("POST /v1/chat/completions (feat/failover harness): status = %d, want 200 (failover-a's 500 must fall through to failover-b), body=%s", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "served by failover-b") { + return fmt.Errorf("POST /v1/chat/completions (feat/failover harness): body does not contain failover-b's own content, want the SECOND provider's response: %s", rec.Body.String()) + } + if atomic.LoadInt64(bHits) != 1 { + return fmt.Errorf("feat/failover harness: failover-b upstream hit count = %d, want exactly 1", atomic.LoadInt64(bHits)) + } + fmt.Println("yaegi-check: failover fell through from failover-a to failover-b and served its response") + return nil +} + func exerciseAttemptAccounting(handler http.Handler) error { chatReq := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader( `{"model":"`+testDataWantModel+`","messages":[{"role":"user","content":"hi"}]}`,