diff --git a/redis_store.go b/redis_store.go index 3ccf796..3adb51e 100644 --- a/redis_store.go +++ b/redis_store.go @@ -3,6 +3,7 @@ package traefikllmgateway import ( "fmt" "strconv" + "sync" "time" ) @@ -11,11 +12,120 @@ import ( // instead of each enforcing limits against its own in-process state. type redisStore struct { client *respClient + // expireSeen and expireOrder implement needsExpire/commitExpire/ + // forgetExpire's bounded, process-lifetime "does this key already + // have a confirmed EXPIRE on Redis" tracker (perf finding 2, + // 2026-08-2x audit) — see needsExpire's own doc comment. Guarded by + // expireMu since incrBy/incrMulti/incrAndGetMulti can run + // concurrently across goroutines sharing one redisStore (the limiter + // calls them from every request's goroutine). + expireSeen map[string]struct{} + expireOrder []string + expireMu sync.Mutex } // newRedisStore returns a redisStore using client for storage. func newRedisStore(client *respClient) *redisStore { - return &redisStore{client: client} + return &redisStore{client: client, expireSeen: make(map[string]struct{})} +} + +// expireOnceCacheMax bounds needsExpire/commitExpire's tracked key set: +// once it holds this many keys, the oldest-inserted is evicted (FIFO) to +// make room for the newest. An evicted key simply becomes "unseen" again +// — its next INCRBY sends a redundant-but-harmless EXPIRE, exactly as if +// this process had never touched it before. This bounds memory for a +// long-lived process against its own ever-rolling counter buckets +// (windowKey embeds a new bucket per window rollover — limits.go), not +// against any single window's real key count, so 4096 is generous +// headroom over any realistic per-process working set of concurrently +// hot counters. +const expireOnceCacheMax = 4096 + +// needsExpire reports whether key does NOT currently have a CONFIRMED +// EXPIRE recorded for it (perf finding 2, 2026-08-2x audit, deleting +// EXPIRE entirely measured as the upper bound: 74->43 cmds/req, p50 +// 867.9->697.9us (-19.6%), conc=16 throughput 1,477->2,336 req/s (+58%)). +// windowKey (limits.go) already embeds each counter's own time bucket in +// its key string, and the TTLs callers pass (dayWindowTTL=35d, +// monthWindowTTL=400d, ...) are deliberately far longer than their +// window's own natural length purely as a stale-key GC margin, not a +// sliding-window mechanism (limits.go's own doc comment on those +// constants) — so once EXPIRE is CONFIRMED sent for an exact key string, +// every further INCRBY against that SAME key happens well within the +// window that key's own bucket belongs to, comfortably inside the TTL's +// margin, and resending EXPIRE for it is redundant, not incorrect. +// +// needsExpire does NOT mark key as seen — unlike this method's +// predecessor (expireOnce), which marked a key seen at CHECK time, +// before its pipeline was ever sent. That was a bug (review finding, +// 2026-08-2x, CRITICAL): a pipeline carrying key's only EXPIRE can still +// fail after the mark — e.g. Redis is briefly unreachable, routine +// during a rollout — leaving this process believing EXPIRE succeeded +// when Redis never received it, and every later INCRBY against that key +// sends no EXPIRE ever again: an immortal, permanently TTL-less key +// (unbounded keyspace growth — precisely what dayWindowTTL/ +// monthWindowTTL exist to prevent — and under maxmemory-policy +// volatile-*, a TTL-less key is not evictable, so it can drive Redis to +// OOM-on-write). Marking now happens ONLY in commitExpire, called by the +// caller after confirming the pipeline that carried EXPIRE actually +// succeeded (commit-after-success). See commitExpire and forgetExpire +// for the other half of the fix: Redis can also independently lose an +// already-EXPIRE-confirmed key (restart, maxmemory eviction, FLUSHDB, a +// failover to a replica missing it) without this process's cache ever +// finding out on its own. +func (s *redisStore) needsExpire(key string) bool { + s.expireMu.Lock() + defer s.expireMu.Unlock() + _, ok := s.expireSeen[key] + return !ok +} + +// commitExpire records key as having a CONFIRMED EXPIRE on Redis — call +// only after the pipeline carrying key's EXPIRE command is known to have +// succeeded (client.pipeline returned a nil error). Bounded FIFO +// eviction, same reasoning as expireOnceCacheMax's own doc comment. +func (s *redisStore) commitExpire(key string) { + s.expireMu.Lock() + defer s.expireMu.Unlock() + if _, ok := s.expireSeen[key]; ok { + return + } + if len(s.expireOrder) >= expireOnceCacheMax { + oldest := s.expireOrder[0] + s.expireOrder = s.expireOrder[1:] + delete(s.expireSeen, oldest) + } + s.expireSeen[key] = struct{}{} + s.expireOrder = append(s.expireOrder, key) +} + +// forgetExpire removes key from the confirmed-EXPIRE set — called when an +// INCRBY reply reveals key did NOT actually exist before this call (its +// returned value equals exactly this call's own delta — Redis treats a +// missing key as 0 for INCRBY), even though this process's cache +// believed EXPIRE was already confirmed for it (review finding, +// 2026-08-2x, CRITICAL, path 2): Redis lost the key — restart, maxmemory +// eviction, FLUSHDB, or a failover to a replica that never had it — and +// this call's own INCRBY just recreated it with NO TTL, because the +// (stale) cache made the caller skip sending EXPIRE this round. +// +// Forgetting key here does not fix THIS round's missing TTL — the +// decision to skip EXPIRE was already made and sent before this reply +// arrived — but it makes the VERY NEXT increment for key resend EXPIRE +// (needsExpire will report true again), bounding the immortal-key window +// to at most one skipped cycle instead of forever. An active counter (the +// only kind this matters for) gets incremented again almost immediately, +// so this self-heals in practice. +func (s *redisStore) forgetExpire(key string) { + s.expireMu.Lock() + defer s.expireMu.Unlock() + delete(s.expireSeen, key) + // expireOrder may still carry a stale entry for key; left as-is. The + // only consequence is commitExpire's FIFO eviction occasionally + // evicting a key one step early if it is re-committed at a new + // position later — itself always safe (a spurious extra EXPIRE, the + // same accepted cost as any other eviction), so a full scan-and-purge + // here would add complexity for no correctness benefit. } // ttlToSeconds converts ttl to whole Redis EXPIRE seconds, rounded up, @@ -32,12 +142,20 @@ func ttlToSeconds(ttl time.Duration) int64 { return s } -// incrBy implements counterStore: INCRBY key n, then EXPIRE key -// ttlToSeconds(ttl), sent as one pipeline so the two commands share a -// single round trip. It returns INCRBY's resulting counter value; -// EXPIRE's reply is not otherwise inspected — a failed EXPIRE right after -// a successful INCRBY on the same key would only leave that key without a -// fresh TTL, not corrupt the count. +// incrBy implements counterStore: INCRBY key n, then (only when +// needsExpire(key) is true) EXPIRE key ttlToSeconds(ttl), sent as one +// pipeline so both commands share a single round trip when EXPIRE is +// included. It returns INCRBY's resulting counter value; EXPIRE's reply +// is not otherwise inspected — a failed EXPIRE right after a successful +// INCRBY on the same key would only leave that key without a fresh TTL, +// not corrupt the count. +// +// expireSeen bookkeeping (perf finding 2, review fix, CRITICAL, +// 2026-08-2x) happens strictly AFTER the pipeline succeeds — never +// before: commitExpire(key) only when this call actually sent (and +// confirmed) EXPIRE; forgetExpire(key) when the INCRBY reply reveals key +// was recreated (v == n) despite this process's cache believing it +// already had a confirmed EXPIRE — see both methods' own doc comments. // // Semantics are deliberately at-least-once, not exactly-once, both in the // conservative direction (never under-counts): respClient.pipeline's @@ -51,10 +169,12 @@ func ttlToSeconds(ttl time.Duration) int64 { // that occasionally over-counts by one request's worth is fail-safe (more // restrictive than reality), never fail-open in the unsafe direction. func (s *redisStore) incrBy(key string, n int64, ttl time.Duration) (int64, error) { - replies, err := s.client.pipeline([][]string{ - {"INCRBY", key, strconv.FormatInt(n, 10)}, - {"EXPIRE", key, strconv.FormatInt(ttlToSeconds(ttl), 10)}, - }) + sendExpire := s.needsExpire(key) + cmds := [][]string{{"INCRBY", key, strconv.FormatInt(n, 10)}} + if sendExpire { + cmds = append(cmds, []string{"EXPIRE", key, strconv.FormatInt(ttlToSeconds(ttl), 10)}) + } + replies, err := s.client.pipeline(cmds) if err != nil { return 0, fmt.Errorf("redisStore: incrBy %q: %w", key, err) } @@ -68,33 +188,50 @@ func (s *redisStore) incrBy(key string, n int64, ttl time.Duration) (int64, erro if !ok { return 0, fmt.Errorf("redisStore: incrBy %q: unexpected INCRBY reply type %T", key, replies[0]) } + if sendExpire { + s.commitExpire(key) + } else if v == n { + s.forgetExpire(key) + } return v, nil } -// incrMulti implements counterStore: every entry's INCRBY+EXPIRE pair sent -// as ONE pipeline (perf review, 2026-08-21) — 2*len(entries) commands, one -// round trip regardless of len(entries), extending incrBy's own -// single-key pipelining to a whole batch of counters at once. Reply i*2 -// is entry i's INCRBY result (returned in out[i]); reply i*2+1 is its -// EXPIRE result, not otherwise inspected — same reasoning as incrBy's own -// doc comment: a failed EXPIRE right after a successful INCRBY only -// leaves that one key without a fresh TTL, not a corrupted count. The -// same at-least-once semantics as incrBy (see its own doc comment) apply -// here too, extended to the whole batch: a lost reply after the server -// already applied the pipeline can cause respClient.pipeline's -// reconnect-once retry to re-send it, over-counting every entry in it by -// its own delta. +// incrMulti implements counterStore: every entry's INCRBY, plus an EXPIRE +// only when needsExpire(e.key) is true, sent as ONE pipeline (perf +// review, 2026-08-21) — one round trip regardless of len(entries) or how +// many of them get an EXPIRE this time, extending incrBy's own +// single-key pipelining to a whole batch of counters at once. +// sendExpire[i] records, per entry, whether THIS call actually sent +// EXPIRE for it — decided once, before the pipeline goes out, and reused +// after the reply arrives to choose commitExpire vs. the v==delta +// forgetExpire check (incrBy's own doc comment covers why both matter, +// review fix, CRITICAL, 2026-08-2x). incrReplyIdx[i] is the index within +// cmds (and so within replies) of entry i's own INCRBY — NOT a fixed i*2 +// stride any more, since a skipped EXPIRE shifts every later entry's +// commands left by one; entry i's own EXPIRE, when included, immediately +// follows its INCRBY and is not otherwise inspected — same reasoning as +// incrBy's own doc comment: a failed EXPIRE right after a successful +// INCRBY only leaves that one key without a fresh TTL, not a corrupted +// count. The same at-least-once semantics as incrBy (see its own doc +// comment) apply here too, extended to the whole batch: a lost reply +// after the server already applied the pipeline can cause +// respClient.pipeline's reconnect-once retry to re-send it, over-counting +// every entry in it by its own delta. func (s *redisStore) incrMulti(entries []counterIncr) ([]int64, error) { if len(entries) == 0 { return nil, nil } cmds := make([][]string, 0, len(entries)*2) - for _, e := range entries { - cmds = append(cmds, - []string{"INCRBY", e.key, strconv.FormatInt(e.delta, 10)}, - []string{"EXPIRE", e.key, strconv.FormatInt(ttlToSeconds(e.ttl), 10)}, - ) + incrReplyIdx := make([]int, len(entries)) + sendExpire := make([]bool, len(entries)) + for i, e := range entries { + incrReplyIdx[i] = len(cmds) + cmds = append(cmds, []string{"INCRBY", e.key, strconv.FormatInt(e.delta, 10)}) + sendExpire[i] = s.needsExpire(e.key) + if sendExpire[i] { + cmds = append(cmds, []string{"EXPIRE", e.key, strconv.FormatInt(ttlToSeconds(e.ttl), 10)}) + } } replies, err := s.client.pipeline(cmds) @@ -107,7 +244,7 @@ func (s *redisStore) incrMulti(entries []counterIncr) ([]int64, error) { out := make([]int64, len(entries)) for i, e := range entries { - reply := replies[i*2] + reply := replies[incrReplyIdx[i]] if re, ok := reply.(error); ok { return nil, fmt.Errorf("redisStore: incrMulti %q: INCRBY failed: %w", e.key, re) } @@ -116,35 +253,50 @@ func (s *redisStore) incrMulti(entries []counterIncr) ([]int64, error) { return nil, fmt.Errorf("redisStore: incrMulti %q: unexpected INCRBY reply type %T", e.key, reply) } out[i] = v + if sendExpire[i] { + s.commitExpire(e.key) + } else if v == e.delta { + s.forgetExpire(e.key) + } } return out, nil } -// incrAndGetMulti implements counterStore: every entry's INCRBY+EXPIRE pair -// AND every read key's GET, sent as ONE pipeline (perf review round 3, -// 2026-08-22) — 2*len(entries)+len(reads) commands, one round trip -// regardless of how many of each there are, fusing checkAndCount's own -// request-counter increments with its token/cost budget reads (limits.go's -// counterStore doc comment explains why that fusion is safe). Reply layout -// mirrors incrMulti's own: entry i's INCRBY reply sits at reply index 2*i -// (EXPIRE's reply at 2*i+1 is not otherwise inspected, same reasoning as -// incrMulti's own doc comment), followed by one GET reply per read key, in -// order — a missing read key (RESP null bulk) maps to 0, matching -// getBatch's own convention. The same at-least-once semantics as incrMulti -// (see its own doc comment) apply to the INCRBY half of this call; the GET -// half is read-only and carries no such caveat. +// incrAndGetMulti implements counterStore: every entry's INCRBY, plus an +// EXPIRE only when needsExpire(e.key) is true (same commit-after-success/ +// forget-on-recreate bookkeeping as incrMulti's own doc comment), AND +// every read key's GET, sent as ONE pipeline (perf review round 3, +// 2026-08-22) — one round trip regardless of how many of each there are, +// fusing checkAndCount's own request-counter increments with its +// token/cost budget reads (limits.go's counterStore doc comment explains +// why that fusion is safe). Reply layout mirrors incrMulti's own: +// incrReplyIdx[i] is entry i's own INCRBY reply index (not a fixed 2*i +// stride — a skipped EXPIRE shifts every later command left by one; its +// EXPIRE, when included, immediately follows and is not otherwise +// inspected, same reasoning as incrMulti's own doc comment), followed by +// one GET reply per read key at base+i, where base is however many +// commands entries actually produced. A missing read key (RESP null +// bulk) maps to 0, matching getBatch's own convention. The same +// at-least-once semantics as incrMulti (see its own doc comment) apply to +// the INCRBY half of this call; the GET half is read-only and carries no +// such caveat. func (s *redisStore) incrAndGetMulti(entries []counterIncr, reads []string) ([]int64, []int64, error) { if len(entries) == 0 && len(reads) == 0 { return nil, nil, nil } cmds := make([][]string, 0, len(entries)*2+len(reads)) - for _, e := range entries { - cmds = append(cmds, - []string{"INCRBY", e.key, strconv.FormatInt(e.delta, 10)}, - []string{"EXPIRE", e.key, strconv.FormatInt(ttlToSeconds(e.ttl), 10)}, - ) + incrReplyIdx := make([]int, len(entries)) + sendExpire := make([]bool, len(entries)) + for i, e := range entries { + incrReplyIdx[i] = len(cmds) + cmds = append(cmds, []string{"INCRBY", e.key, strconv.FormatInt(e.delta, 10)}) + sendExpire[i] = s.needsExpire(e.key) + if sendExpire[i] { + cmds = append(cmds, []string{"EXPIRE", e.key, strconv.FormatInt(ttlToSeconds(e.ttl), 10)}) + } } + base := len(cmds) for _, k := range reads { cmds = append(cmds, []string{"GET", k}) } @@ -159,7 +311,7 @@ func (s *redisStore) incrAndGetMulti(entries []counterIncr, reads []string) ([]i incrVals := make([]int64, len(entries)) for i, e := range entries { - reply := replies[i*2] + reply := replies[incrReplyIdx[i]] if re, ok := reply.(error); ok { return nil, nil, fmt.Errorf("redisStore: incrAndGetMulti %q: INCRBY failed: %w", e.key, re) } @@ -168,9 +320,13 @@ func (s *redisStore) incrAndGetMulti(entries []counterIncr, reads []string) ([]i return nil, nil, fmt.Errorf("redisStore: incrAndGetMulti %q: unexpected INCRBY reply type %T", e.key, reply) } incrVals[i] = v + if sendExpire[i] { + s.commitExpire(e.key) + } else if v == e.delta { + s.forgetExpire(e.key) + } } - base := len(entries) * 2 readVals := make([]int64, len(reads)) for i, k := range reads { reply := replies[base+i] diff --git a/redis_store_test.go b/redis_store_test.go index 20d1bb3..88c7010 100644 --- a/redis_store_test.go +++ b/redis_store_test.go @@ -1,6 +1,7 @@ package traefikllmgateway import ( + "fmt" "testing" "time" @@ -255,6 +256,307 @@ func TestRedisStore_Get_ParsesIntegerValue(t *testing.T) { } } +// --- expireOnce: skip redundant EXPIRE for an already-touched key (perf +// finding 2, 2026-08-2x audit) --- +// +// Every test below builds its store over newRESPClientPool(..., 1), not +// newRESPClient's self-tuned default pool: each test issues MORE THAN ONE +// top-level call against the same store, and runFakeRESPServer scripts a +// single, strictly ordered connection — with the default pool (size >1, +// finding 1) a later call could acquire a different, not-yet-dialled +// slot instead of reusing the first connection, sending a second SELECT +// the fixed script never expects. Production callers (llmgateway.go's +// buildRedisClient) are unaffected — they always go through newRESPClient +// itself. + +// TestRedisStore_IncrBy_SecondCallSameKey_SkipsExpire proves incrBy sends +// EXPIRE only on the first call for a given key on one redisStore +// instance — the second call for the SAME key sends INCRBY alone, which +// runFakeRESPServer's exact script here proves: an unexpected EXPIRE +// command would fail this test by mismatching (or exhausting) the +// script. +func TestRedisStore_IncrBy_SecondCallSameKey_SkipsExpire(t *testing.T) { + ln := newFakeListener(t) + runFakeRESPServer(t, ln, []respStep{ + {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, + {wantArgs: []string{"INCRBY", "k", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "k", "60"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"INCRBY", "k", "1"}, reply: []byte(":2\r\n")}, // no EXPIRE step: none must be sent + }) + + store := newRedisStore(newRESPClientPool(ln.Addr().String(), "", 0, 1)) + v1, err := store.incrBy("k", 1, 60*time.Second) + if err != nil { + t.Fatalf("incrBy (1st): %v", err) + } + if v1 != 1 { + t.Errorf("incrBy (1st) = %d, want 1", v1) + } + + v2, err := store.incrBy("k", 1, 60*time.Second) + if err != nil { + t.Fatalf("incrBy (2nd): %v", err) + } + if v2 != 2 { + t.Errorf("incrBy (2nd) = %d, want 2", v2) + } +} + +// TestRedisStore_IncrBy_DifferentKeys_EachGetsItsOwnFirstExpire asserts +// expireOnce tracks per KEY, not globally — a second, DIFFERENT key still +// gets its own EXPIRE even though this store already touched a different +// key first. +func TestRedisStore_IncrBy_DifferentKeys_EachGetsItsOwnFirstExpire(t *testing.T) { + ln := newFakeListener(t) + runFakeRESPServer(t, ln, []respStep{ + {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, + {wantArgs: []string{"INCRBY", "a", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "a", "60"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"INCRBY", "b", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "b", "60"}, reply: []byte(":1\r\n")}, + }) + + store := newRedisStore(newRESPClientPool(ln.Addr().String(), "", 0, 1)) + if _, err := store.incrBy("a", 1, 60*time.Second); err != nil { + t.Fatalf("incrBy(a): %v", err) + } + if _, err := store.incrBy("b", 1, 60*time.Second); err != nil { + t.Fatalf("incrBy(b): %v", err) + } +} + +// TestRedisStore_IncrMulti_RepeatedKeyAcrossCalls_SkipsRedundantExpireAndKeepsReplyIndexingCorrect +// is the incrMulti counterpart: a mixed batch where one entry's key was +// already touched (no EXPIRE) and another's is new (gets one) proves the +// variable-stride command layout still returns each entry's own INCRBY +// value at the right index, not shifted by the earlier entries' differing +// command counts. +func TestRedisStore_IncrMulti_RepeatedKeyAcrossCalls_SkipsRedundantExpireAndKeepsReplyIndexingCorrect(t *testing.T) { + ln := newFakeListener(t) + runFakeRESPServer(t, ln, []respStep{ + {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, + // First incrMulti: both "hot" and "cold" are new to this store. + {wantArgs: []string{"INCRBY", "hot", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "hot", "60"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"INCRBY", "cold", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "cold", "60"}, reply: []byte(":1\r\n")}, + // Second incrMulti: "hot" was already touched (no EXPIRE, one + // command shorter), "new" was not (gets one) — reply indexing + // for "new" must still land correctly despite "hot" contributing + // only 1 command instead of 2 ahead of it. + {wantArgs: []string{"INCRBY", "hot", "1"}, reply: []byte(":2\r\n")}, + {wantArgs: []string{"INCRBY", "new", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "new", "60"}, reply: []byte(":1\r\n")}, + }) + + store := newRedisStore(newRESPClientPool(ln.Addr().String(), "", 0, 1)) + first, err := store.incrMulti([]counterIncr{ + {key: "hot", delta: 1, ttl: time.Minute}, + {key: "cold", delta: 1, ttl: time.Minute}, + }) + if err != nil { + t.Fatalf("incrMulti (1st): %v", err) + } + assert.Equal(t, []int64{1, 1}, first) + + second, err := store.incrMulti([]counterIncr{ + {key: "hot", delta: 1, ttl: time.Minute}, + {key: "new", delta: 1, ttl: time.Minute}, + }) + if err != nil { + t.Fatalf("incrMulti (2nd): %v", err) + } + assert.Equal(t, []int64{2, 1}, second) +} + +// TestRedisStore_IncrAndGetMulti_MixedFirstAndRepeatedKeys_ReadsStillLandAfterEntries +// proves incrAndGetMulti's reads (GET commands) still land at the correct +// offset (the dynamically computed "base", not a fixed len(entries)*2) +// when some entries skip their EXPIRE: "hot" was already touched by an +// earlier incrBy call on the SAME store (so this call sends no EXPIRE +// for it), "cold" is new to this store (so it gets one) — the "budget" +// read must still land right after however many commands the entries +// actually produced (3, not the naive len(entries)*2 == 4). +func TestRedisStore_IncrAndGetMulti_MixedFirstAndRepeatedKeys_ReadsStillLandAfterEntries(t *testing.T) { + ln := newFakeListener(t) + runFakeRESPServer(t, ln, []respStep{ + {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, + {wantArgs: []string{"INCRBY", "hot", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "hot", "60"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"INCRBY", "hot", "1"}, reply: []byte(":2\r\n")}, + {wantArgs: []string{"INCRBY", "cold", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "cold", "60"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"GET", "budget"}, reply: []byte("$2\r\n42\r\n")}, + }) + + store := newRedisStore(newRESPClientPool(ln.Addr().String(), "", 0, 1)) + if _, err := store.incrBy("hot", 1, time.Minute); err != nil { + t.Fatalf("incrBy (prime hot's expireOnce state): %v", err) + } + + incrVals, readVals, err := store.incrAndGetMulti( + []counterIncr{ + {key: "hot", delta: 1, ttl: time.Minute}, + {key: "cold", delta: 1, ttl: time.Minute}, + }, + []string{"budget"}, + ) + if err != nil { + t.Fatalf("incrAndGetMulti: %v", err) + } + assert.Equal(t, []int64{2, 1}, incrVals) + assert.Equal(t, []int64{42}, readVals) +} + +// TestRedisStore_ExpireOnce_BoundedEviction proves commitExpire evicts +// the oldest-inserted key (FIFO) once it holds expireOnceCacheMax keys, +// making that oldest key "unseen" again — its next INCRBY sends a +// redundant-but-harmless EXPIRE, exactly as if this store had never +// touched it before — rather than growing without bound across a +// long-lived process's ever-rolling counter buckets. +func TestRedisStore_ExpireOnce_BoundedEviction(t *testing.T) { + store := newRedisStore(nil) // needsExpire/commitExpire touch no network at all + + if !store.needsExpire("key-0") { + t.Fatal("needsExpire(key-0) (before ever committed) = false, want true") + } + store.commitExpire("key-0") + if store.needsExpire("key-0") { + t.Fatal("needsExpire(key-0) (after commit) = true, want false (already seen)") + } + + // Fill the tracker past its cap with distinct keys, evicting key-0. + for i := 1; i <= expireOnceCacheMax; i++ { + store.commitExpire(fmt.Sprintf("key-%d", i)) + } + + if got := len(store.expireSeen); got != expireOnceCacheMax { + t.Errorf("len(expireSeen) = %d, want exactly %d (bounded)", got, expireOnceCacheMax) + } + if !store.needsExpire("key-0") { + t.Error("needsExpire(key-0) after the tracker wrapped = false, want true (evicted, so unseen again)") + } +} + +// --- CRITICAL review fix (2026-08-2x): the old expireOnce marked a key +// "seen" at CHECK time, before its EXPIRE was ever confirmed sent — +// leaving a permanently TTL-less, immortal key on either of two paths: +// (1) the pipeline carrying that EXPIRE fails outright (Redis briefly +// unreachable — routine during a rollout); (2) Redis independently loses +// an already-confirmed key (restart, maxmemory eviction, FLUSHDB, +// failover to a replica missing it) and a later INCRBY recreates it with +// no TTL, since this process's stale cache skips sending EXPIRE. Both are +// pinned directly below, over the real wire (fakeRESPServer scripts that +// fail the test outright if an expected EXPIRE is missing, or an +// unexpected one is sent). + +// TestRedisStore_IncrBy_FailedPipeline_DoesNotConsumeExpireCredit is path +// 1: a pipeline that fails outright (dead address — Redis unreachable) +// must NOT mark the key as having a confirmed EXPIRE. Proven two ways: +// directly, the store's own expireSeen must still be empty after the +// failure (needsExpire's commit-after-success contract); and +// behaviorally, the very next call for the same key — now against a live +// server — must still send EXPIRE, exactly as if this were the true +// first touch. +func TestRedisStore_IncrBy_FailedPipeline_DoesNotConsumeExpireCredit(t *testing.T) { + ln := newFakeListener(t) + deadAddr := ln.Addr().String() + require.NoError(t, ln.Close()) // nothing listens at deadAddr from here on + + store := newRedisStore(newRESPClientPool(deadAddr, "", 0, 1)) + if _, err := store.incrBy("k", 1, 60*time.Second); err == nil { + t.Fatal("want an error: nothing listens at deadAddr") + } + if len(store.expireSeen) != 0 { + t.Fatalf("expireSeen = %v after a failed pipeline, want empty (the EXPIRE credit must not be consumed by a failed send)", store.expireSeen) + } + + // Same store, now pointed at a live server: the next call for the + // SAME key must still send EXPIRE, proving the failed attempt above + // left the key genuinely "unseen". + ln2 := newFakeListener(t) + runFakeRESPServer(t, ln2, []respStep{ + {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, + {wantArgs: []string{"INCRBY", "k", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "k", "60"}, reply: []byte(":1\r\n")}, + }) + store.client = newRESPClientPool(ln2.Addr().String(), "", 0, 1) + if _, err := store.incrBy("k", 1, 60*time.Second); err != nil { + t.Fatalf("incrBy after recovery: %v", err) + } +} + +// TestRedisStore_IncrBy_KeyRecreatedAfterLoss_ForgetsAndResendsExpireNextCall +// is path 2: this process believes "k" already has a confirmed EXPIRE +// (committed by the 1st call below), so the 2nd call sends INCRBY alone +// — but the fake server's reply for it (":1\r\n", equal to this call's +// own delta) simulates Redis having lost the key and INCRBY recreating +// it fresh, with no TTL. forgetExpire must fire so the 3rd call resends +// EXPIRE — pinned by the fake server's exact script, which fails the +// test if the 3rd call's EXPIRE is missing. +func TestRedisStore_IncrBy_KeyRecreatedAfterLoss_ForgetsAndResendsExpireNextCall(t *testing.T) { + ln := newFakeListener(t) + runFakeRESPServer(t, ln, []respStep{ + {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, + // 1st call: genuinely new key — sends EXPIRE, commits it. + {wantArgs: []string{"INCRBY", "k", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "k", "60"}, reply: []byte(":1\r\n")}, + // 2nd call: cache says "k" already has a confirmed EXPIRE, so + // only INCRBY is sent — but Redis lost the key, so its reply + // (":1\r\n") equals this call's own delta, the recreate signal. + {wantArgs: []string{"INCRBY", "k", "1"}, reply: []byte(":1\r\n")}, + // 3rd call: forgetExpire must have fired after the 2nd call, so + // this one sends EXPIRE again. + {wantArgs: []string{"INCRBY", "k", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "k", "60"}, reply: []byte(":1\r\n")}, + }) + + store := newRedisStore(newRESPClientPool(ln.Addr().String(), "", 0, 1)) + for i := 0; i < 3; i++ { + if _, err := store.incrBy("k", 1, 60*time.Second); err != nil { + t.Fatalf("incrBy (call %d): %v", i+1, err) + } + } +} + +// TestRedisStore_IncrMulti_KeyRecreatedAfterLoss_ForgetsAndResendsExpireNextCall +// is the incrMulti counterpart, proving the per-entry sendExpire/ +// forgetExpire bookkeeping (not just incrBy's single-key path) reacts to +// a recreate signal correctly inside a batch alongside an unrelated, +// genuinely-still-alive entry. +func TestRedisStore_IncrMulti_KeyRecreatedAfterLoss_ForgetsAndResendsExpireNextCall(t *testing.T) { + ln := newFakeListener(t) + runFakeRESPServer(t, ln, []respStep{ + {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, + // 1st batch: both keys are new — both get EXPIRE, both commit. + {wantArgs: []string{"INCRBY", "lost", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "lost", "60"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"INCRBY", "alive", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "alive", "60"}, reply: []byte(":1\r\n")}, + // 2nd batch: "lost" was evicted from Redis (reply == its own + // delta, the recreate signal) but "alive" genuinely still has + // its counter (reply 2, not equal to delta 1) — only "lost" + // must be forgotten. + {wantArgs: []string{"INCRBY", "lost", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"INCRBY", "alive", "1"}, reply: []byte(":2\r\n")}, + // 3rd batch: "lost" gets a fresh EXPIRE; "alive" still does not. + {wantArgs: []string{"INCRBY", "lost", "1"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"EXPIRE", "lost", "60"}, reply: []byte(":1\r\n")}, + {wantArgs: []string{"INCRBY", "alive", "1"}, reply: []byte(":3\r\n")}, + }) + + store := newRedisStore(newRESPClientPool(ln.Addr().String(), "", 0, 1)) + entries := []counterIncr{ + {key: "lost", delta: 1, ttl: time.Minute}, + {key: "alive", delta: 1, ttl: time.Minute}, + } + for i := 0; i < 3; i++ { + if _, err := store.incrMulti(entries); err != nil { + t.Fatalf("incrMulti (batch %d): %v", i+1, err) + } + } +} + // TestLimiter_FailOpenFalse_DeadRedisAddress_ReturnsStoreDownViolation is // the brief's Step-1 case: a limiter backed by a redisStore pointed at an // address nothing listens on, with failOpen=false, must refuse the