mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-14 05:18:54 +00:00
merge: per-provider discovery circuit breaker and health surfacing
This commit is contained in:
@@ -190,6 +190,7 @@ config accepts them as YAML, which decodes to the same JSON shape.
|
||||
| `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. |
|
||||
| `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. |
|
||||
| `modelAliases` | `map[string]string` | `{}` | Operator-defined alias id → target model id — see [Model aliases](#model-aliases). Omitted or empty means no aliases, byte-identical to a gateway built before this field existed. |
|
||||
| `modelMeta` | `map[string]*ModelMetaConfig` | `{}` | Per-model context-window and cost overrides, keyed by exact `provider/model` or a bare model id — see [Model metadata](#model-metadata). Omitted or empty means no overrides; every model's metadata still resolves through discovery and the built-in table. |
|
||||
@@ -308,6 +309,14 @@ own is governed purely by their group's — see
|
||||
| `ttl` | `string` (Go duration) | `5m` | How long a cached response stays valid. Invalid duration string is a construction error. |
|
||||
| `maxBodyBytes` | `int` | `1048576` (1MiB) | Largest response body still eligible for caching; a larger one is skipped (never stored, never an error). Maximum accepted value is `8388608` (8MiB) — above that is a construction error. |
|
||||
|
||||
### `BreakerConfig`
|
||||
|
||||
| Field | Type | Default | Semantics |
|
||||
|---|---|---|---|
|
||||
| `failureThreshold` | `int` | `3` | Consecutive failed discovery refreshes that open the breaker — see [Provider health](#provider-health-discovery-circuit-breaker). `0` uses the default. Must be between `1` and `100`, or construction fails. |
|
||||
| `openDuration` | `string` (Go duration) | `1m` | Base backoff a newly opened breaker waits before its first half-open probe; doubles on every further failed probe, capped at `maxOpenDuration`. Invalid or non-positive duration string is a construction error. |
|
||||
| `maxOpenDuration` | `string` (Go duration) | `6h` | Ceiling on the backoff `openDuration` doubles into. Must be `>= openDuration` and no more than `24h`, or construction fails. Deliberately longer than the default `discoveryInterval` (1h) — see [Provider health](#provider-health-discovery-circuit-breaker) for why a value at or below `discoveryInterval` suppresses nothing. |
|
||||
|
||||
### `AdminConfig`
|
||||
|
||||
| Field | Type | Default | Semantics |
|
||||
@@ -897,6 +906,66 @@ cached bodies for the same request.
|
||||
rate-limited) and never fails the request; the upstream call still
|
||||
happens and the client still gets a correct answer.
|
||||
|
||||
## Provider health (discovery circuit breaker)
|
||||
|
||||
A per-provider, in-memory, three-state circuit breaker (`closed` →
|
||||
`open` → `half-open`) driven entirely by `discovery` refresh outcomes —
|
||||
never by `/v1/chat/completions` or any other request-path traffic. It
|
||||
exists so a provider whose discovery endpoint fails on every cycle (an
|
||||
expired API key, a typo'd `baseUrl`, an upstream outage) is re-probed
|
||||
LESS often the longer it stays broken, instead of at the same fixed
|
||||
`discoveryInterval` forever, and so the admin dashboard can show that
|
||||
provider as visibly broken instead of indistinguishable from a healthy
|
||||
one.
|
||||
|
||||
- **States**: `closed` is normal — refreshes run on the provider's own
|
||||
`discoveryInterval`, exactly as if this feature did not exist.
|
||||
`breaker.failureThreshold` (default `3`) CONSECUTIVE failed refreshes
|
||||
open it; any success in between resets the count to zero. `open` skips
|
||||
refresh attempts until a backoff window elapses, starting at
|
||||
`breaker.openDuration` (default `1m`) and doubling on every further
|
||||
failed probe, capped at `breaker.maxOpenDuration` (default `6h`).
|
||||
`half-open` is exactly one probe refresh, deciding whether to close the
|
||||
breaker again (success) or reopen it with a doubled backoff (failure).
|
||||
- **The backoff window is an ADDITIONAL gate on top of
|
||||
`discoveryInterval`, never a replacement for it** — whichever of the
|
||||
two is currently longer wins, and the shorter one is masked entirely
|
||||
rather than merely adding a little extra delay. This means
|
||||
`breaker.maxOpenDuration` only ever reduces re-probe frequency once it
|
||||
EXCEEDS `discoveryInterval`; set no higher, it changes nothing
|
||||
observable and the provider is refreshed at the same cadence a
|
||||
deployment without this feature would already see. The shipped
|
||||
defaults are chosen with this in mind: `maxOpenDuration` (6h) is
|
||||
deliberately longer than the default `discoveryInterval` (1h), so a
|
||||
permanently broken provider under stock config backs off 1h, 2h, 4h,
|
||||
6h, 6h, ... — roughly 5 re-probes a day instead of 24. A custom
|
||||
`discoveryInterval` shorter than `maxOpenDuration` keeps this relationship;
|
||||
one set LONGER than `maxOpenDuration` reverts to plain interval-only
|
||||
throttling with no suppression from this feature at all.
|
||||
- **Classification**: any refresh outcome with a non-nil error counts as
|
||||
a failure, including a malformed-`baseUrl` request-build failure — this
|
||||
is deliberately NOT the same classification `retry` above uses for
|
||||
deciding whether to retry a request, since a permanently broken
|
||||
provider must still be able to trip the breaker. A client-canceled
|
||||
context (plugin shutdown, an outer context canceled upstream) is
|
||||
NEUTRAL: it counts as neither a failure nor a success, and never resets
|
||||
or closes an open breaker.
|
||||
- **A provider that never fails is completely unaffected** — every
|
||||
default is chosen so `closed` never diverges from plain
|
||||
`discoveryInterval` gating.
|
||||
- **Scope**: this reflects the DISCOVERY endpoint only. A provider whose
|
||||
model listing 401s while its actual completion endpoint works fine
|
||||
(common behind a proxy that exposes chat but not model listing) is
|
||||
reported unhealthy here despite being entirely usable for real traffic
|
||||
— this signal must never be the sole reason to refuse routing to a
|
||||
provider.
|
||||
- **Per-pod, not distributed.** Each replica learns a provider's
|
||||
discovery health independently — a latency optimization, not a
|
||||
correctness guarantee. Three replicas disagreeing briefly after a
|
||||
provider starts failing is expected, not a bug.
|
||||
- **Dashboard**: the Providers tab shows a provider's current state and,
|
||||
while open, when it will next be probed — see [Admin](#admin).
|
||||
|
||||
## Unified vs. passthrough
|
||||
|
||||
| | Unified (`/v1/...`) | Native passthrough (`/{provider}/...`) |
|
||||
@@ -1057,8 +1126,12 @@ or in CI.
|
||||
per-model minute figures, since nothing displays them; the dashboard's
|
||||
per-model badge falls back to a day-window-only detail instead), and
|
||||
`modelMeta` (the resolved context window and cost per known model —
|
||||
see [Model metadata](#model-metadata)); each alias entry carries its
|
||||
own `modelMeta` too. `usage`
|
||||
see [Model metadata](#model-metadata)), and `healthState`/`openUntil`
|
||||
(the discovery circuit breaker's current state — `"closed"`, `"open"`,
|
||||
or `"half-open"` — and, while open, when it will next attempt a
|
||||
half-open probe; see [Provider health](#provider-health-discovery-circuit-breaker)).
|
||||
`openUntil` is the zero time when `healthState` is not `"open"`. Each
|
||||
alias entry carries its own `modelMeta` too. `usage`
|
||||
returns every user's and every group's
|
||||
current-window counter values — `requestsPerMinute`, `requestsPerDay`,
|
||||
`tokensInPerDay`/`tokensOutPerDay`, `tokensInPerMonth`/
|
||||
|
||||
@@ -261,6 +261,12 @@ func sanitizeProviderErr(lastErr, rawBaseURL string) string {
|
||||
// sanitizeBaseURL before this view is ever built.
|
||||
type adminProviderView struct {
|
||||
LastRefresh time.Time `json:"lastRefresh"`
|
||||
// OpenUntil is when this provider's breaker will next attempt a
|
||||
// half-open probe (feat/provider-health). The zero time.Time when
|
||||
// HealthState is not "open". Grouped with LastRefresh above (both
|
||||
// time.Time) for fieldalignment — see providerState's own doc
|
||||
// comment for the convention this struct follows.
|
||||
OpenUntil time.Time `json:"openUntil"`
|
||||
// ModelMeta is this provider's resolved per-model metadata (feature
|
||||
// v0.23: context window, per-token cost — resolveModelMeta,
|
||||
// modelmeta.go), keyed by upstream model id. One entry per id in
|
||||
@@ -283,7 +289,16 @@ type adminProviderView struct {
|
||||
Type string `json:"type"`
|
||||
BaseURL string `json:"baseUrl"`
|
||||
LastErr string `json:"lastErr,omitempty"`
|
||||
Name string `json:"name"`
|
||||
// HealthState is this provider's discovery circuit breaker state
|
||||
// (feat/provider-health): "closed" (normal), "open" (backing off
|
||||
// after repeated discovery failures — maybeRefresh skips it until
|
||||
// OpenUntil above), or "half-open" (a probe refresh is in flight,
|
||||
// deciding whether to close the breaker again). A provider with
|
||||
// discovery disabled, or one that has never failed a refresh, always
|
||||
// reads "closed" — it has no health signal that could trip the
|
||||
// breaker.
|
||||
HealthState string `json:"healthState"`
|
||||
Name string `json:"name"`
|
||||
// Models is the sorted explicit∪discovered model id set
|
||||
// (registry.go's providerSnapshot.models — provider-model-accordion
|
||||
// task, Vue admin panel): the dashboard's expandable provider row.
|
||||
@@ -494,6 +509,8 @@ func (g *Gateway) buildAdminOverview() adminOverviewResponse {
|
||||
LastRefresh: s.lastRefresh,
|
||||
LastErr: sanitizeProviderErr(s.lastErr, s.baseURL),
|
||||
DiscoveryEnabled: s.discoveryEnabled,
|
||||
HealthState: s.health.String(),
|
||||
OpenUntil: s.openUntil,
|
||||
AttemptsDay: pc.attemptsDay,
|
||||
FailuresDay: pc.failuresDay,
|
||||
AttemptsMinute: pc.attemptsMinute,
|
||||
|
||||
+10
-10
File diff suppressed because one or more lines are too long
@@ -0,0 +1,95 @@
|
||||
package traefikllmgateway
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// findAdminProvider returns the entry named name from got.Providers, or
|
||||
// fails the test — every admin_health_test.go case looks up exactly one
|
||||
// provider by name out of newAdminTestConfig's fixed "zeta"/"alpha" pair.
|
||||
func findAdminProvider(t *testing.T, got adminOverviewResponse, name string) *adminProviderView {
|
||||
t.Helper()
|
||||
for i := range got.Providers {
|
||||
if got.Providers[i].Name == name {
|
||||
return &got.Providers[i]
|
||||
}
|
||||
}
|
||||
t.Fatalf("providers missing %q: %+v", name, got.Providers)
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestAdminOverview_HealthState_DefaultClosed proves GET
|
||||
// /admin/api/overview reports "closed" for every provider in a freshly
|
||||
// constructed, never-failing Gateway (feat/provider-health) — the
|
||||
// dashboard's default view for a healthy deployment is unaffected by this
|
||||
// feature existing.
|
||||
func TestAdminOverview_HealthState_DefaultClosed(t *testing.T) {
|
||||
t.Parallel()
|
||||
cfg := newAdminTestConfig()
|
||||
h, _ := newAdminGatewayHandle(t, cfg)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, adminRequest(http.MethodGet, adminOverviewPath, "sk-admin1"))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
var got adminOverviewResponse
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
|
||||
for _, name := range []string{"zeta", "alpha"} {
|
||||
p := findAdminProvider(t, got, name)
|
||||
if p.HealthState != "closed" {
|
||||
t.Errorf("provider %q HealthState = %q, want %q", name, p.HealthState, "closed")
|
||||
}
|
||||
if !p.OpenUntil.IsZero() {
|
||||
t.Errorf("provider %q OpenUntil = %v, want zero", name, p.OpenUntil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminOverview_HealthState_ReflectsOpenBreaker proves GET
|
||||
// /admin/api/overview surfaces a tripped breaker (feat/provider-health) —
|
||||
// the dashboard's whole reason for carrying these two fields — without
|
||||
// affecting a healthy sibling provider's own view.
|
||||
func TestAdminOverview_HealthState_ReflectsOpenBreaker(t *testing.T) {
|
||||
t.Parallel()
|
||||
cfg := newAdminTestConfig()
|
||||
cfg.Breaker = BreakerConfig{FailureThreshold: 2, OpenDuration: "1m", MaxOpenDuration: "10m"}
|
||||
h, gw := newAdminGatewayHandle(t, cfg)
|
||||
|
||||
boom := errors.New("boom: discovery unreachable")
|
||||
st := gw.registry.states["alpha"]
|
||||
st.finishRefresh(gw.registry.now(), nil, boom)
|
||||
st.finishRefresh(gw.registry.now(), nil, boom) // reaches FailureThreshold: 2, opens the breaker
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, adminRequest(http.MethodGet, adminOverviewPath, "sk-admin1"))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
var got adminOverviewResponse
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
|
||||
alpha := findAdminProvider(t, got, "alpha")
|
||||
if alpha.HealthState != "open" {
|
||||
t.Errorf("alpha HealthState = %q, want %q", alpha.HealthState, "open")
|
||||
}
|
||||
if alpha.OpenUntil.IsZero() {
|
||||
t.Error("alpha OpenUntil = zero, want a non-zero time while the breaker is open")
|
||||
}
|
||||
|
||||
zeta := findAdminProvider(t, got, "zeta")
|
||||
if zeta.HealthState != "closed" {
|
||||
t.Errorf("zeta HealthState = %q, want %q — a sibling provider's breaker must not affect it", zeta.HealthState, "closed")
|
||||
}
|
||||
}
|
||||
+58
-12
@@ -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
|
||||
|
||||
+487
-32
@@ -28,6 +28,54 @@ 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. Deliberately LONGER than
|
||||
// defaultDiscoveryInterval (1h) — design ruling, adversarial review round
|
||||
// 3: a breaker whose backoff can never exceed the plain interval cannot
|
||||
// suppress anything, since tryBeginRefresh's interval gate (see its own
|
||||
// doc comment) already applies unconditionally and only the LONGER of the
|
||||
// two ever binds. An earlier value here (30m, shorter than the 1h
|
||||
// interval) made the breaker cadence-neutral under stock config — exactly
|
||||
// as strict an interval-only gate would already produce, with none of the
|
||||
// suppression this feature exists to provide. At 6h, a permanently broken
|
||||
// provider under every default backs off 1h, 2h, 4h, 6h, 6h, ... — roughly
|
||||
// 5 attempts a day instead of 24.
|
||||
const defaultBreakerMaxOpenDuration = 6 * time.Hour
|
||||
|
||||
// maxBreakerOpenDuration bounds BreakerConfig.MaxOpenDuration — the same
|
||||
// ceiling-on-a-config-value precedent maxBreakerFailureThreshold already
|
||||
// sets above, applied here because openBreakerLocked's backoff doubling
|
||||
// has no overflow guard of its own: it relies on breakerOpenMax staying
|
||||
// small enough that doubling a value already at or below it can never
|
||||
// approach time.Duration's int64-nanosecond range. A pathological operator
|
||||
// value with no ceiling (adversarial-review finding, round 2) would let
|
||||
// repeated doublings wrap negative after ~28 probes, silently DISABLING
|
||||
// backoff (a negative openUntil is always in the past). 24 hours is
|
||||
// already an extremely generous cap — far longer than any deployment
|
||||
// should ever want a single provider re-probed — and doubling twice from
|
||||
// it (worst case before the cap check fires) lands nowhere near int64
|
||||
// overflow, so the ceiling alone is the fix; openBreakerLocked itself
|
||||
// needs no separate runtime guard.
|
||||
const maxBreakerOpenDuration = 24 * time.Hour
|
||||
|
||||
// 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 +87,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 +158,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,16 +244,53 @@ 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.
|
||||
//
|
||||
// The interval gate (now far enough past lastRefresh, or this is the first
|
||||
// refresh) applies UNCONDITIONALLY, closed or open alike — fix for the
|
||||
// adversarial-review blocker (round 2): an EARLIER version of this method
|
||||
// let an open breaker's own backoff window (openUntil) REPLACE the
|
||||
// interval gate rather than add to it. Since that round's default
|
||||
// maxOpenDuration was shorter than the default interval, a permanently
|
||||
// broken provider got probed roughly TWICE as often as before this
|
||||
// feature existed — the exact opposite of "back off". Backoff can only
|
||||
// ever make retries LESS frequent than the plain interval, never more: a
|
||||
// closed breaker is gated by interval alone (exactly as before this
|
||||
// feature existed — a provider that never fails never touches the
|
||||
// breaker branch below at all); an open breaker is additionally gated by
|
||||
// openUntil, so whichever of the two — interval or backoff — is currently
|
||||
// longer wins. Reaching both 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.
|
||||
//
|
||||
// PRACTICAL CONSEQUENCE, design ruling (round 3): whichever of interval
|
||||
// and maxOpenDuration is larger is the one that actually governs a
|
||||
// persistently broken provider's re-probe cadence — the smaller of the
|
||||
// two is masked entirely, never merely "adds a little more delay". A
|
||||
// deployment whose maxOpenDuration does not exceed its discoveryInterval
|
||||
// gets NO suppression benefit from this feature at all; see
|
||||
// defaultBreakerMaxOpenDuration's own doc comment for why the shipped
|
||||
// defaults are chosen to avoid exactly that trap.
|
||||
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 now.Sub(st.lastRefresh) < st.interval {
|
||||
return false
|
||||
}
|
||||
if st.health == breakerOpen {
|
||||
if now.Before(st.openUntil) {
|
||||
return false
|
||||
}
|
||||
st.health = breakerHalfOpen
|
||||
}
|
||||
st.inFlight = true
|
||||
return true
|
||||
}
|
||||
@@ -152,7 +301,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 +312,159 @@ 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.
|
||||
//
|
||||
// context.Canceled is NEUTRAL — checked first, before any classification —
|
||||
// per adversarial-review ruling (round 2): a canceled discovery fetch
|
||||
// reflects the CALLER walking away (plugin shutdown, an outer context
|
||||
// canceled upstream), not the provider's fault. It must not count as a
|
||||
// FAILURE (would wrongly trip an otherwise-healthy provider's breaker on a
|
||||
// well-timed shutdown) and — the sharper bug an earlier version of this
|
||||
// method had — must not count as a SUCCESS either: treating it as success
|
||||
// would CLOSE an already-open breaker and wipe its backoff/counters purely
|
||||
// because a half-open probe got interrupted before it could complete,
|
||||
// reporting a still-broken provider as healthy again. Every OTHER field
|
||||
// this method could touch is left exactly as it was; only lastRefresh/
|
||||
// lastErr (finishRefresh, above, unconditionally) and inFlight change.
|
||||
//
|
||||
// A canceled probe while halfOpen is the one exception (round 3, MEDIUM
|
||||
// finding): health reverts to OPEN with its backoff/openUntil left
|
||||
// completely untouched, rather than staying at halfOpen. discoveryHealthy
|
||||
// reports true for halfOpen (by design — a provider actively being
|
||||
// re-probed must stay routable), so leaving a canceled probe's state at
|
||||
// halfOpen indefinitely would report a still-broken provider healthy
|
||||
// until the next attempt happens to land, which the plain interval gate
|
||||
// alone does not bound tightly. Reverting to open costs nothing extra —
|
||||
// tryBeginRefresh's own interval gate already runs before this is ever
|
||||
// reached, so the NEXT attempt is admitted at the identical instant
|
||||
// either way; only what discoveryHealthy reports in the meantime differs.
|
||||
//
|
||||
// context.DeadlineExceeded is NOT given the same neutral treatment as
|
||||
// Canceled — deliberately: it counts as a plain failure, below, matching
|
||||
// recordProviderAttempt's own SHOULD-1 ruling (limits.go) that a deadline
|
||||
// the GATEWAY set and the upstream never answered inside is a genuine
|
||||
// provider-health signal, not a caller-walked-away one. Three consecutive
|
||||
// backgroundRefreshTimeout (30s) overruns trip the breaker exactly like
|
||||
// three plain errors would.
|
||||
//
|
||||
// failed classifies every OTHER error explicitly for provider health,
|
||||
// deliberately NOT reusing isTransient's retry-shaped semantics (retry.go)
|
||||
// wholesale — an earlier version of this function did, and adversarial
|
||||
// review found the gap: isTransient returns false for errRequestBuildFailed
|
||||
// (retrying a malformed request build within the SAME attempt would never
|
||||
// help, correct for RETRY purposes), so a provider whose baseURL is
|
||||
// malformed — nothing validates it at construction — built errRequestBuildFailed
|
||||
// on every single discovery cycle forever, and the old code routed that
|
||||
// straight to closeBreakerLocked, wiping state every time: the breaker
|
||||
// could never open despite continuous, permanent failure. For HEALTH
|
||||
// purposes there is no such nuance to preserve: any non-nil, non-canceled
|
||||
// error means this discovery attempt did not succeed, full stop.
|
||||
//
|
||||
// While closed, breakerThreshold consecutive failures open the breaker at
|
||||
// its base backoff (breakerOpenBase); any success resets the counter to
|
||||
// zero. breakerThreshold <= 0 (unreachable through newModelRegistry, which
|
||||
// always resolves it to >= 1 via validateBreakerConfig, but reachable from
|
||||
// a providerState assembled directly — every test in this file does that)
|
||||
// disables the breaker entirely rather than tripping on the very first
|
||||
// failure with a zero backoff: this is what makes providerState's own
|
||||
// documented "zero value behaves exactly as before" claim (its own doc
|
||||
// comment, above) actually true, not just aspirational.
|
||||
//
|
||||
// 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. This ONLY
|
||||
// suppresses re-probing when the resulting backoff exceeds discoveryInterval
|
||||
// — design ruling, round 3: tryBeginRefresh's interval gate (its own doc
|
||||
// comment) applies regardless of breaker state, so a maxOpenDuration
|
||||
// shorter than discoveryInterval is masked entirely and this escalation
|
||||
// changes nothing observable. defaultBreakerMaxOpenDuration is deliberately
|
||||
// set longer than defaultDiscoveryInterval so this DOES suppress re-probing
|
||||
// under stock config — see that constant's own doc comment for the
|
||||
// resulting cadence.
|
||||
func (st *providerState) recordHealthLocked(now time.Time, err error) {
|
||||
if errors.Is(err, context.Canceled) {
|
||||
if st.health == breakerHalfOpen {
|
||||
st.health = breakerOpen
|
||||
}
|
||||
return
|
||||
}
|
||||
st.lastErr = ""
|
||||
discovered := make(map[string]bool, len(ids))
|
||||
for _, id := range ids {
|
||||
discovered[id] = true
|
||||
failed := err != nil
|
||||
if st.health == breakerHalfOpen {
|
||||
if failed {
|
||||
st.openBreakerLocked(now, true)
|
||||
} else {
|
||||
st.closeBreakerLocked()
|
||||
}
|
||||
return
|
||||
}
|
||||
st.discovered = discovered
|
||||
if !failed {
|
||||
st.closeBreakerLocked()
|
||||
return
|
||||
}
|
||||
if st.breakerThreshold <= 0 {
|
||||
return
|
||||
}
|
||||
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)
|
||||
}
|
||||
|
||||
// discoveryHealthy reports whether st's DISCOVERY circuit breaker is
|
||||
// currently closed or half-open (not open) — see modelRegistry.
|
||||
// discoveryHealthy's doc comment for the full contract this backs, in
|
||||
// particular the scope limitation its name is deliberately chosen to
|
||||
// signal: this reflects the discovery endpoint only.
|
||||
func (st *providerState) discoveryHealthy() 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 +488,35 @@ 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.
|
||||
//
|
||||
// openUntil is NOT returned raw (round 3 MEDIUM finding): since
|
||||
// tryBeginRefresh's interval gate applies unconditionally (its own doc
|
||||
// comment), the REAL next probe time is max(st.openUntil,
|
||||
// lastRefresh+interval), which can differ sharply from st.openUntil alone
|
||||
// — under stock defaults, a fresh trip sets st.openUntil to +1m while the
|
||||
// first admitted probe is actually at +1h, a 59-minute skew an operator
|
||||
// watching the dashboard would otherwise see as a broken countdown
|
||||
// (formatUntil renders "in 60s", nothing happens, then the timestamp goes
|
||||
// stale and the countdown vanishes for the remaining 59 minutes). This
|
||||
// correction only ever applies while OPEN — st.openUntil is always the
|
||||
// zero time while closed (closeBreakerLocked) or a past instant while
|
||||
// half-open (a probe only ever starts once now has already reached it),
|
||||
// neither of which this dashboard-facing value should be adjusted for.
|
||||
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
|
||||
effectiveOpenUntil := st.openUntil
|
||||
if st.health == breakerOpen {
|
||||
if floor := st.lastRefresh.Add(st.interval); floor.After(effectiveOpenUntil) {
|
||||
effectiveOpenUntil = floor
|
||||
}
|
||||
}
|
||||
return st.knownIDsLocked(), st.lastRefresh, st.lastErr, st.health, effectiveOpenUntil
|
||||
}
|
||||
|
||||
// modelRegistry aggregates every configured provider's explicit and
|
||||
@@ -285,6 +607,71 @@ 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)
|
||||
}
|
||||
// Ceiling (adversarial-review finding, round 2): see
|
||||
// maxBreakerOpenDuration's own doc comment for why an unbounded
|
||||
// value here is an overflow risk, not merely an odd config choice.
|
||||
if d > maxBreakerOpenDuration {
|
||||
return breakerConfig{}, fmt.Errorf("llmgateway: breaker.maxOpenDuration must be at most %s, got %q", maxBreakerOpenDuration, 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 +679,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 +712,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 +735,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 +833,44 @@ func (m *modelRegistry) now() time.Time {
|
||||
return m.nowFn()
|
||||
}
|
||||
|
||||
// discoveryHealthy reports whether name's DISCOVERY circuit breaker is
|
||||
// currently closed or half-open (feat/provider-health's queryable health
|
||||
// signal). 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, reports false.
|
||||
//
|
||||
// SCOPE, per adversarial-review ruling (round 2) — read this before wiring
|
||||
// it into any routing decision: this reflects the DISCOVERY endpoint
|
||||
// (/v1/models-equivalent) ONLY. It is driven exclusively by
|
||||
// finishRefresh's own outcomes and knows nothing about
|
||||
// /v1/chat/completions or any other request-path traffic. A provider
|
||||
// whose discovery listing 401s or 404s while its actual completion
|
||||
// endpoint serves requests perfectly fine — common behind a proxy that
|
||||
// exposes chat but not model listing — reports false here despite being
|
||||
// entirely usable for real traffic. feat/failover MUST NOT treat a false
|
||||
// result here as a sole reason to refuse routing; request-path health is
|
||||
// a SEPARATE signal, fed by limiter.recordProviderAttempt (limits.go),
|
||||
// deliberately kept apart so discovery outages and request-path outages
|
||||
// are never conflated under one name.
|
||||
//
|
||||
// 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) discoveryHealthy(name string) bool {
|
||||
st, ok := m.states[name]
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
return st.discoveryHealthy()
|
||||
}
|
||||
|
||||
// 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 +1542,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 +1616,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 +1633,8 @@ func (m *modelRegistry) snapshot() []providerSnapshot {
|
||||
lastRefresh: lastRefresh,
|
||||
lastErr: lastErr,
|
||||
discoveryEnabled: m.states[name].discoveryEnabled,
|
||||
health: health,
|
||||
openUntil: openUntil,
|
||||
})
|
||||
}
|
||||
return out
|
||||
|
||||
@@ -0,0 +1,709 @@
|
||||
package traefikllmgateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"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,
|
||||
},
|
||||
{
|
||||
name: "maxOpenDuration at the ceiling is accepted",
|
||||
bc: BreakerConfig{MaxOpenDuration: "24h"},
|
||||
wantThresh: defaultBreakerFailureThreshold,
|
||||
wantOpen: defaultBreakerOpenDuration,
|
||||
wantMaxOpen: 24 * time.Hour,
|
||||
},
|
||||
{
|
||||
name: "maxOpenDuration above the ceiling is an error",
|
||||
bc: BreakerConfig{MaxOpenDuration: "25h"},
|
||||
wantErr: true,
|
||||
},
|
||||
}
|
||||
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.discoveryHealthy() {
|
||||
t.Fatalf("after %d failure(s) (threshold 3): discoveryHealthy() = false, want true — must not open early", i+1)
|
||||
}
|
||||
}
|
||||
|
||||
st.finishRefresh(now, nil, boom) // 3rd consecutive failure reaches the threshold
|
||||
if st.discoveryHealthy() {
|
||||
t.Fatal("after reaching the failure threshold: discoveryHealthy() = 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.discoveryHealthy() {
|
||||
t.Error("discoveryHealthy() = false, want true — the intervening success must have reset the consecutive-failure streak")
|
||||
}
|
||||
}
|
||||
|
||||
// TestProviderState_Breaker_ErrRequestBuildFailed_CountsAsFailure is the
|
||||
// adversarial-review IMPORTANT-2 regression: an earlier version of
|
||||
// recordHealthLocked reused isTransient's RETRY-shaped classification
|
||||
// wholesale (retry.go) — which deliberately excludes errRequestBuildFailed,
|
||||
// since retrying a malformed request build within the same attempt would
|
||||
// never help. That is correct for retrying, but wrong for HEALTH: a
|
||||
// provider whose baseURL is malformed (nothing validates it at
|
||||
// construction) produces this exact error on every discovery cycle,
|
||||
// forever. The old code routed that straight to a "success" outcome,
|
||||
// resetting the breaker every time — it could never open despite
|
||||
// continuous, permanent failure. It must now count as a plain failure.
|
||||
func TestProviderState_Breaker_ErrRequestBuildFailed_CountsAsFailure(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := newHealthTestState(time.Hour, 3, time.Minute, 10*time.Minute)
|
||||
now := time.Now()
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
st.finishRefresh(now, nil, errRequestBuildFailed)
|
||||
}
|
||||
|
||||
if st.discoveryHealthy() {
|
||||
t.Fatal("discoveryHealthy() = true after 3 consecutive errRequestBuildFailed outcomes, want false — a persistent request-build failure is a provider fault the breaker must open on")
|
||||
}
|
||||
}
|
||||
|
||||
// TestProviderState_Breaker_ContextDeadlineExceeded_CountsAsFailure proves
|
||||
// context.DeadlineExceeded is a plain failure, NOT given Canceled's
|
||||
// neutral treatment — deliberate, matching recordProviderAttempt's own
|
||||
// SHOULD-1 ruling (limits.go): a deadline the GATEWAY set and the upstream
|
||||
// never answered inside is a genuine provider-health signal, unlike a
|
||||
// caller-canceled context. Three consecutive backgroundRefreshTimeout
|
||||
// (30s) overruns against a merely slow provider trip the breaker exactly
|
||||
// like three plain errors would.
|
||||
func TestProviderState_Breaker_ContextDeadlineExceeded_CountsAsFailure(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := newHealthTestState(time.Hour, 3, time.Minute, 10*time.Minute)
|
||||
now := time.Now()
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
st.finishRefresh(now, nil, context.DeadlineExceeded)
|
||||
}
|
||||
|
||||
if st.discoveryHealthy() {
|
||||
t.Fatal("discoveryHealthy() = true after 3 consecutive context.DeadlineExceeded outcomes, want false — a deadline the gateway set and the upstream never answered inside is a provider fault, not a neutral cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
// --- breaker: the zero value never trips (adversarial-review finding,
|
||||
// round 2 — providerState's own doc comment claims "zero value behaves
|
||||
// exactly as before this feature existed"; without this guard that claim
|
||||
// was false: a zero-value breaker opened on the FIRST failure with a
|
||||
// ZERO backoff, after which the interval throttle was gone too) ---
|
||||
|
||||
func TestProviderState_Breaker_ZeroThreshold_NeverOpens(t *testing.T) {
|
||||
t.Parallel()
|
||||
// A providerState assembled directly (bypassing newModelRegistry,
|
||||
// which always resolves breakerThreshold to >= 1 via
|
||||
// validateBreakerConfig — every test in this file up to here does
|
||||
// exactly that, through newHealthTestState) leaves every breaker
|
||||
// field at its Go zero value: this is that case, deliberately.
|
||||
st := &providerState{
|
||||
explicit: make(map[string]bool),
|
||||
discovered: make(map[string]bool),
|
||||
interval: time.Millisecond,
|
||||
}
|
||||
now := time.Now()
|
||||
boom := errors.New("boom")
|
||||
|
||||
for i := 0; i < 50; i++ {
|
||||
if !st.tryBeginRefresh(now) {
|
||||
t.Fatalf("iteration %d: tryBeginRefresh = false, want true — a zero-value breaker must behave like plain interval-only gating, never blocked by its own (disabled) backoff", i)
|
||||
}
|
||||
st.finishRefresh(now, nil, boom)
|
||||
if !st.discoveryHealthy() {
|
||||
t.Fatalf("iteration %d: discoveryHealthy() = false, want true — a zero-value breaker must never open at all", i)
|
||||
}
|
||||
now = now.Add(time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// --- breaker: discovery backoff must actually suppress re-probing under
|
||||
// DEFAULT config, strictly, not merely stay no-worse-than plain interval
|
||||
// throttling (BLOCKING adversarial-review findings, rounds 2 and 3) ---
|
||||
|
||||
// TestProviderState_Breaker_DefaultCadence_StrictlyFewerThanPreFeatureBaseline
|
||||
// simulates 24 hours of a permanently-failing provider under every DEFAULT
|
||||
// setting (the shipped configuration, the xiaomi case this feature exists
|
||||
// for) and counts actual ADMITTED refresh attempts.
|
||||
//
|
||||
// Round 2's earlier version of tryBeginRefresh let an open breaker's own
|
||||
// backoff window (openUntil) REPLACE the plain interval gate instead of
|
||||
// adding to it. Since that round's defaultBreakerMaxOpenDuration (30m) was
|
||||
// shorter than defaultDiscoveryInterval (1h), that measured 50 admitted
|
||||
// attempts here — MORE than the 24 a plain, breaker-less interval gate
|
||||
// would ever produce for the same 24h window: the exact inverse of this
|
||||
// feature's purpose. The round-2 fix (interval gate applies
|
||||
// unconditionally) closed that regression, but left the feature CADENCE-
|
||||
// NEUTRAL: with defaultBreakerMaxOpenDuration still shorter than
|
||||
// defaultDiscoveryInterval, the interval alone remained the sole binding
|
||||
// constraint and this test measured EXACTLY 24 — no worse than baseline,
|
||||
// but no better either, doing nothing a plain interval gate would not
|
||||
// already do.
|
||||
//
|
||||
// Round 3's design ruling raised defaultBreakerMaxOpenDuration to 6h
|
||||
// (longer than the 1h interval) specifically so the breaker's backoff can
|
||||
// eventually become the binding constraint instead of the interval — see
|
||||
// that constant's own doc comment for the resulting cadence (roughly 5
|
||||
// attempts a day). This test now asserts the STRONGER, actually-intended
|
||||
// property: strictly fewer admits than the pre-feature baseline. Reverting
|
||||
// EITHER the round-2 interval-gate fix OR the round-3 default bump makes
|
||||
// this fail — the first by exceeding the baseline, the second by merely
|
||||
// equaling it.
|
||||
func TestProviderState_Breaker_DefaultCadence_StrictlyFewerThanPreFeatureBaseline(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := newHealthTestState(defaultDiscoveryInterval, defaultBreakerFailureThreshold, defaultBreakerOpenDuration, defaultBreakerMaxOpenDuration)
|
||||
base := time.Now()
|
||||
boom := errors.New("boom: persistent 401")
|
||||
|
||||
var admits int
|
||||
for minute := 0; minute < 24*60; minute++ {
|
||||
now := base.Add(time.Duration(minute) * time.Minute)
|
||||
if st.tryBeginRefresh(now) {
|
||||
admits++
|
||||
st.finishRefresh(now, nil, boom)
|
||||
}
|
||||
}
|
||||
|
||||
// The pre-feature baseline: a plain interval-only gate (no breaker at
|
||||
// all) admits exactly one attempt per defaultDiscoveryInterval (1h)
|
||||
// over a 24h window for a provider that never succeeds.
|
||||
const preFeatureBaseline = 24
|
||||
if admits >= preFeatureBaseline {
|
||||
t.Errorf("admitted %d discovery attempts over 24h with default config and a permanently failing provider, want STRICTLY FEWER than %d (the pre-feature, interval-only baseline) — the breaker must actually suppress re-probing under stock config, not merely match plain interval throttling", admits, preFeatureBaseline)
|
||||
}
|
||||
}
|
||||
|
||||
// --- breaker: the admin-facing openUntil accounts for the interval floor
|
||||
// (MEDIUM adversarial-review finding, round 3) ---
|
||||
|
||||
// TestProviderState_Snapshot_OpenUntil_ReflectsIntervalFloor proves
|
||||
// snapshot()'s emitted openUntil is the REAL next probe time —
|
||||
// max(raw openUntil, lastRefresh+interval) — not the raw backoff-only
|
||||
// value. Since tryBeginRefresh's interval gate applies unconditionally,
|
||||
// a fresh trip's raw openUntil (lastRefresh + breakerOpenBase) can be far
|
||||
// EARLIER than the actual next admitted attempt whenever breakerOpenBase
|
||||
// is shorter than interval — measured on pure defaults, a 59-minute skew
|
||||
// (raw +1m versus the real +1h). An operator watching the dashboard would
|
||||
// otherwise see "open (in 60s)", nothing happen, and then the countdown
|
||||
// silently vanish once formatUntil (webui) reads the now-past raw value.
|
||||
func TestProviderState_Snapshot_OpenUntil_ReflectsIntervalFloor(t *testing.T) {
|
||||
t.Parallel()
|
||||
// interval (1h) far longer than the backoff base (1m) — the exact
|
||||
// shape of the shipped defaults on a fresh trip.
|
||||
st := newHealthTestState(time.Hour, 1, time.Minute, 10*time.Minute)
|
||||
base := time.Now()
|
||||
st.tryBeginRefresh(base)
|
||||
st.finishRefresh(base, nil, errors.New("boom")) // opens: raw openUntil = base+1m
|
||||
|
||||
_, lastRefresh, _, health, openUntil := st.snapshot()
|
||||
if health != breakerOpen {
|
||||
t.Fatalf("health = %v, want breakerOpen", health)
|
||||
}
|
||||
want := lastRefresh.Add(time.Hour)
|
||||
if !openUntil.Equal(want) {
|
||||
t.Errorf("snapshot openUntil = %v, want %v (lastRefresh + interval, not the raw +1m backoff — tryBeginRefresh will not actually admit a probe before then)", openUntil, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestProviderState_Snapshot_OpenUntil_UnchangedWhenBackoffExceedsInterval
|
||||
// is the complementary case: once backoff is the LONGER of the two (the
|
||||
// steady state this feature is designed to reach — see
|
||||
// defaultBreakerMaxOpenDuration's own doc comment), snapshot's openUntil
|
||||
// is exactly the raw value, with no floor adjustment needed.
|
||||
func TestProviderState_Snapshot_OpenUntil_UnchangedWhenBackoffExceedsInterval(t *testing.T) {
|
||||
t.Parallel()
|
||||
// interval (1m) far shorter than the backoff base (1h) — backoff is
|
||||
// already the binding constraint from the very first trip.
|
||||
st := newHealthTestState(time.Minute, 1, time.Hour, 10*time.Hour)
|
||||
base := time.Now()
|
||||
st.tryBeginRefresh(base)
|
||||
st.finishRefresh(base, nil, errors.New("boom")) // opens: raw openUntil = base+1h
|
||||
|
||||
_, _, _, health, openUntil := st.snapshot()
|
||||
if health != breakerOpen {
|
||||
t.Fatalf("health = %v, want breakerOpen", health)
|
||||
}
|
||||
want := base.Add(time.Hour)
|
||||
if !openUntil.Equal(want) {
|
||||
t.Errorf("snapshot openUntil = %v, want %v (the raw backoff value, unchanged — it already exceeds the interval floor)", openUntil, want)
|
||||
}
|
||||
}
|
||||
|
||||
// --- 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.discoveryHealthy() {
|
||||
t.Fatal("discoveryHealthy() = 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.discoveryHealthy() {
|
||||
t.Error("discoveryHealthy() 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.discoveryHealthy() {
|
||||
t.Error("discoveryHealthy() 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.discoveryHealthy() {
|
||||
t.Fatal("discoveryHealthy() 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: context.Canceled is NEUTRAL, neither failure nor success
|
||||
// (IMPORTANT adversarial-review finding, round 2) ---
|
||||
|
||||
// TestProviderState_Breaker_ContextCanceled_IsNeutral_NotSuccess replaces
|
||||
// a vacuous predecessor that only asserted discoveryHealthy()==true after
|
||||
// ONE Canceled outcome on a threshold-1 breaker — true identically whether
|
||||
// Canceled is ignored (correct) or treated as a success (the bug). This
|
||||
// sequence distinguishes the two: fail, fail, Canceled, fail on a
|
||||
// threshold-3 breaker opens it only if Canceled did NOT reset the
|
||||
// consecutive-failure streak.
|
||||
func TestProviderState_Breaker_ContextCanceled_IsNeutral_NotSuccess(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := newHealthTestState(time.Second, 3, time.Minute, 10*time.Minute)
|
||||
now := time.Now()
|
||||
boom := errors.New("boom")
|
||||
|
||||
st.finishRefresh(now, nil, boom) // failure 1
|
||||
st.finishRefresh(now, nil, boom) // failure 2
|
||||
st.finishRefresh(now, nil, context.Canceled) // neutral: must NOT reset the streak
|
||||
st.finishRefresh(now, nil, boom) // failure 3: reaches the threshold only if Canceled was neutral
|
||||
|
||||
if st.discoveryHealthy() {
|
||||
t.Fatal("discoveryHealthy() = true after fail,fail,Canceled,fail on a threshold-3 breaker, want false — Canceled must not reset the consecutive-failure streak (that would treat cancellation as a success)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestProviderState_Breaker_ContextCanceled_IsNeutral_NotFailure is the
|
||||
// other direction of the bidirectional check (round 3 finding): its
|
||||
// _NotSuccess sibling above proves Canceled does not reset the streak,
|
||||
// but says nothing about whether Canceled ITSELF is wrongly counted as a
|
||||
// failure — a mutant that classified Canceled as a failure would still
|
||||
// pass that test unchanged, since either way the threshold is reached by
|
||||
// the trailing plain failure. A LONE Canceled call on a threshold-1
|
||||
// breaker settles it directly: if Canceled counted as a failure, one call
|
||||
// would be enough to trip a threshold of 1.
|
||||
func TestProviderState_Breaker_ContextCanceled_IsNeutral_NotFailure(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.discoveryHealthy() {
|
||||
t.Fatal("discoveryHealthy() = false after a single context.Canceled outcome on a threshold-1 breaker, want true — Canceled must not be counted as a failure")
|
||||
}
|
||||
}
|
||||
|
||||
// TestProviderState_Breaker_ContextCanceled_DuringHalfOpen_PreservesBackoff
|
||||
// is the sharper half of the same finding: an earlier version treated a
|
||||
// Canceled half-open probe as a SUCCESS, which CLOSED an already-open
|
||||
// breaker and wiped its backoff — reporting a still-401ing provider
|
||||
// healthy purely because a probe was interrupted at shutdown. This proves
|
||||
// two things: discoveryHealthy reports the provider unhealthy immediately
|
||||
// (round 3 finding: a canceled probe reverts to OPEN rather than being
|
||||
// left stuck at halfOpen, which would report healthy), and the backoff
|
||||
// survives — the NEXT real failure must escalate from the ORIGINAL
|
||||
// backoff, not restart fresh at the base.
|
||||
func TestProviderState_Breaker_ContextCanceled_DuringHalfOpen_PreservesBackoff(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := newHealthTestState(time.Second, 1, time.Minute, 10*time.Minute)
|
||||
base := time.Now()
|
||||
st.tryBeginRefresh(base)
|
||||
st.finishRefresh(base, nil, errors.New("boom")) // opens, backoff = 1m, openUntil = base+1m
|
||||
|
||||
probeAt := base.Add(61 * time.Second)
|
||||
if !st.tryBeginRefresh(probeAt) {
|
||||
t.Fatal("tryBeginRefresh after the backoff window = false, want true")
|
||||
}
|
||||
st.finishRefresh(probeAt, nil, context.Canceled) // probe interrupted at shutdown: must be neutral
|
||||
|
||||
if st.discoveryHealthy() {
|
||||
t.Fatal("discoveryHealthy() = true immediately after a canceled half-open probe, want false — a still-broken provider must not read healthy just because its re-probe was interrupted")
|
||||
}
|
||||
|
||||
// The next real attempt must see the ORIGINAL backoff still in
|
||||
// force, not a wiped/reset one: if Canceled had (incorrectly) closed
|
||||
// the breaker, this next failure would re-open it FRESH at the 1m
|
||||
// base instead of escalating to 2m.
|
||||
retryAt := probeAt.Add(2 * time.Second)
|
||||
if !st.tryBeginRefresh(retryAt) {
|
||||
t.Fatal("tryBeginRefresh right after a canceled probe = false, want true (interval elapsed, not blocked by the cancel)")
|
||||
}
|
||||
st.finishRefresh(retryAt, nil, errors.New("still broken"))
|
||||
|
||||
if ok := st.tryBeginRefresh(retryAt.Add(119 * time.Second)); ok {
|
||||
t.Error("tryBeginRefresh before the escalated 2m backoff = true, want false — a canceled probe must not have reset the backoff to its 1m base")
|
||||
}
|
||||
if ok := st.tryBeginRefresh(retryAt.Add(121 * time.Second)); !ok {
|
||||
t.Error("tryBeginRefresh after the escalated 2m backoff = false, want true")
|
||||
}
|
||||
}
|
||||
|
||||
// --- 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()
|
||||
const interval = time.Hour
|
||||
st := newHealthTestState(interval, 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.discoveryHealthy() {
|
||||
t.Fatalf("cycle %d: discoveryHealthy() = false, want true — a never-failing provider must never trip its breaker", i)
|
||||
}
|
||||
// Behavioral proxy for "the breaker never opened" (assert
|
||||
// behavior, not the internal openUntil field, per the brief):
|
||||
// an opened breaker would admit again as soon as its own
|
||||
// (likely shorter) backoff elapsed. Checking one instant BEFORE
|
||||
// the plain interval elapses proves nothing shorter is gating
|
||||
// this provider.
|
||||
if ok := st.tryBeginRefresh(now.Add(interval - time.Nanosecond)); ok {
|
||||
t.Fatalf("cycle %d: tryBeginRefresh one nanosecond before the interval elapses = true, want false", i)
|
||||
}
|
||||
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 * interval) // past the interval for the next cycle
|
||||
}
|
||||
}
|
||||
|
||||
// --- breaker: concurrent access is race-free (IMPORTANT adversarial-
|
||||
// review finding, round 2 — no prior test in this file ever exercised the
|
||||
// mutex-guarded breaker fields from more than one goroutine, so -race had
|
||||
// nothing to catch a locking bug with; the locking was correct, but
|
||||
// unproven) ---
|
||||
|
||||
// TestProviderState_Breaker_ConcurrentRefreshes_RaceFree is primarily a
|
||||
// -race and panic net (round 2), but round 3 found it carried ZERO
|
||||
// assertions and so survived every one of the seven fix-reversion
|
||||
// mutations tried against this file — useful for what it catches, but not
|
||||
// a behavioral test of its own. It now asserts two invariants on every
|
||||
// iteration, under the SAME lock discoveryHealthy/tryBeginRefresh already
|
||||
// take, so the read is race-free too: backoff never exceeds breakerOpenMax
|
||||
// (the overflow-ceiling property, round 2 finding 7, checked here under
|
||||
// real concurrent mutation rather than only sequentially), and health is
|
||||
// always one of the three declared states — breakerState is a bare int32
|
||||
// under the hood, so nothing enforces that at compile time.
|
||||
func TestProviderState_Breaker_ConcurrentRefreshes_RaceFree(t *testing.T) {
|
||||
t.Parallel()
|
||||
st := newHealthTestState(time.Millisecond, 3, 2*time.Millisecond, 20*time.Millisecond)
|
||||
base := time.Now()
|
||||
|
||||
const workers = 8
|
||||
const iterations = 500
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < workers; w++ {
|
||||
wg.Add(1)
|
||||
go func(worker int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < iterations; i++ {
|
||||
now := base.Add(time.Duration(worker*iterations+i) * time.Microsecond)
|
||||
if st.tryBeginRefresh(now) {
|
||||
var err error
|
||||
if (worker+i)%2 == 0 {
|
||||
err = errors.New("boom")
|
||||
}
|
||||
st.finishRefresh(now, []string{"m1"}, err)
|
||||
}
|
||||
_ = st.discoveryHealthy()
|
||||
|
||||
st.mu.Lock()
|
||||
gotBackoff, gotMax, gotHealth := st.backoff, st.breakerOpenMax, st.health
|
||||
st.mu.Unlock()
|
||||
if gotBackoff > gotMax {
|
||||
t.Errorf("worker %d iteration %d: backoff = %v, want <= breakerOpenMax %v", worker, i, gotBackoff, gotMax)
|
||||
}
|
||||
switch gotHealth {
|
||||
case breakerClosed, breakerOpen, breakerHalfOpen:
|
||||
default:
|
||||
t.Errorf("worker %d iteration %d: health = %v, want one of closed/open/half-open", worker, i, gotHealth)
|
||||
}
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// --- modelRegistry.discoveryHealthy: the exported predicate feat/failover
|
||||
// consumes ---
|
||||
|
||||
func TestModelRegistry_DiscoveryHealthy(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.discoveryHealthy("flaky") || !reg.discoveryHealthy("healthy") {
|
||||
t.Fatal("a freshly constructed registry's providers must both be discovery-healthy")
|
||||
}
|
||||
if !reg.discoveryHealthy("does-not-exist") {
|
||||
t.Error(`discoveryHealthy("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.discoveryHealthy("flaky") {
|
||||
t.Error(`discoveryHealthy("flaky") = true after tripping its breaker, want false`)
|
||||
}
|
||||
if !reg.discoveryHealthy("healthy") {
|
||||
t.Error(`discoveryHealthy("healthy") = false, want true — a sibling provider's breaker must not affect it`)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
+150
-2
@@ -45,6 +45,7 @@ import (
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/traefik/yaegi/interp"
|
||||
@@ -293,7 +294,36 @@ func run() error {
|
||||
return fmt.Errorf("read builtin context tokens for %q from pricing_data_gen.go: %w", builtinLookupModelID, err)
|
||||
}
|
||||
|
||||
attemptAccountingOverride := `{"providers":{"openai":{"type":"openai","baseUrl":"` + upstream.URL + `","apiKey":"sk-up","models":["` + testDataWantModel + `","` + builtinLookupModelID + `"]},"` +
|
||||
// --- breaker probe (feat/provider-health): a provider whose
|
||||
// /v1/models 401s, exactly the xiaomi production case, driven under
|
||||
// the REAL interpreter. brkHealthy flips the upstream from always-401
|
||||
// to serving real data once the probe wants to observe recovery;
|
||||
// brkModelsHits counts every /v1/models hit so exerciseBreaker can
|
||||
// prove the open window actually suppresses discovery traffic (a
|
||||
// count staying flat), not merely that healthState reads "open".
|
||||
brkModelsHits := new(int64)
|
||||
brkHealthy := new(int64)
|
||||
brkUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if strings.HasSuffix(r.URL.Path, "/v1/models") {
|
||||
atomic.AddInt64(brkModelsHits, 1)
|
||||
if atomic.LoadInt64(brkHealthy) == 0 {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
_, _ = w.Write([]byte(`{"error":{"message":"invalid api key","type":"invalid_request_error"}}`))
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"object":"list","data":[{"id":"brk-model"}]}`))
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"id":"c3","object":"chat.completion","model":"brk-model","choices":[],"usage":{}}`))
|
||||
}))
|
||||
defer brkUpstream.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"},"` +
|
||||
slowProviderName + `":{"type":"openai","baseUrl":"` + slowUpstream.URL + `","apiKey":"sk-up","models":["` + slowProviderModel + `"]}},` +
|
||||
// modelMeta (feature v0.23): a config-override entry for
|
||||
// testDataWantModel, so exerciseHandler's GET /v1/models
|
||||
@@ -345,7 +375,20 @@ func run() error {
|
||||
return fmt.Errorf("%s.New's returned value does not implement http.Handler", pkgName)
|
||||
}
|
||||
|
||||
return exerciseHandler(handler, builtinLookupContextTokens)
|
||||
if err := exerciseHandler(handler, builtinLookupContextTokens); err != nil {
|
||||
return err
|
||||
}
|
||||
// exerciseBreaker (feat/provider-health, adversarial-review round 2):
|
||||
// runs after exerciseHandler, against the SAME interpreted handler —
|
||||
// proves the discovery circuit breaker's open/half-open/closed state
|
||||
// machine, its new interpreted breakerState (a named int32 with a
|
||||
// String() method), and recordHealthLocked's classification call all
|
||||
// run correctly under Yaegi, not merely compiled. limits.go's own
|
||||
// matchesSentinel doc comment (Trap-4/reverse-direction history) is
|
||||
// exactly why this class of divergence needs its own interpreted
|
||||
// probe: a compiled `go test` pass here would prove nothing about the
|
||||
// interpreter.
|
||||
return exerciseBreaker(handler, brkModelsHits, brkHealthy)
|
||||
}
|
||||
|
||||
// builtinLookupModelID is a real, stable entry in the generated
|
||||
@@ -447,6 +490,111 @@ func readBuiltinContextTokens(repoRoot, modelID string) (int, error) {
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// readHealth returns provider name's healthState string from the admin
|
||||
// overview, decoded generically like readProviderAttemptCounters does
|
||||
// (feat/provider-health).
|
||||
func readHealth(handler http.Handler, name string) (string, error) {
|
||||
req := httptest.NewRequest(http.MethodGet, "/admin/api/overview", nil)
|
||||
req.Header.Set("Authorization", "Bearer "+attemptAccountingAdminAPIKey)
|
||||
rec := httptest.NewRecorder()
|
||||
handler.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
return "", fmt.Errorf("GET /admin/api/overview: status = %d, body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var body map[string]any
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||
return "", err
|
||||
}
|
||||
providers, _ := body["providers"].([]any)
|
||||
for _, raw := range providers {
|
||||
p, ok := raw.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if p["name"] == name {
|
||||
s, _ := p["healthState"].(string)
|
||||
return s, nil
|
||||
}
|
||||
}
|
||||
return "", fmt.Errorf("provider %q not in overview: %s", name, rec.Body.String())
|
||||
}
|
||||
|
||||
// drive fires n authenticated GET /v1/models requests, each of which runs
|
||||
// maybeRefresh at ServeHTTP entry, with a small gap so the 1ms discovery
|
||||
// interval always permits a fresh attempt when the breaker is closed
|
||||
// (feat/provider-health).
|
||||
func drive(handler http.Handler, n int) {
|
||||
for i := 0; i < n; i++ {
|
||||
req := httptest.NewRequest(http.MethodGet, "/v1/models", nil)
|
||||
req.Header.Set("Authorization", "Bearer "+testDataUserAPIKey)
|
||||
handler.ServeHTTP(httptest.NewRecorder(), req)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// exerciseBreaker drives the "brk" provider (attemptAccountingOverride,
|
||||
// run()) through the discovery circuit breaker's full state cycle —
|
||||
// closed -> open -> half-open -> closed again — against the REAL
|
||||
// interpreter, proving under Yaegi what registry_health_test.go already
|
||||
// proves compiled: tryBeginRefresh/finishRefresh/recordHealthLocked, the
|
||||
// interpreted breakerState named-int32 type and its String() method, and
|
||||
// the classification call recordHealthLocked makes all behave correctly
|
||||
// interpreted, not merely compiled (feat/provider-health, adversarial-
|
||||
// review round 2 — limits.go's matchesSentinel doc comment is exactly why
|
||||
// this class of divergence needs its own interpreted probe).
|
||||
func exerciseBreaker(handler http.Handler, hits, healthy *int64) error {
|
||||
// Phase 1: hammer with the upstream 401ing. The breaker must open.
|
||||
deadline := time.Now().Add(10 * time.Second)
|
||||
state := ""
|
||||
for time.Now().Before(deadline) {
|
||||
drive(handler, 5)
|
||||
s, err := readHealth(handler, "brk")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
state = s
|
||||
if s == "open" {
|
||||
break
|
||||
}
|
||||
}
|
||||
if state != "open" {
|
||||
return fmt.Errorf("BREAKER-1: healthState = %q after sustained 401s, want %q (upstream /v1/models hits=%d)", state, "open", atomic.LoadInt64(hits))
|
||||
}
|
||||
openedAtHits := atomic.LoadInt64(hits)
|
||||
fmt.Printf("yaegi-check: breaker opened after %d upstream /v1/models hits\n", openedAtHits)
|
||||
|
||||
// Phase 2: while open, further traffic must NOT reach the upstream.
|
||||
// openDuration is 3s; hammer for ~1s and require zero new hits.
|
||||
drive(handler, 30)
|
||||
duringOpen := atomic.LoadInt64(hits) - openedAtHits
|
||||
if duringOpen != 0 {
|
||||
return fmt.Errorf("BREAKER-2: %d upstream /v1/models hits during the open window, want 0 — backoff did not suppress discovery", duringOpen)
|
||||
}
|
||||
fmt.Println("yaegi-check: breaker open window suppressed all discovery hits")
|
||||
|
||||
// Phase 3: provider recovers. After the backoff window elapses the
|
||||
// half-open probe must run and CLOSE the breaker again.
|
||||
atomic.StoreInt64(healthy, 1)
|
||||
recovered := false
|
||||
recoverDeadline := time.Now().Add(20 * time.Second)
|
||||
for time.Now().Before(recoverDeadline) {
|
||||
drive(handler, 5)
|
||||
s, err := readHealth(handler, "brk")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if s == "closed" {
|
||||
recovered = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !recovered {
|
||||
return fmt.Errorf("BREAKER-3: breaker never returned to closed after the provider recovered (hits=%d)", atomic.LoadInt64(hits))
|
||||
}
|
||||
fmt.Println("yaegi-check: recovered provider closed the breaker again")
|
||||
return nil
|
||||
}
|
||||
|
||||
// exerciseHandler runs two real requests against the interpreted
|
||||
// handler, drawn from the plugin's own .traefik.yml testData: an
|
||||
// authenticated GET /v1/models, which must return the configured
|
||||
|
||||
@@ -21,7 +21,7 @@ import {
|
||||
CardTitle,
|
||||
} from '@/components/ui/card'
|
||||
import { useSearchQuery } from '@/composables/useSearchQuery'
|
||||
import { formatAgo, formatContextWindow, refreshDetailLabel, refreshLabel, routableModelId } from '@/lib/format'
|
||||
import { formatAgo, formatContextWindow, formatUntil, refreshDetailLabel, refreshLabel, routableModelId } from '@/lib/format'
|
||||
import { isModelDegraded } from '@/lib/provider-rate'
|
||||
import { type ExpandState, clearExpandOverrides, computeExpandedItems, toggleItemExpand } from '@/lib/search-expand'
|
||||
import { useDashboardStore } from '@/stores/dashboard'
|
||||
@@ -169,6 +169,36 @@ function modelMetaFor(p: AdminProviderView, model: string): AdminModelMetaView {
|
||||
return p.modelMeta?.[model] ?? ZERO_MODEL_META
|
||||
}
|
||||
|
||||
// --- discovery circuit breaker (feat/provider-health) ---
|
||||
//
|
||||
// Badge shown ONLY for 'open'/'half-open' — mirroring modelIsDegraded's
|
||||
// own "no badge when healthy" convention above: a 'closed' provider (the
|
||||
// common case) renders no breaker badge at all, exactly like a
|
||||
// non-degraded model gets no ProviderRateBadge.
|
||||
|
||||
/** healthBadgeVariant maps healthState to the shadcn-vue Badge variant: destructive (backing off) for 'open', secondary (in progress) for 'half-open', undefined (no badge) for 'closed'. */
|
||||
function healthBadgeVariant(state: AdminProviderView['healthState']): 'destructive' | 'secondary' | undefined {
|
||||
switch (state) {
|
||||
case 'open':
|
||||
return 'destructive'
|
||||
case 'half-open':
|
||||
return 'secondary'
|
||||
default:
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/** healthBadgeLabel renders the breaker badge's text: "open (in Ns)" while backing off (formatUntil omits the parenthetical once openUntil has passed or is unset), or "half-open" while probing. */
|
||||
function healthBadgeLabel(p: AdminProviderView): string {
|
||||
const until = formatUntil(p.openUntil)
|
||||
return until ? `${p.healthState} (${until})` : p.healthState
|
||||
}
|
||||
|
||||
/** healthDetailClass matches the accordion-content detail row's text color to healthBadgeVariant's own trigger-row Badge color, rather than hardcoding text-destructive for every non-closed state: destructive (red) for 'open', muted (grey, the same neutral tone secondary conveys on the Badge) for 'half-open'. */
|
||||
function healthDetailClass(state: AdminProviderView['healthState']): string {
|
||||
return state === 'open' ? 'text-destructive' : 'text-muted-foreground'
|
||||
}
|
||||
|
||||
// --- model aliases (sortable DataTable, operator directive) ---
|
||||
//
|
||||
// The alias id gets the ModelChip copy treatment (an alias name IS the
|
||||
@@ -295,6 +325,21 @@ const aliasEmptyMessage = computed(() =>
|
||||
:attempts-minute="p.attemptsMinute"
|
||||
:failures-minute="p.failuresMinute"
|
||||
/>
|
||||
<!--
|
||||
Discovery circuit breaker (feat/provider-health): no
|
||||
badge at all while 'closed' (the healthy, common case),
|
||||
mirroring modelIsDegraded's own convention for the
|
||||
per-model rate badge above.
|
||||
-->
|
||||
<Badge
|
||||
v-if="p.healthState !== 'closed'"
|
||||
as="span"
|
||||
:variant="healthBadgeVariant(p.healthState)"
|
||||
class="font-normal"
|
||||
:title="p.healthState === 'open' ? 'discovery is backing off after repeated failures' : 'a discovery probe is deciding whether to recover'"
|
||||
>
|
||||
{{ healthBadgeLabel(p) }}
|
||||
</Badge>
|
||||
<FontAwesomeIcon
|
||||
v-if="p.lastErr"
|
||||
:icon="faTriangleExclamation"
|
||||
@@ -314,6 +359,10 @@ const aliasEmptyMessage = computed(() =>
|
||||
<dt class="text-xs text-muted-foreground">Last refresh</dt>
|
||||
<dd>{{ refreshDetailLabel(p.discoveryEnabled, p.lastRefresh) }}</dd>
|
||||
</div>
|
||||
<div v-if="p.healthState !== 'closed'">
|
||||
<dt class="text-xs text-muted-foreground">Discovery health</dt>
|
||||
<dd :class="healthDetailClass(p.healthState)">{{ healthBadgeLabel(p) }}</dd>
|
||||
</div>
|
||||
<div v-if="p.lastErr">
|
||||
<dt class="text-xs text-muted-foreground">Last error</dt>
|
||||
<dd class="text-destructive">{{ p.lastErr }}</dd>
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
import { formatCompactCount, formatContextWindow, formatModelCostHover, refreshDetailLabel, refreshLabel } from './format'
|
||||
import { formatCompactCount, formatContextWindow, formatModelCostHover, formatUntil, refreshDetailLabel, refreshLabel } from './format'
|
||||
|
||||
const ZERO_TIME = '0001-01-01T00:00:00Z'
|
||||
|
||||
@@ -98,6 +98,30 @@ describe('formatModelCostHover', () => {
|
||||
})
|
||||
})
|
||||
|
||||
// Discovery circuit breaker (feat/provider-health, round 3): openUntil's
|
||||
// future-time counterpart to formatAgo's past-time "(Ns ago)". Only
|
||||
// caller today is ProvidersView.vue's healthBadgeLabel, rendering the
|
||||
// breaker's live "open (in Ns)" countdown.
|
||||
describe('formatUntil', () => {
|
||||
it('renders "" for the unset zero-time sentinel', () => {
|
||||
expect(formatUntil(ZERO_TIME)).toBe('')
|
||||
})
|
||||
|
||||
it('renders "" for undefined', () => {
|
||||
expect(formatUntil(undefined)).toBe('')
|
||||
})
|
||||
|
||||
it('renders "in Ns" for a future timestamp', () => {
|
||||
const inFortyTwoSeconds = new Date(Date.now() + 42_000).toISOString()
|
||||
expect(formatUntil(inFortyTwoSeconds)).toMatch(/^in \d+s$/)
|
||||
})
|
||||
|
||||
it('renders "" for a timestamp already in the past — a stale poll must not show a negative countdown', () => {
|
||||
const oneMinuteAgo = new Date(Date.now() - 60_000).toISOString()
|
||||
expect(formatUntil(oneMinuteAgo)).toBe('')
|
||||
})
|
||||
})
|
||||
|
||||
// Usage-count compact display (v0.23 addendum): SI-style, 3 significant
|
||||
// figures, floor-not-round (the same honesty rule formatRatePercent
|
||||
// already applies to the success-rate badge — a count must never appear
|
||||
|
||||
@@ -156,6 +156,24 @@ export function formatAgo(iso: string | undefined): string {
|
||||
return ` (${secs}s ago)`
|
||||
}
|
||||
|
||||
/**
|
||||
* formatUntil renders a FUTURE ISO timestamp as "in Ns" — the discovery
|
||||
* circuit breaker's openUntil (feat/provider-health, admin.go's
|
||||
* adminProviderView.OpenUntil) is the only caller — or "" for the unset
|
||||
* zero-time sentinel, an unparseable value, or a time already in the
|
||||
* past (a stale poll reading an openUntil the breaker has since moved
|
||||
* past; the next 5s refresh corrects it, so this just omits the detail
|
||||
* rather than showing a negative/nonsensical countdown).
|
||||
*/
|
||||
export function formatUntil(iso: string | undefined): string {
|
||||
if (!iso || iso === ZERO_TIME) return ''
|
||||
const then = new Date(iso).getTime()
|
||||
if (Number.isNaN(then)) return ''
|
||||
const secs = Math.round((then - Date.now()) / 1000)
|
||||
if (secs <= 0) return ''
|
||||
return `in ${secs}s`
|
||||
}
|
||||
|
||||
/** formatTimestamp renders an ISO timestamp for display, or "never" for the unset zero-time sentinel. */
|
||||
export function formatTimestamp(iso: string | undefined): string {
|
||||
if (!iso || iso === ZERO_TIME) return 'never'
|
||||
|
||||
@@ -64,6 +64,25 @@ export interface AdminProviderView {
|
||||
* zero lastRefresh.
|
||||
*/
|
||||
discoveryEnabled: boolean
|
||||
/**
|
||||
* healthState is the provider's DISCOVERY circuit breaker state
|
||||
* (admin.go: adminProviderView.HealthState — feat/provider-health):
|
||||
* 'closed' (normal), 'open' (backing off after repeated discovery
|
||||
* failures — openUntil below is when it next probes), or 'half-open'
|
||||
* (a probe is in flight, deciding whether to close again). Reflects
|
||||
* the discovery endpoint ONLY, not request-path health — a provider
|
||||
* whose model listing 401s while its actual completion endpoint works
|
||||
* fine still reads 'open' here. A provider with discovery disabled, or
|
||||
* one that has never failed a refresh, always reads 'closed'.
|
||||
*/
|
||||
healthState: 'closed' | 'open' | 'half-open'
|
||||
/**
|
||||
* openUntil is when this provider's breaker will next attempt a
|
||||
* half-open probe (admin.go: adminProviderView.OpenUntil). The unset
|
||||
* zero-time sentinel (formatAgo/formatTimestamp's ZERO_TIME) when
|
||||
* healthState is not 'open'.
|
||||
*/
|
||||
openUntil: string
|
||||
/**
|
||||
* attemptsDay/failuresDay/attemptsMinute/failuresMinute are this
|
||||
* provider's current-window upstream-attempt counters (admin.go:
|
||||
|
||||
Reference in New Issue
Block a user