From 6e6744f0fc9493c5ff0e70fb13006456b5062c50 Mon Sep 17 00:00:00 2001 From: Luis C Date: Fri, 30 Oct 2026 15:16:05 -0400 Subject: [PATCH 1/3] test(cache): cover serving a value after the source reports it gone The cache keeps serving a cached value after the fetcher returns ErrDoesNotExist for it: only the short negative sentinel is written, so each time the sentinel expires the next Get is a soft miss served from the old data. These tests fail on that behavior for a single backend with and without negative caching, for multiple backends, and for the shadow client, and pin that the fetcher's error is still returned when Redis rejects the eviction. --- cache/cache_test.go | 171 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 171 insertions(+) diff --git a/cache/cache_test.go b/cache/cache_test.go index 8df1f52..9920c3b 100644 --- a/cache/cache_test.go +++ b/cache/cache_test.go @@ -5,12 +5,17 @@ import ( "encoding/json" "errors" "fmt" + "sync/atomic" "testing" "time" + "github.com/alicebob/miniredis/v2" "github.com/go-redis/redismock/v9" "github.com/redis/go-redis/v9" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/replicate/go/test" ) type testObj struct { @@ -760,3 +765,169 @@ func TestCacheOperationsSucceedWhenShadowClientFails(t *testing.T) { return nil }, "shadow mock expectations not met or did not error as expected") } + +const ( + goneFresh = 10 * time.Second + goneStale = 30 * time.Minute + goneNegative = 10 * time.Second +) + +type revocableSource struct { + gone atomic.Bool + calls atomic.Int64 +} + +func (s *revocableSource) fetch(_ context.Context, _ string) (testObj, error) { + s.calls.Add(1) + if s.gone.Load() { + return testObj{}, fmt.Errorf("revoked: %w", ErrDoesNotExist) + } + return testObj{Value: "alice"}, nil +} + +func fastForward(d time.Duration, mrs ...*miniredis.Miniredis) { + for _, mr := range mrs { + mr.FastForward(d) + } +} + +func expireFreshAndAwaitRefresh(t *testing.T, c *Cache[testObj], key string, source *revocableSource, mrs ...*miniredis.Miniredis) { + t.Helper() + + callsBefore := source.calls.Load() + fastForward(goneFresh+time.Second, mrs...) + _, _ = c.Get(context.Background(), key, source.fetch) + + lockKey := c.keysFor(key).lock + assertEventually(t, func() error { + for _, mr := range mrs { + if mr.Exists(lockKey) { + return errors.New("refresh still in flight") + } + } + return nil + }, "background refresh did not finish") + require.EqualValues(t, 1, source.calls.Load()-callsBefore, "refresh did not consult the source") +} + +func assertGoneAcrossNegativeExpiries(t *testing.T, c *Cache[testObj], key string, source *revocableSource, mrs ...*miniredis.Miniredis) { + t.Helper() + + for i := range 3 { + for range 2 { + v, err := c.Get(context.Background(), key, source.fetch) + require.Zero(t, v, "round %d: Get served the old value", i) + require.ErrorIs(t, err, ErrDoesNotExist, "round %d", i) + } + fastForward(goneNegative+time.Second, mrs...) + } +} + +func TestCacheStopsServingValueOnceSourceReportsItGone(t *testing.T) { + mr, rdb := test.MiniRedis(t) + c := NewCache[testObj](rdb, "auth", goneFresh, goneStale, WithNegativeCaching(goneNegative)) + var source revocableSource + + require.NoError(t, c.Set(context.Background(), "tok", testObj{Value: "alice"})) + source.gone.Store(true) + + expireFreshAndAwaitRefresh(t, c, "tok", &source, mr) + assertGoneAcrossNegativeExpiries(t, c, "tok", &source, mr) +} + +func TestCacheStopsServingValueOnceSourceReportsItGoneWithoutNegativeCaching(t *testing.T) { + mr, rdb := test.MiniRedis(t) + c := NewCache[testObj](rdb, "auth", goneFresh, goneStale) + var source revocableSource + + require.NoError(t, c.Set(context.Background(), "tok", testObj{Value: "alice"})) + source.gone.Store(true) + + expireFreshAndAwaitRefresh(t, c, "tok", &source, mr) + assertGoneAcrossNegativeExpiries(t, c, "tok", &source, mr) +} + +func TestCacheStopsServingValueFromAnyBackendOnceSourceReportsItGone(t *testing.T) { + mr1, rdb1 := test.MiniRedis(t) + mr2, rdb2 := test.MiniRedis(t) + c := NewCacheMultipleBackends[testObj]([]redis.Cmdable{rdb1, rdb2}, "auth", goneFresh, goneStale, WithNegativeCaching(goneNegative)) + var source revocableSource + + require.NoError(t, c.Set(context.Background(), "tok", testObj{Value: "alice"})) + require.True(t, mr1.Exists(c.keysFor("tok").data)) + require.True(t, mr2.Exists(c.keysFor("tok").data)) + source.gone.Store(true) + + expireFreshAndAwaitRefresh(t, c, "tok", &source, mr1, mr2) + assertGoneAcrossNegativeExpiries(t, c, "tok", &source, mr1, mr2) +} + +func TestCacheShadowClientDropsValueOnceSourceReportsItGone(t *testing.T) { + mr, rdb := test.MiniRedis(t) + shadowMr, shadowRdb := test.MiniRedis(t) + c := NewCache[testObj](rdb, "auth", goneFresh, goneStale, + WithNegativeCaching(goneNegative), + WithShadowWriteClient(shadowRdb, time.Second)) + var source revocableSource + keys := c.keysFor("tok") + + require.NoError(t, c.Set(context.Background(), "tok", testObj{Value: "alice"})) + assertEventually(t, func() error { + if !shadowMr.Exists(keys.data) { + return errors.New("shadow write has not landed") + } + return nil + }, "shadow client never received the value") + source.gone.Store(true) + + expireFreshAndAwaitRefresh(t, c, "tok", &source, mr, shadowMr) + + assertEventually(t, func() error { + if shadowMr.Exists(keys.data) { + return errors.New("shadow client still holds the value") + } + return nil + }, "shadow client kept the value after the source reported it gone") +} + +type failDeletes struct{ armed atomic.Bool } + +func (*failDeletes) DialHook(next redis.DialHook) redis.DialHook { return next } + +func (*failDeletes) ProcessHook(next redis.ProcessHook) redis.ProcessHook { return next } + +func (h *failDeletes) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook { + return func(ctx context.Context, cmds []redis.Cmder) error { + if h.armed.Load() { + for _, cmd := range cmds { + if cmd.Name() == "del" { + return errors.New("boom") + } + } + } + return next(ctx, cmds) + } +} + +func TestCacheReturnsSourceErrorWhenRedisRejectsEviction(t *testing.T) { + for name, opts := range map[string][]Option{ + "with negative caching": {WithNegativeCaching(goneNegative)}, + "without negative caching": nil, + } { + t.Run(name, func(t *testing.T) { + _, rdb := test.MiniRedis(t) + hook := &failDeletes{} + rdb.AddHook(hook) + hook.armed.Store(true) + c := NewCache[testObj](rdb, "auth", goneFresh, goneStale, opts...) + var source revocableSource + source.gone.Store(true) + + v, err := c.Get(context.Background(), "tok", source.fetch) + + assert.ErrorIs(t, err, ErrDoesNotExist) + assert.NotContains(t, fmt.Sprint(err), "boom") + assert.Zero(t, v) + }) + } +} From 7d0b50a4725ce52d93cbd747aad64f205bc5a6bb Mon Sep 17 00:00:00 2001 From: Luis C Date: Fri, 30 Oct 2026 15:19:36 -0400 Subject: [PATCH 2/3] fix(cache): stop serving cached values after the source reports them gone When a fetcher returned ErrDoesNotExist, the cache only wrote the short negative sentinel and left the data key with its full stale TTL. Each time the sentinel expired, the next Get was a soft miss and served the old value again, so a revoked item kept coming back until the stale TTL ran out. fill now also removes the data and fresh keys from every backend and the shadow client in one transaction per client, even when negative caching is off, and still returns the fetcher's error if that eviction fails. --- cache/cache.go | 61 ++++++++++++++++++++++++++++++--------------- cache/cache_test.go | 13 ++++++++-- 2 files changed, 52 insertions(+), 22 deletions(-) diff --git a/cache/cache.go b/cache/cache.go index 62f07a6..43d7e15 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -38,8 +38,9 @@ var ( // ErrDoesNotExist is returned if negative caching is enabled and the // non-existence of the specified key has been cached. It must also be - // returned by cache fetchers when the specified key does not exist and - // negative caching is wanted. + // returned by cache fetchers when the specified key does not exist. The + // cache then evicts any value it holds for the key, and with negative + // caching enabled it remembers the non-existence. ErrDoesNotExist = errors.New("requested item does not exist") // ErrDisallowedCacheValue is thrown if a client attempts to set an entry in @@ -220,12 +221,11 @@ func (c *Cache[T]) fill(ctx context.Context, key string, fetcher Fetcher[T]) (va value, err = fetcher(ctx, key) if errors.Is(err, ErrDoesNotExist) { span.SetAttributes(attribute.String("cache.result", "negative (caching nonexistence)")) - if cacheSetErr := c.setNegative(ctx, key); cacheSetErr != nil { - // Errors encountered while filling the cache are not returned to the - // caller: we don't want a cache availability problem to be exposed if the - // value was already successfully fetched. - recordError(ctx, cacheSetErr) - log.Warnw("cache set negative failed", "error", cacheSetErr) + if evictErr := c.evict(ctx, key); evictErr != nil { + // The caller needs the fetcher's ErrDoesNotExist, not a cache + // availability problem, so eviction errors are not returned. + recordError(ctx, evictErr) + log.Warnw("cache evict failed", "error", evictErr) } // we return the original err, so the caller can handle the // ErrDoesNotExist @@ -325,19 +325,40 @@ func (c *Cache[T]) set(ctx context.Context, key string, value T) error { return errors.Join(errs...) } -func (c *Cache[T]) setNegative(ctx context.Context, key string) error { - // If negative caching is not enabled, this is a no-op. - if c.opts.Negative == 0 { - return nil +func (c *Cache[T]) evictOnClient(ctx context.Context, client redis.Cmdable, keys keys, negative bool) error { + pipe := client.TxPipeline() + pipe.Del(ctx, keys.data, keys.fresh) + if negative { + pipe.Set(ctx, keys.negative, 1, c.opts.Negative) } + _, err := pipe.Exec(ctx) + return err +} + +func (c *Cache[T]) evict(ctx context.Context, key string) error { log := logger.With(logging.GetFields(ctx)...).Sugar() keys := c.keysFor(key) + negative := c.opts.Negative > 0 + + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + l, err := c.acquireIfMultipleRedises(ctx, keys.lockMultiple, 5*time.Second) + if err != nil { + return err + } + defer func() { + err := l.Release(ctx) + if err != nil { + recordError(ctx, fmt.Errorf("error releasing update lock: %w", err)) + } + }() - // Record non-existence sentinel in the cache - err := c.clients[0].Set(ctx, keys.negative, 1, c.opts.Negative).Err() + errs := []error{} + for i, client := range c.clients { + errs = append(errs, c.evictOnClient(ctx, client, keys, negative && i == 0)) + } - // Shadow write if configured if c.opts.ShadowWriteClient != nil { // Fire-and-forget shadow write go func() { @@ -348,21 +369,21 @@ func (c *Cache[T]) setNegative(ctx context.Context, key string) error { ctx, "cache.shadow_write", trace.WithAttributes(c.spanAttributes(key)...), - trace.WithAttributes(attribute.String("cache.operation", "setNegative")), + trace.WithAttributes(attribute.String("cache.operation", "evict")), ) defer span.End() - log.Debugw("performing shadow negative write", "key", key) + log.Debugw("performing shadow evict", "key", key) - if err := c.opts.ShadowWriteClient.Set(ctx, keys.negative, 1, c.opts.Negative).Err(); err != nil { + if err := c.evictOnClient(ctx, c.opts.ShadowWriteClient, keys, negative); err != nil { recordError(ctx, err) - log.Warnw("shadow negative write failed", "key", key, "error", err) + log.Warnw("shadow write failed", "key", key, "error", err, "operation", "evict") } }() } - return err + return errors.Join(errs...) } type _nullLock struct{} diff --git a/cache/cache_test.go b/cache/cache_test.go index 9920c3b..4c0a0c7 100644 --- a/cache/cache_test.go +++ b/cache/cache_test.go @@ -109,7 +109,10 @@ func (m mockWrapper) ExpectCacheFillWithLockErr(key string, err error) { } func (m mockWrapper) ExpectCacheFillNegative(key string) { + m.ExpectTxPipeline() + m.ExpectDel("cache:data:"+m.name+":"+key, "cache:fresh:"+m.name+":"+key).SetVal(0) m.ExpectSet("cache:negative:"+m.name+":"+key, 1, m.negative).SetVal("OK") + m.ExpectTxPipelineExec() } // Helper function for polling mock expectations @@ -626,8 +629,11 @@ func TestCacheWithShadowWriteClientNegative(t *testing.T) { cacheMock.ExpectCacheFetchEmpty("elephant") cacheMock.ExpectCacheFillNegative("elephant") - // Expectations for shadow client (negative write only) + // Expectations for shadow client (evict) + shadowMock.ExpectTxPipeline() + shadowMock.ExpectDel("cache:data:objects:elephant", "cache:fresh:objects:elephant").SetVal(0) shadowMock.ExpectSet("cache:negative:objects:elephant", 1, negative).SetVal("OK") + shadowMock.ExpectTxPipelineExec() // Trigger Get with fetcher returning ErrDoesNotExist _, err := cache.Get(ctx, "elephant", func(_ context.Context, _ string) (t testObj, err error) { @@ -718,8 +724,11 @@ func TestCacheOperationsSucceedWhenShadowClientFails(t *testing.T) { shadowMock.Regexp().ExpectSet(".*", 1, fresh).SetErr(expectedErr) shadowMock.ExpectTxPipelineExec().SetErr(expectedErr) - // For cache.Get("missing", ...) -> setNegative + // For cache.Get("missing", ...) -> evict + shadowMock.ExpectTxPipeline() + shadowMock.Regexp().ExpectDel(".*", ".*").SetErr(expectedErr) shadowMock.Regexp().ExpectSet(".*", 1, negative).SetErr(expectedErr) + shadowMock.ExpectTxPipelineExec().SetErr(expectedErr) // Setup primary client expectations upfront cacheMock.ExpectCacheFill("elephant", testObj{Value: "shadow_failure_test"}) From e4e63267995683dbac9e811698bc5c1ee9cb29d1 Mon Sep 17 00:00:00 2001 From: Luis C Date: Tue, 3 Nov 2026 10:30:25 -0500 Subject: [PATCH 3/3] fix(cache): delete a value before recording that it is gone Eviction ran DEL and the negative-marker SET in one MULTI. Redis aborts the whole transaction when it rejects a queued command, such as a SET at maxmemory, so the value survived. The delete now runs first on its own. The eviction-failure hook now also rejects a DEL sent outside a pipeline, and the test requires it to fire. --- cache/cache.go | 16 ++++---- cache/cache_test.go | 91 +++++++++++++++++++++++++++++++++++++-------- 2 files changed, 85 insertions(+), 22 deletions(-) diff --git a/cache/cache.go b/cache/cache.go index 43d7e15..abf0e58 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -325,15 +325,17 @@ func (c *Cache[T]) set(ctx context.Context, key string, value T) error { return errors.Join(errs...) } +// evictOnClient deletes the value before it records non-existence, and not in +// one transaction. Redis aborts a whole MULTI when it rejects a queued command, +// for example a SET at maxmemory, which would keep the value. func (c *Cache[T]) evictOnClient(ctx context.Context, client redis.Cmdable, keys keys, negative bool) error { - pipe := client.TxPipeline() - pipe.Del(ctx, keys.data, keys.fresh) - if negative { - pipe.Set(ctx, keys.negative, 1, c.opts.Negative) + if err := client.Del(ctx, keys.data, keys.fresh).Err(); err != nil { + return err } - - _, err := pipe.Exec(ctx) - return err + if !negative { + return nil + } + return client.Set(ctx, keys.negative, 1, c.opts.Negative).Err() } func (c *Cache[T]) evict(ctx context.Context, key string) error { diff --git a/cache/cache_test.go b/cache/cache_test.go index 4c0a0c7..7115e0c 100644 --- a/cache/cache_test.go +++ b/cache/cache_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "sync/atomic" "testing" "time" @@ -109,10 +110,8 @@ func (m mockWrapper) ExpectCacheFillWithLockErr(key string, err error) { } func (m mockWrapper) ExpectCacheFillNegative(key string) { - m.ExpectTxPipeline() m.ExpectDel("cache:data:"+m.name+":"+key, "cache:fresh:"+m.name+":"+key).SetVal(0) m.ExpectSet("cache:negative:"+m.name+":"+key, 1, m.negative).SetVal("OK") - m.ExpectTxPipelineExec() } // Helper function for polling mock expectations @@ -630,10 +629,8 @@ func TestCacheWithShadowWriteClientNegative(t *testing.T) { cacheMock.ExpectCacheFillNegative("elephant") // Expectations for shadow client (evict) - shadowMock.ExpectTxPipeline() shadowMock.ExpectDel("cache:data:objects:elephant", "cache:fresh:objects:elephant").SetVal(0) shadowMock.ExpectSet("cache:negative:objects:elephant", 1, negative).SetVal("OK") - shadowMock.ExpectTxPipelineExec() // Trigger Get with fetcher returning ErrDoesNotExist _, err := cache.Get(ctx, "elephant", func(_ context.Context, _ string) (t testObj, err error) { @@ -724,11 +721,8 @@ func TestCacheOperationsSucceedWhenShadowClientFails(t *testing.T) { shadowMock.Regexp().ExpectSet(".*", 1, fresh).SetErr(expectedErr) shadowMock.ExpectTxPipelineExec().SetErr(expectedErr) - // For cache.Get("missing", ...) -> evict - shadowMock.ExpectTxPipeline() + // For cache.Get("missing", ...) -> evict stops at the failed delete shadowMock.Regexp().ExpectDel(".*", ".*").SetErr(expectedErr) - shadowMock.Regexp().ExpectSet(".*", 1, negative).SetErr(expectedErr) - shadowMock.ExpectTxPipelineExec().SetErr(expectedErr) // Setup primary client expectations upfront cacheMock.ExpectCacheFill("elephant", testObj{Value: "shadow_failure_test"}) @@ -899,19 +893,34 @@ func TestCacheShadowClientDropsValueOnceSourceReportsItGone(t *testing.T) { }, "shadow client kept the value after the source reported it gone") } -type failDeletes struct{ armed atomic.Bool } +// failDeletes rejects any DEL of cache data, sent alone or in a pipeline. +type failDeletes struct { + armed atomic.Bool + hits atomic.Int32 +} func (*failDeletes) DialHook(next redis.DialHook) redis.DialHook { return next } -func (*failDeletes) ProcessHook(next redis.ProcessHook) redis.ProcessHook { return next } +func (h *failDeletes) rejects(cmd redis.Cmder) bool { + return h.armed.Load() && cmd.Name() == "del" && strings.Contains(fmt.Sprint(cmd.Args()), "cache:data:") +} + +func (h *failDeletes) ProcessHook(next redis.ProcessHook) redis.ProcessHook { + return func(ctx context.Context, cmd redis.Cmder) error { + if h.rejects(cmd) { + h.hits.Add(1) + return errors.New("boom") + } + return next(ctx, cmd) + } +} func (h *failDeletes) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook { return func(ctx context.Context, cmds []redis.Cmder) error { - if h.armed.Load() { - for _, cmd := range cmds { - if cmd.Name() == "del" { - return errors.New("boom") - } + for _, cmd := range cmds { + if h.rejects(cmd) { + h.hits.Add(1) + return errors.New("boom") } } return next(ctx, cmds) @@ -937,6 +946,58 @@ func TestCacheReturnsSourceErrorWhenRedisRejectsEviction(t *testing.T) { assert.ErrorIs(t, err, ErrDoesNotExist) assert.NotContains(t, fmt.Sprint(err), "boom") assert.Zero(t, v) + assert.Positive(t, hook.hits.Load(), "the delete never reached Redis") }) } } + +// rejectNegativeMarker rejects SET on a negative key, as Redis does for writes +// at maxmemory. Redis aborts a whole MULTI when it rejects a queued command, so +// a transaction that holds the rejected SET runs none of its commands. +type rejectNegativeMarker struct{ hits atomic.Int32 } + +func (*rejectNegativeMarker) DialHook(next redis.DialHook) redis.DialHook { return next } + +func (*rejectNegativeMarker) rejects(cmd redis.Cmder) bool { + args := cmd.Args() + return cmd.Name() == "set" && len(args) > 1 && strings.HasPrefix(fmt.Sprint(args[1]), "cache:negative:") +} + +func (h *rejectNegativeMarker) ProcessHook(next redis.ProcessHook) redis.ProcessHook { + return func(ctx context.Context, cmd redis.Cmder) error { + if h.rejects(cmd) { + h.hits.Add(1) + err := errors.New("OOM command not allowed when used memory > 'maxmemory'") + cmd.SetErr(err) + return err + } + return next(ctx, cmd) + } +} + +func (h *rejectNegativeMarker) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook { + return func(ctx context.Context, cmds []redis.Cmder) error { + for _, cmd := range cmds { + if h.rejects(cmd) { + h.hits.Add(1) + return errors.New("EXECABORT Transaction discarded because of previous errors") + } + } + return next(ctx, cmds) + } +} + +func TestCacheDropsValueWhenRedisRejectsNegativeMarker(t *testing.T) { + mr, rdb := test.MiniRedis(t) + hook := &rejectNegativeMarker{} + rdb.AddHook(hook) + c := NewCache[testObj](rdb, "auth", goneFresh, goneStale, WithNegativeCaching(goneNegative)) + require.NoError(t, c.Set(context.Background(), "tok", testObj{Value: "alice"})) + var source revocableSource + source.gone.Store(true) + + expireFreshAndAwaitRefresh(t, c, "tok", &source, mr) + + require.Positive(t, hook.hits.Load(), "the negative marker write never reached Redis") + require.False(t, mr.Exists(c.keysFor("tok").data), "a rejected negative marker kept the value") +}