diff --git a/CHANGELOG.md b/CHANGELOG.md index 39055db..0084fd4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed +- `GetMulti` no longer spins (re-locking and re-invoking the callback) until the + context deadline when the callback omits a requested key. It now fails fast + with an error naming the missing keys. Callbacks must return a value for every + key they are given (use an empty string for a known-absent entity). +- `SetMulti` now reports keys the callback locked but never returned a value for + via `BatchError` instead of silently reverting them and returning `nil`. +- Refresh-ahead callbacks are no longer capped at `LockTTL`; a callback slower + than `LockTTL` was previously cancelled mid-flight and mis-reported as an + error, silently disabling refresh for slow-to-recompute values. The successful + back-write now runs under a fresh context so a callback that consumed most of + its budget still persists its value. +- The distributed refresh lock is released with an ownership CAS (and now holds a + unique token instead of a fixed value), so a refresh that overran its TTL can + no longer delete a concurrent refresher's lock. +- `tryGetMulti` guards against a truncated `DoMultiCache` response length, + matching `waitForReadLocks`, so a short pipeline fails loud instead of silently + dropping keys. + +### Added +- `RefreshTimeout` option: bounds a refresh-ahead callback's compute time + independently of `LockTTL`. Defaults to the value's per-call `ttl`. +- Per-call `ttl` validation. `Get`, `GetMulti`, `Set`, `SetMulti`, `Touch`, + `TouchMulti`, `ForceSet`, and `ForceSetMulti` reject a non-positive `ttl` up + front with a clear error instead of running the callback and surfacing an + opaque Redis "invalid expire time" — and `Touch`/`TouchMulti` no longer delete + the key via `PEXPIRE 0`. + +### Changed +- `ForceSetMulti` wraps its first error with key context (matches `DelMulti`). + ## [v0.2.0] - 2026-05-04 ### Breaking diff --git a/README.md b/README.md index 0db1495..bb6c490 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ A cache-aside implementation for Redis, built on the [rueidis](https://github.co ## Requirements -- Go 1.23+ +- Go 1.24+ - Redis 7+ ## Installation @@ -149,6 +149,7 @@ func (r Repository) GetByIDs(ctx context.Context, keys []string) (map[string]str | `RefreshLockPrefix` | `string` | `"__redcache:refresh:"` | Prefix for distributed refresh-ahead locks. | | `RefreshAfterFraction` | `float64` | `0` (disabled) | Fraction of TTL after which a refresh-ahead is triggered. Must be in `[0, 1)`. | | `RefreshBeta` | `float64` | `0` (XFetch off) | Scales the XFetch probabilistic-refresh window. `0` keeps refresh deterministic at the floor; `1.0` is the canonical XFetch beta from Vattani et al. Higher values trigger refresh earlier within the floor, weighted by how long the value took to compute. | +| `RefreshTimeout` | `time.Duration` | `0` (uses the value's ttl) | Compute budget for a refresh-ahead callback, decoupled from `LockTTL`. Without it a callback slower than `LockTTL` is cancelled and reported as an error. Must not be negative. | | `RefreshWorkers` | `int` | `4` (when refresh enabled) | Background workers processing refresh jobs. | | `RefreshQueueSize` | `int` | `64` (when refresh enabled) | Capacity of the refresh job queue. Jobs are silently dropped when full; the stale value continues to serve. | @@ -166,6 +167,10 @@ client, err := redcache.NewRedCacheAside( RefreshQueueSize: 64, }, ) +// Close() drains the refresh workers on shutdown. It does not close the rueidis +// client, so close that separately. +defer client.Close() +defer client.Client().Close() ``` ### XFetch probabilistic refresh @@ -191,7 +196,7 @@ For workloads that already use `RefreshAfterFraction` + `RefreshBeta`, XFetch ha Implement `Metrics` (or embed `NoopMetrics` and override the methods you care about) to wire counters into Prometheus, OpenTelemetry, or any other backend. Methods are called on the hot path and must be concurrent-safe. -High-volume events (`CacheHits`, `CacheMisses`, `LockContended`, `RefreshTriggered`, `RefreshSkipped`, `RefreshDropped`) are aggregated per operation and emitted once with a count rather than once per key. Diagnostic events (`LockLost`, `RefreshError`, `RefreshPanicked`) carry the affected key. +High-volume events (`CacheHits`, `CacheMisses`, `LockContended`, `RefreshTriggered`, `RefreshSkipped`, `RefreshDropped`) are aggregated per operation and emitted once with a count rather than once per key. `LockWaitDuration` fires once per resolved lock wait. Diagnostic events (`LockLost`, `RefreshError`, `RefreshPanicked`) carry the affected key; `InvalidationError` fires when an invalidation message can't be parsed (no key). ```go type myMetrics struct { diff --git a/cacheaside.go b/cacheaside.go index 0de613b..3298003 100644 --- a/cacheaside.go +++ b/cacheaside.go @@ -34,8 +34,9 @@ // executes the callback function for a given key at a time. Other goroutines will wait // for the lock to be released and then return the cached value. // -// Locks are implemented using Redis SET NX with a configurable TTL. Lock values use -// UUIDv7 for uniqueness and are prefixed (default: "__redcache:lock:") to avoid +// Locks are implemented using Redis SET NX with a configurable TTL. Lock values +// combine a per-instance UUIDv7 with an atomic counter (avoiding a UUID +// allocation per lock) and are prefixed (default: "__redcache:lock:") to avoid // collisions with application data. // // # Context and Timeouts @@ -63,6 +64,7 @@ import ( "errors" "fmt" "log/slog" + "slices" "strconv" "strings" "sync" @@ -156,6 +158,7 @@ type CacheAside struct { lockPrefix string refreshAfter float64 // 0 = disabled refreshBeta float64 // XFetch beta; 0 = simple floor only + refreshTimeout time.Duration // refresh callback budget; 0 = use per-call ttl refreshing syncx.Map[string, struct{}] // dedup in-flight refreshes (local) refreshPrefix string // prefix for distributed refresh lock keys refreshQueue chan refreshJob // worker pool job queue (nil when disabled) @@ -219,6 +222,14 @@ type CacheAsideOption struct { // refresh it relative to a serial path — the tradeoff for not having to // instrument each per-key compute time. RefreshBeta float64 + // RefreshTimeout bounds how long a refresh-ahead callback may run before it + // is cancelled, decoupling refresh compute budget from LockTTL. Without it a + // callback slower than LockTTL is cancelled mid-flight and reported as an + // error, silently disabling refresh-ahead for the slow-to-recompute values + // it exists to help. Defaults to the per-call ttl of the value being + // refreshed when 0. Must not be negative. Only meaningful when + // RefreshAfterFraction > 0. + RefreshTimeout time.Duration // RefreshWorkers is the number of background workers that process refresh-ahead // jobs. Defaults to 4 when RefreshAfterFraction > 0. Must be > 0 when refresh // is enabled. @@ -272,6 +283,9 @@ func validateRefreshDefaults(caOption *CacheAsideOption) error { if caOption.RefreshBeta < 0 { return errors.New("RefreshBeta must not be negative") } + if caOption.RefreshTimeout < 0 { + return errors.New("RefreshTimeout must not be negative") + } if caOption.RefreshAfterFraction == 0 { return nil } @@ -290,6 +304,18 @@ func validateRefreshDefaults(caOption *CacheAsideOption) error { return nil } +// validateTTL rejects a non-positive per-call ttl before any Redis work or user +// callback runs. Redis SET PX / PEXPIRE require a positive millisecond value, so +// a ttl that rounds to <= 0ms would otherwise surface as an opaque Redis +// "invalid expire time" error on a miss (Get/GetMulti/Set/SetMulti) or silently +// delete the key via PEXPIRE 0 (Touch/TouchMulti). +func validateTTL(ttl time.Duration) error { + if ttl.Milliseconds() <= 0 { + return fmt.Errorf("redcache: ttl must be at least 1ms, got %s", ttl) + } + return nil +} + // NewRedCacheAside creates a CacheAside with the given Redis client and cache-aside options. func NewRedCacheAside(clientOption rueidis.ClientOption, caOption CacheAsideOption) (*CacheAside, error) { if err := validateAndApplyDefaults(clientOption, &caOption); err != nil { @@ -311,6 +337,7 @@ func NewRedCacheAside(clientOption rueidis.ClientOption, caOption CacheAsideOpti lockPrefix: caOption.LockPrefix, refreshAfter: caOption.RefreshAfterFraction, refreshBeta: caOption.RefreshBeta, + refreshTimeout: caOption.RefreshTimeout, refreshPrefix: caOption.RefreshLockPrefix, } // Force a single connection per node so client-side cache reads and the @@ -552,6 +579,9 @@ func (rca *CacheAside) Get( key string, fn func(ctx context.Context, key string) (val string, err error), ) (string, error) { + if err := validateTTL(ttl); err != nil { + return "", err + } retry: wait, leader := rca.register(key) res, err := rca.tryGet(ctx, ttl, key) @@ -649,6 +679,9 @@ func (rca *CacheAside) DelMulti(ctx context.Context, keys ...string) error { // Use Touch to implement sliding-TTL semantics (sessions, tokens) without // re-running the origin function. func (rca *CacheAside) Touch(ctx context.Context, ttl time.Duration, key string) error { + if err := validateTTL(ttl); err != nil { + return err + } ttlMs := strconv.FormatInt(ttl.Milliseconds(), 10) if err := touchScript.Exec(ctx, rca.client, []string{key}, []string{ttlMs, rca.lockPrefix}).Error(); err != nil { return fmt.Errorf("touch key %q: %w", key, err) @@ -664,6 +697,9 @@ func (rca *CacheAside) TouchMulti(ctx context.Context, ttl time.Duration, keys . if len(keys) == 0 { return nil } + if err := validateTTL(ttl); err != nil { + return err + } stmtsBySlot := rca.groupTouchExecs(ttl, keys) if firstErrKey, firstErr := rca.runTouchSlots(ctx, stmtsBySlot); firstErr != nil { return fmt.Errorf("touch key %q: %w", firstErrKey, firstErr) @@ -847,15 +883,26 @@ func (rca *CacheAside) unlock(ctx context.Context, key string, lock string) erro // GetMulti returns cached values for the given keys, populating any misses by calling fn. // SET operations are grouped by Redis cluster slot for efficient batching. +// +// fn is invoked with the subset of keys that missed and must return a value for +// every key it is given (use an empty string for a known-absent entity). If fn +// omits a requested key, GetMulti returns an error naming the missing keys +// rather than retrying indefinitely. +// +// ttl must be positive (at least 1ms); a non-positive ttl returns an error +// before fn runs. func (rca *CacheAside) GetMulti( ctx context.Context, ttl time.Duration, keys []string, - fn func(ctx context.Context, key []string) (val map[string]string, err error), + fn func(ctx context.Context, keys []string) (val map[string]string, err error), ) (map[string]string, error) { if len(keys) == 0 { return map[string]string{}, nil } + if err := validateTTL(ttl); err != nil { + return nil, err + } res := make(map[string]string, len(keys)) // Parallel slices: pending[i] is an unresolved key, chans[i] is its wait @@ -944,7 +991,7 @@ func (rca *CacheAside) runLeaderSets( ctx context.Context, ttl time.Duration, leaderKeys []string, - fn func(ctx context.Context, key []string) (val map[string]string, err error), + fn func(ctx context.Context, keys []string) (val map[string]string, err error), res map[string]string, ) error { n := 0 @@ -995,6 +1042,11 @@ func (rca *CacheAside) tryGetMulti(ctx context.Context, ttl time.Duration, keys } } resps := rca.client.DoMultiCache(ctx, multi...) + // Defensive: a truncated pipeline response would otherwise misalign resps[i] + // with keys[i] and silently drop keys. Mirrors waitForReadLocks' guard. + if len(resps) != len(keys) { + return needRefresh, fmt.Errorf("tryGetMulti: got %d responses for %d keys", len(resps), len(keys)) + } for i, resp := range resps { val, err := resp.ToString() @@ -1021,7 +1073,7 @@ func (rca *CacheAside) trySetMultiKeyFn( ctx context.Context, ttl time.Duration, keys []string, - fn func(ctx context.Context, key []string) (val map[string]string, err error), + fn func(ctx context.Context, keys []string) (val map[string]string, err error), res map[string]string, ) error { lockVals, err := rca.tryLockMulti(ctx, keys) @@ -1052,6 +1104,14 @@ func (rca *CacheAside) trySetMultiKeyFn( if err != nil { return err } + // Enforce the callback contract: fn must return a value for every key it was + // given. A locked key the callback omits never lands in res, which would make + // GetMulti re-lock the key and re-invoke fn each retry until the caller's + // context deadline expires. Fail fast with a key-specific error instead; the + // deferred unlock releases the locks we acquired. + if missing := keysMissingFrom(lockVals, vals); len(missing) > 0 { + return fmt.Errorf("callback returned no value for keys %q", missing) + } // XFetch metadata: amortise total fn time across the values it produced. // Skewed estimate (computing 100 keys in parallel reports the same per-key // delta as a serial loop), but it's better than 0: it gives slow batches @@ -1086,6 +1146,19 @@ func perValueDelta(total time.Duration, n int) time.Duration { return total / time.Duration(n) } +// keysMissingFrom returns the keys present in locked but absent from result, +// sorted so the resulting error message is stable. +func keysMissingFrom(locked map[string]string, result map[string]string) []string { + var missing []string + for k := range locked { + if _, ok := result[k]; !ok { + missing = append(missing, k) + } + } + slices.Sort(missing) + return missing +} + func (rca *CacheAside) tryLockMulti(ctx context.Context, keys []string) (map[string]string, error) { lockVals := make(map[string]string, len(keys)) cmdsP := commandsPool.GetCap(len(keys)) diff --git a/cacheaside_refresh.go b/cacheaside_refresh.go index 98ed1cf..9eb68e6 100644 --- a/cacheaside_refresh.go +++ b/cacheaside_refresh.go @@ -6,11 +6,25 @@ import ( "math/rand/v2" "runtime/debug" "strconv" + "sync" "time" "github.com/redis/rueidis" + + "github.com/dcbickfo/redcache/internal/cmdx" + "github.com/dcbickfo/redcache/internal/mapsx" ) +// refreshBudget returns how long a refresh-ahead callback may run. RefreshTimeout +// overrides the per-call ttl when set; otherwise the value's own ttl is the +// compute budget. +func (rca *CacheAside) refreshBudget(ttl time.Duration) time.Duration { + if rca.refreshTimeout > 0 { + return rca.refreshTimeout + } + return ttl +} + // refreshJob is a unit of work for the refresh worker pool. Holding fields // directly (rather than a closure) avoids an allocation per trigger on the // hot path. Exactly one of singleFn / multiFn is set; runRefreshJob dispatches @@ -170,15 +184,19 @@ func (rca *CacheAside) doSingleRefresh( key string, fn func(ctx context.Context, key string) (string, error), ) { - refreshCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), rca.lockTTL) + budget := rca.refreshBudget(ttl) + refreshCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget) defer cancel() // Distributed dedup: SET NX on a separate refresh lock key. IsRedisNil // signals "another node is refreshing" (healthy contention); other errors // are real Redis problems and must be reported separately so operators can - // distinguish a healthy dedup signal from a broken Redis. + // distinguish a healthy dedup signal from a broken Redis. The lock value is a + // unique token (not a fixed "1") so release can CAS-check ownership, and its + // TTL is the compute budget so it covers a long refresh without expiring. refreshKey := rca.refreshKeyFor(key) - err := rca.client.Do(refreshCtx, rca.client.B().Set().Key(refreshKey).Value("1").Nx().Px(rca.lockTTL).Build()).Error() + token := rca.lockPool.Generate() + err := rca.client.Do(refreshCtx, rca.client.B().Set().Key(refreshKey).Value(token).Nx().Px(budget).Build()).Error() if err != nil { if rueidis.IsRedisNil(err) { rca.emitRefreshSkipped(1) @@ -191,7 +209,9 @@ func (rca *CacheAside) doSingleRefresh( defer func() { cleanupCtx, cleanupCancel := rca.cleanupCtx(ctx) defer cleanupCancel() - if delErr := rca.client.Do(cleanupCtx, rca.client.B().Del().Key(refreshKey).Build()).Error(); delErr != nil { + // CAS release: delete only if we still hold our own token, so a refresh + // that overran its TTL can't delete a successor's lock. + if delErr := delKeyLua.Exec(cleanupCtx, rca.client, []string{refreshKey}, []string{token}).Error(); delErr != nil { rca.logger.Error("refresh-ahead lock release failed", "key", key, "refreshKey", refreshKey, "error", delErr) } }() @@ -205,8 +225,13 @@ func (rca *CacheAside) doSingleRefresh( } wrapped := wrapEnvelope(val, time.Since(start)) + // Write under a fresh cleanup context, not refreshCtx: a callback that + // consumed most of its budget would otherwise hit an expired context here and + // lose its successfully-computed value. + writeCtx, writeCancel := rca.cleanupCtx(ctx) + defer writeCancel() ttlMs := strconv.FormatInt(ttl.Milliseconds(), 10) - if err := refreshAheadSetScript.Exec(refreshCtx, rca.client, []string{key}, []string{wrapped, ttlMs, rca.lockPrefix}).Error(); err != nil { + if err := refreshAheadSetScript.Exec(writeCtx, rca.client, []string{key}, []string{wrapped, ttlMs, rca.lockPrefix}).Error(); err != nil { rca.logger.Error("refresh-ahead set failed", "key", key, "error", err) rca.emitRefreshError(key) } @@ -256,14 +281,16 @@ func (rca *CacheAside) doMultiRefresh( keys []string, fn func(ctx context.Context, keys []string) (map[string]string, error), ) { - refreshCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), rca.lockTTL) + budget := rca.refreshBudget(ttl) + refreshCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget) defer cancel() - lockedKeys := rca.acquireRefreshLocks(refreshCtx, keys) - if len(lockedKeys) == 0 { + lockedTokens := rca.acquireRefreshLocks(refreshCtx, keys, budget) + if len(lockedTokens) == 0 { return } - defer rca.deleteRefreshLocks(ctx, lockedKeys) + defer rca.deleteRefreshLocks(ctx, lockedTokens) + lockedKeys := mapsx.Keys(lockedTokens) start := time.Now() vals, err := fn(refreshCtx, lockedKeys) @@ -275,22 +302,30 @@ func (rca *CacheAside) doMultiRefresh( return } - rca.setRefreshedValues(refreshCtx, ttl, vals, perValueDelta(time.Since(start), len(vals))) + // Write under a fresh cleanup context (see doSingleRefresh) so a slow-but- + // successful callback still persists its values. + writeCtx, writeCancel := rca.cleanupCtx(ctx) + defer writeCancel() + rca.setRefreshedValues(writeCtx, ttl, vals, perValueDelta(time.Since(start), len(vals))) } -// acquireRefreshLocks batch-acquires distributed SET NX locks for refresh keys. -// Distinguishes IsRedisNil (healthy dedup → RefreshSkipped) from real Redis -// errors (→ RefreshError + log). -func (rca *CacheAside) acquireRefreshLocks(ctx context.Context, keys []string) []string { +// acquireRefreshLocks batch-acquires distributed SET NX locks for refresh keys, +// returning a map of data key to the unique token held for that key (used by +// deleteRefreshLocks to CAS-release). Each lock's TTL is the compute budget so a +// long refresh does not lose its dedup lock mid-flight. Distinguishes IsRedisNil +// (healthy dedup → RefreshSkipped) from real Redis errors (→ RefreshError + log). +func (rca *CacheAside) acquireRefreshLocks(ctx context.Context, keys []string, budget time.Duration) map[string]string { cmdsP := commandsPool.Get(len(keys)) defer commandsPool.Put(cmdsP) cmds := *cmdsP + tokens := make([]string, len(keys)) for i, key := range keys { - cmds[i] = rca.client.B().Set().Key(rca.refreshKeyFor(key)).Value("1").Nx().Px(rca.lockTTL).Build() + tokens[i] = rca.lockPool.Generate() + cmds[i] = rca.client.B().Set().Key(rca.refreshKeyFor(key)).Value(tokens[i]).Nx().Px(budget).Build() } resps := rca.client.DoMulti(ctx, cmds...) - var locked []string + locked := make(map[string]string, len(keys)) var skipped int for i, resp := range resps { if err := resp.Error(); err != nil { @@ -302,30 +337,55 @@ func (rca *CacheAside) acquireRefreshLocks(ctx context.Context, keys []string) [ } continue } - locked = append(locked, keys[i]) + locked[keys[i]] = tokens[i] } rca.emitRefreshSkipped(skipped) return locked } -// deleteRefreshLocks removes distributed refresh lock keys (best effort). -// Failures are logged so operators can investigate; a stuck refresh lock -// disables refresh-ahead for that key for one lockTTL window. -func (rca *CacheAside) deleteRefreshLocks(ctx context.Context, keys []string) { +// deleteRefreshLocks CAS-releases distributed refresh locks (best effort), +// deleting each key only if it still holds our token so an overran refresh can't +// delete a successor's lock. Lua scripts run per cluster slot (refresh keys are +// hash-tagged to their data key's slot), fanning out across slots in parallel — +// mirroring unlockMulti. Failures are logged; a stuck lock disables refresh for +// that key until its TTL expires. +func (rca *CacheAside) deleteRefreshLocks(ctx context.Context, tokens map[string]string) { + if len(tokens) == 0 { + return + } cleanupCtx, cleanupCancel := rca.cleanupCtx(ctx) defer cleanupCancel() - delCmdsP := commandsPool.Get(len(keys)) - defer commandsPool.Put(delCmdsP) - delCmds := *delCmdsP - for i, key := range keys { - delCmds[i] = rca.client.B().Del().Key(rca.refreshKeyFor(key)).Build() + type keyedExec struct { + key string + exec rueidis.LuaExec } - resps := rca.client.DoMulti(cleanupCtx, delCmds...) - for i, resp := range resps { - if err := resp.Error(); err != nil { - rca.logger.Error("refresh-ahead lock release failed", "key", keys[i], "error", err) - } + bySlot := make(map[uint16][]keyedExec) + for key, token := range tokens { + refreshKey := rca.refreshKeyFor(key) + slot := cmdx.Slot(refreshKey) + bySlot[slot] = append(bySlot[slot], keyedExec{ + key: key, + exec: rueidis.LuaExec{Keys: []string{refreshKey}, Args: []string{token}}, + }) + } + var wg sync.WaitGroup + for _, stmts := range bySlot { + wg.Add(1) + go func() { + defer wg.Done() + execs := make([]rueidis.LuaExec, len(stmts)) + for i, s := range stmts { + execs[i] = s.exec + } + resps := delKeyLua.ExecMulti(cleanupCtx, rca.client, execs...) + for i, resp := range resps { + if err := resp.Error(); err != nil { + rca.logger.Error("refresh-ahead lock release failed", "key", stmts[i].key, "error", err) + } + } + }() } + wg.Wait() } // setRefreshedValues writes refreshed values via a CAS-style Lua script that diff --git a/cacheaside_test.go b/cacheaside_test.go index e429c41..87a3cf7 100644 --- a/cacheaside_test.go +++ b/cacheaside_test.go @@ -1639,6 +1639,15 @@ func TestRefreshAhead_FractionValidation(t *testing.T) { require.Error(t, err) assert.Contains(t, err.Error(), "RefreshQueueSize") }) + t.Run("negative RefreshTimeout", func(t *testing.T) { + t.Parallel() + _, err := redcache.NewRedCacheAside( + rueidis.ClientOption{InitAddress: addr}, + redcache.CacheAsideOption{RefreshAfterFraction: 0.8, RefreshTimeout: -time.Second}, + ) + require.Error(t, err) + assert.Contains(t, err.Error(), "RefreshTimeout") + }) t.Run("custom workers and queue", func(t *testing.T) { t.Parallel() client, err := redcache.NewRedCacheAside( diff --git a/callback_contract_test.go b/callback_contract_test.go new file mode 100644 index 0000000..2b120d0 --- /dev/null +++ b/callback_contract_test.go @@ -0,0 +1,133 @@ +package redcache_test + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/redis/rueidis" + "github.com/stretchr/testify/require" + + "github.com/dcbickfo/redcache" +) + +// GetMulti must fail fast with a key-specific error when the callback omits a +// requested key, rather than spinning (re-locking + re-invoking fn) until the +// caller's context deadline expires. +func TestCacheAside_GetMulti_CallbackOmitsKey_FailsFast(t *testing.T) { + t.Parallel() + client := makeClient(t, addr) + defer client.Client().Close() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + present := "k:present:" + uuid.New().String() + omitted := "k:omitted:" + uuid.New().String() + keys := []string{present, omitted} + + start := time.Now() + _, err := client.GetMulti(ctx, 10*time.Second, keys, func(ctx context.Context, ks []string) (map[string]string, error) { + res := make(map[string]string) + for _, k := range ks { + if k == omitted { + continue // violate the contract: omit a requested key + } + res[k] = "val:" + k + } + return res, nil + }) + require.Error(t, err) + require.NotErrorIs(t, err, context.DeadlineExceeded, "must not hang until the deadline") + require.ErrorContains(t, err, omitted, "error should name the omitted key") + require.Less(t, time.Since(start), 1500*time.Millisecond, "should fail fast, not spin to the deadline") +} + +// SetMulti must surface keys the callback omits (locked but never returned) in +// the BatchError rather than silently reverting them and returning nil. +func TestPrimeableCacheAside_SetMulti_CallbackOmitsKey_ReportedInBatchError(t *testing.T) { + t.Parallel() + client := makePrimeableClient(t, addr) + defer client.Client().Close() + ctx := context.Background() + + present := "k:present:" + uuid.New().String() + omitted := "k:omitted:" + uuid.New().String() + keys := []string{present, omitted} + + err := client.SetMulti(ctx, 10*time.Second, keys, func(ctx context.Context, ks []string) (map[string]string, error) { + res := make(map[string]string) + for _, k := range ks { + if k == omitted { + continue // violate the contract: omit a locked key + } + res[k] = "val:" + k + } + return res, nil + }) + require.Error(t, err) + var be *redcache.BatchError + require.ErrorAs(t, err, &be) + require.True(t, be.HasError(omitted), "omitted key must be reported as failed") + require.False(t, be.HasError(present), "present key must not be reported as failed") + + // The present key was cached (Get is a hit); the omitted key was reverted + // (Get is a miss that runs the callback). + presentCalled := false + v, err := client.Get(ctx, 10*time.Second, present, func(ctx context.Context, k string) (string, error) { + presentCalled = true + return "x", nil + }) + require.NoError(t, err) + require.False(t, presentCalled, "present key should be a cache hit") + require.Equal(t, "val:"+present, v) + + omittedCalled := false + _, err = client.Get(ctx, 10*time.Second, omitted, func(ctx context.Context, k string) (string, error) { + omittedCalled = true + return "y", nil + }) + require.NoError(t, err) + require.True(t, omittedCalled, "omitted key should be a cache miss (it was reverted)") +} + +// Get must reject a non-positive ttl up front with a clear error rather than +// running the callback and surfacing an opaque Redis "invalid expire time". +func TestCacheAside_Get_RejectsNonPositiveTTL(t *testing.T) { + t.Parallel() + client := makeClient(t, addr) + defer client.Client().Close() + ctx := context.Background() + + called := false + _, err := client.Get(ctx, 0, "k:"+uuid.New().String(), func(ctx context.Context, k string) (string, error) { + called = true + return "v", nil + }) + require.Error(t, err) + require.ErrorContains(t, err, "ttl") + require.False(t, called, "callback must not run when ttl is invalid") +} + +// Touch must reject a non-positive ttl rather than issuing PEXPIRE 0, which +// would delete the key. +func TestCacheAside_Touch_RejectsNonPositiveTTL(t *testing.T) { + t.Parallel() + client := makeClient(t, addr) + defer client.Client().Close() + ctx := context.Background() + + key := "k:" + uuid.New().String() + _, err := client.Get(ctx, 10*time.Second, key, func(ctx context.Context, k string) (string, error) { + return "v", nil + }) + require.NoError(t, err) + + err = client.Touch(ctx, 0, key) + require.Error(t, err) + require.ErrorContains(t, err, "ttl") + + raw := client.Client() + getErr := raw.Do(ctx, raw.B().Get().Key(key).Build()).Error() + require.False(t, rueidis.IsRedisNil(getErr), "Touch(ttl=0) must not delete the key") +} diff --git a/docs/design-docs/2026-05-05-typed-cache-aside-design.md b/docs/design-docs/2026-05-05-typed-cache-aside-design.md new file mode 100644 index 0000000..6a7c072 --- /dev/null +++ b/docs/design-docs/2026-05-05-typed-cache-aside-design.md @@ -0,0 +1,380 @@ +# Typed Cache-Aside (Generics + Codec) + +**Status:** Draft — pending implementation +**Author:** David +**Date:** 2026-05-05 + +## Summary + +Add typed views over `*CacheAside` and `*PrimeableCacheAside` so callers can store and retrieve arbitrary value types (and arbitrary key types) without hand-serializing at every call site. The existing string-typed public API is preserved unchanged. Generics live at the wrapper boundary only; the hot path inside the library still operates on strings. The lock / refresh / multi-slot machinery is unchanged. The only internal change is a small refactor of the multi-set error collector so it can serve both the legacy `BatchError` and the new `BatchKeyError[K]` from a single source — see **Implementation notes**. + +## Goals + +- Let users hold typed values (`User`, `Order`, etc.) and typed keys (`UserID`, composite ID structs, etc.) in cache-aside calls. +- Provide a sensible default value codec (JSON) so users who don't care about codecs get a working setup with zero configuration. +- Allow users to plug in their own value or key codec when JSON is wrong (msgpack, protobuf, custom binary, etc.). +- Keep the existing string-based public surface (`*CacheAside`, `*PrimeableCacheAside`, `BatchError`) byte-for-byte source-compatible. +- Keep the existing hot path (lock acquisition, Lua CAS-set, multi-slot batching, refresh-ahead) untouched. + +## Non-goals + +- Library-level absence / negative-cache semantics. Callers handle missing values via `*T`, `sql.Null[T]`, or a domain-specific sentinel inside `V`. The library does not introduce an `ErrNotFound` sentinel. +- Generic-ifying the underlying `CacheAside` type itself. Methods on the existing string types remain string-typed; users who want type safety construct a `Typed[K,V]` view. +- Any architecture / refactoring work in the existing string code. That work is captured in the **Follow-up work** section and lands after this PR. +- A rich type-safe key model beyond what `KeyCodec[K]` already provides. Users who want algebraic key types build them in their own packages. + +## Constraints and prior decisions + +The design pins down each of these explicitly. Each was settled during brainstorming. + +| Decision | Choice | +|---|---| +| API shape | Wrapper view (`Typed[K,V]` / `PrimeableTyped[K,V]`) over the existing string clients | +| Codec payload | `[]byte` | +| Default value codec | JSON | +| Absence semantics | Caller-owned (no library sentinel) | +| String API after this lands | Keeps both string and typed views exported | +| Generics depth | Boundary only — hot path stays string-based | +| Type parameter count | Two: `K` and `V` | +| Mapping to existing types | Two parallel wrappers (`Typed`, `PrimeableTyped`) mirror `CacheAside` / `PrimeableCacheAside` | +| `KeyCodec.EncodeKey` signature | Returns `(string, error)` | +| Codec shape | Interface (with provided concrete types) | +| String-key convenience | Provide `NewStringTyped` / `NewPrimeableStringTyped` | +| Encode-fail in user fn | Returned as callback error | +| Decode-fail on read | Returned wrapped (`ErrDecode` sentinel); no auto-evict | +| Encode-fail on `ForceSet` | Returned before any Redis call | +| `BatchError` evolution | Keep `BatchError` non-generic for the string API; add new `BatchKeyError[K comparable]` for typed; share internals | +| `KeyPrefix` view option | Dropped — users put prefix logic in their `KeyCodec` | + +## Public surface + +All new types live in the existing `redcache` package alongside `CacheAside` and `PrimeableCacheAside`. + +### Codec interfaces + +```go +// Codec encodes and decodes typed values to and from byte slices that the +// library stores inside the existing envelope payload. +// +// Implementations must be safe for concurrent use; the library invokes +// Encode and Decode from arbitrary goroutines (callbacks, refresh workers, +// multi-set fan-out). +// +// Slice-ownership contract: the slice returned from Encode is owned by the +// library after return — implementations must not retain or mutate it. The +// slice passed to Decode is owned by the caller (the library) — implementations +// must not retain or mutate it past the Decode call. +type Codec[V any] interface { + Encode(V) ([]byte, error) + Decode([]byte) (V, error) +} + +// KeyCodec converts a typed key to the Redis key string used to address +// the entry. Implementations must be deterministic: the same K must always +// produce the same string within the lifetime of a process. The output +// must be a valid Redis key (non-empty; the library does not validate +// further). Implementations must be safe for concurrent use. +type KeyCodec[K any] interface { + EncodeKey(K) (string, error) +} +``` + +### Provided concrete codecs + +```go +// JSONCodec serializes V via encoding/json. The default value codec. +type JSONCodec[V any] struct{} + +// BytesCodec is the identity codec for V = []byte. +type BytesCodec struct{} + +// StringCodec is the identity codec for V = string. +type StringCodec struct{} + +// StringKeyCodec is the identity codec for K = string. +type StringKeyCodec struct{} + +// KeyCodecFunc adapts a function value to a KeyCodec[K]. Callers supply +// `redcache.KeyCodecFunc[UserID](func(u UserID) (string, error) { ... })`. +type KeyCodecFunc[K any] func(K) (string, error) +func (f KeyCodecFunc[K]) EncodeKey(k K) (string, error) { return f(k) } +``` + +### Typed wrappers + +```go +// Typed is a type-safe view over a *CacheAside. Construct with NewTyped. +// One *CacheAside can be shared across many Typed views with different +// K / V instantiations and codecs. +type Typed[K comparable, V any] struct { + cache *CacheAside + keyCodec KeyCodec[K] + valCodec Codec[V] +} + +// PrimeableTyped is a type-safe view over a *PrimeableCacheAside, adding +// the Set/ForceSet methods to the read/delete/touch surface from Typed. +type PrimeableTyped[K comparable, V any] struct { + Typed[K, V] + primeable *PrimeableCacheAside +} +``` + +### Constructors + +```go +func NewTyped[K comparable, V any]( + c *CacheAside, + keyCodec KeyCodec[K], + valCodec Codec[V], +) *Typed[K, V] + +func NewPrimeableTyped[K comparable, V any]( + c *PrimeableCacheAside, + keyCodec KeyCodec[K], + valCodec Codec[V], +) *PrimeableTyped[K, V] + +// String-key convenience constructors. Equivalent to passing +// StringKeyCodec{} explicitly. +func NewStringTyped[V any](c *CacheAside, valCodec Codec[V]) *Typed[string, V] +func NewPrimeableStringTyped[V any](c *PrimeableCacheAside, valCodec Codec[V]) *PrimeableTyped[string, V] +``` + +### Typed methods (read / delete / touch — `*Typed[K,V]`) + +```go +func (t *Typed[K, V]) Get( + ctx context.Context, ttl time.Duration, k K, + fn func(ctx context.Context, k K) (V, error), +) (V, error) + +func (t *Typed[K, V]) GetMulti( + ctx context.Context, ttl time.Duration, keys []K, + fn func(ctx context.Context, missing []K) (map[K]V, error), +) (map[K]V, error) + +func (t *Typed[K, V]) Del(ctx context.Context, k K) error +func (t *Typed[K, V]) DelMulti(ctx context.Context, keys ...K) error +func (t *Typed[K, V]) Touch(ctx context.Context, ttl time.Duration, k K) error +func (t *Typed[K, V]) TouchMulti(ctx context.Context, ttl time.Duration, keys ...K) error +``` + +### Primeable methods (write — `*PrimeableTyped[K,V]`) + +```go +func (p *PrimeableTyped[K, V]) Set( + ctx context.Context, ttl time.Duration, k K, + fn func(ctx context.Context, k K) (V, error), +) error + +func (p *PrimeableTyped[K, V]) SetMulti( + ctx context.Context, ttl time.Duration, keys []K, + fn func(ctx context.Context, keys []K) (map[K]V, error), +) error + +func (p *PrimeableTyped[K, V]) ForceSet( + ctx context.Context, ttl time.Duration, k K, v V, +) error + +func (p *PrimeableTyped[K, V]) ForceSetMulti( + ctx context.Context, ttl time.Duration, values map[K]V, +) error +``` + +### Errors + +```go +// BatchError is unchanged. The string-API SetMulti / ForceSetMulti continue +// to return *BatchError for source compatibility. +type BatchError struct { + Failed map[string]error + Succeeded []string +} + +// BatchKeyError is the typed counterpart returned from PrimeableTyped's +// multi-set methods. Same method shape as BatchError but keyed by K. +// Reachable via errors.As, paralleling how BatchError is exposed today. +type BatchKeyError[K comparable] struct { + Failed map[K]error + Succeeded []K +} +func (e *BatchKeyError[K]) Error() string +func (e *BatchKeyError[K]) HasFailures() bool +func (e *BatchKeyError[K]) ErrorFor(k K) error +func (e *BatchKeyError[K]) HasError(k K) bool + +// NewBatchKeyError mirrors NewBatchError: returns nil (untyped) when there +// are no failures, so call sites can return it directly as `error`. +func NewBatchKeyError[K comparable](failed map[K]error, succeeded []K) error + +// Sentinel returned (wrapped) from typed reads when a stored value cannot +// be decoded. Library does not auto-evict; the caller decides whether to +// log, Del, or retry. +var ErrDecode = errors.New("redcache: decode failed") +``` + +The internal multi-set collector that powers both `BatchError` and `BatchKeyError` is shared. Two presentations, one source of truth — see **Implementation notes** below. + +## Data flow + +### Single-key Get + +``` +caller k + → EncodeKey(k) // returns (s string, err); err short-circuits, no Redis I/O + → cache.Get(ctx, ttl, s, wrappedFn) + wrappedFn: // closes over original k, valCodec + v, err := userFn(ctx, k) + if err != nil { return "", err } + b, err := valCodec.Encode(v) + if err != nil { return "", err } // surfaces as Get-callback error + return unsafeBytesToString(b), nil + ← raw string from cache + → valCodec.Decode(unsafeStringToBytes(raw)) + ← (V, ErrDecode-wrapped or nil) +``` + +### Multi-key Get + +``` +caller keys []K + → for each k: encode → encKeys []string, byEnc map[string]K + (any EncodeKey error short-circuits before Redis I/O) + → cache.GetMulti(ctx, ttl, encKeys, wrappedFn) + wrappedFn(missing []string): + missingK := look up each missing string in byEnc + result, err := userFn(ctx, missingK) + if err != nil { return nil, err } + out := make(map[string]string, len(result)) + for k, v := range result { + s, err := keyCodec.EncodeKey(k); if err != nil { return nil, err } + b, err := valCodec.Encode(v); if err != nil { return nil, err } + out[s] = unsafeBytesToString(b) + } + return out, nil + ← map[string]string from cache + → for each (s, raw): byEnc[s] → original K, valCodec.Decode(raw) → V + ← map[K]V (or first ErrDecode-wrapped error) +``` + +### Set / SetMulti / ForceSet / ForceSetMulti + +Same encode-on-write pattern. The public return type is `error`; multi-key partial failures expose `*BatchKeyError[K]` reachable via `errors.As`, paralleling how `*BatchError` is exposed in the existing string API. Encode failures inside the user `fn` are propagated as callback errors (lock released cleanly, nothing cached). Encode failures in `ForceSet*` short-circuit before any Redis I/O. + +### Del / Touch / DelMulti / TouchMulti + +Key-only operations. Encode each `K`, forward to the underlying string method. + +### Refresh-ahead transparency + +Refresh-ahead operates inside `*CacheAside` on the envelope-wrapped string payload. The typed wrapper's user `fn` is captured in the closure passed to the underlying `Get` / `GetMulti` call, and that closure is what the refresh worker invokes when the floor / XFetch fires. The wrapper does not need any refresh-specific code path — encoding happens inside the closure regardless of who triggers it. + +## Error semantics + +| Failure | Behavior | +|---|---| +| `KeyCodec.EncodeKey` returns error (any path) | Return error to caller before any Redis I/O. No lock attempt, no partial state. | +| User `fn` returns error in `Get` / `Set` / their multi variants | Forwarded as callback error from the underlying string call. Lock released; nothing cached. | +| `Codec.Encode` returns error inside the user `fn` wrapper | Treated identically to a callback error: surfaced to the caller, lock released, nothing cached. | +| `Codec.Encode` returns error in `ForceSet` / `ForceSetMulti` | Returned before any Redis I/O. `ForceSetMulti` collects per-key encode failures into `BatchKeyError[K].Failed` and proceeds with the rest. | +| `Codec.Decode` returns error on read (`Get` / `GetMulti`) | Returned wrapped: `fmt.Errorf("...: %w", ErrDecode)`. Library does not call `Del`. The cached value remains. | +| Multi-key partial failure (some keys CAS-mismatch / lock-lost) | Reported through `BatchKeyError[K]` exactly as the existing string `SetMulti` does through `BatchError`. | + +## Implementation notes + +### Same package + +Everything new lives in `package redcache` next to the existing files. New files: + +- `typed.go` — `Typed`, `PrimeableTyped`, constructors, single-key methods. +- `typed_multi.go` — multi-key methods (Get/Set/Del/Touch). +- `codec.go` — `Codec`, `KeyCodec`, `KeyCodecFunc`, `JSONCodec`, `BytesCodec`, `StringCodec`, `StringKeyCodec`. +- `errors.go` — extend with `BatchKeyError[K]` and `ErrDecode`. + +Tests: + +- `typed_test.go` — typed Get/GetMulti hot path through real Redis, mirroring the existing `cacheaside_test.go` structure. +- `typed_primeable_test.go` — Set / SetMulti / ForceSet / ForceSetMulti, including partial-failure → `BatchKeyError`. +- `typed_unit_test.go` — codec round-trip, `BatchKeyError` accessors, key-codec error short-circuit, decode-error wrapping. +- `typed_bench_test.go` — at least one Get + one GetMulti benchmark for codec overhead measurement. + +### Shared multi-set error collection + +Internally there is exactly one multi-set error collection helper. It builds a `BatchKeyError[string]` (the typed form is used internally because the type system makes it the natural choice; `K=string` for the legacy path is just an instantiation). At the legacy string-API boundary (in the existing `PrimeableCacheAside.SetMulti` / `ForceSetMulti` returns), the helper output is converted to `*BatchError` so the public type does not change. The typed wrapper uses `BatchKeyError[K]` directly. + +The conversion at the boundary is structural: `BatchError{Failed: bke.Failed, Succeeded: bke.Succeeded}`. Both types share the same `Error()` formatter logic via an unexported helper to avoid drift. + +### Unsafe string ↔ []byte at the codec boundary + +The library converts between `string` (rueidis) and `[]byte` (codec) at the typed wrapper boundary using `unsafe.String` / `unsafe.Slice`. Each conversion is read-only on the source side, so this is sound: the codec promises not to mutate the byte slice it receives, and the library never mutates a returned `string`. This avoids two allocations per Get and two per Set on the hot path. Document the contract on `Codec.Decode` ("must not mutate the input slice"). + +### Key constraint + +`K comparable` is required to use `K` as a Go map key (`map[K]V`, `map[string]K`). This excludes types containing slices, maps, or functions. Most realistic key types (numeric IDs, strings, struct compositions of comparable fields) satisfy this. + +### Concurrency + +`Typed[K,V]` and `PrimeableTyped[K,V]` are stateless apart from their codec and underlying client pointers. They are safe for concurrent use to the same extent the underlying `*CacheAside` and codec implementations are. No new goroutines, no new locks. + +### Allocation budget + +Per single-key Get, additional allocations relative to the string API: one for `EncodeKey` if it allocates, plus whatever `Codec.Encode` / `Codec.Decode` allocate. The wrapper itself adds zero. Per multi-key Get with `n` keys, additional allocations are `O(n)` for the `byEnc` map and the encoded-key slice, plus codec costs. + +## Testing strategy + +- **Codec unit tests.** Round-trip JSON for primitives, structs, slices, nested maps. Identity codecs verify byte equality. `KeyCodecFunc` adapter exercise. +- **Integration tests against real Redis** mirroring the structure of the existing `cacheaside_test.go`. Two parameterizations: `Typed[string, exampleStruct]` and `Typed[int64, exampleStruct]` to exercise both string-key and numeric-key paths. +- **Targeted regression tests:** + - Encode-fail in `fn` releases the lock (next caller acquires immediately) and does not cache anything. + - Decode-fail on read returns wrapped `ErrDecode` and leaves the key alive in Redis. + - GetMulti with a partial encode failure inside `fn` propagates the error and leaves no partial cache state. + - Key-codec error short-circuits before any Redis I/O (verified by checking Redis was not contacted via a mock or by call counts). + - Refresh-ahead through a `Typed` view works end-to-end (configure `RefreshAfterFraction`, observe re-fetch firing). +- **`BatchKeyError[K]` tests** mirroring `errors_test.go`: nil-safe accessors, `HasFailures`, `ErrorFor`, formatted output, sorted determinism. +- **Benchmarks.** Add `BenchmarkTypedGet` and `BenchmarkTypedGetMulti` for codec overhead measurement; compare against the existing string-API benchmarks. + +## Migration + +For external users of the library, this PR is **fully source-compatible**. The existing `*CacheAside`, `*PrimeableCacheAside`, `BatchError`, every option, every method signature, and every Lua script is unchanged. A user who never touches the new typed surface sees no diff. + +A user who wants to adopt the typed API does: + +```go +client, err := redcache.NewPrimeableCacheAside(ropts, caopts) +if err != nil { /* ... */ } + +users := redcache.NewPrimeableStringTyped[User](client, redcache.JSONCodec[User]{}) + +u, err := users.Get(ctx, time.Minute, "u-123", + func(ctx context.Context, key string) (User, error) { + // load from db + return user, nil + }, +) +``` + +## Follow-up work (out of scope for this PR) + +These were surfaced during the architecture audit during brainstorming and are deliberately **not** included in this PR. Each is independently shippable after the typed work lands. + +### Worth doing — clear ROI + +- **Metric-emit boilerplate consolidation.** `cacheaside.go` defines 13 near-identical 6-line helpers (`emitCacheHits`, `emitCacheMisses`, etc.) totaling ~80 lines. Collapse into a single helper or wrap the metrics interface. Pure refactor; no behavior change. +- **Lua-script contract tests.** Lua scripts (`lua_scripts.go`) are string literals; `KEYS[N]` / `ARGV[N]` drift between script body and Go-side wiring is only caught at runtime. Add a static test that scans each script for `KEYS[N]` / `ARGV[N]` references and asserts the call sites pass exactly that count. Cheap insurance against future bugs. +- **Slot-aware command executor extraction.** `primeable_writelock_helpers.go` repeats the slot-group → per-slot Lua exec → result-collect pattern across roughly five functions and ~130 LoC. Extract into a dedicated `slotExec` helper that consolidates retry / error parsing in one place. Estimated ~50 LoC saved; touches the multi-set hot path so requires careful regression coverage. + +### Lower priority + +- **Split `cacheaside.go`** (currently 1,322 lines) into single-key, multi-key, and lifecycle files. Pure reorganization; muddies git blame for limited gain. +- **Replace `goto retry` with a `for` loop** in `Get`. Cosmetic; the existing form is clear and efficient. + +### Acceptable as-is — no action + +- `syncx.Map[K,V]` interface{}-cast overhead. Intrinsic to `sync.Map`; the typed wrapper is already a net cohesion win. +- `lockEntry`'s `sync.Once` race-avoidance pattern. Already correct and well-commented. +- `internal/poolx` slice-reuse helpers. Well-bounded; no smell. + +## Open questions + +None. All decisions captured in the **Constraints and prior decisions** table above. \ No newline at end of file diff --git a/errors.go b/errors.go index 003e866..3f75142 100644 --- a/errors.go +++ b/errors.go @@ -11,6 +11,11 @@ import ( // This can occur if the lock TTL expires during callback execution or if Redis invalidates the lock. var ErrLockLost = errors.New("lock was lost or expired before value could be set") +// errCallbackNoValue marks a key that a multi-key callback locked but never +// returned a value for. SetMulti surfaces it via BatchError so the caller learns +// the key was not cached (it is rolled back to its prior value). +var errCallbackNoValue = errors.New("callback returned no value for key") + // BatchError represents partial failures in a multi-key operation. // Some keys may have succeeded while others failed. // diff --git a/internal/mapsx/maps.go b/internal/mapsx/maps.go index 653281e..1a28b90 100644 --- a/internal/mapsx/maps.go +++ b/internal/mapsx/maps.go @@ -9,12 +9,3 @@ func Keys[M ~map[K]V, K comparable, V any](m M) []K { } return keys } - -// Values returns the values of the map m in unspecified order. -func Values[M ~map[K]V, K comparable, V any](m M) []V { - values := make([]V, 0, len(m)) - for _, v := range m { - values = append(values, v) - } - return values -} diff --git a/internal/mapsx/maps_test.go b/internal/mapsx/maps_test.go index 575755e..1b690e6 100644 --- a/internal/mapsx/maps_test.go +++ b/internal/mapsx/maps_test.go @@ -29,25 +29,3 @@ func TestKeys(t *testing.T) { assert.ElementsMatch(t, expectedIntKeys, intKeys, "expected keys to match") } - -func TestValues(t *testing.T) { - t.Parallel() - // Test with an empty map - emptyMap := map[string]int{} - values := mapsx.Values(emptyMap) - assert.Lenf(t, values, 0, "expected no values for empty map") - - // Test with a map with some elements - sampleMap := map[string]int{"a": 1, "b": 2, "c": 3} - values = mapsx.Values(sampleMap) - expectedValues := []int{1, 2, 3} - - assert.ElementsMatch(t, expectedValues, values, "expected values to match") - - // Test with a map with different value types - intKeyMap := map[int]string{1: "one", 2: "two", 3: "three"} - strValues := mapsx.Values(intKeyMap) - expectedStrValues := []string{"one", "two", "three"} - - assert.ElementsMatch(t, expectedStrValues, strValues, "expected values to match") -} diff --git a/lua_scripts.go b/lua_scripts.go index 68649c3..d52e725 100644 --- a/lua_scripts.go +++ b/lua_scripts.go @@ -127,6 +127,12 @@ var ( // a real (non-lock) value. Skips if the key is missing (let normal Get-on-miss // handle population) or holds a lock value (a Get/Set is already in progress // and will write its own current value). Returns 1 on write, 0 if skipped. + // + // The write is NOT a value-CAS: it has no ownership token to compare against, + // so a lock-free ForceSet that lands between this script's GET and SET can be + // overwritten by the (slightly staler) recomputed value. Refresh-ahead is + // best-effort — the staleness is bounded by ttl and self-heals on the next + // refresh; callers needing strict last-writer-wins should use Set/ForceSet. refreshAheadSetScript = rueidis.NewLuaScript(` local cur = redis.call("GET", KEYS[1]) if cur == false then diff --git a/primeable_cacheaside.go b/primeable_cacheaside.go index 037dd3c..360925a 100644 --- a/primeable_cacheaside.go +++ b/primeable_cacheaside.go @@ -55,6 +55,9 @@ func (pca *PrimeableCacheAside) Set( key string, fn func(ctx context.Context, key string) (string, error), ) error { + if err := validateTTL(ttl); err != nil { + return err + } lockVal := pca.lockPool.Generate() for { @@ -145,6 +148,12 @@ func (pca *PrimeableCacheAside) acquireSingleWriteLock( // // The callback receives currently-held lock keys in undefined order (map // iteration). Callers needing a stable order should sort the slice before use. +// fn must return a value for every key it is given; a key the callback omits is +// rolled back to its prior value (not cached) and reported as failed in the +// returned *BatchError. +// +// ttl must be positive (at least 1ms); a non-positive ttl returns an error +// before any lock is taken. // // On partial CAS failure, returns a *BatchError listing succeeded and failed keys. // On full success, returns nil. @@ -157,6 +166,9 @@ func (pca *PrimeableCacheAside) SetMulti( if len(keys) == 0 { return nil } + if err := validateTTL(ttl); err != nil { + return err + } // Wait for any existing read locks on these keys. if err := pca.waitForReadLocks(ctx, keys); err != nil { @@ -192,10 +204,25 @@ func (pca *PrimeableCacheAside) SetMulti( return nil } - // Roll back any keys that weren't successfully written. Restore (rather - // than unlock) so a CAS transport/parse error preserves the prior real - // value captured during acquire. For lock-lost keys the restore Lua's - // CAS-check fails harmlessly (stealer's value stays). + failed = pca.rollbackUnsetKeys(ctx, lockValues, savedValues, vals, succeeded, failed) + return NewBatchError(failed, succeeded) +} + +// rollbackUnsetKeys reverts every locked key that was not successfully written +// and returns the per-key failure map. "Not written" covers two cases: a CAS +// failure (already in failed) and a key the callback omitted (absent from vals) +// — the latter is surfaced as errCallbackNoValue so it doesn't vanish from the +// BatchError. Restore (rather than unlock) preserves the prior real value +// captured during acquire; for lock-lost keys the restore Lua's CAS-check fails +// harmlessly (the stealer's value stays). +func (pca *PrimeableCacheAside) rollbackUnsetKeys( + ctx context.Context, + lockValues map[string]string, + savedValues map[string]savedValue, + vals map[string]string, + succeeded []string, + failed map[string]error, +) map[string]error { succeededSet := make(map[string]struct{}, len(succeeded)) for _, s := range succeeded { succeededSet[s] = struct{}{} @@ -209,8 +236,15 @@ func (pca *PrimeableCacheAside) SetMulti( if len(toRestore) > 0 { pca.restoreMultiValues(ctx, toRestore, savedValues) } - - return NewBatchError(failed, succeeded) + for key := range lockValues { + if _, ok := vals[key]; !ok { + if failed == nil { + failed = make(map[string]error) + } + failed[key] = errCallbackNoValue + } + } + return failed } // ForceSet unconditionally writes a value to Redis, bypassing all locks. @@ -224,6 +258,9 @@ func (pca *PrimeableCacheAside) SetMulti( // sampling treats it as "no compute-time information" — which falls back to // the simple floor check, preserving prior behaviour for unconditional writes. func (pca *PrimeableCacheAside) ForceSet(ctx context.Context, ttl time.Duration, key, value string) error { + if err := validateTTL(ttl); err != nil { + return err + } return pca.client.Do(ctx, pca.client.B().Set().Key(key).Value(wrapEnvelope(value, 0)).Px(ttl).Build()).Error() } @@ -240,6 +277,9 @@ func (pca *PrimeableCacheAside) ForceSetMulti(ctx context.Context, ttl time.Dura if len(values) == 0 { return nil } + if err := validateTTL(ttl); err != nil { + return err + } cmdsP := commandsPool.GetCap(len(values)) defer commandsPool.Put(cmdsP) keyOrder := make([]string, 0, len(values)) @@ -249,15 +289,19 @@ func (pca *PrimeableCacheAside) ForceSetMulti(ctx context.Context, ttl time.Dura } resps := pca.client.DoMulti(ctx, *cmdsP...) var firstErr error + var firstErrKey string for i, resp := range resps { if err := resp.Error(); err != nil { pca.logger.Error("ForceSetMulti key failed", "key", keyOrder[i], "error", err) if firstErr == nil { - firstErr = err + firstErr, firstErrKey = err, keyOrder[i] } } } - return firstErr + if firstErr != nil { + return fmt.Errorf("force set key %q: %w", firstErrKey, firstErr) + } + return nil } // waitForReadLocks registers all keys, batch-reads them, and waits for any that diff --git a/refresh_timeout_test.go b/refresh_timeout_test.go new file mode 100644 index 0000000..3d5bc71 --- /dev/null +++ b/refresh_timeout_test.go @@ -0,0 +1,72 @@ +package redcache_test + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + "github.com/redis/rueidis" + "github.com/stretchr/testify/require" + + "github.com/dcbickfo/redcache" +) + +// A refresh-ahead callback slower than LockTTL must still complete and persist +// its value when RefreshTimeout gives it a larger compute budget — instead of +// being cancelled at LockTTL and reported as an error. +func TestRefreshAhead_RefreshTimeoutExtendsCallbackBudget(t *testing.T) { + t.Parallel() + metrics := &capturingMetrics{} + client, err := redcache.NewRedCacheAside( + rueidis.ClientOption{InitAddress: addr}, + redcache.CacheAsideOption{ + LockTTL: 200 * time.Millisecond, + RefreshAfterFraction: 0.01, + RefreshTimeout: 3 * time.Second, + Metrics: metrics, + }, + ) + require.NoError(t, err) + t.Cleanup(func() { + client.Close() + client.Client().Close() + }) + + ctx := context.Background() + key := "refresh-timeout:" + uuid.New().String() + var calls atomic.Int32 + + // First call populates quickly; refresh calls take longer than LockTTL but + // well under RefreshTimeout, and honour context cancellation. + cb := func(ctx context.Context, _ string) (string, error) { + if calls.Add(1) == 1 { + return "initial", nil + } + select { + case <-time.After(400 * time.Millisecond): + return "refreshed", nil + case <-ctx.Done(): + return "", ctx.Err() + } + } + + const dataTTL = 5 * time.Second + _, err = client.Get(ctx, dataTTL, key, cb) + require.NoError(t, err) + + // Age the value past the refresh floor, then trigger a refresh. + time.Sleep(100 * time.Millisecond) + _, err = client.Get(ctx, dataTTL, key, cb) + require.NoError(t, err) + + // The slow refresh should land its new value (it would be cancelled at + // LockTTL=200ms without RefreshTimeout). + require.Eventually(t, func() bool { + v, gErr := client.Get(ctx, dataTTL, key, cb) + return gErr == nil && v == "refreshed" + }, 2*time.Second, 20*time.Millisecond, "refresh should have persisted the new value") + + require.Zero(t, metrics.errs.Load(), "no RefreshError should fire when the callback finishes within RefreshTimeout") +}