From ce39b8eb37ab1f5cd0f727dd9439b78ee72030b4 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Thu, 24 Sep 2026 06:04:43 +0100 Subject: [PATCH] fix(registry): retry failed startup discovery once in the background A provider whose discovery failed during the warm fill at pod start went without discovered models for a full discovery interval, which is 1 hour by default. It also stayed that way until a request arrived after the interval. Production logs showed this for openai-audio and anthropic when outbound calls were slow at startup. A provider that fails the warm fill now gets one background retry after 10s. The retry has its own timeout and uses the same refresh bookkeeping. It is skipped when a refresh is already in flight or has finished since the warm fill, and it is not scheduled when the parent context was canceled. A recovery logs one INFO line. If the retry also fails, the existing breaker behaviour applies unchanged. The retry delay and the warm-fill timeout can be set per registry for tests. yaegi-check probes the retry under the interpreter. Claude-Session: https://claude.ai/code/session_01V5jYJNxpDFdJkyyExpADrv --- README.md | 16 ++ llmgateway.go | 4 + registry.go | 341 +++++++++++++++++++++++++++++++++-- registry_test.go | 367 ++++++++++++++++++++++++++++++++++++++ tools/yaegi-check/main.go | 131 +++++++++++++- 5 files changed, 846 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 2512784..b34b432 100644 --- a/README.md +++ b/README.md @@ -1551,6 +1551,22 @@ one. - **Dashboard**: the Models page's provider health panel shows a provider's current state and, while open, when it will next be probed — see [Admin](#admin). +- **Cold-start retry**: when a provider's synchronous warm-fill fetch + fails at startup, the gateway schedules ONE background retry after a + fixed 10-second delay (`warmFillRetryDelay`, not configurable). The + retry uses the same `finishRefresh` bookkeeping as every other + refresh, so the breaker state stays consistent with it. This recovers + a transient cold-start blip, such as slow egress on the very first + outbound call, not a provider that stays broken. If the retry also + fails, the provider falls back to the breaker behaviour described + above, unchanged. A successful retry logs one INFO line naming the + provider. A failed retry logs the existing ERROR line. The retry + shares its in-flight guard with the regular refresh cycle. It skips + if a refresh is already running, or if one already completed since + the warm-fill failure, so it never runs a second, overlapping + request against the same provider. A warm fill that fails because + the gateway itself is shutting down (its parent context canceled) + schedules no retry at all. ## Unified vs. passthrough diff --git a/llmgateway.go b/llmgateway.go index 2f60eba..53de021 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -1146,6 +1146,10 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str // Routes the self-resolving conditions (a model id two providers both // serve) to WARN instead of ERROR — see the registry's warn field. registry.warn = g.warnf + // Routes retryWarmFill's "recovered after retry" success line to INFO + // instead of the constructor's default ERROR (g.errorf, passed to + // newModelRegistry above) — see the registry's info field doc comment. + registry.info = g.logf // ctx is passed through unwrapped, not re-bounded to warmFillTimeout here: // warmFill already gives each discovery-enabled provider its own fresh // warmFillTimeout budget per provider (registry.go). Wrapping ctx to a diff --git a/registry.go b/registry.go index 03e5b1b..faa5f38 100644 --- a/registry.go +++ b/registry.go @@ -22,6 +22,27 @@ const defaultDiscoveryInterval = time.Hour // construction indefinitely. const warmFillTimeout = 5 * time.Second +// warmFillRetryDelay is how long warmFill waits, after a provider's +// synchronous initial discovery fetch fails, before retrying that ONE +// provider once in the background. Production incident (three separate +// pod-start rollouts, "openai-audio" and "anthropic" both observed): +// occasional cold-start egress latency lets a provider's listModels call +// exceed warmFillTimeout (5s) even though the same provider answers fine +// a few seconds later — without a retry, the provider is left with zero +// discovered models until the next request-driven maybeRefresh window, +// which tryBeginRefresh's interval gate can put up to the provider's own +// discoveryInterval away (defaultDiscoveryInterval: 1h), since +// finishRefresh advances lastRefresh on a FAILED attempt too, so the +// clock starts from the failed warm fill, not from whenever a request +// happens to arrive. +// +// A short, fixed delay — not a config knob — because this recovers a +// one-off cold-start blip, not a steady-state retry policy: a provider +// that fails again after this single retry falls back to the existing +// breaker behavior unchanged (recordHealthLocked, reached through the +// same finishRefresh bookkeeping the retry itself uses). +const warmFillRetryDelay = 10 * time.Second + // backgroundRefreshTimeout bounds one background refresh goroutine spawned // by maybeRefresh. Not specified by the task brief; chosen generously // enough for a provider's models-list call while still bounding a leaked @@ -306,6 +327,54 @@ func (st *providerState) tryBeginRefresh(now time.Time) bool { return true } +// tryBeginWarmFillRetry reports whether warmFill's one-shot background +// retry (scheduleWarmFillRetry/retryWarmFill) is still allowed to run +// against st, and if so marks it inFlight — the SAME field, and the SAME +// mutual-exclusion contract, tryBeginRefresh's own doc comment above +// already establishes for maybeRefresh's request-driven path. Fix 2/3 +// (verify-retry.md): an EARLIER version of retryWarmFill called neither +// this nor tryBeginRefresh at all, so its listModels call could run +// CONCURRENTLY with an ordinary request-driven refresh already in flight +// against the identical provider — two in-progress fetches racing to +// finishRefresh in arbitrary completion order, plus (with a half-open +// breaker) a failed retry double-counting as the half-open probe. Sharing +// st.inFlight with tryBeginRefresh, under the SAME st.mu, makes the two +// mutually exclusive unconditionally: whichever caller's check-and-set +// runs first wins; the other's own guard (this one, or tryBeginRefresh's) +// observes inFlight already true and refuses — never a race window where +// both believe they won. +// +// sinceWarmFill is the exact finishRefresh timestamp the warm fill that +// scheduled THIS retry was recorded with (warmFillFailure.at, set once in +// warmFill and threaded through scheduleWarmFillRetry/ +// scheduleOneWarmFillRetry unchanged). A refresh whose lastRefresh has +// already moved PAST that instant by the time this retry's timer fires — +// most plausibly a request-driven refresh admitted by a very short +// discoveryInterval, but any refresh path qualifies — has already +// recorded a result superseding whatever this retry would fetch, so the +// retry is redundant and skips even though inFlight is currently false +// again. This is also what makes tools/yaegi-check's own "brk" provider +// probe safe (fix 3, verify-retry.md): its 1ms discoveryInterval means a +// request-driven refresh reliably beats the 10s retry delay, so the +// retry always finds lastRefresh already past sinceWarmFill and skips, +// rather than racing exerciseBreaker's own hit-count assertions. +// +// The caller must pair a true result with a later finishRefresh call — +// identical contract to tryBeginRefresh's own — which clears inFlight +// unconditionally regardless of which of the two guards set it. +func (st *providerState) tryBeginWarmFillRetry(sinceWarmFill time.Time) bool { + st.mu.Lock() + defer st.mu.Unlock() + if st.inFlight { + return false + } + if st.lastRefresh.After(sinceWarmFill) { + return false + } + st.inFlight = true + return true +} + // finishRefresh records the outcome of a refresh attempt started by // tryBeginRefresh (or, for the synchronous warm fill, run unconditionally). // lastRefresh always advances to now, on success or failure alike — that is @@ -570,6 +639,22 @@ type modelRegistry struct { // type: call it only as m.warn("%s", fmt.Sprintf(...)), never with two // or more variadic arguments. warn func(string, ...any) + // info is the info-level counterpart of log/warn (g.logf), used only + // by retryWarmFill's own "recovered after retry" success line: an + // operator watching INFO-level logs during a rollout should see a + // transient cold-start discovery failure self-heal, not silence. + // Set by the production caller AFTER newModelRegistry returns, the + // same "not a fourth constructor parameter" convention the warn + // field's own doc comment explains above. A nil info falls back to + // log (infof, below) — the same fallback shape warnf already has — + // so a registry built without one (every existing test) keeps + // compiling and reports the recovery at ERROR severity instead of + // panicking or silently dropping the line. + // + // The same Yaegi CFG restriction the log/warn fields document above + // applies here too: call it only as m.info("%s", fmt.Sprintf(...)), + // never with two or more variadic arguments. + info func(string, ...any) // aliases is the validated alias->target map from Config.ModelAliases // (spec §5, v0.2), built once by newModelRegistry via // validateModelAliases and never mutated afterward — resolve checks @@ -608,7 +693,30 @@ type modelRegistry struct { // finishRefresh writes it from whichever goroutine (warmFill's caller, // or maybeRefresh's background goroutine) is recording that refresh. modelsGen int64 - warnedMu sync.Mutex + // warmFillRetryDelay is the injectable form of the warmFillRetryDelay + // constant above (its own doc comment has the full rationale) — + // resolved once at construction (newModelRegistry, below) to that + // constant, the same "baked in, never reassigned outside a test" + // convention nowFn already establishes, so a test can shorten it + // directly (registry_test.go) without a fake clock: unlike interval/ + // lastRefresh, this genuinely sleeps on a real timer (time.AfterFunc), + // which m.nowFn's overridable fake clock cannot accelerate. Grouped + // down here with the other pointer-free fields (fieldalignment), not + // beside nowFn above, despite the doc-comment cross-reference. + warmFillRetryDelay time.Duration + // warmFillTimeout is the injectable form of the warmFillTimeout + // constant above (its own doc comment has the full rationale) — + // resolved once at construction (newModelRegistry, below) to that + // constant, the same "baked in, never reassigned outside a test" + // convention warmFillRetryDelay above already establishes. warmFill, + // retryWarmFill, and captureModelMetadata's warm-fill budget (both + // call sites, within warmFill and retryWarmFill) all read this field + // rather than the package constant, so a test can shorten the whole + // warm-fill budget directly (registry_test.go) without waiting out + // the real 5s default — refreshProvider's own steady-state refresh + // stays on backgroundRefreshTimeout, unaffected. + warmFillTimeout time.Duration + warnedMu sync.Mutex // modelsCacheMu guards modelsCache. modelsCacheMu sync.Mutex } @@ -698,13 +806,15 @@ func newModelRegistry(adapters map[string]providerAdapter, cfg *Config, log func } m := &modelRegistry{ - adapters: adapters, - states: make(map[string]*providerState, len(adapters)), - log: log, - nowFn: time.Now, - warned: make(map[string]bool), - providerNames: make([]string, 0, len(adapters)), - modelsCache: make(map[string]modelsCacheEntry), + adapters: adapters, + states: make(map[string]*providerState, len(adapters)), + log: log, + nowFn: time.Now, + warmFillRetryDelay: warmFillRetryDelay, + warmFillTimeout: warmFillTimeout, + warned: make(map[string]bool), + providerNames: make([]string, 0, len(adapters)), + modelsCache: make(map[string]modelsCacheEntry), } for name := range adapters { m.providerNames = append(m.providerNames, name) @@ -912,23 +1022,60 @@ func (m *modelRegistry) finishRefresh(st *providerState, now time.Time, ids []st atomic.AddInt64(&m.modelsGen, 1) } +// warmFillFailure records ONE provider whose synchronous warmFill +// discovery fetch just failed and is eligible for scheduleWarmFillRetry's +// one-shot background retry (production incident fix, warmFillRetryDelay's +// own doc comment). at is the exact finishRefresh timestamp that failed +// attempt was recorded with — the SAME m.now() value warmFill itself used +// for THIS provider's finishRefresh call, threaded unchanged through +// scheduleWarmFillRetry/scheduleOneWarmFillRetry to retryWarmFill's own +// tryBeginWarmFillRetry guard (fix 2/3, verify-retry.md) as sinceWarmFill: +// a refresh whose lastRefresh has already moved past this exact instant +// by the time the retry's timer fires has already superseded whatever +// this retry would fetch, so the retry skips rather than racing or +// duplicating it. +type warmFillFailure struct { + at time.Time + name string +} + // warmFill performs newGateway's synchronous first discovery fill: for // every discovery-enabled provider, it fetches listModels once, bounded by // warmFillTimeout, and records the result via finishRefresh. A fetch error // is logged and otherwise non-fatal — construction still succeeds, and the // provider's known model set is whatever its explicit config already // provides (empty, if it configured neither explicit models nor a -// reachable discovery endpoint) until the next maybeRefresh window. +// reachable discovery endpoint) until either the one-shot retry below +// recovers it, or the next maybeRefresh window does. +// +// Production incident fix: every provider whose fetch failed here is +// handed to scheduleWarmFillRetry (below), which retries each of them +// exactly once, in the background, after warmFillRetryDelay — see that +// constant's own doc comment for why a plain interval-gated maybeRefresh +// alone leaves a cold-start blip unrecovered for up to an hour. This adds +// no cost to warmFill itself: scheduling is a single time.AfterFunc call +// per failed provider, not a blocking wait. +// +// A fetch that failed because the PARENT ctx was canceled (plugin +// construction itself aborted — shutdown, not a slow/broken upstream) is +// excluded from the retry list (fix 4a, verify-retry.md): retryWarmFill +// has no shutdown signal of its own to listen on (its own doc comment +// explains why), so scheduling one anyway would keep retrying a provider +// nobody is waiting on any more. Mirrors recordHealthLocked's own +// "Canceled is neutral, not a failure" treatment of the breaker, extended +// here to the separate scheduling decision. func (m *modelRegistry) warmFill(ctx context.Context) { + var failed []warmFillFailure for _, name := range m.providerNames { st := m.states[name] if !st.discoveryEnabled { continue } - fctx, cancel := context.WithTimeout(ctx, warmFillTimeout) + fctx, cancel := context.WithTimeout(ctx, m.warmFillTimeout) ids, err := m.adapters[name].listModels(fctx) cancel() - m.finishRefresh(st, m.now(), ids, err) + now := m.now() + m.finishRefresh(st, now, ids, err) if err != nil { // sanitizeProviderErr (admin.go): the error commonly embeds the // dialed URL verbatim, so a baseUrl an operator misconfigured @@ -938,6 +1085,11 @@ func (m *modelRegistry) warmFill(ctx context.Context) { // key material" contract. Security audit run-1, finding F-2. m.log("%s", fmt.Sprintf("model registry: initial discovery for provider %q failed: %v", name, sanitizeProviderErr(err.Error(), m.adapters[name].base()))) + // Fix 4a (verify-retry.md): see this function's own doc + // comment above — a parent-ctx cancellation gets no retry. + if !errors.Is(err, context.Canceled) { + failed = append(failed, warmFillFailure{name: name, at: now}) + } } // Own timeout budget, review fix (SHOULD-4): fctx above is // spent by listModels — reusing it here would hand the metadata @@ -945,10 +1097,13 @@ func (m *modelRegistry) warmFill(ctx context.Context) { // timed-out listModels) context every warm fill, guaranteeing a // spurious deadline error and log line even for a healthy // provider whose model-list call merely took a while. - mctx, mcancel := context.WithTimeout(ctx, warmFillTimeout) + mctx, mcancel := context.WithTimeout(ctx, m.warmFillTimeout) m.captureModelMetadata(mctx, name, m.adapters[name], st) mcancel() } + if len(failed) > 0 { + m.scheduleWarmFillRetry(failed) + } } // modelMetadataFetcher is an OPTIONAL providerAdapter capability (feature @@ -1027,6 +1182,168 @@ func (m *modelRegistry) captureModelMetadata(ctx context.Context, name string, a st.setDiscoveredContext(meta) } +// infof writes msg through the info field, falling back to log when no +// info was injected (mirrors warnf's own fallback, above) — same +// single-variadic-argument Yaegi-safety shape the log/warn/warnf call +// sites already document. +func (m *modelRegistry) infof(msg string) { + if m.info != nil { + m.info("%s", msg) + return + } + m.log("%s", msg) +} + +// scheduleWarmFillRetry schedules exactly ONE background retry, after +// m.warmFillRetryDelay, for each provider name in failed — providers +// whose synchronous warmFill discovery fetch (above) just failed. +// warmFillRetryDelay's own doc comment has the full production-incident +// rationale; this method is purely the scheduling mechanics. +// +// Each retry runs on its own goroutine via time.AfterFunc, self-bounded +// by context.Background() plus its own warmFillTimeout budget in +// retryWarmFill (below) — the SAME self-terminating shape refreshProvider +// already uses for its maybeRefresh-spawned goroutine, and for the +// identical reason documented there: this plugin has no wired shutdown/ +// Close hook a background goroutine could listen on instead (Close, +// llmgateway.go, only releases redisClient — its own doc comment explains +// why nothing in this codebase calls it automatically, and Traefik's own +// plugin contract gives a discarded instance no teardown signal at +// config-reload swap time either). Binding to a fixed delay+timeout +// instead of an external cancel signal means a Gateway discarded mid- +// flight (a config reload landing inside the retry window) leaks at most +// one goroutine per still-failed provider, for at most +// warmFillRetryDelay+2×warmFillTimeout (worst case ~20s at the shipped +// defaults: the retry's own listModels call budgeted by warmFillTimeout, +// PLUS captureModelMetadata's own separate warmFillTimeout budget within +// retryWarmFill, below — not merely the first of the two) — bounded, not +// a steady-state leak, matching refreshProvider's own accepted worst case +// (its own doc comment, below). +// +// The retry goes through the SAME finishRefresh bookkeeping warmFill/ +// refreshProvider already use (retryWarmFill, below), so breaker state +// (recordHealthLocked) stays consistent: a successful retry is recorded +// exactly like any other successful discovery attempt, closing an +// already-open breaker if the failed warm fill alone had somehow tripped +// one; a failed retry is simply a second consecutive failure toward the +// ordinary breaker threshold — existing breaker behavior, unchanged. +// +// Each iteration calls scheduleOneWarmFillRetry (below) rather than +// inlining the time.AfterFunc closure directly in this loop — the SAME +// "close over named parameters, not loop variables" defense +// refreshProvider's own doc comment already documents for maybeRefresh, +// below. This is not merely stylistic: an EARLIER version of this method +// built the closure right here, over this loop's own range variable, and +// tools/yaegi-check caught it misbehaving under the real interpreter — +// with two failed providers, BOTH scheduled timers ended up firing +// against whichever provider's state the range variable held LAST, +// double-retrying it and never touching the other at all. Compiled Go +// (go.mod: go 1.22) never showed this: the range variable is per- +// iteration there, so the bug was invisible to `go test`, only to Yaegi. +func (m *modelRegistry) scheduleWarmFillRetry(failed []warmFillFailure) { + for _, f := range failed { + m.scheduleOneWarmFillRetry(f.name, f.at) + } +} + +// scheduleOneWarmFillRetry schedules name's single retry timer +// (scheduleWarmFillRetry, above). name is a plain function PARAMETER +// here, not a loop variable of any kind, so the closure below captures a +// binding that is unambiguously fresh per call under every Go version — +// and, per this method's own existence, under Yaegi's interpretation of +// one too. +func (m *modelRegistry) scheduleOneWarmFillRetry(name string, at time.Time) { + st := m.states[name] + adapter := m.adapters[name] + time.AfterFunc(m.warmFillRetryDelay, func() { + m.retryWarmFill(name, st, adapter, at) + }) +} + +// retryWarmFill runs one deferred warm-fill retry (scheduleWarmFillRetry, +// above) for a single provider whose synchronous warm fill just failed. +// Mirrors warmFill's own per-provider body — including the same +// captureModelMetadata step — rather than refreshProvider's shape below: +// this is still a first-DISCOVERY attempt, budgeted by warmFillTimeout, +// not a steady-state background refresh budgeted by +// backgroundRefreshTimeout. +// +// On success, logs an INFO "recovered after retry" line (m.infof) so an +// operator watching a rollout sees the transient cold-start failure +// self-heal instead of staying silent; on failure, logs the existing +// ERROR line unchanged and leaves the provider exactly where the +// ordinary breaker/refresh path already would. +// +// The deferred recover mirrors refreshProvider's own, for the identical +// reason: this runs off a time.AfterFunc goroutine with no ServeHTTP +// caller to unwind into, so an unrecovered panic here would crash the +// whole Traefik process, not just fail one request. finishRefresh still +// runs on the panic path (with a synthetic error) — but ONLY once the +// guard below has actually started something: sinceWarmFill's own +// tryBeginWarmFillRetry call is this function's FIRST statement, +// deliberately ahead of the recover-guarded section, so a guard refusal +// returns before the deferred finishRefresh is even registered. +// +// Fix 2/3 (verify-retry.md) — corrected: an EARLIER version of this +// function never called tryBeginRefresh (or any guard) at all, so its +// listModels call could run CONCURRENTLY with an ordinary request-driven +// refresh already in flight against the identical provider. It now calls +// tryBeginWarmFillRetry (providerState, above) — the SAME st.inFlight +// field and st.mu, so the two are mutually exclusive exactly like two +// request-driven refreshes already were — and, on a true result, is +// bound by the identical "pair with a later finishRefresh" contract +// tryBeginRefresh's own doc comment establishes; finishRefresh clears +// inFlight unconditionally regardless of which of the two guards set it. +func (m *modelRegistry) retryWarmFill(name string, st *providerState, adapter providerAdapter, sinceWarmFill time.Time) { + // Guard (fix 2/3, verify-retry.md): skip entirely, before touching + // anything else, when a refresh is already in flight or one has + // already completed since the warm fill that scheduled this retry — + // tryBeginWarmFillRetry's own doc comment has the full contract. A + // skip here means NOTHING was attempted, so finishRefresh below must + // not run either: there is no outcome to record, and calling it + // anyway would wrongly overwrite whatever the winning refresh already + // recorded with a synthetic no-op result. + if !st.tryBeginWarmFillRetry(sinceWarmFill) { + return + } + + var ids []string + var err error + defer func() { + if rec := recover(); rec != nil { + err = fmt.Errorf("panic: %v", rec) + } + m.finishRefresh(st, m.now(), ids, err) + if err != nil { + // Scrubbed for the same reason as warmFill's own + // discovery-failure line (security audit run-1, finding F-2). + m.log("%s", fmt.Sprintf("model registry: warm-fill retry for provider %q failed: %v", + name, sanitizeProviderErr(err.Error(), adapter.base()))) + return + } + m.infof(fmt.Sprintf("model registry: initial discovery for provider %q recovered after retry", name)) + }() + + // Both cancels are deferred, mirroring refreshProvider's own reasoning + // below: this function has a recover above, so a panicking listModels + // would skip an inline cancel() and hold the timer until + // warmFillTimeout fires. Deferring costs nothing and survives the + // panic path. + fctx, cancel := context.WithTimeout(context.Background(), m.warmFillTimeout) + defer cancel() + ids, err = adapter.listModels(fctx) + if err != nil { + return + } + + // Own timeout budget, same reasoning as warmFill/refreshProvider's + // identical comment: reusing fctx here would hand the metadata fetch + // an already-spent context whenever listModels itself ran long. + mctx, mcancel := context.WithTimeout(context.Background(), m.warmFillTimeout) + defer mcancel() + m.captureModelMetadata(mctx, name, adapter, st) +} + // maybeRefresh is called at every ServeHTTP entry. For each discovery- // enabled provider whose last refresh is older than its interval (and that // is not already mid-refresh), it spawns exactly one background goroutine diff --git a/registry_test.go b/registry_test.go index 31ccbb3..919103d 100644 --- a/registry_test.go +++ b/registry_test.go @@ -8,6 +8,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "runtime" "strings" "sync" "sync/atomic" @@ -74,6 +75,17 @@ func (r *recordingLog) count() int { return len(r.lines) } +// linesCopy returns a snapshot of every line recorded so far — for a test +// that needs to assert on a specific line's CONTENT (e.g. the +// warm-fill-retry "recovered after retry" text), not merely its count. +func (r *recordingLog) linesCopy() []string { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]string, len(r.lines)) + copy(out, r.lines) + return out +} + // waitUntil polls cond every millisecond until it returns true, failing the // test if timeout elapses first. Used to synchronize against background // refresh goroutines without a fixed sleep. @@ -917,6 +929,361 @@ func TestModelRegistry_WarmFill_EachProviderGetsFullBudget_NotSharedAcrossProvid } } +// --- warmFill retry: one-shot background recovery for a cold-start blip +// (production incident: "openai-audio"/"anthropic" initial discovery +// occasionally exceeds warmFillTimeout on pod start) --- + +// TestModelRegistry_WarmFillRetry_FailedThenSuccessful_PopulatesModelsAndClosesBreaker +// is the core recovery case: warmFill's synchronous fetch fails, tripping +// a threshold-1 breaker, and the scheduled retry succeeds — the provider +// must end up with the retried models AND a closed breaker, exactly as if +// its very first discovery attempt had simply succeeded a little late. +func TestModelRegistry_WarmFillRetry_FailedThenSuccessful_PopulatesModelsAndClosesBreaker(t *testing.T) { + t.Parallel() + fa := newFakeAdapter("openai") + var calls int32 + fa.listModelsFn = func(context.Context) ([]string, error) { + if atomic.AddInt32(&calls, 1) == 1 { + return nil, fmt.Errorf("cold-start egress timeout") + } + return []string{"gpt-warm"}, nil + } + adapters := map[string]providerAdapter{"openai": fa} + cfg := &Config{ + Providers: map[string]*ProviderConfig{"openai": {Discovery: true}}, + // FailureThreshold 1: the failed warm fill alone trips the breaker, + // so a successful retry closing it again is actually exercised — + // at the package default (3) a single failure would never open it. + Breaker: BreakerConfig{FailureThreshold: 1, OpenDuration: "1m", MaxOpenDuration: "10m"}, + } + rl := &recordingLog{} + reg, err := newModelRegistry(adapters, cfg, rl.fn) + if err != nil { + t.Fatalf("newModelRegistry: %v", err) + } + infoLog := &recordingLog{} + reg.info = infoLog.fn + // Widened from an earlier 10ms (verify-retry.md item 4): the two + // "before the retry has had a chance to run" checks just below run + // synchronously, right after warmFill returns, with no synchronization + // of their own against the scheduled timer — a heavily loaded (e.g. + // -race) machine could in principle delay the test goroutine past a + // 10ms window and see the retry's own result instead of the pre-retry + // state. 200ms is generous headroom for that gap while the overall + // waitUntil below still bounds the test's total time. + reg.warmFillRetryDelay = 200 * time.Millisecond + + reg.warmFill(context.Background()) + + if reg.discoveryHealthy("openai") { + t.Fatal(`discoveryHealthy("openai") = true immediately after the failed warm fill (threshold 1), want false`) + } + if reg.states["openai"].hasModel("gpt-warm") { + t.Fatal(`hasModel("gpt-warm") = true before the retry has had a chance to run`) + } + + waitUntil(t, 2*time.Second, func() bool { + return reg.states["openai"].hasModel("gpt-warm") + }) + + if !reg.discoveryHealthy("openai") { + t.Error(`discoveryHealthy("openai") = false after a successful retry, want true — the retry must close the breaker it tripped`) + } + _, _, lastErr, _, _ := reg.states["openai"].snapshot() + if lastErr != "" { + t.Errorf("lastErr = %q after a successful retry, want empty", lastErr) + } + if got := atomic.LoadInt32(&calls); got != 2 { + t.Errorf("listModels called %d times, want exactly 2 (the failed warm fill plus one retry)", got) + } + // Fix 1 (verify-retry.md): wait for the log line itself, not just the + // state change that precedes it — retryWarmFill's own deferred func + // calls finishRefresh (which flips hasModel/discoveryHealthy, above) + // BEFORE it calls m.infof, on the SAME goroutine but with no + // synchronization the test can observe from outside; a single read + // immediately after the waitUntil above can race that write and see + // no line yet. + waitUntil(t, 2*time.Second, func() bool { + for _, line := range infoLog.linesCopy() { + if strings.Contains(line, "recovered after retry") { + return true + } + } + return false + }) +} + +// TestModelRegistry_WarmFillRetry_RetryAlsoFails_FallsBackToExistingBreakerBehavior +// proves the retry adds no special-casing to the breaker: two consecutive +// failures (the warm fill, then its retry) against a threshold-2 breaker +// open it exactly as two ordinary consecutive discovery failures already +// would, and the existing ERROR line (not the success INFO line) is what +// logs. +func TestModelRegistry_WarmFillRetry_RetryAlsoFails_FallsBackToExistingBreakerBehavior(t *testing.T) { + t.Parallel() + fa := newFakeAdapter("openai") + var calls int32 + fa.listModelsFn = func(context.Context) ([]string, error) { + atomic.AddInt32(&calls, 1) + return nil, fmt.Errorf("upstream still down") + } + adapters := map[string]providerAdapter{"openai": fa} + cfg := &Config{ + Providers: map[string]*ProviderConfig{"openai": {Models: []string{"gpt-a"}, Discovery: true}}, + Breaker: BreakerConfig{FailureThreshold: 2, OpenDuration: "1m", MaxOpenDuration: "10m"}, + } + rl := &recordingLog{} + reg, err := newModelRegistry(adapters, cfg, rl.fn) + if err != nil { + t.Fatalf("newModelRegistry: %v", err) + } + infoLog := &recordingLog{} + reg.info = infoLog.fn + reg.warmFillRetryDelay = 10 * time.Millisecond + + reg.warmFill(context.Background()) + + waitUntil(t, 2*time.Second, func() bool { + return atomic.LoadInt32(&calls) == 2 + }) + waitUntil(t, 2*time.Second, func() bool { + return !reg.discoveryHealthy("openai") + }) + + if reg.states["openai"].hasModel("gpt-a") == false { + t.Error(`hasModel("gpt-a") = false, want true — the explicit model must survive two failed discovery attempts (stale-while-error)`) + } + _, _, lastErr, _, _ := reg.states["openai"].snapshot() + if lastErr == "" { + t.Error("lastErr = \"\" after the retry also failed, want the retry's own error message") + } + // Fix 1 (verify-retry.md): wait for the ERROR line itself rather than + // reading it once right after the state-change waits above — + // retryWarmFill's deferred func writes the log line AFTER finishRefresh + // (which is what discoveryHealthy/lastErr above already observed), on + // the SAME goroutine but with no synchronization the test can see from + // outside; a single read can race that write and find nothing yet. + waitUntil(t, 2*time.Second, func() bool { + for _, line := range rl.linesCopy() { + if strings.Contains(line, "warm-fill retry") && strings.Contains(line, "failed") { + return true + } + } + return false + }) + if len(infoLog.linesCopy()) != 0 { + t.Errorf("info log = %v, want no INFO line — the retry itself failed, nothing recovered", infoLog.linesCopy()) + } +} + +// TestModelRegistry_WarmFillRetry_NoRetry_WhenWarmFillSucceeds proves +// scheduleWarmFillRetry is never invoked for a provider whose synchronous +// warm fill already succeeded: listModels must not be called a second +// time just because a retry window happened to elapse. +func TestModelRegistry_WarmFillRetry_NoRetry_WhenWarmFillSucceeds(t *testing.T) { + t.Parallel() + fa := newFakeAdapter("openai") + var calls int32 + fa.listModelsFn = func(context.Context) ([]string, error) { + atomic.AddInt32(&calls, 1) + return []string{"gpt-a"}, nil + } + adapters := map[string]providerAdapter{"openai": fa} + cfg := &Config{Providers: map[string]*ProviderConfig{"openai": {Discovery: true}}} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + if err != nil { + t.Fatalf("newModelRegistry: %v", err) + } + reg.warmFillRetryDelay = 15 * time.Millisecond + + reg.warmFill(context.Background()) + + // Sleep comfortably past the retry window: proving an ABSENCE of a + // second call needs to outlast the delay that would have fired it, + // not just poll for a positive condition. + time.Sleep(10 * reg.warmFillRetryDelay) + + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("listModels called %d times after a successful warm fill, want exactly 1 (no retry scheduled)", got) + } +} + +// TestModelRegistry_WarmFillRetry_BoundedByOwnTimeout_NoGoroutineLeak +// proves the retry goroutine is bounded by its own context timeout +// (m.warmFillTimeout) rather than depending on any external shutdown +// signal — this plugin has none (llmgateway.go's Close doc comment) — so +// an upstream that never responds at all still cannot leak the goroutine +// forever: retryWarmFill's own context.WithTimeout cuts it off and +// finishRefresh still runs. +func TestModelRegistry_WarmFillRetry_BoundedByOwnTimeout_NoGoroutineLeak(t *testing.T) { + t.Parallel() + fa := newFakeAdapter("openai") + var calls int32 + fa.listModelsFn = func(ctx context.Context) ([]string, error) { + if atomic.AddInt32(&calls, 1) == 1 { + return nil, fmt.Errorf("cold-start egress timeout") + } + // The retry attempt: never returns on its own, exactly like a + // hung upstream connection — only ctx's own timeout can end it. + <-ctx.Done() + return nil, ctx.Err() + } + adapters := map[string]providerAdapter{"openai": fa} + cfg := &Config{Providers: map[string]*ProviderConfig{"openai": {Discovery: true}}} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + if err != nil { + t.Fatalf("newModelRegistry: %v", err) + } + reg.warmFillRetryDelay = 20 * time.Millisecond + // Shortened from the package default (5s, registry.go's warmFillTimeout + // const) via the injectable m.warmFillTimeout field: this test only + // needs the retry's own context to expire, not the production budget, + // so there is no reason to actually wait out 5s of real time here. + reg.warmFillTimeout = 50 * time.Millisecond + + // Fix 4b (verify-retry.md): capture the goroutine count BEFORE the + // retry's own goroutine (time.AfterFunc, fired warmFillRetryDelay from + // now) exists at all, so the "back to baseline" poll below actually + // proves it exited rather than merely proving something else did. + runtime.GC() + before := runtime.NumGoroutine() + + reg.warmFill(context.Background()) + + // warmFillRetryDelay (20ms) + reg.warmFillTimeout (50ms, shortened + // above) + slack for a loaded (e.g. -race) test machine. + waitUntil(t, 2*time.Second, func() bool { + _, _, lastErr, _, _ := reg.states["openai"].snapshot() + return strings.Contains(lastErr, context.DeadlineExceeded.Error()) + }) + + // The state change above (finishRefresh recording lastErr) happens + // BEFORE retryWarmFill's goroutine actually returns and exits — its + // deferred log call still runs after. Poll NumGoroutine back down to + // (at most) the baseline instead of asserting immediately, so the + // test proves the goroutine itself ends, not just that its result was + // recorded. + waitUntil(t, 2*time.Second, func() bool { + runtime.GC() + return runtime.NumGoroutine() <= before + }) +} + +// TestModelRegistry_WarmFillRetry_GuardSkipsWhenRefreshInFlightOrDone proves +// fix 2/3 (verify-retry.md): retryWarmFill's tryBeginWarmFillRetry guard +// makes it skip entirely — never calling listModels a second time — both +// while a request-driven refresh (maybeRefresh/refreshProvider, via +// tryBeginRefresh) is already in flight against the SAME provider, and +// after one has already completed since the warm fill that scheduled this +// retry. Exercised by calling refreshProvider/retryWarmFill directly +// (not by racing real timers against each other), so the assertion is +// deterministic rather than timing-dependent — this is also, structurally, +// the "concurrent request-driven refresh + retry" scenario: both paths +// gate on the identical st.mu-guarded inFlight field, so at most one +// listModels call is ever in progress for this provider at a time. +func TestModelRegistry_WarmFillRetry_GuardSkipsWhenRefreshInFlightOrDone(t *testing.T) { + t.Parallel() + fa := newFakeAdapter("openai") + inProgress := make(chan struct{}) + release := make(chan struct{}) + var calls, current, maxConcurrent int32 + fa.listModelsFn = func(context.Context) ([]string, error) { + atomic.AddInt32(&calls, 1) + n := atomic.AddInt32(¤t, 1) + for { + old := atomic.LoadInt32(&maxConcurrent) + if n <= old || atomic.CompareAndSwapInt32(&maxConcurrent, old, n) { + break + } + } + close(inProgress) + <-release + atomic.AddInt32(¤t, -1) + return []string{"gpt-a"}, nil + } + adapters := map[string]providerAdapter{"openai": fa} + cfg := &Config{Providers: map[string]*ProviderConfig{"openai": {Discovery: true}}} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + if err != nil { + t.Fatalf("newModelRegistry: %v", err) + } + st := reg.states["openai"] + warmFillTime := reg.now() + + // Simulate a request-driven refresh already claiming inFlight — exactly + // what maybeRefresh's own tryBeginRefresh call, ahead of spawning + // refreshProvider, already does in production — and let it block + // mid-fetch so a concurrent retry has something real to race against. + if !st.tryBeginRefresh(reg.now()) { + t.Fatal("tryBeginRefresh on a fresh provider must succeed") + } + go reg.refreshProvider("openai", st, fa) + + select { + case <-inProgress: + case <-time.After(2 * time.Second): + t.Fatal("refreshProvider's listModels call never started") + } + + // The retry must skip: a refresh is already inFlight. + reg.retryWarmFill("openai", st, fa, warmFillTime) + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("calls = %d after the in-flight guard should have skipped the retry, want 1 (only the request-driven refresh)", got) + } + + close(release) + waitUntil(t, 2*time.Second, func() bool { + return st.hasModel("gpt-a") + }) + + // The retry must ALSO skip now that the refresh has completed: its + // lastRefresh has already moved past warmFillTime. + reg.retryWarmFill("openai", st, fa, warmFillTime) + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("calls = %d after the retry should have skipped (a refresh already completed since the warm fill), want 1", got) + } + if got := atomic.LoadInt32(&maxConcurrent); got > 1 { + t.Errorf("max concurrent listModels calls = %d, want at most 1 — the retry must never overlap a request-driven refresh", got) + } +} + +// TestModelRegistry_WarmFillRetry_NoRetry_WhenParentCtxCanceled proves fix +// 4a (verify-retry.md): a warm fill that fails because the PARENT ctx +// (newGateway's own construction context) was already canceled — a +// shutdown mid-construction, not a slow or broken upstream — must not +// schedule a retry at all. +func TestModelRegistry_WarmFillRetry_NoRetry_WhenParentCtxCanceled(t *testing.T) { + t.Parallel() + fa := newFakeAdapter("openai") + var calls int32 + fa.listModelsFn = func(ctx context.Context) ([]string, error) { + atomic.AddInt32(&calls, 1) + return nil, ctx.Err() + } + adapters := map[string]providerAdapter{"openai": fa} + cfg := &Config{Providers: map[string]*ProviderConfig{"openai": {Discovery: true}}} + reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {}) + if err != nil { + t.Fatalf("newModelRegistry: %v", err) + } + reg.warmFillRetryDelay = 15 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // parent already canceled — simulates a shutdown mid-construction + + reg.warmFill(ctx) + + // Sleep comfortably past the retry window: proving an ABSENCE of a + // scheduled retry needs to outlast the delay that would have fired + // it, not just poll for a positive condition — same shape as + // NoRetry_WhenWarmFillSucceeds above. + time.Sleep(10 * reg.warmFillRetryDelay) + + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("listModels called %d times after a parent-ctx-canceled warm fill, want exactly 1 (no retry scheduled)", got) + } +} + // --- maybeRefresh: throttling, in-flight guard, stale-while-error --- func TestModelRegistry_MaybeRefresh_Throttled_OnePerInterval(t *testing.T) { diff --git a/tools/yaegi-check/main.go b/tools/yaegi-check/main.go index 7edbed8..4dc784d 100644 --- a/tools/yaegi-check/main.go +++ b/tools/yaegi-check/main.go @@ -505,6 +505,31 @@ func run() error { })) defer brkUpstream.Close() + // warmRetryUpstream backs exerciseWarmFillRetry (production incident + // fix, registry.go's warmFillRetryDelay): 401s on the FIRST /v1/models + // hit only — the synchronous warmFill call inside New() itself — then + // serves real data on every hit after, so the scheduled background + // retry (and ONLY the retry) succeeds. Failing fast rather than + // sleeping past warmFillTimeout keeps this probe's cost down to + // roughly warmFillRetryDelay, not warmFillTimeout+warmFillRetryDelay. + warmRetryHits := new(int64) + warmRetryUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/v1/models") { + if atomic.AddInt64(warmRetryHits, 1) == 1 { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte(`{"error":{"message":"cold-start timeout","type":"invalid_request_error"}}`)) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"object":"list","data":[{"id":"warmretry-model"}]}`)) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"c-warmretry","object":"chat.completion","model":"warmretry-model","choices":[],"usage":{}}`)) + })) + defer warmRetryUpstream.Close() + // failoverAUpstream/failoverBUpstream back exerciseFailover (feat/ // failover, below): two providers configured with the IDENTICAL bare // model id (failoverModelID) — the exact shape registry.go's bareWinner @@ -674,7 +699,15 @@ func run() error { attemptAccountingOverride := `{"failover":{"enabled":true},"breaker":{"failureThreshold":2,"openDuration":"3s","maxOpenDuration":"6s"},` + `"providers":{"openai":{"type":"openai","baseUrl":"` + upstream.URL + `","apiKey":"sk-up","models":["` + testDataWantModel + `","` + builtinLookupModelID + `"]},` + - `"brk":{"type":"openai","baseUrl":"` + brkUpstream.URL + `","apiKey":"sk-up","discovery":true,"discoveryInterval":"1ms"},"` + + `"brk":{"type":"openai","baseUrl":"` + brkUpstream.URL + `","apiKey":"sk-up","discovery":true,"discoveryInterval":"1ms"},` + + // "warmretry" backs exerciseWarmFillRetry (production incident fix, + // registry.go's warmFillRetryDelay) — deliberately NO + // discoveryInterval override: the default (1h) keeps maybeRefresh's + // own request-driven interval gate from ever firing again during + // this probe, once warmFill's failed synchronous attempt (or the + // retry that follows it) has set lastRefresh, so the retry is the + // ONLY thing that can produce warmRetryUpstream's second hit. + `"warmretry":{"type":"openai","baseUrl":"` + warmRetryUpstream.URL + `","apiKey":"sk-up","discovery":true},"` + slowProviderName + `":{"type":"openai","baseUrl":"` + slowUpstream.URL + `","apiKey":"sk-up","models":["` + slowProviderModel + `"]},` + // timeoutProviderName (feature: request timeout) — see its own // const block doc comment above for why this provider exists @@ -935,6 +968,14 @@ func run() error { if err := exerciseBreaker(handler, brkModelsHits, brkHealthy); err != nil { return err } + // exerciseWarmFillRetry (production incident fix, registry.go's + // warmFillRetryDelay) runs right after exerciseBreaker: it drives its + // own, dedicated "warmretry" provider/upstream, so it cannot perturb + // exerciseBreaker's own "brk"-specific hit counting, and nothing below + // depends on "warmretry"'s own hit count either. + if err := exerciseWarmFillRetry(handler, warmRetryHits); err != nil { + return err + } // exerciseMetricsRoute runs after exerciseBreaker has already settled: // GET /metrics runs registry.maybeRefresh like every other route // (ServeHTTP's own unconditional call at entry), and exerciseBreaker's @@ -1337,6 +1378,43 @@ func readHealth(handler http.Handler, name string) (string, error) { return "", fmt.Errorf("provider %q not in overview: %s", name, rec.Body.String()) } +// readProviderModels returns provider name's known model id set from the +// admin overview (registry.go's providerSnapshot.models, +// provider-model-accordion task) — decoded generically like readHealth +// above, for exerciseWarmFillRetry's own poll loop below. +func readProviderModels(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 nil, 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 nil, err + } + providers, _ := body["providers"].([]any) + for _, raw := range providers { + p, ok := raw.(map[string]any) + if !ok { + continue + } + if p["name"] != name { + continue + } + rawModels, _ := p["models"].([]any) + models := make([]string, 0, len(rawModels)) + for _, rm := range rawModels { + if s, ok := rm.(string); ok { + models = append(models, s) + } + } + return models, nil + } + return nil, 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 @@ -1413,6 +1491,57 @@ func exerciseBreaker(handler http.Handler, hits, healthy *int64) error { return nil } +// exerciseWarmFillRetry proves registry.go's warm-fill retry (production +// incident fix: a provider whose SYNCHRONOUS initial discovery fetch +// fails, at pod start, is left with zero discovered models until a +// scheduled background retry recovers it) actually schedules and runs +// under the REAL interpreter — time.AfterFunc, the retry's own +// context.WithTimeout budget, and the finishRefresh bookkeeping it shares +// with warmFill/refreshProvider all proven interpreted, not merely +// compiled, per this harness's own stated purpose. +// +// "warmretry" (attemptAccountingOverride, run()) 401s on its very first +// /v1/models hit — warmFill's own synchronous call, spent inside New() +// itself, well before this probe (or any other) runs — then serves real +// data on every hit after. Only the scheduled retry, firing after +// registry.go's warmFillRetryDelay (10s, not overridable from outside +// the package), can ever produce a second hit and populate +// "warmretry-model"; this probe does not assume anything about how much +// wall-clock time the OTHER probes ahead of it in run()'s orchestration +// already burned relative to that 10s window — by the time this runs the +// retry may already have completed, or may still be pending — it only +// polls until the recovered state is reached (or times out) and then +// checks hits landed on exactly 2, not more. +func exerciseWarmFillRetry(handler http.Handler, hits *int64) error { + // warmFillRetryDelay (10s) + warmFillTimeout (5s) + slack: the same + // budget shape exerciseBreaker's own phase-3 recovery wait above + // uses, just against this feature's fixed, non-configurable delay + // instead of a config-driven breaker backoff. + var models []string + var err error + recovered := false + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + models, err = readProviderModels(handler, "warmretry") + if err != nil { + return err + } + if len(models) == 1 && models[0] == "warmretry-model" { + recovered = true + break + } + time.Sleep(100 * time.Millisecond) + } + if !recovered { + return fmt.Errorf("WARMFILLRETRY-1: provider %q never showed the retried model %q within 20s (hits=%d, models=%v)", "warmretry", "warmretry-model", atomic.LoadInt64(hits), models) + } + if got := atomic.LoadInt64(hits); got != 2 { + return fmt.Errorf("WARMFILLRETRY-2: warmRetryUpstream saw %d /v1/models hits once recovered, want exactly 2 (the failed warm fill plus one retry — no maybeRefresh interference)", got) + } + fmt.Println("yaegi-check: warm-fill retry recovered the provider's discovered models after the scheduled delay") + return nil +} + // exerciseMetricsRoute drives GET /metrics against the interpreted // handler and proves both its auth gate and its rendering run correctly // under Yaegi — the metrics feature's own gate requirement: compiled