mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-13 04:58:50 +00:00
fix(registry): recover refresh goroutine, per-provider warm-fill budget, exact-glob model authorization
1. refreshProvider (the plugin's only background goroutine) had no recover(): a panicking adapter's listModels would crash the whole Traefik process. Restructured around deferred recover + finishRefresh, so a panic is turned into a logged error, inFlight always releases, and the next interval retries. Documented as a scoped exception to limits.go's no-background-goroutine rule. 2. newGateway wrapped ctx in an outer 5s WithTimeout before warmFill, but warmFill already derives a fresh 5s budget per provider from whatever ctx it receives. The outer wrap made that a single budget shared across every discovery-enabled provider instead of a per-provider one - providers after the first could get a near-expired context and fail instantly. Dropped the outer wrapper; construction's worst case is now N * 5s, accepted. 3. group.allowsModel stripped at any first slash, so a glob like 'deepseek-*' could spuriously match a bare model id that merely contains a slash (e.g. an upstream's own 'uni/deepseek-v4-flash-0731' naming) even though the leading segment names no configured provider. allowsModel is now an exact glob with no internal splitting; resolveAgainst and listFor generate the provider-prefixed and bare-suffix candidates themselves, only when a prefix legitimately names a configured provider. Updated the Task-3 TestIdentify assertion to match. 4. Annotated the discarded json.Encode error in handleModels. Gate: go test ./... -count=1, -race, gofmt, vet, fieldalignment, golangci-lint (incl gosec) all clean.
This commit is contained in:
@@ -32,18 +32,19 @@ func (grp *group) allowsProvider(name string) bool {
|
||||
}
|
||||
|
||||
// allowsModel reports whether id matches one of the group's model glob
|
||||
// patterns. An empty pattern list allows every model. A pattern is checked
|
||||
// against id as given and, when id has a "provider/model" form, against the
|
||||
// part after the "/" — so a pattern like "gpt-5*" matches both "gpt-5-mini"
|
||||
// and "openai/gpt-5-mini".
|
||||
// patterns, exactly as given. An empty pattern list allows every model.
|
||||
//
|
||||
// allowsModel does no "provider/model" prefix-splitting of its own — a
|
||||
// pattern like "deepseek-*" must not spuriously match a bare model id that
|
||||
// merely happens to contain a slash (e.g. an upstream's own
|
||||
// "uni/deepseek-v4-flash-0731" naming, where "uni" is not a configured
|
||||
// provider: there is no legitimate prefix to strip there at all). A caller
|
||||
// that genuinely knows an id's leading segment names a configured provider
|
||||
// (modelRegistry.resolveAgainst, modelRegistry.listFor) generates both the
|
||||
// full and bare-suffix candidate strings itself and calls allowsModel once
|
||||
// per candidate.
|
||||
func (grp *group) allowsModel(id string) bool {
|
||||
if matchesGlob(grp.models, id) {
|
||||
return true
|
||||
}
|
||||
if _, bare, found := strings.Cut(id, "/"); found {
|
||||
return matchesGlob(grp.models, bare)
|
||||
}
|
||||
return false
|
||||
return matchesGlob(grp.models, id)
|
||||
}
|
||||
|
||||
// allowsMCP reports whether name matches one of the group's MCP-server glob
|
||||
|
||||
+8
-2
@@ -30,8 +30,14 @@ func TestIdentify(t *testing.T) {
|
||||
if !grp.allowsModel("gpt-5-mini") {
|
||||
t.Error("glob should match")
|
||||
}
|
||||
if !grp.allowsModel("openai/gpt-5-mini") {
|
||||
t.Error("prefixed form should match")
|
||||
// allowsModel is an exact glob against the given string only — it no
|
||||
// longer strips a "provider/" prefix itself (fix(registry) ruling 3): a
|
||||
// pattern like "gpt-5*" reaching a provider-prefixed request like
|
||||
// "openai/gpt-5-mini" is now the caller's job (modelRegistry.
|
||||
// resolveAgainst / listFor build the bare-suffix candidate themselves),
|
||||
// covered at the registry level in registry_test.go, not here.
|
||||
if grp.allowsModel("openai/gpt-5-mini") {
|
||||
t.Error("allowsModel must not match a provider-prefixed id on its own; prefix-stripping now lives in the caller")
|
||||
}
|
||||
if grp.allowsModel("claude-4") {
|
||||
t.Error("should not match")
|
||||
|
||||
+9
-3
@@ -153,9 +153,15 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
warmCtx, warmCancel := context.WithTimeout(ctx, warmFillTimeout)
|
||||
defer warmCancel()
|
||||
registry.warmFill(warmCtx)
|
||||
// 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
|
||||
// single shared warmFillTimeout here would make that per-provider budget
|
||||
// a lie — provider 2..N would inherit whatever is left of the first
|
||||
// provider's deadline instead of a full one. Construction's worst case is
|
||||
// therefore N * warmFillTimeout (N = discovery-enabled providers), which
|
||||
// is accepted.
|
||||
registry.warmFill(ctx)
|
||||
g.registry = registry
|
||||
|
||||
setPricingWarnFn(func(msg string) {
|
||||
|
||||
+47
-11
@@ -239,14 +239,38 @@ func (m *modelRegistry) maybeRefresh(ctx context.Context) {
|
||||
// already marked inFlight by tryBeginRefresh, and records its outcome via
|
||||
// finishRefresh. Split out from maybeRefresh so the goroutine closes over
|
||||
// named parameters, not loop variables.
|
||||
//
|
||||
// This goroutine (maybeRefresh's only "go" statement in the whole plugin)
|
||||
// is a deliberate, narrow exception to memoryStore's no-background-goroutine
|
||||
// rule (limits.go:144-149): unlike a ticker that would outlive a Yaegi
|
||||
// middleware instance rebuilt on every config reload, this goroutine is
|
||||
// self-terminating — at most one per provider per interval
|
||||
// (tryBeginRefresh's inFlight guard), bounded to backgroundRefreshTimeout
|
||||
// (30s). A rebuild mid-flight can transiently leak one goroutine for at
|
||||
// most 30s; that is accepted, not a steady-state leak.
|
||||
//
|
||||
// The deferred recover below is required precisely because this runs off
|
||||
// any request's goroutine: an unrecovered panic here has no ServeHTTP
|
||||
// caller to unwind into and would crash the whole Traefik process, not just
|
||||
// fail one request. finishRefresh still runs on the panic path (with a
|
||||
// synthetic error) so inFlight is always released and the next interval
|
||||
// retries, exactly as a normal fetch failure would.
|
||||
func (m *modelRegistry) refreshProvider(name string, st *providerState, adapter providerAdapter) {
|
||||
var ids []string
|
||||
var err error
|
||||
defer func() {
|
||||
if rec := recover(); rec != nil {
|
||||
err = fmt.Errorf("panic: %v", rec)
|
||||
}
|
||||
st.finishRefresh(m.now(), ids, err)
|
||||
if err != nil {
|
||||
m.log("model registry: discovery refresh for provider %q failed: %v", name, err)
|
||||
}
|
||||
}()
|
||||
|
||||
fctx, cancel := context.WithTimeout(context.Background(), backgroundRefreshTimeout)
|
||||
defer cancel()
|
||||
ids, err := adapter.listModels(fctx)
|
||||
st.finishRefresh(m.now(), ids, err)
|
||||
if err != nil {
|
||||
m.log("model registry: discovery refresh for provider %q failed: %v", name, err)
|
||||
}
|
||||
ids, err = adapter.listModels(fctx)
|
||||
}
|
||||
|
||||
// splitConfiguredProvider reports whether id has "prefix/rest" form where
|
||||
@@ -314,15 +338,24 @@ func (m *modelRegistry) resolve(id string, grp *group) (providerAdapter, string,
|
||||
// upstreamModel is actually known to that provider, then checks
|
||||
// authorization, in that order — so a real model behind a provider the
|
||||
// group cannot use reports errModelDenied, not errModelUnknown.
|
||||
// requestedID is the id form passed to grp.allowsModel; it is the
|
||||
// provider-prefixed form for a direct request, or the bare id for a
|
||||
// bare-id request — allowsModel already checks both a pattern's full-string
|
||||
// and bare-suffix match, so either form is checked correctly either way.
|
||||
//
|
||||
// allowsModel is an exact glob match (auth.go) with no prefix-stripping of
|
||||
// its own, so resolveAgainst generates both candidate strings itself:
|
||||
// requestedID (the provider-prefixed form for a direct request, e.g.
|
||||
// "openai/gpt-test") and upstreamModel (its bare suffix, e.g. "gpt-test"),
|
||||
// and allows if either matches — that is what lets a pattern like "gpt-*"
|
||||
// reach a provider-prefixed request. For a bare-id request the caller
|
||||
// passes the same string as both arguments, so the two checks collapse to
|
||||
// one: no bare-suffix candidate is invented for an id that was never
|
||||
// legitimately provider-prefixed in the first place (ruling: a pattern like
|
||||
// "deepseek-*" must not match a bare id that merely contains a slash, e.g.
|
||||
// "uni/deepseek-v4-flash-0731", when "uni" is not a configured provider).
|
||||
func (m *modelRegistry) resolveAgainst(providerName, upstreamModel, requestedID string, grp *group) (providerAdapter, string, string, error) {
|
||||
if !m.states[providerName].hasModel(upstreamModel) {
|
||||
return nil, "", "", errModelUnknown
|
||||
}
|
||||
if !grp.allowsModel(requestedID) || !grp.allowsProvider(providerName) {
|
||||
allowed := grp.allowsModel(requestedID) || grp.allowsModel(upstreamModel)
|
||||
if !allowed || !grp.allowsProvider(providerName) {
|
||||
return nil, "", "", errModelDenied
|
||||
}
|
||||
return m.adapters[providerName], upstreamModel, providerName + "/" + upstreamModel, nil
|
||||
@@ -370,7 +403,10 @@ func (m *modelRegistry) listFor(grp *group) []map[string]any {
|
||||
m.warnCollisionOnce(id, winner, provs)
|
||||
for _, p := range provs {
|
||||
pid := p + "/" + id
|
||||
if grp.allowsModel(pid) && grp.allowsProvider(p) {
|
||||
// allowsModel does no prefix-stripping (auth.go): check both the
|
||||
// prefixed form and its bare suffix, same as resolveAgainst does
|
||||
// for the equivalent client request.
|
||||
if (grp.allowsModel(pid) || grp.allowsModel(id)) && grp.allowsProvider(p) {
|
||||
out = append(out, modelObject(pid, p))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -228,6 +228,59 @@ func TestModelRegistry_Resolve_AllowedByBothProviderAndModel_Succeeds(t *testing
|
||||
}
|
||||
}
|
||||
|
||||
// --- resolve: exact-glob model authorization, no spurious prefix-strip
|
||||
// (fix(registry) ruling 3) ---
|
||||
|
||||
// TestModelRegistry_Resolve_BareIDWithSlash_ModelGlobMustNotMatchViaSpuriousStrip
|
||||
// is the regression: allowsModel used to strip at any first slash, so a
|
||||
// pattern like "deepseek-*" would spuriously match a bare model id that
|
||||
// merely contains a slash — an upstream's own "uni/deepseek-v4-flash-0731"
|
||||
// naming, where "uni" is not a configured provider and there is no
|
||||
// legitimate prefix to strip at all. resolveAgainst must not invent that
|
||||
// candidate for a request that was never provider-prefixed in the first
|
||||
// place.
|
||||
func TestModelRegistry_Resolve_BareIDWithSlash_ModelGlobMustNotMatchViaSpuriousStrip(t *testing.T) {
|
||||
t.Parallel()
|
||||
adapters := map[string]providerAdapter{"acme": newFakeAdapter("acme")}
|
||||
cfg := &Config{Providers: map[string]*ProviderConfig{"acme": {Models: []string{"uni/deepseek-v4-flash-0731"}}}}
|
||||
reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {})
|
||||
if err != nil {
|
||||
t.Fatalf("newModelRegistry: %v", err)
|
||||
}
|
||||
|
||||
const id = "uni/deepseek-v4-flash-0731"
|
||||
grp := &group{name: "narrow", models: []string{"deepseek-*"}}
|
||||
if _, _, _, err := reg.resolve(id, grp); err != errModelDenied {
|
||||
t.Errorf("err = %v, want errModelDenied (deepseek-* must not spuriously match via a bare-suffix strip)", err)
|
||||
}
|
||||
|
||||
// Positive control: a pattern that legitimately targets the full bare
|
||||
// id, slash included, still matches via plain exact-glob.
|
||||
grp.models = []string{"uni/deepseek*"}
|
||||
if _, _, _, err := reg.resolve(id, grp); err != nil {
|
||||
t.Errorf("resolve with a full-string-matching pattern: %v, want success", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestModelRegistry_Resolve_ProviderPrefixed_ModelGlobMatchesViaBareCandidate
|
||||
// is the positive half of the same ruling: a provider-prefixed request's
|
||||
// bare suffix is still checked against the model glob — resolveAgainst
|
||||
// generates that candidate itself now that allowsModel no longer strips.
|
||||
func TestModelRegistry_Resolve_ProviderPrefixed_ModelGlobMatchesViaBareCandidate(t *testing.T) {
|
||||
t.Parallel()
|
||||
adapters := map[string]providerAdapter{"openai": newFakeAdapter("openai")}
|
||||
cfg := &Config{Providers: map[string]*ProviderConfig{"openai": {Models: []string{"gpt-test"}}}}
|
||||
reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {})
|
||||
if err != nil {
|
||||
t.Fatalf("newModelRegistry: %v", err)
|
||||
}
|
||||
|
||||
grp := &group{name: "narrow", models: []string{"gpt-*"}}
|
||||
if _, _, _, err := reg.resolve("openai/gpt-test", grp); err != nil {
|
||||
t.Errorf("resolve: %v, want success (gpt-* must match via the bare-suffix candidate)", err)
|
||||
}
|
||||
}
|
||||
|
||||
// --- resolve: collision precedence (ruling g) ---
|
||||
|
||||
func TestModelRegistry_Resolve_BareIDCollision_PicksSortedFirstProvider(t *testing.T) {
|
||||
@@ -374,6 +427,54 @@ func TestModelRegistry_WarmFill_NonDiscoveryProvider_NeverCallsListModels(t *tes
|
||||
}
|
||||
}
|
||||
|
||||
// TestModelRegistry_WarmFill_EachProviderGetsFullBudget_NotSharedAcrossProviders
|
||||
// is the regression for llmgateway.go's old outer 5s wrap around the whole
|
||||
// warmFill call (fix(registry) ruling 2): that outer wrap made
|
||||
// warmFillTimeout a total budget shared across every discovery-enabled
|
||||
// provider instead of a fresh one per provider, so a slow first provider
|
||||
// ate into a later provider's deadline and could leave it resolving
|
||||
// nothing for the whole first interval. warmFill derives each provider's
|
||||
// timeout via context.WithTimeout(ctx, warmFillTimeout) fresh per
|
||||
// iteration; this proves that yields close to the full budget for a later
|
||||
// provider even after an earlier one took real wall-clock time, as long as
|
||||
// ctx itself (as llmgateway.go now passes it) carries no deadline of its
|
||||
// own.
|
||||
func TestModelRegistry_WarmFill_EachProviderGetsFullBudget_NotSharedAcrossProviders(t *testing.T) {
|
||||
t.Parallel()
|
||||
slow := newFakeAdapter("a-slow")
|
||||
slow.listModelsFn = func(context.Context) ([]string, error) {
|
||||
time.Sleep(1200 * time.Millisecond) // wall-clock cost, not budget shared with b-fast
|
||||
return []string{"slow-model"}, nil
|
||||
}
|
||||
var fastRemaining time.Duration
|
||||
fast := newFakeAdapter("b-fast")
|
||||
fast.listModelsFn = func(ctx context.Context) ([]string, error) {
|
||||
if dl, ok := ctx.Deadline(); ok {
|
||||
fastRemaining = time.Until(dl)
|
||||
}
|
||||
return []string{"fast-model"}, nil
|
||||
}
|
||||
adapters := map[string]providerAdapter{"a-slow": slow, "b-fast": fast}
|
||||
cfg := &Config{Providers: map[string]*ProviderConfig{
|
||||
// Sorted provider-name order drives warmFill's iteration: "a-slow"
|
||||
// runs before "b-fast" for this test to be meaningful.
|
||||
"a-slow": {Discovery: true},
|
||||
"b-fast": {Discovery: true},
|
||||
}}
|
||||
reg, err := newModelRegistry(adapters, cfg, func(string, ...any) {})
|
||||
if err != nil {
|
||||
t.Fatalf("newModelRegistry: %v", err)
|
||||
}
|
||||
|
||||
// Mirrors llmgateway.go's fixed call site: an unbounded parent ctx, not
|
||||
// one pre-wrapped to warmFillTimeout.
|
||||
reg.warmFill(context.Background())
|
||||
|
||||
if fastRemaining < 4*time.Second {
|
||||
t.Errorf("b-fast's received context deadline was %v from expiry, want >4s (its 5s warm-fill budget must not be reduced by a-slow's 1.2s elapsed time)", fastRemaining)
|
||||
}
|
||||
}
|
||||
|
||||
// --- maybeRefresh: throttling, in-flight guard, stale-while-error ---
|
||||
|
||||
func TestModelRegistry_MaybeRefresh_Throttled_OnePerInterval(t *testing.T) {
|
||||
@@ -452,6 +553,59 @@ func TestModelRegistry_MaybeRefresh_InFlightGuard_BlocksConcurrentFetch(t *testi
|
||||
waitUntil(t, time.Second, func() bool { return atomic.LoadInt32(&calls) == 1 && reg.states["openai"].hasModel("m1") })
|
||||
}
|
||||
|
||||
// TestModelRegistry_RefreshProvider_AdapterPanics_RecoversAndReleasesInFlight
|
||||
// is the regression for the missing recover() in refreshProvider (fix
|
||||
// (registry) ruling 1): refreshProvider's "go" statement is the plugin's
|
||||
// only background goroutine, running off any request's stack. An
|
||||
// unrecovered panic there would crash the whole Traefik process, not just
|
||||
// fail one refresh — there is no ServeHTTP caller above it to catch it.
|
||||
// Reaching the end of this test at all is part of the proof: if recover()
|
||||
// were missing, the panicking listModels call below would have already
|
||||
// crashed the test binary.
|
||||
func TestModelRegistry_RefreshProvider_AdapterPanics_RecoversAndReleasesInFlight(t *testing.T) {
|
||||
t.Parallel()
|
||||
var calls int32
|
||||
fa := newFakeAdapter("openai")
|
||||
fa.listModelsFn = func(context.Context) ([]string, error) {
|
||||
n := atomic.AddInt32(&calls, 1)
|
||||
if n == 1 {
|
||||
panic("boom: adapter blew up")
|
||||
}
|
||||
return []string{"m1"}, nil
|
||||
}
|
||||
adapters := map[string]providerAdapter{"openai": fa}
|
||||
cfg := &Config{Providers: map[string]*ProviderConfig{"openai": {Discovery: true, DiscoveryInterval: "1h"}}}
|
||||
rl := &recordingLog{}
|
||||
reg, err := newModelRegistry(adapters, cfg, rl.fn)
|
||||
if err != nil {
|
||||
t.Fatalf("newModelRegistry: %v", err)
|
||||
}
|
||||
now := time.Now()
|
||||
reg.nowFn = func() time.Time { return now }
|
||||
|
||||
reg.maybeRefresh(context.Background())
|
||||
// The log write happens strictly after finishRefresh inside
|
||||
// refreshProvider's deferred recover handler (both under the same
|
||||
// goroutine, in that order), so waiting for it also proves inFlight was
|
||||
// released and lastRefresh advanced — not merely that the panic
|
||||
// occurred. recordingLog's own mutex gives this a proper happens-before
|
||||
// edge with the write, so this is race-safe without touching
|
||||
// providerState's fields directly.
|
||||
waitUntil(t, time.Second, func() bool { return rl.count() > 0 })
|
||||
if got := atomic.LoadInt32(&calls); got != 1 {
|
||||
t.Fatalf("listModels called %d times, want exactly 1 before the panic", got)
|
||||
}
|
||||
|
||||
// The guard was released, not left stuck true: a second attempt after
|
||||
// the interval elapses retries rather than being throttled forever.
|
||||
now = now.Add(2 * time.Hour)
|
||||
reg.maybeRefresh(context.Background())
|
||||
waitUntil(t, time.Second, func() bool { return reg.states["openai"].hasModel("m1") })
|
||||
if got := atomic.LoadInt32(&calls); got != 2 {
|
||||
t.Errorf("listModels called %d times after the retry window, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestModelRegistry_MaybeRefresh_DiscoveryError_KeepsPreviousDiscoveredSet(t *testing.T) {
|
||||
t.Parallel()
|
||||
var fail atomic.Bool
|
||||
|
||||
+1
-1
@@ -16,7 +16,7 @@ func (g *Gateway) handleModels(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{ // headers already committed; nothing useful to do on encode failure
|
||||
"object": "list",
|
||||
"data": g.registry.listFor(grp),
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user