From 5510c8a3a2324aa0ba3e395d05627db51b78e499 Mon Sep 17 00:00:00 2001 From: Mohammad Samimi Date: Wed, 10 Jun 2026 15:38:15 +0330 Subject: [PATCH] feat: add memory-authoritative write-behind mode for Redis Adds an optional "memory-authoritative" mode where Redis is used purely as an async write-behind layer, keeping the hot path fully in-memory after warmup. All changes are backward-compatible (defaults preserve existing read-through behavior). - Single round-trip Redis reads via optional RedisGetTTLClient interface (implemented by v8/v9 adapters with a GET+TTL pipeline); falls back to separate Get+TTL otherwise. - WithWriteThroughOnly(): disables synchronous Redis reads on local miss for Get/getFallback/ModifyNumeric/SetNX; writes still flow async. - WarmFromRedis(keys): loads state from Redis into memory at startup for durability across restarts without runtime synchronous reads. - Local-authoritative SetNX in write-through-only mode: atomicity via the local shard lock, async SETNX to Redis; default behavior preserved. - Fix silent drop-on-full: centralize async sends through enqueueRedis, count drops via DroppedWrites(), and add WithBlockOnFull(d) to block briefly instead of dropping. - Tests for all new behavior (miniredis-backed, no external Redis needed). - README updated with new features and clarified read-through semantics. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 77 ++++++++++++- cache.go | 249 ++++++++++++++++++++++++++++++++++++------ cache_redis_test.go | 20 ++++ redis/v8/adapter.go | 23 ++++ redis/v9/adapter.go | 23 ++++ sharded.go | 60 ++++++++++ sharded_redis_test.go | 240 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 658 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 75fc020..05060ef 100644 --- a/README.md +++ b/README.md @@ -10,13 +10,17 @@ go-cache is an in-memory key:value store/cache similar to memcached that is suit * **Sharding**: Reduces lock contention for high-concurrency workloads. * **Redis Integration**: Optional L2 caching and persistence layer using `go-redis` (supports v8 and v9). +* **Memory-Authoritative Mode**: Optional write-behind-only mode where Redis is *never* read synchronously on the hot path — after warmup, reads are fully in-memory while writes still persist asynchronously. +* **Single Round-Trip Reads**: Redis reads fetch the value and its TTL in one round-trip (pipeline / `GETEX`) when the adapter supports it, instead of a separate `GET` + `TTL`. +* **Startup Warmup**: `WarmFromRedis` rebuilds in-memory state from Redis at boot, so durability survives restarts without any synchronous reads at runtime. * **Capacity Management**: Internal LRU-like eviction when memory limits are reached. * **Generics**: Type-safe API (Go 1.18+). -* **SetNX**: Atomic "Set if Not Exists" operation, seamlessly synchronized with Redis. +* **SetNX**: Atomic "Set if Not Exists" operation, either synchronized with Redis (default) or local-authoritative with async persistence. * **Numeric Operations**: Atomic increment/decrement support for numeric types, persisted to Redis. * **Set Cache**: Track unique members per key, each with its own TTL — ideal for counting active sessions/devices per user. * **Graceful Shutdown**: Ensures pending Redis operations are completed before exit. * **Sync**: Force refresh items from Redis. +* **Back-Pressure Control**: Observe dropped async writes via `DroppedWrites`, and optionally block briefly instead of dropping when the write-behind queue is full. * **Configurable Timeouts**: Fine-tune Redis L2 operation timeouts for all cache types. * **Performance**: Extremely low latency local operations (see [BENCHMARKS.md](BENCHMARKS.md)). @@ -28,6 +32,11 @@ go-cache is an in-memory key:value store/cache similar to memcached that is suit ## Recent Updates +* **Memory-Authoritative / Write-Behind-Only Mode**: `WithWriteThroughOnly()` keeps the hot path (`Get`/`ModifyNumeric`/`SetNX`) fully in-memory — Redis is used purely as an async write-behind layer and is never read synchronously on a local miss. +* **Single Round-Trip Redis Reads**: Added the optional `RedisGetTTLClient` interface (implemented by the bundled v8/v9 adapters via a pipeline). When available, a Redis read fetches value + TTL in one round-trip instead of separate `GET` and `TTL` calls. +* **Startup Warmup**: `WarmFromRedis(keys)` loads existing Redis state into local memory at startup, for durability across restarts without runtime synchronous reads. +* **Local-Authoritative SetNX**: In write-through-only mode, `SetNX` guarantees atomicity via the local shard lock and pushes the Redis `SETNX` asynchronously. The original cluster-wide synchronous behavior remains the default. +* **Back-Pressure Metrics & Blocking**: Async write drops (when the queue is full) are now counted and exposed via `DroppedWrites()`. `WithBlockOnFull(d)` blocks briefly instead of dropping immediately. * **SetNX Support**: Added atomic `SetNX` (Set if Not Exists) operations seamlessly synchronized with Redis. * **Automatic Redis Fetching**: Enabled automatic Redis fetching for local cache misses and expired items, enhancing multi-instance synchronization. * **Configurable Timeouts**: Fine-tune Redis L2 operation timeouts for all cache types. @@ -121,6 +130,72 @@ c.WithRedisTimeout(500 * time.Millisecond) val, err := c.Sync("key") ``` +#### How reads work (default: read-through) + +By default the cache is **read-through**: a `Get` first checks local memory, and on a local miss (or a locally-expired item) it transparently falls back to Redis. If the key exists in Redis, the value is pulled back into local memory and returned. This keeps multiple instances loosely synchronized — a value written by one worker becomes visible to others on their next miss. + +The Redis read fetches both the value and its remaining TTL. When the configured adapter implements `RedisGetTTLClient` (the bundled `redisv8` / `redisv9` adapters do, via a pipeline), this is a **single round-trip**; otherwise it falls back to a separate `GET` + `TTL`. + +--- + +### Memory-Authoritative Mode (Write-Behind Only) + +For latency-critical, high-throughput workloads you can make memory the source of truth and demote Redis to a pure **async write-behind** layer. In this mode the hot path is *never* blocked by a synchronous Redis read. + +Enable it with `WithWriteThroughOnly()` (available on `Cache`, `NumericCache`, `ShardedCache`, and `ShardedNumericCache`): + +```go +c := cache.NewShardedCache[MyStruct](16, 5*time.Minute, 10*time.Minute) +c.WithRedis(redisv9.New(rdb)). + WithRedisTimeout(500 * time.Millisecond). + WithWriteThroughOnly() +defer c.Close() +``` + +What changes in this mode: + +* **Reads stay in memory.** `Get`, `ModifyNumeric`, `SetNX`, and the `SetCache` getters no longer fall back to Redis on a local miss — a miss is a miss. After warmup, the hot path is 100% in-memory. +* **Writes still persist.** `Set`, `Delete`, `Incr`/`Decr`, etc. are still propagated to Redis asynchronously through the write-behind worker. +* **`SetNX` becomes local-authoritative.** Atomicity is guaranteed by the local shard lock (exactly one winner per process), and the Redis `SETNX` is sent asynchronously. The signature and return value are unchanged. *Note:* cluster-wide uniqueness is no longer enforced synchronously — use the default mode if you need that guarantee. +* **Explicit reads still hit Redis.** `Sync(key)` and `WarmFromRedis(keys)` deliberately bypass this mode, since they are not on the hot path. + +#### Warming up from Redis at startup + +To get durability across restarts *without* paying for synchronous reads at runtime, load your state from Redis once at boot: + +```go +// Rebuild in-memory state for known keys (e.g. after a restart). +loaded, err := c.WarmFromRedis([]string{"user:1", "user:2", "user:3"}) +fmt.Printf("warmed %d keys from Redis\n", loaded) +``` + +Keys missing from Redis are skipped silently; `loaded` is the number of keys actually restored. For `ShardedCache`, keys are automatically routed to the correct shard. An error is returned only if Redis is not configured. + +--- + +### Back-Pressure: Dropped Writes & Blocking + +Async Redis writes flow through a buffered queue. If writes are produced faster than the Redis worker can drain them, the queue fills up. By default an overflowing write is **dropped** (to avoid stalling the application), but it is now **counted** rather than silently lost: + +```go +// Number of async Redis writes dropped because the queue was full. +// On ShardedCache this is summed across all shards. +dropped := c.DroppedWrites() +if dropped > 0 { + log.Printf("WARNING: %d cache writes were dropped before reaching Redis", dropped) +} +``` + +If losing a write is unacceptable (for example, an item about to be evicted from memory), configure the producer to **block briefly** instead of dropping: + +```go +// When the queue is full, block for up to 50ms waiting for room. +// If still full after the timeout, the write is dropped and counted. +c.WithBlockOnFull(50 * time.Millisecond) +``` + +> **Trade-off:** blocking applies back-pressure to the calling goroutine (and, for `Set`/`Delete`, briefly while holding the shard lock), so keep the timeout short. A zero or negative duration restores the default drop-immediately behavior. + --- ### Capacity & Eviction diff --git a/cache.go b/cache.go index 705228e..fa0d4b3 100644 --- a/cache.go +++ b/cache.go @@ -9,6 +9,7 @@ import ( "os" "runtime" "sync" + "sync/atomic" "time" ) @@ -22,6 +23,18 @@ type RedisClient interface { TTL(ctx context.Context, key string) (time.Duration, error) } +// RedisGetTTLClient is an optional extension of RedisClient. If a configured +// Redis client also implements this interface, the cache uses GetWithTTL to +// fetch a value and its remaining TTL in a single round-trip instead of issuing +// a separate Get and TTL call. Adapters typically implement this with a Redis +// pipeline (GET + TTL) or GETEX. +type RedisGetTTLClient interface { + // GetWithTTL returns the raw value bytes and the remaining TTL for key. + // A negative ttl means "no expiration"; an error means the key is missing + // or unreadable. + GetWithTTL(ctx context.Context, key string) (value []byte, ttl time.Duration, err error) +} + type Item[V any] struct { Object V Expiration time.Time @@ -70,6 +83,18 @@ type cache[V any] struct { ctx context.Context wg sync.WaitGroup redisTimeout time.Duration + // writeThroughOnly enables "memory-authoritative" mode: Redis is used only + // as an async write-behind layer and is never read on a local miss in the + // hot path (Get/getFallback/ModifyNumeric/SetNX). Explicit reads (Sync, + // WarmFromRedis) still hit Redis. Default false keeps the original + // read-through-on-miss behavior. + writeThroughOnly bool + // redisBlockTimeout, when > 0, makes async Redis enqueues block for up to + // this duration when redisCh is full instead of dropping immediately. + redisBlockTimeout time.Duration + // droppedWrites counts async Redis writes dropped because redisCh was full + // (and could not be enqueued within redisBlockTimeout, if set). + droppedWrites atomic.Uint64 } type redisOp int @@ -77,6 +102,7 @@ type redisOp int const ( redisOpSet redisOp = iota redisOpDel + redisOpSetNX ) type redisItem[V any] struct { @@ -97,12 +123,37 @@ func (c *cache[V]) Set(k string, x V, d time.Duration) { // SetNX adds an item to the cache only if it doesn't already exist. // Returns true if the item was added, false otherwise. -// If Redis is configured, it performs a synchronous SETNX on Redis. +// +// By default, when Redis is configured, atomicity is guaranteed cluster-wide by +// performing a synchronous SETNX on Redis (the original behavior). +// +// In write-through-only mode (see WithWriteThroughOnly), SetNX is +// local-authoritative: atomicity is guaranteed solely by the local shard lock, +// the SETNX on Redis is performed asynchronously, and the method never blocks on +// Redis. This keeps the hot path fully in-memory at the cost of cluster-wide +// uniqueness guarantees. func (c *cache[V]) SetNX(k string, x V, d time.Duration) bool { if d == DefaultExpiration { d = c.defaultExpiration } + // Local-authoritative path: decide entirely under the shard lock and push + // the SETNX to Redis asynchronously. + if c.writeThroughOnly { + c.mu.Lock() + item, found := c.items[k] + if found && !item.Expired() { + c.mu.Unlock() + return false + } + c.setLocal(k, x, d) + c.mu.Unlock() + + // Best-effort async persistence; keeps the same op semantics on Redis. + c.enqueueRedis(redisItem[V]{k: k, v: x, d: d, op: redisOpSetNX}) + return true + } + c.mu.Lock() item, found := c.items[k] if found && !item.Expired() { @@ -159,16 +210,49 @@ func (c *cache[V]) set(k string, x V, d time.Duration) { c.evict() // Async Redis Write - if c.redisClient != nil && c.redisCh != nil { - // Use non-blocking send or buffered - select { - case c.redisCh <- redisItem[V]{k: k, v: x, d: d, op: redisOpSet}: - default: - // Channel full, drop write to avoid blocking app - } + c.enqueueRedis(redisItem[V]{k: k, v: x, d: d, op: redisOpSet}) +} + +// enqueueRedis hands an operation to the async Redis worker. By default it is +// non-blocking: if redisCh is full the write is dropped and droppedWrites is +// incremented (use DroppedWrites to observe this). When a block timeout has been +// configured via WithBlockOnFull, it instead blocks for up to that duration +// before falling back to dropping the write. +func (c *cache[V]) enqueueRedis(item redisItem[V]) { + if c.redisClient == nil || c.redisCh == nil { + return + } + + // Fast path: try a non-blocking send first. + select { + case c.redisCh <- item: + return + default: + } + + if c.redisBlockTimeout <= 0 { + // Drop-on-full, but account for it instead of silently losing the write. + c.droppedWrites.Add(1) + return + } + + // Block briefly rather than drop immediately. + timer := time.NewTimer(c.redisBlockTimeout) + defer timer.Stop() + select { + case c.redisCh <- item: + case <-timer.C: + c.droppedWrites.Add(1) } } +// DroppedWrites returns the number of async Redis writes that have been dropped +// because the write-behind queue was full. A persistently non-zero and growing +// value indicates the Redis worker cannot keep up with the write rate. +func (c *cache[V]) DroppedWrites() uint64 { + return c.droppedWrites.Load() +} + // evict clears 25% of the cache if it's full. // This prevents thrashing (add 1, delete 1, add 1, delete 1...) func (c *cache[V]) evict() { @@ -309,7 +393,8 @@ func (c *cache[V]) getFallbackWithExpiration(k string) (V, time.Time, bool) { // Sync fetches the value for the given key from Redis (if available) and updates the local cache. // It returns the value and an error if the key was not found in Redis or if Redis is not configured. func (c *cache[V]) Sync(k string) (V, error) { - val, ttl, found := c.fetchFromRedis(k) + // Sync is an explicit read, so it bypasses write-through-only mode. + val, ttl, found := c.fetchFromRedisForce(k) if !found { var zero V return zero, fmt.Errorf("item %s not found in Redis", k) @@ -322,6 +407,33 @@ func (c *cache[V]) Sync(k string) (V, error) { return val, nil } +// WarmFromRedis loads the given keys from Redis into the local cache. It is +// intended to be called at startup to rebuild in-memory state (e.g. dedup state) +// after a restart, so the hot path can run fully in-memory afterwards — this is +// especially useful together with WithWriteThroughOnly, which disables +// synchronous Redis reads at runtime. +// +// It returns the number of keys successfully loaded. Keys missing from Redis are +// skipped silently. An error is returned only when Redis is not configured. +func (c *cache[V]) WarmFromRedis(keys []string) (int, error) { + if c.redisClient == nil { + return 0, fmt.Errorf("redis is not configured") + } + + loaded := 0 + for _, k := range keys { + val, ttl, found := c.fetchFromRedisForce(k) + if !found { + continue + } + c.mu.Lock() + c.setLocal(k, val, ttl) + c.mu.Unlock() + loaded++ + } + return loaded, nil +} + func (c *cache[V]) get(k string) (V, bool) { item, found := c.items[k] if !found || item.Expired() { @@ -336,8 +448,23 @@ func (c *cache[V]) get(k string) (V, bool) { return item.Object, true } -// Fetches from Redis without modifying local cache. Safe to call without lock. +// fetchFromRedis fetches from Redis without modifying the local cache. Safe to +// call without a lock. In write-through-only mode it short-circuits and reports +// a miss, so callers on the hot path (Get/getFallback/ModifyNumeric/SetNX) never +// perform a synchronous Redis read. Explicit reads (Sync, WarmFromRedis) use +// fetchFromRedisForce to bypass this. func (c *cache[V]) fetchFromRedis(k string) (V, time.Duration, bool) { + if c.writeThroughOnly { + var zero V + return zero, 0, false + } + return c.fetchFromRedisForce(k) +} + +// fetchFromRedisForce always reads from Redis (ignoring write-through-only mode). +// It uses a single round-trip via RedisGetTTLClient when the configured client +// supports it, otherwise it falls back to a separate Get + TTL. +func (c *cache[V]) fetchFromRedisForce(k string) (V, time.Duration, bool) { var zero V if c.redisClient == nil { return zero, 0, false @@ -350,18 +477,36 @@ func (c *cache[V]) fetchFromRedis(k string) (V, time.Duration, bool) { defer cancel() } - val, err := c.redisClient.Get(ctx, k) - if err != nil { - return zero, 0, false + var ( + val []byte + ttl time.Duration + err error + ) + if gt, ok := c.redisClient.(RedisGetTTLClient); ok { + // Single round-trip: value + remaining TTL together. + val, ttl, err = gt.GetWithTTL(ctx, k) + if err != nil { + return zero, 0, false + } + } else { + // Fallback: two round-trips for clients that don't support GetWithTTL. + val, err = c.redisClient.Get(ctx, k) + if err != nil { + return zero, 0, false + } + ttl, err = c.redisClient.TTL(ctx, k) + if err != nil { + ttl = DefaultExpiration + } } + if ttl < 0 { + ttl = DefaultExpiration + } + var obj V if err := json.Unmarshal(val, &obj); err != nil { return zero, 0, false } - ttl, err := c.redisClient.TTL(ctx, k) - if err != nil || ttl < 0 { - ttl = DefaultExpiration - } return obj, ttl, true } @@ -434,6 +579,51 @@ func (c *cache[V]) withRedisTimeout(d time.Duration) { c.mu.Unlock() } +func (c *cache[V]) withWriteThroughOnly() { + c.mu.Lock() + c.writeThroughOnly = true + c.mu.Unlock() +} + +func (c *cache[V]) withBlockOnFull(d time.Duration) { + c.mu.Lock() + c.redisBlockTimeout = d + c.mu.Unlock() +} + +// WithWriteThroughOnly switches the cache into "memory-authoritative" mode: +// Redis becomes a pure async write-behind layer. After warmup, the hot path +// (Get/getFallback/ModifyNumeric/SetNX) is served entirely from memory and never +// reads Redis synchronously on a local miss. Writes are still propagated to Redis +// asynchronously via the write-behind worker, and explicit reads (Sync, +// WarmFromRedis) still consult Redis. +func (c *Cache[V]) WithWriteThroughOnly() *Cache[V] { + c.cache.withWriteThroughOnly() + return c +} + +// WithBlockOnFull makes async Redis enqueues block for up to d when the +// write-behind queue is full, instead of dropping the write immediately. If the +// write still cannot be enqueued within d it is dropped and counted by +// DroppedWrites. A zero or negative d restores the default drop-immediately +// behavior. +func (c *Cache[V]) WithBlockOnFull(d time.Duration) *Cache[V] { + c.cache.withBlockOnFull(d) + return c +} + +// WithWriteThroughOnly enables memory-authoritative mode. See Cache.WithWriteThroughOnly. +func (c *NumericCache[N]) WithWriteThroughOnly() *NumericCache[N] { + c.cache.withWriteThroughOnly() + return c +} + +// WithBlockOnFull configures blocking-on-full behavior. See Cache.WithBlockOnFull. +func (c *NumericCache[N]) WithBlockOnFull(d time.Duration) *NumericCache[N] { + c.cache.withBlockOnFull(d) + return c +} + // ModifyNumeric atomically modifies a numeric item in the cache. // If the item does not exist or is expired, it is set to `operand`. // If isIncrement is true, `operand` is added to the existing value. @@ -489,10 +679,7 @@ func (c *NumericCache[N]) ModifyNumeric(k string, operand N, isIncrement bool) ( d = -1 // NoExpiration } - select { - case c.redisCh <- redisItem[N]{k: k, v: newVal, d: d, op: redisOpSet}: - default: - } + c.enqueueRedis(redisItem[N]{k: k, v: newVal, d: d, op: redisOpSet}) } return newVal, nil @@ -525,23 +712,13 @@ func (c *cache[V]) delete(k string) (V, bool) { if v, found := c.items[k]; found { delete(c.items, k) // Redis Async Delete - if c.redisClient != nil && c.redisCh != nil { - select { - case c.redisCh <- redisItem[V]{k: k, op: redisOpDel}: - default: - } - } + c.enqueueRedis(redisItem[V]{k: k, op: redisOpDel}) return v.Object, true } } delete(c.items, k) // Redis Async Delete (even if onEvicted is nil) - if c.redisClient != nil && c.redisCh != nil { - select { - case c.redisCh <- redisItem[V]{k: k, op: redisOpDel}: - default: - } - } + c.enqueueRedis(redisItem[V]{k: k, op: redisOpDel}) var zero V return zero, false } @@ -790,6 +967,12 @@ func (c *cache[V]) redisWorker(ch chan redisItem[V]) { if err == nil { c.redisClient.Set(c.ctx, item.k, data, item.d) } + case redisOpSetNX: + // Best-effort async SETNX for local-authoritative SetNX. + data, err := json.Marshal(item.v) + if err == nil { + c.redisClient.SetNX(c.ctx, item.k, data, item.d) + } case redisOpDel: c.redisClient.Del(c.ctx, item.k) } diff --git a/cache_redis_test.go b/cache_redis_test.go index 89faf56..02482f5 100644 --- a/cache_redis_test.go +++ b/cache_redis_test.go @@ -18,6 +18,26 @@ func (r *RedisAdapter) Get(ctx context.Context, key string) ([]byte, error) { return r.client.Get(ctx, key).Bytes() } +// GetWithTTL fetches value + remaining TTL in a single round-trip via a pipeline, +// implementing cache.RedisGetTTLClient. +func (r *RedisAdapter) GetWithTTL(ctx context.Context, key string) ([]byte, time.Duration, error) { + pipe := r.client.Pipeline() + getCmd := pipe.Get(ctx, key) + ttlCmd := pipe.TTL(ctx, key) + if _, err := pipe.Exec(ctx); err != nil { + return nil, 0, err + } + b, err := getCmd.Bytes() + if err != nil { + return nil, 0, err + } + ttl, err := ttlCmd.Result() + if err != nil { + ttl = 0 + } + return b, ttl, nil +} + func (r *RedisAdapter) Set(ctx context.Context, key string, value any, expiration time.Duration) error { return r.client.Set(ctx, key, value, expiration).Err() } diff --git a/redis/v8/adapter.go b/redis/v8/adapter.go index d9ec817..5288600 100644 --- a/redis/v8/adapter.go +++ b/redis/v8/adapter.go @@ -20,6 +20,29 @@ func (a *Adapter) Get(ctx context.Context, key string) ([]byte, error) { return a.client.Get(ctx, key).Bytes() } +// GetWithTTL fetches the value and its remaining TTL in a single round-trip +// using a pipeline (GET + TTL). It implements cache.RedisGetTTLClient so the +// cache can avoid issuing two separate round-trips on a Redis read. +func (a *Adapter) GetWithTTL(ctx context.Context, key string) ([]byte, time.Duration, error) { + pipe := a.client.Pipeline() + getCmd := pipe.Get(ctx, key) + ttlCmd := pipe.TTL(ctx, key) + if _, err := pipe.Exec(ctx); err != nil { + // Exec returns the first command error (e.g. redis.Nil when the key + // is missing), which we surface as a miss. + return nil, 0, err + } + b, err := getCmd.Bytes() + if err != nil { + return nil, 0, err + } + ttl, err := ttlCmd.Result() + if err != nil { + ttl = 0 + } + return b, ttl, nil +} + func (a *Adapter) Set(ctx context.Context, key string, value any, expiration time.Duration) error { if expiration < time.Millisecond { expiration = time.Millisecond diff --git a/redis/v9/adapter.go b/redis/v9/adapter.go index 7883698..0ed6f21 100644 --- a/redis/v9/adapter.go +++ b/redis/v9/adapter.go @@ -20,6 +20,29 @@ func (a *Adapter) Get(ctx context.Context, key string) ([]byte, error) { return a.client.Get(ctx, key).Bytes() } +// GetWithTTL fetches the value and its remaining TTL in a single round-trip +// using a pipeline (GET + TTL). It implements cache.RedisGetTTLClient so the +// cache can avoid issuing two separate round-trips on a Redis read. +func (a *Adapter) GetWithTTL(ctx context.Context, key string) ([]byte, time.Duration, error) { + pipe := a.client.Pipeline() + getCmd := pipe.Get(ctx, key) + ttlCmd := pipe.TTL(ctx, key) + if _, err := pipe.Exec(ctx); err != nil { + // Exec returns the first command error (e.g. redis.Nil when the key + // is missing), which we surface as a miss. + return nil, 0, err + } + b, err := getCmd.Bytes() + if err != nil { + return nil, 0, err + } + ttl, err := ttlCmd.Result() + if err != nil { + ttl = 0 + } + return b, ttl, nil +} + func (a *Adapter) Set(ctx context.Context, key string, value any, expiration time.Duration) error { if expiration < time.Millisecond { expiration = time.Millisecond diff --git a/sharded.go b/sharded.go index 9f196db..e9d0fab 100644 --- a/sharded.go +++ b/sharded.go @@ -268,3 +268,63 @@ func (sc *ShardedNumericCache[N]) WithRedisTimeout(d time.Duration) *ShardedNume sc.ShardedCache.WithRedisTimeout(d) return sc } + +// WithWriteThroughOnly switches every shard into memory-authoritative mode. +// See Cache.WithWriteThroughOnly. +func (sc *ShardedCache[V]) WithWriteThroughOnly() *ShardedCache[V] { + for _, c := range sc.cs { + c.withWriteThroughOnly() + } + return sc +} + +// WithBlockOnFull configures blocking-on-full behavior on every shard. +// See Cache.WithBlockOnFull. +func (sc *ShardedCache[V]) WithBlockOnFull(d time.Duration) *ShardedCache[V] { + for _, c := range sc.cs { + c.withBlockOnFull(d) + } + return sc +} + +// WarmFromRedis loads the given keys from Redis into the appropriate shards. +// See cache.WarmFromRedis. It returns the total number of keys loaded. +func (sc *ShardedCache[V]) WarmFromRedis(keys []string) (int, error) { + // Group keys by shard so each shard is warmed in one call. + byShard := make(map[int][]string) + for _, k := range keys { + idx := shardKey(k, sc.numBuckets) + byShard[idx] = append(byShard[idx], k) + } + + total := 0 + for idx, shardKeys := range byShard { + n, err := sc.cs[idx].WarmFromRedis(shardKeys) + total += n + if err != nil { + return total, err + } + } + return total, nil +} + +// DroppedWrites returns the total number of async Redis writes dropped across all +// shards because their write-behind queues were full. +func (sc *ShardedCache[V]) DroppedWrites() (total uint64) { + for _, c := range sc.cs { + total += c.DroppedWrites() + } + return +} + +// WithWriteThroughOnly enables memory-authoritative mode. See Cache.WithWriteThroughOnly. +func (sc *ShardedNumericCache[N]) WithWriteThroughOnly() *ShardedNumericCache[N] { + sc.ShardedCache.WithWriteThroughOnly() + return sc +} + +// WithBlockOnFull configures blocking-on-full behavior. See Cache.WithBlockOnFull. +func (sc *ShardedNumericCache[N]) WithBlockOnFull(d time.Duration) *ShardedNumericCache[N] { + sc.ShardedCache.WithBlockOnFull(d) + return sc +} diff --git a/sharded_redis_test.go b/sharded_redis_test.go index 43972cb..bef4dab 100644 --- a/sharded_redis_test.go +++ b/sharded_redis_test.go @@ -277,6 +277,246 @@ func TestRedisIntegration_NumericAndSet(t *testing.T) { }) } +// newMiniRedisAdapter spins up an in-process miniredis and returns an adapter +// plus the *miniredis.Miniredis handle (for direct inspection) and a cleanup. +func newMiniRedisAdapter(t *testing.T) (*RedisAdapter, *miniredis.Miniredis) { + t.Helper() + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("Could not start miniredis: %v", err) + } + t.Cleanup(mr.Close) + + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { rdb.Close() }) + return &RedisAdapter{client: rdb}, mr +} + +// TestWriteThroughOnly_NoSyncRead verifies that in memory-authoritative mode the +// hot path never reads Redis on a local miss, while async writes still land in +// Redis and explicit Sync/WarmFromRedis still read it. +func TestWriteThroughOnly_NoSyncRead(t *testing.T) { + adapter, _ := newMiniRedisAdapter(t) + + sc := NewShardedCache[RedisTestStruct](4, 5*time.Minute, 0) + sc.WithRedis(adapter).WithRedisTimeout(time.Second).WithWriteThroughOnly() + defer sc.Close() + + key := "wto_key" + val := RedisTestStruct{Name: "wto", Value: 7} + + // Write goes to local immediately and to Redis asynchronously. + sc.Set(key, val, 0) + + // Hot-path read hits local cache. + if got, found := sc.Get(key); found != Found || got != val { + t.Fatalf("expected local hit, got %v found=%v", got, found) + } + + // Wait for async write-behind to reach Redis. + ctx := context.Background() + deadline := time.Now().Add(2 * time.Second) + for { + if _, err := adapter.Get(ctx, key); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("async write never reached Redis") + } + time.Sleep(10 * time.Millisecond) + } + + // Flush local cache: in write-through-only mode the hot path must NOT + // fall back to Redis, so Get should now be a Miss even though the key + // still exists in Redis. + sc.Flush() + if _, found := sc.Get(key); found != Miss { + t.Errorf("expected Miss in write-through-only mode after flush, got found=%v", found) + } + + // Explicit Sync bypasses write-through-only and reads Redis. + got, err := sc.Sync(key) + if err != nil { + t.Fatalf("Sync should read Redis even in write-through-only mode: %v", err) + } + if got != val { + t.Errorf("Sync returned %v, want %v", got, val) + } + + // After Sync the value is back in local cache and served from memory. + if got, found := sc.Get(key); found != Found || got != val { + t.Errorf("expected local hit after Sync, got %v found=%v", got, found) + } +} + +// TestWarmFromRedis verifies that startup warmup loads existing Redis state into +// local memory, routed to the correct shards. +func TestWarmFromRedis(t *testing.T) { + adapter, _ := newMiniRedisAdapter(t) + ctx := context.Background() + + // Pre-populate Redis directly (as if a previous process had persisted it). + keys := []string{"warm_a", "warm_b", "warm_c"} + for i, k := range keys { + data, _ := json.Marshal(RedisTestStruct{Name: k, Value: i}) + if err := adapter.Set(ctx, k, data, time.Hour); err != nil { + t.Fatalf("seed Redis: %v", err) + } + } + + sc := NewShardedCache[RedisTestStruct](4, 5*time.Minute, 0) + sc.WithRedis(adapter).WithRedisTimeout(time.Second).WithWriteThroughOnly() + defer sc.Close() + + // Nothing in local cache yet, and the hot path won't read Redis. + if _, found := sc.Get("warm_a"); found != Miss { + t.Fatalf("expected Miss before warmup") + } + + // Warm a subset (plus one missing key, which must be skipped). + loaded, err := sc.WarmFromRedis([]string{"warm_a", "warm_b", "missing"}) + if err != nil { + t.Fatalf("WarmFromRedis: %v", err) + } + if loaded != 2 { + t.Errorf("expected 2 keys loaded, got %d", loaded) + } + + // Warmed keys are now served from memory. + if got, found := sc.Get("warm_a"); found != Found || got.Name != "warm_a" { + t.Errorf("warm_a not loaded into local cache: got %v found=%v", got, found) + } + if got, found := sc.Get("warm_b"); found != Found || got.Name != "warm_b" { + t.Errorf("warm_b not loaded into local cache: got %v found=%v", got, found) + } + // Un-warmed key remains a local miss (no sync read). + if _, found := sc.Get("warm_c"); found != Miss { + t.Errorf("warm_c should not be loaded, got found=%v", found) + } +} + +// TestSetNX_WriteThroughOnly_LocalAuthoritative verifies that in +// memory-authoritative mode SetNX is decided by the local shard lock (exactly +// one winner) and the Redis SETNX is performed asynchronously. +func TestSetNX_WriteThroughOnly_LocalAuthoritative(t *testing.T) { + adapter, _ := newMiniRedisAdapter(t) + + sc := NewShardedCache[string](4, DefaultExpiration, 0) + sc.WithRedis(adapter).WithRedisTimeout(time.Second).WithWriteThroughOnly() + defer sc.Close() + + key := "wto_setnx_key" + val := "wto_setnx_val" + + var wg sync.WaitGroup + var mu sync.Mutex + successCount := 0 + numGoroutines := 100 + startCh := make(chan struct{}) + wg.Add(numGoroutines) + for i := 0; i < numGoroutines; i++ { + go func() { + defer wg.Done() + <-startCh + if sc.SetNX(key, val, DefaultExpiration) { + mu.Lock() + successCount++ + mu.Unlock() + } + }() + } + close(startCh) + wg.Wait() + + if successCount != 1 { + t.Errorf("expected exactly 1 local-authoritative SetNX winner, got %d", successCount) + } + + // Value is present in local cache immediately. + if v, found := sc.Get(key); found != Found || v != val { + t.Errorf("expected local value %q, got %q found=%v", val, v, found) + } + + // The async SETNX should eventually persist to Redis. + ctx := context.Background() + deadline := time.Now().Add(2 * time.Second) + for { + if data, err := adapter.Get(ctx, key); err == nil { + var res string + json.Unmarshal(data, &res) + if res == val { + break + } + } + if time.Now().After(deadline) { + t.Fatalf("async SETNX never reached Redis") + } + time.Sleep(10 * time.Millisecond) + } +} + +// TestDroppedWrites_Counter verifies that overflowing the async write-behind +// queue increments the dropped-writes counter instead of silently losing writes. +func TestDroppedWrites_Counter(t *testing.T) { + // Block the worker by using an adapter whose Set sleeps, so the buffered + // channel fills up and subsequent non-blocking enqueues are dropped. + adapter, _ := newMiniRedisAdapter(t) + slow := &slowSetAdapter{RedisAdapter: adapter, delay: 2 * time.Millisecond} + + c := New[int](NoExpiration, 0) + c.WithRedis(slow) + defer c.Close() + + // Fire far more writes than the channel buffer (1000) can absorb while the + // worker is stalled, so some must be dropped. + for i := 0; i < 5000; i++ { + c.Set(fmt.Sprintf("k-%d", i), i, 0) + } + + if c.DroppedWrites() == 0 { + t.Errorf("expected some dropped writes when the queue overflows, got 0") + } +} + +// TestBlockOnFull_ReducesDrops verifies WithBlockOnFull lets a saturated queue +// drain rather than dropping immediately. With blocking enabled, the producer is +// throttled to the worker's rate, so nothing is dropped. +func TestBlockOnFull_ReducesDrops(t *testing.T) { + adapter, _ := newMiniRedisAdapter(t) + slow := &slowSetAdapter{RedisAdapter: adapter, delay: time.Millisecond} + + c := New[int](NoExpiration, 0) + c.WithRedis(slow).WithBlockOnFull(5 * time.Second) + defer c.Close() + + // 1500 writes > 1000 buffer. With a generous block timeout the producer + // waits for room instead of dropping. + for i := 0; i < 1500; i++ { + c.Set(fmt.Sprintf("bk-%d", i), i, 0) + } + + if dropped := c.DroppedWrites(); dropped != 0 { + t.Errorf("expected 0 dropped writes with WithBlockOnFull, got %d", dropped) + } +} + +// slowSetAdapter wraps RedisAdapter and slows down Set/SetNX to simulate a Redis +// worker that cannot keep up with the producer. +type slowSetAdapter struct { + *RedisAdapter + delay time.Duration +} + +func (s *slowSetAdapter) Set(ctx context.Context, key string, value any, expiration time.Duration) error { + time.Sleep(s.delay) + return s.RedisAdapter.Set(ctx, key, value, expiration) +} + +func (s *slowSetAdapter) SetNX(ctx context.Context, key string, value any, expiration time.Duration) (bool, error) { + time.Sleep(s.delay) + return s.RedisAdapter.SetNX(ctx, key, value, expiration) +} + func TestShardedCache_SetNX_RedisConcurrency(t *testing.T) { mr, err := miniredis.Run() if err != nil {