diff --git a/CHANGELOG.md b/CHANGELOG.md index c339e9c..fb4dfc5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,10 @@ APIs change between minors without a deprecation period. an endpoint it does not measure as a queue, so an external model (OpenAI, another llmgw cluster) was held to what a saturation of 0.85 allows of "in flight / 4"; the rest waited and timed out with 504. An outside provider queues for itself and says 429 when full, which its breaker already hears: it no longer counts towards saturation. +- **A valid key told to slow down under load, every thirty seconds.** When a busy key's cached entry expired, every + request in flight missed the cache together and each took one of the address's allowance for unknown keys: past a + hundred or so streams at once, a valid key got 429 "too many unrecognised api keys". Requests missing the same key + at once now share one read and one allowance. ## v0.16.1 — 2026-10-30 diff --git a/internal/auth/auth.go b/internal/auth/auth.go index c425554..2eb8c98 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -15,6 +15,7 @@ import ( "time" "gitea.mynodes.xyz/ilker/llmgw/internal/store" + "golang.org/x/sync/singleflight" ) // Principal is who a request is, once authenticated. @@ -118,6 +119,12 @@ type Authenticator struct { // lookups rations the database reads of keys not in the cache, per // client address. lookups *buckets + + // resolving shares one database read between the requests that miss the + // cache for the same key at once: when a busy key's entry expired, every + // request in flight missed together, each took from the client's ration, + // and a valid key was told to slow down. + resolving singleflight.Group } type entry struct { @@ -166,26 +173,37 @@ func (a *Authenticator) AuthenticateFrom(ctx context.Context, presented, client return e.principal, e.err } - if client != "" && !a.lookups.allow(client, a.now()) { - return Principal{}, ErrThrottled - } - p, err := a.resolve(ctx, hash) - if err != nil && !isRejection(err) { - // A database problem is not an authentication answer, so it must not be - // cached as one. Failing the request is right — the alternative is - // serving traffic whose authorization is unknown. + // One read for every request missing this key at once, rationed once. + v, err, _ := a.resolving.Do(id, func() (any, error) { + if e, ok := a.lookup(id); ok { // filled while this waited its turn + return e, nil + } + if client != "" && !a.lookups.allow(client, a.now()) { + return nil, ErrThrottled + } + p, err := a.resolve(ctx, hash) + if err != nil && !isRejection(err) { + // A database problem is not an authentication answer, so it must + // not be cached as one. Failing the request is right — the + // alternative is serving traffic whose authorization is unknown. + return nil, err + } + ttl := a.cfg.TTL + if err != nil { + ttl = a.cfg.NegativeTTL + } + e := entry{principal: p, err: err, expires: a.now().Add(ttl)} + a.store_(id, e) + return e, nil + }) + if err != nil { return Principal{}, err } - - ttl := a.cfg.TTL - if err != nil { - ttl = a.cfg.NegativeTTL + e := v.(entry) + if e.err == nil { + a.markUsed(e.principal.KeyID) } - a.store_(id, entry{principal: p, err: err, expires: a.now().Add(ttl)}) - if err == nil { - a.markUsed(p.KeyID) - } - return p, err + return e.principal, e.err } func (a *Authenticator) resolve(ctx context.Context, hash []byte) (Principal, error) { diff --git a/internal/auth/auth_test.go b/internal/auth/auth_test.go index f335158..14ea2ab 100644 --- a/internal/auth/auth_test.go +++ b/internal/auth/auth_test.go @@ -478,3 +478,41 @@ func TestTheCacheDoesNotGrowWithoutBound(t *testing.T) { t.Errorf("%d entries", n) } } + +// A busy key whose entry has expired: every request in flight misses at once, +// and they share one read and one of the address's allowance. Each taking +// its own told a valid key to slow down, every thirty seconds, under load. +func TestABusyKeyMissingTheCacheTogetherIsReadOnceAndNotThrottled(t *testing.T) { + s, secret, _ := fixture(t) + c := &slowStore{countingStore: countingStore{Store: s}, delay: 50 * time.Millisecond} + a := New(c, DefaultConfig()) + var wg sync.WaitGroup + var throttled atomic.Int64 + for range 200 { + wg.Add(1) + go func() { + defer wg.Done() + if _, err := a.AuthenticateFrom(context.Background(), secret, "203.0.113.9"); errors.Is(err, ErrThrottled) { + throttled.Add(1) + } else if err != nil { + t.Error(err) + } + }() + } + wg.Wait() + if throttled.Load() != 0 || c.lookups.Load() != 1 { + t.Errorf("%d throttled, %d database reads", throttled.Load(), c.lookups.Load()) + } +} + +// slowStore is a countingStore whose reads take a while, so the requests +// missing together are still waiting when the first one asks. +type slowStore struct { + countingStore + delay time.Duration +} + +func (s *slowStore) KeyByHash(ctx context.Context, hash []byte) (store.Key, error) { + time.Sleep(s.delay) + return s.countingStore.KeyByHash(ctx, hash) +}