mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-20 05:56:56 +00:00
perf(redis_store): send EXPIRE only once per key per process lifetime
31 of 74 commands per request in the interpreted-harness audit were EXPIRE, emitted unconditionally after every INCRBY. Deleting EXPIRE entirely (upper bound) measured 74->43 cmds/req, p50 867.9->697.9us (-19.6%), conc=16 throughput 1,477->2,336 req/s (+58%), allocs -26%. Route chosen: a bounded first-touch-per-key cache (expireOnce), not EVALSHA+Lua. windowKey (limits.go) already embeds each counter's own time bucket in its key string, and every TTL callers pass is deliberately far longer than its window's own natural length purely as a stale-key GC margin (limits.go's own doc comment on dayWindowTTL/monthWindowTTL/etc.), not a sliding-window mechanism — so once EXPIRE has been sent for an exact key, every further INCRBY against that SAME key happens well within its own bucket's window, comfortably inside the TTL's margin. Skipping the redundant resend cannot leave a key un-expiring: a bucket's key simply stops being written once its window rolls over (a new bucket string takes its place). expireOnce is bounded (expireOnceCacheMax=4096, FIFO eviction) so a long-lived process's ever-rolling buckets cannot grow it unboundedly; an evicted key just gets one redundant-but-harmless EXPIRE again, exactly as if never touched before. incrBy/incrMulti/incrAndGetMulti now build a variable-stride command list (each entry contributes its own INCRBY, plus EXPIRE only when expireOnce(key) is true) with an explicit incrReplyIdx/base tracking each entry's actual reply position — no longer a fixed i*2 stride, since a skipped EXPIRE shifts every later command left by one. EVALSHA+Lua was rejected as the shippable route here: it would rewrite every command-building and reply-index path, requires NOSCRIPT fallback handling, and every existing wire-level test (dozens of INCRBY+EXPIRE-scripted fakeRESPServer cases) would need rebuilding around SHA1 digests for no additional correctness benefit over the simpler first-touch cache.
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user