Spaces:
Running
Running
| // Package cache provides multi-tier caching implementation with local LRU and Redis. | |
| // Production-grade caching with L1 (in-memory LRU), L2 (Redis), and stampede protection. | |
| package cache | |
| import ( | |
| "context" | |
| "encoding/json" | |
| "fmt" | |
| "strings" | |
| "sync" | |
| "time" | |
| "github.com/redis/go-redis/v9" | |
| "golang.org/x/sync/singleflight" | |
| ) | |
| const ( | |
| DefaultRegion = "nbo" | |
| DefaultTier = "L1" | |
| KeySeparator = ":" | |
| ) | |
| var ( | |
| ErrCacheMiss = &CacheMissError{} | |
| ErrCircuitOpen = fmt.Errorf("circuit breaker is open") | |
| ) | |
| type CacheMissError struct{} | |
| func (e *CacheMissError) Error() string { return "cache miss" } | |
| type CacheMetrics struct { | |
| mu sync.RWMutex | |
| LocalHits int64 | |
| LocalMisses int64 | |
| RedisHits int64 | |
| RedisMisses int64 | |
| StampedeHits int64 | |
| } | |
| func (m *CacheMetrics) Snapshot() CacheMetrics { | |
| m.mu.RLock() | |
| defer m.mu.RUnlock() | |
| return CacheMetrics{ | |
| LocalHits: m.LocalHits, | |
| LocalMisses: m.LocalMisses, | |
| RedisHits: m.RedisHits, | |
| RedisMisses: m.RedisMisses, | |
| StampedeHits: m.StampedeHits, | |
| } | |
| } | |
| type Config struct { | |
| RedisURL string `yaml:"redisUrl"` | |
| RedisAddr string `yaml:"redisAddr"` | |
| RedisPassword string `yaml:"redisPassword"` | |
| RedisDB int `yaml:"redisDb"` | |
| LocalSize int `yaml:"localSize"` | |
| TTL time.Duration `yaml:"ttl"` | |
| MaxRetries int `yaml:"maxRetries"` | |
| PoolSize int `yaml:"poolSize"` | |
| MaxEntrySize int `yaml:"maxEntrySize"` | |
| KeyPrefix string `yaml:"keyPrefix"` | |
| Region string `yaml:"region"` | |
| // PrefixIndexEnabled maintains a per-namespace Redis SET index of cache keys | |
| // so DeleteByPrefix can invalidate in O(matched) instead of O(keyspace). | |
| // See docs/architecture/cdn-invalidation-prefix-index.md. | |
| PrefixIndexEnabled bool `yaml:"prefixIndexEnabled"` | |
| } | |
| type CachedResponse struct { | |
| Status int `json:"status"` | |
| Headers map[string]string `json:"headers"` | |
| Body json.RawMessage `json:"body"` | |
| Compressed bool `json:"compressed"` | |
| Timestamp time.Time `json:"timestamp"` | |
| } | |
| func MarshalResponse(status int, headers map[string]string, body []byte) []byte { | |
| entry := CachedResponse{ | |
| Status: status, | |
| Headers: headers, | |
| Body: json.RawMessage(body), | |
| Timestamp: time.Now(), | |
| } | |
| data, _ := json.Marshal(entry) | |
| return data | |
| } | |
| func UnmarshalResponse(data []byte) (*CachedResponse, error) { | |
| var entry CachedResponse | |
| if err := json.Unmarshal(data, &entry); err != nil { | |
| return nil, err | |
| } | |
| return &entry, nil | |
| } | |
| type MultiTierCache struct { | |
| local *LRUCache | |
| redis *redis.Client | |
| ttl time.Duration | |
| keyPrefix string | |
| maxEntrySize int | |
| region string | |
| prefixIndex bool | |
| sf singleflight.Group | |
| metrics *CacheMetrics | |
| healthy bool | |
| mu sync.RWMutex | |
| } | |
| func New(cfg Config) (*MultiTierCache, error) { | |
| region := cfg.Region | |
| if region == "" { | |
| region = DefaultRegion | |
| } | |
| keyPrefix := cfg.KeyPrefix | |
| if keyPrefix == "" { | |
| keyPrefix = fmt.Sprintf("%s%s%s%s", DefaultTier, KeySeparator, region, KeySeparator) | |
| } | |
| ttl := cfg.TTL | |
| if ttl == 0 { | |
| ttl = 5 * time.Minute | |
| } | |
| localSize := cfg.LocalSize | |
| if localSize == 0 { | |
| localSize = 10000 | |
| } | |
| poolSize := cfg.PoolSize | |
| if poolSize == 0 { | |
| poolSize = 20 | |
| } | |
| c := &MultiTierCache{ | |
| local: NewLRUCache(localSize), | |
| ttl: ttl, | |
| keyPrefix: keyPrefix, | |
| maxEntrySize: cfg.MaxEntrySize, | |
| region: region, | |
| prefixIndex: cfg.PrefixIndexEnabled, | |
| metrics: &CacheMetrics{}, | |
| healthy: false, | |
| } | |
| var redisClient *redis.Client | |
| var redisErr error | |
| switch { | |
| case cfg.RedisURL != "": | |
| var opt *redis.Options | |
| opt, redisErr = redis.ParseURL(cfg.RedisURL) | |
| if redisErr == nil { | |
| opt.MaxRetries = cfg.MaxRetries | |
| opt.PoolSize = poolSize | |
| redisClient = redis.NewClient(opt) | |
| } | |
| case cfg.RedisAddr != "": | |
| redisClient = redis.NewClient(&redis.Options{ | |
| Addr: cfg.RedisAddr, | |
| Password: cfg.RedisPassword, | |
| DB: cfg.RedisDB, | |
| MaxRetries: cfg.MaxRetries, | |
| PoolSize: poolSize, | |
| MinIdleConns: 5, | |
| DialTimeout: 5 * time.Second, | |
| ReadTimeout: 3 * time.Second, | |
| WriteTimeout: 3 * time.Second, | |
| }) | |
| } | |
| if redisClient != nil { | |
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | |
| defer cancel() | |
| if err := redisClient.Ping(ctx).Err(); err != nil { | |
| redisClient.Close() | |
| redisClient = nil | |
| } else { | |
| c.healthy = true | |
| } | |
| } | |
| c.redis = redisClient | |
| return c, nil | |
| } | |
| func (c *MultiTierCache) Healthy() bool { | |
| c.mu.RLock() | |
| defer c.mu.RUnlock() | |
| return c.healthy | |
| } | |
| // Keys are stored verbatim (callers build them via NormalizeKey). The previous | |
| // implementation hashed keys a SECOND time here, making the stored key prefix- | |
| // opaque — SCAN-pattern invalidation (DeleteByPrefix, admin invalidate-pattern) | |
| // could never match and silently no-op'd. | |
| func (c *MultiTierCache) Get(ctx context.Context, key string) ([]byte, error) { | |
| if value, found := c.local.Get(key); found { | |
| c.metrics.mu.Lock() | |
| c.metrics.LocalHits++ | |
| c.metrics.mu.Unlock() | |
| EmitCacheHitsLocal(c.region) | |
| return value, nil | |
| } | |
| c.metrics.mu.Lock() | |
| c.metrics.LocalMisses++ | |
| c.metrics.mu.Unlock() | |
| EmitCacheMissesLocal(c.region) | |
| if c.redis != nil { | |
| if !c.Healthy() { | |
| return nil, ErrCacheMiss | |
| } | |
| value, err := c.redis.Get(ctx, key).Bytes() | |
| if err == nil { | |
| c.metrics.mu.Lock() | |
| c.metrics.RedisHits++ | |
| c.metrics.mu.Unlock() | |
| EmitCacheHitsRedis(c.region) | |
| // Mirror into L1 with the REMAINING Redis TTL, not the full default: | |
| // short-TTL entries (e.g. 5s error-caches) must not be re-promoted to | |
| // a 5-minute L1 lifetime, or L1 would serve data past Redis expiry. | |
| promoteTTL := c.ttl | |
| if ttl, terr := c.redis.TTL(ctx, key).Result(); terr == nil && ttl > 0 && ttl < promoteTTL { | |
| promoteTTL = ttl | |
| } | |
| c.local.Set(key, value, promoteTTL) | |
| return value, nil | |
| } | |
| c.metrics.mu.Lock() | |
| c.metrics.RedisMisses++ | |
| c.metrics.mu.Unlock() | |
| EmitCacheMissesRedis(c.region) | |
| } | |
| return nil, ErrCacheMiss | |
| } | |
| func (c *MultiTierCache) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error { | |
| if c.maxEntrySize > 0 && len(value) > c.maxEntrySize { | |
| return nil | |
| } | |
| if ttl == 0 { | |
| ttl = c.ttl | |
| } | |
| c.local.Set(key, value, ttl) | |
| if c.redis != nil && c.Healthy() { | |
| // Index maintenance is atomic with the data write (same MULTI/EXEC) so | |
| // the index can never miss a just-written key. | |
| if c.prefixIndex { | |
| if ns := c.namespaceOf(key); ns != "" { | |
| pipe := c.redis.TxPipeline() | |
| pipe.Set(ctx, key, value, ttl) | |
| pipe.SAdd(ctx, c.indexKey(ns), key) | |
| pipe.Expire(ctx, c.indexKey(ns), ttl) | |
| _, _ = pipe.Exec(ctx) | |
| return nil | |
| } | |
| } | |
| _ = c.redis.Set(ctx, key, value, ttl).Err() | |
| } | |
| return nil | |
| } | |
| func (c *MultiTierCache) Delete(ctx context.Context, key string) error { | |
| c.local.Delete(key) | |
| if c.redis != nil && c.Healthy() { | |
| if c.prefixIndex { | |
| if ns := c.namespaceOf(key); ns != "" { | |
| c.redis.SRem(ctx, c.indexKey(ns), key) | |
| } | |
| } | |
| return c.redis.Del(ctx, key).Err() | |
| } | |
| return nil | |
| } | |
| func (c *MultiTierCache) GetOrSet(ctx context.Context, key string, fn func() ([]byte, error), ttl time.Duration) ([]byte, error) { | |
| if cached, err := c.Get(ctx, key); err == nil { | |
| return cached, nil | |
| } | |
| val, err, _ := c.sf.Do(key, func() (interface{}, error) { | |
| data, fetchErr := fn() | |
| if fetchErr != nil { | |
| return nil, fetchErr | |
| } | |
| c.metrics.mu.Lock() | |
| c.metrics.StampedeHits++ | |
| c.metrics.mu.Unlock() | |
| EmitStampedeHit(c.region) | |
| _ = c.Set(ctx, key, data, ttl) | |
| return data, nil | |
| }) | |
| if err != nil { | |
| return nil, err | |
| } | |
| return val.([]byte), nil | |
| } | |
| // DeleteByPrefix removes every cached entry whose key starts with prefix — from | |
| // both the local L1 tier and Redis. prefix may be a full key prefix | |
| // ("L1:nbo:rag:") or a bare tail ("rag:") resolved against this cache's | |
| // keyPrefix. An empty prefix purges the whole cache keyspace. | |
| // | |
| // When the prefix index is enabled the deletion runs in O(matched) via the | |
| // per-namespace SET index (single atomic Lua call); SCAN remains the fallback | |
| // for patterns with no index entry (e.g. keys written before the flag). | |
| func (c *MultiTierCache) DeleteByPrefix(ctx context.Context, prefix string) error { | |
| pattern := c.prefixPattern(prefix) // glob for Redis SCAN | |
| localPrefix := strings.TrimSuffix(pattern, "*") // literal prefix for L1 | |
| // Always clear the local L1 tier, regardless of Redis health. | |
| c.local.DeletePrefix(localPrefix) | |
| if c.redis == nil || !c.Healthy() { | |
| return nil | |
| } | |
| if c.prefixIndex { | |
| handled, err := c.deleteByPrefixIndexed(ctx, prefix, localPrefix) | |
| if err != nil { | |
| return err | |
| } | |
| if handled { | |
| return nil | |
| } | |
| } | |
| // SCAN fallback (also the default path when the index is disabled). | |
| var cursor uint64 | |
| for { | |
| keys, next, err := c.redis.Scan(ctx, cursor, pattern, 100).Result() | |
| if err != nil { | |
| return err | |
| } | |
| if len(keys) > 0 { | |
| if err := c.redis.Del(ctx, keys...).Err(); err != nil { | |
| return err | |
| } | |
| } | |
| cursor = next | |
| if cursor == 0 { | |
| break | |
| } | |
| } | |
| return nil | |
| } | |
| // indexKey returns the Redis SET key indexing the cache keys of a namespace. | |
| func (c *MultiTierCache) indexKey(namespace string) string { | |
| return "idx:cache:" + c.region + ":" + namespace | |
| } | |
| // namespaceOf extracts the logical namespace (the segment right after the key | |
| // prefix) from a readable key, e.g. "L1:nbo:rag:..." -> "rag". | |
| func (c *MultiTierCache) namespaceOf(key string) string { | |
| rest := strings.TrimPrefix(key, c.keyPrefix) | |
| if i := strings.Index(rest, ":"); i > 0 { | |
| return rest[:i] | |
| } | |
| return rest | |
| } | |
| // namespaceOfPrefix maps an invalidation prefix to its index namespace. Bare | |
| // tails ("rag:") and full-key prefixes ("L1:nbo:rag:") both resolve to "rag"; | |
| // empty/keyspace-wide prefixes have no namespace (whole-cache SCAN purge). | |
| func (c *MultiTierCache) namespaceOfPrefix(prefix string) string { | |
| if prefix == "" { | |
| return "" | |
| } | |
| if strings.HasPrefix(prefix, c.keyPrefix) { | |
| rest := strings.TrimPrefix(prefix, c.keyPrefix) | |
| if i := strings.Index(rest, ":"); i > 0 { | |
| return rest[:i] | |
| } | |
| return "" | |
| } | |
| if i := strings.Index(prefix, ":"); i > 0 { | |
| return prefix[:i] | |
| } | |
| return prefix | |
| } | |
| // prefixIndexScript atomically deletes cache keys matching a literal prefix and | |
| // removes them from the namespace index in a single round trip. DEL/SREM are | |
| // chunked because Lua's unpack() overflows past ~8000 arguments — a hot legal | |
| // namespace can hold far more keys than that. | |
| var prefixIndexScript = redis.NewScript(` | |
| local members = redis.call('SMEMBERS', KEYS[1]) | |
| local batch = {} | |
| local removed = 0 | |
| local function flush() | |
| if #batch > 0 then | |
| redis.call('DEL', unpack(batch)) | |
| redis.call('SREM', KEYS[1], unpack(batch)) | |
| removed = removed + #batch | |
| batch = {} | |
| end | |
| end | |
| for _, k in ipairs(members) do | |
| if string.find(k, ARGV[1], 1, true) == 1 then | |
| table.insert(batch, k) | |
| if #batch >= 500 then | |
| flush() | |
| end | |
| end | |
| end | |
| flush() | |
| return removed | |
| `) | |
| // deleteByPrefixIndexed removes matching keys through the per-namespace SET | |
| // index in one atomic Lua call. Returns handled=true when the index was | |
| // authoritative (nothing left for SCAN to find); false asks the caller to fall | |
| // back to SCAN (unknown namespace with no index entry). | |
| func (c *MultiTierCache) deleteByPrefixIndexed(ctx context.Context, prefix, localPrefix string) (bool, error) { | |
| ns := c.namespaceOfPrefix(prefix) | |
| if ns == "" { | |
| return false, nil | |
| } | |
| idxKey := c.indexKey(ns) | |
| // Match keys by their RESOLVED literal prefix so bare tails and full-key | |
| // prefixes behave identically to the SCAN path. | |
| n, err := prefixIndexScript.Run(ctx, c.redis, []string{idxKey}, localPrefix).Int() | |
| if err != nil { | |
| // Script failure must not leave stale cache: fall back to SCAN so the | |
| // invalidation still happens (slower, but correct). | |
| return false, nil | |
| } | |
| exists, err := c.redis.Exists(ctx, idxKey).Result() | |
| if err != nil { | |
| return false, nil | |
| } | |
| // n>0: matched & removed. exists>0: index is live with no matches | |
| // (authoritative no-op). Both absent: namespace never indexed — fall back. | |
| return n > 0 || exists > 0, nil | |
| } | |
| // prefixPattern resolves a caller-supplied prefix into a SCAN glob. | |
| func (c *MultiTierCache) prefixPattern(prefix string) string { | |
| if prefix == "" { | |
| return c.keyPrefix + "*" | |
| } | |
| if strings.HasPrefix(prefix, c.keyPrefix) { | |
| return prefix + "*" | |
| } | |
| return c.keyPrefix + prefix + "*" | |
| } | |
| // NormalizeKey builds a READABLE, prefix-preserving key from parts. Keys are | |
| // stored verbatim. The previous implementation hashed the whole key, which | |
| // destroyed the logical prefix — pattern invalidation (DeleteByPrefix, admin | |
| // invalidate-pattern) could never match and silently no-op'd. Readable keys | |
| // make prefix/pattern invalidation work end-to-end (queries, users, models are | |
| // all addressable via glob) and are debuggable in Redis. Redis supports keys up | |
| // to 512MB, and typical cache keys here are well under 1KB. | |
| func (c *MultiTierCache) NormalizeKey(parts ...string) string { | |
| var sb strings.Builder | |
| sb.WriteString(c.keyPrefix) | |
| for i, p := range parts { | |
| if i > 0 { | |
| sb.WriteString(KeySeparator) | |
| } | |
| sb.WriteString(strings.TrimSpace(p)) | |
| } | |
| return sb.String() | |
| } | |
| func (c *MultiTierCache) GetJSON(ctx context.Context, key string, v interface{}) error { | |
| data, err := c.Get(ctx, key) | |
| if err != nil { | |
| return err | |
| } | |
| return json.Unmarshal(data, v) | |
| } | |
| func (c *MultiTierCache) SetJSON(ctx context.Context, key string, v interface{}, ttl time.Duration) error { | |
| data, err := json.Marshal(v) | |
| if err != nil { | |
| return err | |
| } | |
| return c.Set(ctx, key, data, ttl) | |
| } | |
| func (c *MultiTierCache) GetMetrics() CacheMetrics { | |
| return c.metrics.Snapshot() | |
| } | |
| func (c *MultiTierCache) Close() error { | |
| if c.redis != nil { | |
| return c.redis.Close() | |
| } | |
| return nil | |
| } | |
| // LRUCache implements a simple LRU cache with per-entry TTL. Entries are | |
| // evicted when they expire, so L1 can never serve data past its TTL (previously | |
| // L1 held entries indefinitely until capacity eviction — stale responses could | |
| // be served long after their Redis expiry). | |
| type LRUCache struct { | |
| mu sync.RWMutex | |
| capacity int | |
| items map[string]*lruItem | |
| head *lruItem | |
| tail *lruItem | |
| } | |
| type lruItem struct { | |
| key string | |
| value []byte | |
| expiresAt time.Time | |
| prev *lruItem | |
| next *lruItem | |
| } | |
| func NewLRUCache(capacity int) *LRUCache { | |
| return &LRUCache{ | |
| capacity: capacity, | |
| items: make(map[string]*lruItem), | |
| } | |
| } | |
| func (c *LRUCache) Get(key string) ([]byte, bool) { | |
| c.mu.Lock() | |
| defer c.mu.Unlock() | |
| item, found := c.items[key] | |
| if !found { | |
| return nil, false | |
| } | |
| if !item.expiresAt.IsZero() && time.Now().After(item.expiresAt) { | |
| c.removeItem(item) | |
| delete(c.items, key) | |
| return nil, false | |
| } | |
| c.moveToFront(item) | |
| return item.value, true | |
| } | |
| func (c *LRUCache) Set(key string, value []byte, ttl time.Duration) { | |
| c.mu.Lock() | |
| defer c.mu.Unlock() | |
| var expiresAt time.Time | |
| if ttl > 0 { | |
| expiresAt = time.Now().Add(ttl) | |
| } | |
| if item, found := c.items[key]; found { | |
| item.value = value | |
| item.expiresAt = expiresAt | |
| c.moveToFront(item) | |
| return | |
| } | |
| item := &lruItem{key: key, value: value, expiresAt: expiresAt} | |
| c.items[key] = item | |
| c.addToFront(item) | |
| if len(c.items) > c.capacity { | |
| c.evictLRU() | |
| } | |
| } | |
| func (c *LRUCache) Delete(key string) { | |
| c.mu.Lock() | |
| defer c.mu.Unlock() | |
| if item, found := c.items[key]; found { | |
| c.removeItem(item) | |
| delete(c.items, key) | |
| } | |
| } | |
| // DeletePrefix removes all entries whose key starts with prefix. | |
| func (c *LRUCache) DeletePrefix(prefix string) { | |
| c.mu.Lock() | |
| defer c.mu.Unlock() | |
| for key, item := range c.items { | |
| if strings.HasPrefix(key, prefix) { | |
| c.removeItem(item) | |
| delete(c.items, key) | |
| } | |
| } | |
| } | |
| func (c *LRUCache) moveToFront(item *lruItem) { | |
| if item == c.head { | |
| return | |
| } | |
| c.removeItem(item) | |
| c.addToFront(item) | |
| } | |
| func (c *LRUCache) addToFront(item *lruItem) { | |
| item.prev = nil | |
| item.next = c.head | |
| if c.head != nil { | |
| c.head.prev = item | |
| } | |
| c.head = item | |
| if c.tail == nil { | |
| c.tail = item | |
| } | |
| } | |
| func (c *LRUCache) removeItem(item *lruItem) { | |
| if item.prev != nil { | |
| item.prev.next = item.next | |
| } else { | |
| c.head = item.next | |
| } | |
| if item.next != nil { | |
| item.next.prev = item.prev | |
| } else { | |
| c.tail = item.prev | |
| } | |
| } | |
| func (c *LRUCache) evictLRU() { | |
| if c.tail == nil { | |
| return | |
| } | |
| delete(c.items, c.tail.key) | |
| c.removeItem(c.tail) | |
| } | |