feat(19.3-01): add cooler.Cache with otter memory and Redis drivers
- Cache: Get, Set, Has, Forget, Remember (degrades on cache errors), Increment, Flush with per-entry TTLs under <prefix>cache: - memory driver on otter v2.3.0 bounded by cache.memory.max_entries; Redis driver with SET PX, INCRBY and a Flush limited to SCAN MATCH <prefix>cache:* plus UNLINK - Install publishes the Cache (sharing one Redis client with the rate limiter); FromApp serves jobs and commands without a dial - new docs/services/caching.md, configuration rows, cooler README
This commit is contained in:
140
modules/cooler/rediscache.go
Normal file
140
modules/cooler/rediscache.go
Normal file
@@ -0,0 +1,140 @@
|
||||
package cooler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// flushBatch is how many keys one SCAN step asks for during Flush.
|
||||
const flushBatch = 500
|
||||
|
||||
// redisCache is the Cache on Redis, shared by every instance using the same
|
||||
// Redis and prefix. Keys are <prefix>cache:<key>.
|
||||
type redisCache struct {
|
||||
client *redis.Client
|
||||
prefix string
|
||||
warn *warner
|
||||
}
|
||||
|
||||
// NewRedisCache returns a Cache on client with keys <prefix>cache:<key>.
|
||||
// It does not dial; the first call does.
|
||||
func NewRedisCache(client *redis.Client, prefix string) Cache {
|
||||
return &redisCache{client: client, prefix: prefix, warn: newWarner(slog.Default())}
|
||||
}
|
||||
|
||||
func (r *redisCache) key(k string) string { return r.prefix + "cache:" + k }
|
||||
|
||||
func (r *redisCache) setLogger(log *slog.Logger) { r.warn.setLogger(log) }
|
||||
|
||||
func (r *redisCache) ready() error {
|
||||
if r == nil || r.client == nil {
|
||||
return errNilClient
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *redisCache) Get(ctx context.Context, key string) ([]byte, bool, error) {
|
||||
if err := r.ready(); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
b, err := r.client.Get(ctx, r.key(key)).Bytes()
|
||||
if errors.Is(err, redis.Nil) {
|
||||
return nil, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return b, true, nil
|
||||
}
|
||||
|
||||
// Set stores value with SET … PX; a ttl of 0 or less stores it without expiry.
|
||||
func (r *redisCache) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error {
|
||||
if err := r.ready(); err != nil {
|
||||
return err
|
||||
}
|
||||
if value == nil {
|
||||
value = []byte{}
|
||||
}
|
||||
if ttl <= 0 {
|
||||
return r.client.Set(ctx, r.key(key), value, 0).Err()
|
||||
}
|
||||
ms := ttl.Milliseconds()
|
||||
if ms < 1 {
|
||||
ms = 1
|
||||
}
|
||||
return r.client.Do(ctx, "SET", r.key(key), value, "PX", ms).Err()
|
||||
}
|
||||
|
||||
func (r *redisCache) Has(ctx context.Context, key string) (bool, error) {
|
||||
if err := r.ready(); err != nil {
|
||||
return false, err
|
||||
}
|
||||
n, err := r.client.Exists(ctx, r.key(key)).Result()
|
||||
return n > 0, err
|
||||
}
|
||||
|
||||
func (r *redisCache) Forget(ctx context.Context, key string) error {
|
||||
if err := r.ready(); err != nil {
|
||||
return err
|
||||
}
|
||||
return r.client.Unlink(ctx, r.key(key)).Err()
|
||||
}
|
||||
|
||||
func (r *redisCache) Remember(ctx context.Context, key string, ttl time.Duration, fn func(context.Context) ([]byte, error)) ([]byte, error) {
|
||||
return remember(ctx, r, r.warn, key, ttl, fn)
|
||||
}
|
||||
|
||||
// Increment runs INCRBY: a missing key starts at 0 without a TTL, an existing
|
||||
// key keeps its expiry.
|
||||
func (r *redisCache) Increment(ctx context.Context, key string, by int64) (int64, error) {
|
||||
if err := r.ready(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return r.client.IncrBy(ctx, r.key(key), by).Result()
|
||||
}
|
||||
|
||||
// Flush removes only this application's cache keys: it walks SCAN MATCH
|
||||
// <prefix>cache:* in batches and removes each batch with UNLINK. It never
|
||||
// empties the database, which would also drop rate-limit counters and other
|
||||
// applications' keys.
|
||||
func (r *redisCache) Flush(ctx context.Context) error {
|
||||
if err := r.ready(); err != nil {
|
||||
return err
|
||||
}
|
||||
match := escapeGlob(r.prefix+"cache:") + "*"
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, next, err := r.client.Scan(ctx, cursor, match, flushBatch).Result()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(keys) > 0 {
|
||||
if err := r.client.Unlink(ctx, keys...).Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if next == 0 {
|
||||
return nil
|
||||
}
|
||||
cursor = next
|
||||
}
|
||||
}
|
||||
|
||||
// escapeGlob escapes the characters SCAN MATCH treats as a pattern, so a
|
||||
// prefix is matched literally.
|
||||
func escapeGlob(s string) string {
|
||||
var b strings.Builder
|
||||
for _, r := range s {
|
||||
switch r {
|
||||
case '*', '?', '[', ']', '\\':
|
||||
b.WriteByte('\\')
|
||||
}
|
||||
b.WriteRune(r)
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
Reference in New Issue
Block a user