diff --git a/llmgateway.go b/llmgateway.go index 91f513d..3a67eb5 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -18,19 +18,23 @@ import ( // Config is the plugin's dynamic configuration, populated by Traefik from // the middleware's YAML/testData. Every limit or list field follows the // convention: zero/empty/omitted means unlimited/all. +// 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 pointer-first/scalar-last convention this follows. type Config struct { - Providers map[string]*ProviderConfig `json:"providers,omitempty"` - Groups map[string]*GroupConfig `json:"groups,omitempty"` - Pricing map[string]*ModelPricing `json:"pricing,omitempty"` - MCPServers map[string]*TargetConfig `json:"mcpServers,omitempty"` - Agents map[string]*AgentConfig `json:"agents,omitempty"` - Users *UsersConfig `json:"users,omitempty"` - Redis *RedisConfig `json:"redis,omitempty"` + Redis *RedisConfig `json:"redis,omitempty"` // Admin gates the read-only admin dashboard (spec §4, v0.2): nil or // Admin.Enabled false means the /admin* routes are not registered at // all — ServeHTTP falls through to its existing 404/passthroughUnknown // handling for those paths, preserving v0.1 behavior exactly. - Admin *AdminConfig `json:"admin,omitempty"` + Admin *AdminConfig `json:"admin,omitempty"` + Pricing map[string]*ModelPricing `json:"pricing,omitempty"` + MCPServers map[string]*TargetConfig `json:"mcpServers,omitempty"` + Agents map[string]*AgentConfig `json:"agents,omitempty"` + Users *UsersConfig `json:"users,omitempty"` + Groups map[string]*GroupConfig `json:"groups,omitempty"` // ModelAliases maps an operator-defined alias id to a target model id // (spec §5, v0.2) — e.g. {"aliased/coding": "anthropic/claude-sonnet-4-5"}. // An exact alias match wins resolution before any other rule @@ -39,7 +43,8 @@ type Config struct { // exactly: resolve never consults an empty alias map, so no existing // model id's resolution changes. Validated at construction // (validateModelAliases, registry.go). - ModelAliases map[string]string `json:"modelAliases,omitempty"` + ModelAliases map[string]string `json:"modelAliases,omitempty"` + Providers map[string]*ProviderConfig `json:"providers,omitempty"` // ModelMeta declares per-model metadata overrides (feature v0.23): // context window size and per-token cost, keyed by an exact // "provider/model" id or a bare model id (applying wherever that bare @@ -54,6 +59,18 @@ type Config struct { // discovery/builtin/absent, same as before this feature existed. // Validated at construction (validateModelMeta, modelmeta.go). ModelMeta map[string]*ModelMetaConfig `json:"modelMeta,omitempty"` + // Breaker configures the discovery circuit breaker (feat/provider- + // health): how many consecutive discovery-refresh failures open a + // provider's breaker, and how long it then backs off before probing + // again. A struct value, not a pointer, for the same reason as Retry/ + // Cache below — its fields are individually zero-means-default + // (validateBreakerConfig, registry.go), not gated by one Enabled + // flag, so "omitempty" on this tag would be a no-op that misleadingly + // implies otherwise. The zero value (Config{} with no breaker block + // 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"` // 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": @@ -63,8 +80,7 @@ type Config struct { Retry RetryConfig `json:"retry"` // Cache is a struct value, not a pointer, for the same reason as Retry // above: its own Enabled field is the on/off signal. - Cache CacheConfig `json:"cache"` - PassthroughUnknown bool `json:"passthroughUnknown,omitempty"` + Cache CacheConfig `json:"cache"` // MaxInFlightBodyRequests caps how many unified and media JSON/ // multipart requests (routes_unified.go's runUnified via // readAndDecodeUnifiedBody; routes_media.go's decodeMediaJSONRequest/ @@ -85,7 +101,8 @@ type Config struct { // maxExplicitBodyAdmissionCap (llmgateway.go): "wins" means it // overrides the self-tuned value, not that an operator can silently // disable the bound entirely with an unbounded number. - MaxInFlightBodyRequests int `json:"maxInFlightBodyRequests,omitempty"` + MaxInFlightBodyRequests int `json:"maxInFlightBodyRequests,omitempty"` + PassthroughUnknown bool `json:"passthroughUnknown,omitempty"` } // ProviderConfig describes one upstream LLM provider. @@ -278,6 +295,35 @@ type CacheConfig struct { Enabled bool `json:"enabled,omitempty"` } +// BreakerConfig configures the per-provider discovery circuit breaker +// (feat/provider-health): closed -> open -> half-open, driven entirely by +// discovery-refresh outcomes (registry.go's providerState.finishRefresh), +// never by request-path traffic. Every field is individually zero-means- +// default (validateBreakerConfig, registry.go) — there is no Enabled +// flag, unlike Retry/Cache above, because the breaker is always live: +// its defaults are chosen so a provider that never fails never notices +// it exists, so there is nothing meaningful to "disable". Field order +// (both strings before the int) is fieldalignment-sensitive, the same +// convention providerState's own doc comment explains. +type BreakerConfig struct { + // OpenDuration is the base backoff a newly opened breaker waits + // before its first half-open probe (a Go duration string, e.g. + // "1m"). Doubles on every further failed probe, capped at + // MaxOpenDuration. Empty (the default) uses + // defaultBreakerOpenDuration (registry.go). + OpenDuration string `json:"openDuration,omitempty"` + // MaxOpenDuration caps the exponential backoff OpenDuration above + // doubles into. Empty (the default) uses + // defaultBreakerMaxOpenDuration (registry.go). Must be >= + // OpenDuration when both are set. + MaxOpenDuration string `json:"maxOpenDuration,omitempty"` + // FailureThreshold is how many CONSECUTIVE failed discovery refreshes + // open the breaker. 0 (the default) uses defaultBreakerFailureThreshold + // (registry.go). A value outside 1..maxBreakerFailureThreshold is a + // construction error. + FailureThreshold int `json:"failureThreshold,omitempty"` +} + // AdminConfig configures the read-only admin dashboard (spec §4, v0.2). // The zero value (Enabled: false) disables it, preserving v0.1 behavior // exactly: no /admin* route is registered, so those paths fall through diff --git a/registry.go b/registry.go index fa0785f..5ffdf1c 100644 --- a/registry.go +++ b/registry.go @@ -28,6 +28,28 @@ const warmFillTimeout = 5 * time.Second // goroutine's lifetime if an upstream hangs. const backgroundRefreshTimeout = 30 * time.Second +// defaultBreakerFailureThreshold is how many CONSECUTIVE failed discovery +// refreshes open a provider's circuit breaker (feat/provider-health), +// applied when BreakerConfig.FailureThreshold is left at 0. Chosen so a +// single transient blip (one bad refresh) never trips it — only a +// provider that is persistently broken, like xiaomi's ongoing 401s, does. +const defaultBreakerFailureThreshold = 3 + +// maxBreakerFailureThreshold bounds BreakerConfig.FailureThreshold — a +// generous ceiling; there is no legitimate reason to require more +// consecutive failures than this before backing off at all. +const maxBreakerFailureThreshold = 100 + +// defaultBreakerOpenDuration is the base backoff a newly opened breaker +// waits before its first half-open probe, applied when BreakerConfig. +// OpenDuration is left empty. +const defaultBreakerOpenDuration = time.Minute + +// defaultBreakerMaxOpenDuration caps the exponential backoff the breaker +// doubles into on every further failed probe, applied when BreakerConfig. +// MaxOpenDuration is left empty. +const defaultBreakerMaxOpenDuration = 30 * time.Minute + // errModelUnknown is returned by modelRegistry.resolve when id does not // match any provider's known model set (explicit config plus the last // successful discovery fetch) — 404 semantics for the caller. @@ -39,15 +61,59 @@ var errModelUnknown = errors.New("llmgateway: unknown model") // such model" from "that model exists, you cannot use it". var errModelDenied = errors.New("llmgateway: model access denied") +// breakerState is one provider's discovery circuit breaker state +// (feat/provider-health): closed (normal — refreshes run on the +// configured interval, exactly like before this feature existed), open +// (discovery refreshes are skipped until openUntil, backing off +// exponentially on repeated failure), or halfOpen (exactly one probe +// refresh is in flight, deciding whether to close the breaker again). +// The zero value is closed, so a providerState built without ever +// touching health fields (every existing construction path, and every +// provider that never fails) behaves exactly as it did before this type +// existed. +type breakerState int32 + +const ( + breakerClosed breakerState = iota + breakerOpen + breakerHalfOpen +) + +// String renders bs for logs and the admin dashboard (feat/ +// provider-health) — lower-case, hyphenated for the two-word state, to +// read naturally as a status word in either place. +func (bs breakerState) String() string { + switch bs { + case breakerOpen: + return "open" + case breakerHalfOpen: + return "half-open" + default: + return "closed" + } +} + // providerState tracks one provider's known model ids and discovery // bookkeeping. explicit is set once at construction from // ProviderConfig.Models and never mutated afterward; discovered is // stale-while-error — a failed refresh leaves the previous discovered set // in place rather than clearing it. mu guards every mutable field below it. +// +// Field order below is fieldalignment-sensitive (golangci-lint's govet +// enable-all): every pointer-containing field (time.Time — its loc +// *Location field carries a pointer — plus every map/string) is grouped +// first, every pointer-free field (durations, sync.Mutex, the breaker's +// own int/int32 counters, and the two trailing bools) last. Keep new +// fields in the matching group rather than appending at the very end. type providerState struct { lastRefresh time.Time - explicit map[string]bool - discovered map[string]bool + // openUntil is when an open breaker (feat/provider-health) next + // allows a half-open probe — see recordHealthLocked/tryBeginRefresh. + // Grouped with lastRefresh above (both time.Time) rather than beside + // the rest of the breaker fields further down, which are pointer-free. + openUntil time.Time + explicit map[string]bool + discovered map[string]bool // discoveredContext holds discovery-captured per-model context // lengths (feature v0.23, ProviderConfig.MetadataPath — currently // only openai-type adapters ever populate this, via @@ -66,11 +132,31 @@ type providerState struct { // outcome, not the stale-while-error discovered set: a provider can // show a non-empty lastErr while modelCount still reflects its last // successful discovery. - lastErr string - interval time.Duration - mu sync.Mutex - discoveryEnabled bool - inFlight bool + lastErr string + interval time.Duration + mu sync.Mutex + // --- feat/provider-health: discovery circuit breaker, guarded by mu + // like every other mutable field above. consecutiveFailures counts + // unbroken failures while closed, toward breakerThreshold. backoff is + // the duration that produced openUntil above (doubled on the next + // failed probe, capped at breakerOpenMax). breakerThreshold/ + // breakerOpenBase/breakerOpenMax are resolved once, at construction + // (newModelRegistry, from Config.Breaker via validateBreakerConfig), + // and never change afterward — copied onto every providerState + // rather than shared from one place, the same "baked in at + // construction" convention interval above already uses, so + // recordHealthLocked/tryBeginRefresh never need a second parameter or + // a pointer back to the registry. health is the breaker's current + // state, declared last among these since breakerState is a 4-byte + // int32, smaller than its 8-byte neighbors above. + consecutiveFailures int + backoff time.Duration + breakerThreshold int + breakerOpenBase time.Duration + breakerOpenMax time.Duration + health breakerState + discoveryEnabled bool + inFlight bool } // hasModel reports whether id is in this provider's explicit or discovered @@ -132,14 +218,36 @@ func (st *providerState) setDiscoveredContext(m map[string]int) { st.discoveredContext = m } -// tryBeginRefresh reports whether now is far enough past lastRefresh (or -// this is the first refresh) to start a new one, and if so marks the -// provider inFlight so a concurrent caller cannot start a second one. The -// caller must pair a true result with a later finishRefresh call. +// tryBeginRefresh reports whether now permits starting a new refresh, and +// if so marks the provider inFlight so a concurrent caller cannot start a +// second one. The caller must pair a true result with a later finishRefresh +// call. +// +// A closed breaker (feat/provider-health) behaves EXACTLY as before this +// feature existed: now far enough past lastRefresh (or this is the first +// refresh) is the only gate — a provider that never fails never touches the +// breaker branch below at all. An open breaker replaces that interval gate +// with its own backoff window: refresh stays blocked until now reaches +// openUntil, regardless of interval, and reaching it starts exactly one +// half-open probe (state moves to halfOpen here, before the caller's fetch +// even runs) rather than resuming normal per-interval refreshing outright — +// recordHealthLocked decides whether that probe closes the breaker or +// reopens it with a longer backoff. func (st *providerState) tryBeginRefresh(now time.Time) bool { st.mu.Lock() defer st.mu.Unlock() - if st.inFlight || now.Sub(st.lastRefresh) < st.interval { + if st.inFlight { + return false + } + if st.health == breakerOpen { + if now.Before(st.openUntil) { + return false + } + st.health = breakerHalfOpen + st.inFlight = true + return true + } + if now.Sub(st.lastRefresh) < st.interval { return false } st.inFlight = true @@ -152,7 +260,10 @@ func (st *providerState) tryBeginRefresh(now time.Time) bool { // what throttles a failing provider to one attempt per interval instead of // retrying on every call. On success (err == nil) ids replaces the // discovered set; on failure the previous discovered set is kept -// (stale-while-error). +// (stale-while-error). Either way, recordHealthLocked (feat/provider-health) +// updates the discovery circuit breaker from this same outcome — a healthy +// provider's breaker never leaves closed, so this adds no new behavior for +// the common case, only for a provider that is actually failing. func (st *providerState) finishRefresh(now time.Time, ids []string, err error) { st.mu.Lock() defer st.mu.Unlock() @@ -160,14 +271,97 @@ func (st *providerState) finishRefresh(now time.Time, ids []string, err error) { st.lastRefresh = now if err != nil { st.lastErr = err.Error() + } else { + st.lastErr = "" + discovered := make(map[string]bool, len(ids)) + for _, id := range ids { + discovered[id] = true + } + st.discovered = discovered + } + st.recordHealthLocked(now, err) +} + +// recordHealthLocked updates st's discovery circuit breaker (feat/ +// provider-health) from one finished refresh attempt's outcome. Caller must +// already hold st.mu. +// +// failed reuses limits.go's existing recordProviderAttempt classification — +// isTransient(resp, err) with a nil *http.Response (a discovery fetch has no +// response to inspect on a transport-level failure, unlike a proxied +// request) plus isDeadlineExceeded(err) — rather than inventing a second +// one, per this feature's own design mandate. Net effect: any non-nil error +// except a caller-canceled context counts against the provider, the same +// line recordProviderAttempt already draws for request-path attempts. +// +// While closed, breakerThreshold consecutive failures open the breaker at +// its base backoff (breakerOpenBase); any success resets the counter to +// zero. While halfOpen (the only other state finishRefresh can observe — +// tryBeginRefresh never starts a refresh from open without first moving to +// halfOpen), a successful probe closes the breaker outright; a failed one +// reopens it with a DOUBLED backoff, capped at breakerOpenMax — the +// exponential escalation that keeps a persistently broken provider (like +// xiaomi's ongoing 401s) from being re-probed every single interval +// forever. +func (st *providerState) recordHealthLocked(now time.Time, err error) { + failed := isTransient(nil, err) || isDeadlineExceeded(err) + if st.health == breakerHalfOpen { + if failed { + st.openBreakerLocked(now, true) + } else { + st.closeBreakerLocked() + } return } - st.lastErr = "" - discovered := make(map[string]bool, len(ids)) - for _, id := range ids { - discovered[id] = true + if !failed { + st.closeBreakerLocked() + return } - st.discovered = discovered + st.consecutiveFailures++ + if st.consecutiveFailures >= st.breakerThreshold { + st.openBreakerLocked(now, false) + } +} + +// closeBreakerLocked resets st to a healthy, closed breaker — reachable +// from a successful attempt while closed (nothing to do in practice, since +// consecutiveFailures is already 0, but harmless to reset) or a successful +// half-open probe (the "a provider that recovers must never stay dead" +// requirement). Caller must already hold st.mu. +func (st *providerState) closeBreakerLocked() { + st.health = breakerClosed + st.consecutiveFailures = 0 + st.backoff = 0 + st.openUntil = time.Time{} +} + +// openBreakerLocked opens st's breaker as of now, computing the backoff +// window a subsequent tryBeginRefresh must wait out before its next +// half-open probe. escalate true (a failed half-open probe re-opening) +// doubles the PREVIOUS backoff; escalate false (the closed->open trip) +// starts fresh at breakerOpenBase — either way capped at breakerOpenMax so +// a persistently broken provider's backoff grows but never exceeds it. +// Caller must already hold st.mu. +func (st *providerState) openBreakerLocked(now time.Time, escalate bool) { + if escalate && st.backoff > 0 { + st.backoff *= 2 + } else { + st.backoff = st.breakerOpenBase + } + if st.backoff > st.breakerOpenMax { + st.backoff = st.breakerOpenMax + } + st.health = breakerOpen + st.openUntil = now.Add(st.backoff) +} + +// usable reports whether st's discovery circuit breaker currently permits +// routing to this provider — see modelRegistry.providerUsable's doc comment +// for the full contract this backs. +func (st *providerState) usable() bool { + st.mu.Lock() + defer st.mu.Unlock() + return st.health != breakerOpen } // snapshot returns st's read-only view for the admin dashboard (spec §4, @@ -191,10 +385,15 @@ func (st *providerState) finishRefresh(now time.Time, ids []string, err error) { // previous discovered set rather than clearing it. This method applies // no special handling for that case; it is simply the same set every // other snapshot consumer (modelCount, listFor) already reads. -func (st *providerState) snapshot() (models []string, lastRefresh time.Time, lastErr string) { +// health and openUntil (feat/provider-health) join the same locked pass for +// the identical reason the doc comment above already gives for models/ +// lastRefresh/lastErr: a concurrent finishRefresh must never be able to +// hand the admin dashboard a breaker state that disagrees with the +// lastErr/models it is shown alongside. +func (st *providerState) snapshot() (models []string, lastRefresh time.Time, lastErr string, health breakerState, openUntil time.Time) { st.mu.Lock() defer st.mu.Unlock() - return st.knownIDsLocked(), st.lastRefresh, st.lastErr + return st.knownIDsLocked(), st.lastRefresh, st.lastErr, st.health, st.openUntil } // modelRegistry aggregates every configured provider's explicit and @@ -285,6 +484,65 @@ type modelsCacheEntry struct { gen int64 } +// breakerConfig is Config.Breaker (feat/provider-health), validated and +// defaulted once by validateBreakerConfig — the resolved values +// newModelRegistry copies onto every provider's providerState at +// construction (providerState's own doc comment explains why copied, +// not shared). +type breakerConfig struct { + failureThreshold int + openBase time.Duration + openMax time.Duration +} + +// validateBreakerConfig validates bc and returns the resolved breakerConfig +// (feat/provider-health) — mirrors newRetryPolicy (retry.go) and +// validateCacheConfig (cache.go)'s own zero-means-default, validated-and- +// resolved shape. bc's zero value (BreakerConfig{}, an operator who +// configures no breaker block at all) resolves to every default constant +// above unchanged: a provider that never fails never has a reason to +// consult any of these three numbers, so their defaults exist only to +// bound how a FAILING provider backs off, never to alter a healthy one. +func validateBreakerConfig(bc BreakerConfig) (breakerConfig, error) { + threshold := bc.FailureThreshold + if threshold == 0 { + threshold = defaultBreakerFailureThreshold + } + if threshold < 1 || threshold > maxBreakerFailureThreshold { + return breakerConfig{}, fmt.Errorf("llmgateway: breaker.failureThreshold must be between 1 and %d, got %d", maxBreakerFailureThreshold, threshold) + } + + openBase := defaultBreakerOpenDuration + if bc.OpenDuration != "" { + d, err := time.ParseDuration(bc.OpenDuration) + if err != nil { + return breakerConfig{}, fmt.Errorf("llmgateway: breaker.openDuration %q is invalid: %w", bc.OpenDuration, err) + } + if d <= 0 { + return breakerConfig{}, fmt.Errorf("llmgateway: breaker.openDuration must be positive, got %q", bc.OpenDuration) + } + openBase = d + } + + openMax := defaultBreakerMaxOpenDuration + if bc.MaxOpenDuration != "" { + d, err := time.ParseDuration(bc.MaxOpenDuration) + if err != nil { + return breakerConfig{}, fmt.Errorf("llmgateway: breaker.maxOpenDuration %q is invalid: %w", bc.MaxOpenDuration, err) + } + if d <= 0 { + return breakerConfig{}, fmt.Errorf("llmgateway: breaker.maxOpenDuration must be positive, got %q", bc.MaxOpenDuration) + } + openMax = d + } + + if openMax < openBase { + return breakerConfig{}, fmt.Errorf("llmgateway: breaker.maxOpenDuration (%s) must be >= breaker.openDuration (%s)", openMax, openBase) + } + + return breakerConfig{failureThreshold: threshold, openBase: openBase, openMax: openMax}, nil +} + // newModelRegistry builds a modelRegistry from adapters and cfg's matching // ProviderConfig entries. It does not perform discovery itself — the // caller (newGateway) runs the synchronous first fill separately via @@ -292,6 +550,11 @@ type modelsCacheEntry struct { // provider's ProviderConfig.DiscoveryInterval that fails time.ParseDuration // is a constructor error; an empty one defaults to defaultDiscoveryInterval. func newModelRegistry(adapters map[string]providerAdapter, cfg *Config, log func(string, ...any)) (*modelRegistry, error) { + breaker, err := validateBreakerConfig(cfg.Breaker) + if err != nil { + return nil, err + } + m := &modelRegistry{ adapters: adapters, states: make(map[string]*providerState, len(adapters)), @@ -320,9 +583,14 @@ func newModelRegistry(adapters map[string]providerAdapter, cfg *Config, log func interval := defaultDiscoveryInterval if pc.DiscoveryInterval != "" { - d, err := time.ParseDuration(pc.DiscoveryInterval) - if err != nil { - return nil, fmt.Errorf("llmgateway: provider %q: invalid discoveryInterval %q: %w", name, pc.DiscoveryInterval, err) + // parseErr, not err: newModelRegistry's own outer err (from + // validateBreakerConfig above) is already in scope here, and + // govet's shadow check (golangci-lint's govet enable-all) + // flags a nested ":=" reusing that identifier even though it + // is scoped to this if-block only. + d, parseErr := time.ParseDuration(pc.DiscoveryInterval) + if parseErr != nil { + return nil, fmt.Errorf("llmgateway: provider %q: invalid discoveryInterval %q: %w", name, pc.DiscoveryInterval, parseErr) } interval = d } @@ -338,6 +606,9 @@ func newModelRegistry(adapters map[string]providerAdapter, cfg *Config, log func discovered: make(map[string]bool), discoveryEnabled: pc.Discovery, interval: interval, + breakerThreshold: breaker.failureThreshold, + breakerOpenBase: breaker.openBase, + breakerOpenMax: breaker.openMax, } } @@ -433,6 +704,32 @@ func (m *modelRegistry) now() time.Time { return m.nowFn() } +// providerUsable reports whether name's discovery circuit breaker +// currently permits routing to it (feat/provider-health's queryable +// health signal — the predicate feat/failover consults before sending a +// request to a provider). false only while the breaker is OPEN: an +// unconfigured provider name, one with discovery disabled (it has no +// health signal to trip the breaker with, so it stays closed forever), +// and one whose breaker is closed OR half-open all report true — a +// half-open provider is actively being re-probed by the next discovery +// cycle and must stay routable, per this feature's own "a provider that +// recovers must never stay dead" requirement; only an OPEN breaker, still +// backing off, is unusable. +// +// This is a single mutex-guarded read of state finishRefresh/ +// tryBeginRefresh already maintain — no round trip, no store lookup — per +// this feature's own design: per-pod, in-memory, a latency optimization +// rather than a correctness guarantee. Three pods learning independently +// (each with its own view of a provider's health) is accepted and +// intended, not a bug. +func (m *modelRegistry) providerUsable(name string) bool { + st, ok := m.states[name] + if !ok { + return true + } + return st.usable() +} + // finishRefresh records the outcome of a refresh attempt (st.finishRefresh) // and invalidates modelsJSON's cache by bumping modelsGen — the ONLY two // places that ever mutate st.discovered or st.discoveredContext both @@ -1104,14 +1401,28 @@ func (m *modelRegistry) resolveMetaForAlias(alias, targetProvider, targetModel s // model id names no credential, and this same set is already public // through /v1/models (listFor) to any authenticated caller the group // allows. +// Field order below is fieldalignment-sensitive, the same convention +// providerState's own doc comment explains: pointer-containing fields +// (time.Time, string, []string) grouped first, pointer-free fields +// (modelCount, health, discoveryEnabled) last. type providerSnapshot struct { lastRefresh time.Time - name string - typeName string - baseURL string - lastErr string - models []string - modelCount int + // openUntil mirrors providerState's own field of the same name + // (feat/provider-health) — see providerState.snapshot's doc comment + // for why it is read in the same locked pass as the rest of this + // struct's fields. + openUntil time.Time + name string + typeName string + baseURL string + lastErr string + models []string + modelCount int + // health mirrors providerState's own field of the same name (feat/ + // provider-health): the admin dashboard's breaker-state column. Every + // provider reads closed until a discovery failure actually trips its + // breaker — a healthy deployment's snapshot is unaffected. + health breakerState // discoveryEnabled mirrors providerState.discoveryEnabled (Feature B, // v0.22 last-refresh label honesty): read directly rather than through // providerState.snapshot(), because it is set once at construction @@ -1164,13 +1475,14 @@ func (m *modelRegistry) snapshot() []providerSnapshot { out := make([]providerSnapshot, 0, len(m.providerNames)) for _, name := range m.providerNames { adapter := m.adapters[name] - // Three-value multi-assign straight from the call, no intermediate + // Five-value multi-assign straight from the call, no intermediate // named variables reused across a loop iteration boundary — Yaegi // has had multi-assign edge cases around reused/shadowed loop // variables in other parts of this codebase's history; this shape // (fresh locals every iteration, assigned once, read once) avoids - // that class of trap entirely. - models, lastRefresh, lastErr := m.states[name].snapshot() + // that class of trap entirely. health/openUntil (feat/provider- + // health) widen the same tuple rather than a second locked call. + models, lastRefresh, lastErr, health, openUntil := m.states[name].snapshot() out = append(out, providerSnapshot{ name: adapter.name(), typeName: adapter.typeName(), @@ -1180,6 +1492,8 @@ func (m *modelRegistry) snapshot() []providerSnapshot { lastRefresh: lastRefresh, lastErr: lastErr, discoveryEnabled: m.states[name].discoveryEnabled, + health: health, + openUntil: openUntil, }) } return out diff --git a/registry_health_test.go b/registry_health_test.go new file mode 100644 index 0000000..502e53a --- /dev/null +++ b/registry_health_test.go @@ -0,0 +1,366 @@ +package traefikllmgateway + +import ( + "context" + "errors" + "testing" + "time" +) + +// --- validateBreakerConfig: defaults and validation table --- + +func TestValidateBreakerConfig_Table(t *testing.T) { + tests := []struct { + name string + bc BreakerConfig + wantThresh int + wantOpen time.Duration + wantMaxOpen time.Duration + wantErr bool + }{ + { + name: "zero value uses every default", + bc: BreakerConfig{}, + wantThresh: defaultBreakerFailureThreshold, + wantOpen: defaultBreakerOpenDuration, + wantMaxOpen: defaultBreakerMaxOpenDuration, + }, + { + name: "explicit valid overrides used verbatim", + bc: BreakerConfig{FailureThreshold: 5, OpenDuration: "10s", MaxOpenDuration: "2m"}, + wantThresh: 5, + wantOpen: 10 * time.Second, + wantMaxOpen: 2 * time.Minute, + }, + { + name: "threshold at the max boundary is accepted", + bc: BreakerConfig{FailureThreshold: maxBreakerFailureThreshold}, + wantThresh: maxBreakerFailureThreshold, + wantOpen: defaultBreakerOpenDuration, wantMaxOpen: defaultBreakerMaxOpenDuration, + }, + { + name: "threshold above max is an error", + bc: BreakerConfig{FailureThreshold: maxBreakerFailureThreshold + 1}, + wantErr: true, + }, + { + name: "negative threshold is an error", + bc: BreakerConfig{FailureThreshold: -1}, + wantErr: true, + }, + { + name: "unparseable openDuration is an error", + bc: BreakerConfig{OpenDuration: "not-a-duration"}, + wantErr: true, + }, + { + name: "zero-valued openDuration string is an error", + bc: BreakerConfig{OpenDuration: "0s"}, + wantErr: true, + }, + { + name: "negative openDuration is an error", + bc: BreakerConfig{OpenDuration: "-1s"}, + wantErr: true, + }, + { + name: "unparseable maxOpenDuration is an error", + bc: BreakerConfig{MaxOpenDuration: "not-a-duration"}, + wantErr: true, + }, + { + name: "maxOpenDuration below openDuration is an error", + bc: BreakerConfig{OpenDuration: "5m", MaxOpenDuration: "1m"}, + wantErr: true, + }, + { + name: "maxOpenDuration equal to openDuration is accepted", + bc: BreakerConfig{OpenDuration: "5m", MaxOpenDuration: "5m"}, + wantThresh: defaultBreakerFailureThreshold, + wantOpen: 5 * time.Minute, + wantMaxOpen: 5 * time.Minute, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := validateBreakerConfig(tt.bc) + if (err != nil) != tt.wantErr { + t.Fatalf("validateBreakerConfig(%+v): err = %v, wantErr %v", tt.bc, err, tt.wantErr) + } + if tt.wantErr { + return + } + if got.failureThreshold != tt.wantThresh { + t.Errorf("failureThreshold = %d, want %d", got.failureThreshold, tt.wantThresh) + } + if got.openBase != tt.wantOpen { + t.Errorf("openBase = %v, want %v", got.openBase, tt.wantOpen) + } + if got.openMax != tt.wantMaxOpen { + t.Errorf("openMax = %v, want %v", got.openMax, tt.wantMaxOpen) + } + }) + } +} + +// newHealthTestState builds a providerState with the breaker fields set +// directly (no newModelRegistry round trip needed — the state machine +// under test is pure providerState methods), and every other field at a +// harmless zero/empty value. interval is set deliberately SHORT relative +// to openBase in the tests below so a passing assertion can only be +// explained by the breaker's own backoff window, never by the ordinary +// per-interval throttle it lives alongside. +func newHealthTestState(interval time.Duration, threshold int, openBase, openMax time.Duration) *providerState { + return &providerState{ + explicit: make(map[string]bool), + discovered: make(map[string]bool), + interval: interval, + breakerThreshold: threshold, + breakerOpenBase: openBase, + breakerOpenMax: openMax, + } +} + +// --- breaker: opens after the configured threshold --- + +func TestProviderState_Breaker_OpensAfterThreshold(t *testing.T) { + t.Parallel() + st := newHealthTestState(time.Hour, 3, time.Minute, 10*time.Minute) + now := time.Now() + boom := errors.New("boom") + + for i := 0; i < 2; i++ { + st.finishRefresh(now, nil, boom) + if !st.usable() { + t.Fatalf("after %d failure(s) (threshold 3): usable() = false, want true — must not open early", i+1) + } + } + + st.finishRefresh(now, nil, boom) // 3rd consecutive failure reaches the threshold + if st.usable() { + t.Fatal("after reaching the failure threshold: usable() = true, want false (breaker must be open)") + } +} + +// TestProviderState_Breaker_ASuccessResetsTheFailureCounter proves the +// counter is CONSECUTIVE, not cumulative: two failures, a success, then two +// more failures must not open a threshold-3 breaker, because the success +// in between reset the count back to zero. +func TestProviderState_Breaker_ASuccessResetsTheFailureCounter(t *testing.T) { + t.Parallel() + st := newHealthTestState(time.Hour, 3, time.Minute, 10*time.Minute) + now := time.Now() + boom := errors.New("boom") + + st.finishRefresh(now, nil, boom) + st.finishRefresh(now, nil, boom) + st.finishRefresh(now, []string{"m1"}, nil) // success: resets the streak + st.finishRefresh(now, nil, boom) + st.finishRefresh(now, nil, boom) + + if !st.usable() { + t.Error("usable() = false, want true — the intervening success must have reset the consecutive-failure streak") + } +} + +// --- breaker: stays open for the backoff window, superseding interval --- + +func TestProviderState_Breaker_StaysOpenForBackoffWindow(t *testing.T) { + t.Parallel() + // interval is deliberately much shorter than openBase: a bare pass + // through tryBeginRefresh would allow a retry after 5s. Only the + // breaker's own backoff should be blocking it past that point. + st := newHealthTestState(5*time.Second, 1, time.Minute, 10*time.Minute) + base := time.Now() + + if ok := st.tryBeginRefresh(base); !ok { + t.Fatal("first tryBeginRefresh on a fresh provider = false, want true") + } + st.finishRefresh(base, nil, errors.New("boom")) // trips the breaker (threshold 1) + if st.usable() { + t.Fatal("usable() = true after the trip, want false") + } + + if ok := st.tryBeginRefresh(base.Add(6 * time.Second)); ok { + t.Error("tryBeginRefresh past the plain interval (5s) but well inside the 1m backoff = true, want false — backoff must supersede interval") + } + if ok := st.tryBeginRefresh(base.Add(59 * time.Second)); ok { + t.Error("tryBeginRefresh just before the backoff window elapses = true, want false") + } + if ok := st.tryBeginRefresh(base.Add(61 * time.Second)); !ok { + t.Fatal("tryBeginRefresh once the backoff window has elapsed = false, want true (half-open probe should start)") + } + if !st.usable() { + t.Error("usable() during the half-open probe = false, want true — a provider being re-probed must stay routable") + } +} + +// --- breaker: half-open probe success closes it again --- + +func TestProviderState_Breaker_HalfOpenProbe_SuccessCloses(t *testing.T) { + t.Parallel() + st := newHealthTestState(5*time.Second, 1, time.Minute, 10*time.Minute) + base := time.Now() + st.tryBeginRefresh(base) + st.finishRefresh(base, nil, errors.New("boom")) // open + + probeAt := base.Add(61 * time.Second) + if ok := st.tryBeginRefresh(probeAt); !ok { + t.Fatal("tryBeginRefresh after the backoff window = false, want true") + } + st.finishRefresh(probeAt, []string{"m1"}, nil) // probe succeeds + + if !st.usable() { + t.Error("usable() after a successful half-open probe = false, want true — a recovered provider must not stay dead") + } + if !st.hasModel("m1") { + t.Error("hasModel(m1) = false after a successful probe, want true — discovery still updates the model set normally") + } + + // Closed again means ORDINARY interval gating resumed, not a fresh + // backoff window: immediately after the probe, another attempt must + // still wait out the plain 5s interval, and no longer than that. + if ok := st.tryBeginRefresh(probeAt); ok { + t.Error("tryBeginRefresh at the same instant as the closing probe = true, want false (interval not yet elapsed)") + } + if ok := st.tryBeginRefresh(probeAt.Add(6 * time.Second)); !ok { + t.Error("tryBeginRefresh past the plain interval after closing = false, want true") + } +} + +// --- breaker: half-open probe failure reopens it, with escalating backoff --- + +func TestProviderState_Breaker_HalfOpenProbe_FailureReopens_Escalates(t *testing.T) { + t.Parallel() + st := newHealthTestState(5*time.Second, 1, time.Minute, 10*time.Minute) + base := time.Now() + st.tryBeginRefresh(base) + st.finishRefresh(base, nil, errors.New("boom")) // opens, backoff = 1m + + probeAt := base.Add(61 * time.Second) + if ok := st.tryBeginRefresh(probeAt); !ok { + t.Fatal("tryBeginRefresh after the first backoff window = false, want true") + } + st.finishRefresh(probeAt, nil, errors.New("still broken")) // probe fails: reopen, backoff doubles to 2m + + if st.usable() { + t.Fatal("usable() after a failed probe = true, want false") + } + if ok := st.tryBeginRefresh(probeAt.Add(61 * time.Second)); ok { + t.Error("tryBeginRefresh after only the ORIGINAL (1m) backoff = true, want false — a failed probe must double it to 2m") + } + if ok := st.tryBeginRefresh(probeAt.Add(121 * time.Second)); !ok { + t.Fatal("tryBeginRefresh after the doubled (2m) backoff = false, want true") + } +} + +// TestProviderState_Breaker_BackoffCapsAtOpenMax proves repeated failed +// probes double the backoff only up to breakerOpenMax, never past it. +func TestProviderState_Breaker_BackoffCapsAtOpenMax(t *testing.T) { + t.Parallel() + st := newHealthTestState(time.Second, 1, time.Minute, 3*time.Minute) // base 1m, cap 3m: 1m -> 2m -> capped at 3m (not 4m) + base := time.Now() + st.tryBeginRefresh(base) + st.finishRefresh(base, nil, errors.New("boom")) // open, backoff 1m + + at := base.Add(61 * time.Second) + st.tryBeginRefresh(at) + st.finishRefresh(at, nil, errors.New("boom")) // reopen, backoff doubles to 2m + + at = at.Add(121 * time.Second) + st.tryBeginRefresh(at) + st.finishRefresh(at, nil, errors.New("boom")) // reopen, doubling to 4m would exceed the 3m cap + + if ok := st.tryBeginRefresh(at.Add(2*time.Minute + 59*time.Second)); ok { + t.Error("tryBeginRefresh just before the capped 3m backoff = true, want false") + } + if ok := st.tryBeginRefresh(at.Add(3*time.Minute + 1*time.Second)); !ok { + t.Error("tryBeginRefresh just past the capped 3m backoff = false, want true — backoff must not exceed breakerOpenMax") + } +} + +// --- breaker: a context.Canceled discovery error is not the provider's +// fault, and must not count toward the failure threshold (reuses limits.go's +// isTransient/isDeadlineExceeded classification, which draws this same +// line for request-path attempts) --- + +func TestProviderState_Breaker_ContextCanceled_NeverCountsAsFailure(t *testing.T) { + t.Parallel() + st := newHealthTestState(time.Hour, 1, time.Minute, 10*time.Minute) + now := time.Now() + + st.finishRefresh(now, nil, context.Canceled) + if !st.usable() { + t.Error("usable() = false after a context.Canceled discovery error, want true — a canceled attempt is not a provider fault") + } +} + +// --- breaker: a provider that never fails is unaffected (default- +// preserving: today's interval-only gating, unchanged) --- + +func TestProviderState_Breaker_HealthyProviderNeverAffected(t *testing.T) { + t.Parallel() + st := newHealthTestState(time.Hour, defaultBreakerFailureThreshold, defaultBreakerOpenDuration, defaultBreakerMaxOpenDuration) + now := time.Now() + + for i := 0; i < 5; i++ { + if ok := st.tryBeginRefresh(now); !ok { + t.Fatalf("cycle %d: tryBeginRefresh = false, want true (interval already elapsed)", i) + } + st.finishRefresh(now, []string{"m1"}, nil) + if !st.usable() { + t.Fatalf("cycle %d: usable() = false, want true — a never-failing provider must never trip its breaker", i) + } + if !st.openUntil.IsZero() { + t.Fatalf("cycle %d: openUntil = %v, want zero — breaker must never have opened", i, st.openUntil) + } + // Immediately after: interval-gated exactly like before this + // feature existed, no breaker-induced extra wait. + if ok := st.tryBeginRefresh(now); ok { + t.Fatalf("cycle %d: tryBeginRefresh immediately after a fresh refresh = true, want false (interval not elapsed)", i) + } + now = now.Add(2 * time.Hour) // past the interval for the next cycle + } +} + +// --- modelRegistry.providerUsable: the exported predicate feat/failover +// consumes --- + +func TestModelRegistry_ProviderUsable(t *testing.T) { + t.Parallel() + adapters := map[string]providerAdapter{ + "flaky": newFakeAdapter("flaky"), + "healthy": newFakeAdapter("healthy"), + } + cfg := &Config{ + Providers: map[string]*ProviderConfig{ + "flaky": {Models: []string{"m1"}}, + "healthy": {Models: []string{"m2"}}, + }, + Breaker: BreakerConfig{FailureThreshold: 2, OpenDuration: "1m", MaxOpenDuration: "10m"}, + } + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + if err != nil { + t.Fatalf("newModelRegistry: %v", err) + } + now := time.Now() + reg.nowFn = func() time.Time { return now } + + if !reg.providerUsable("flaky") || !reg.providerUsable("healthy") { + t.Fatal("a freshly constructed registry's providers must both be usable") + } + if !reg.providerUsable("does-not-exist") { + t.Error(`providerUsable("does-not-exist") = false, want true — an unconfigured name is not a reason to block routing`) + } + + boom := errors.New("boom") + reg.states["flaky"].finishRefresh(now, nil, boom) + reg.states["flaky"].finishRefresh(now, nil, boom) // reaches FailureThreshold: 2 + + if reg.providerUsable("flaky") { + t.Error(`providerUsable("flaky") = true after tripping its breaker, want false`) + } + if !reg.providerUsable("healthy") { + t.Error(`providerUsable("healthy") = false, want true — a sibling provider's breaker must not affect it`) + } +} diff --git a/registry_modelmeta_test.go b/registry_modelmeta_test.go index 4612bc0..e9ef17f 100644 --- a/registry_modelmeta_test.go +++ b/registry_modelmeta_test.go @@ -367,7 +367,7 @@ func TestModelRegistry_RefreshProvider_MetadataPanicDoesNotFailDiscovery(t *test if !reg.states["lmstudio"].hasModel("m2") { t.Error("a panic in metadata capture must not discard the successful discovery result") } - _, _, lastErr := reg.states["lmstudio"].snapshot() + _, _, lastErr, _, _ := reg.states["lmstudio"].snapshot() if lastErr != "" { t.Errorf("lastErr = %q, want empty — discovery itself succeeded; only metadata capture panicked", lastErr) }