fix(cache): stop serving a value after the source reports it gone

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,42 @@ 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 {
+// 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 {
+	if err := client.Del(ctx, keys.data, keys.fresh).Err(); err != nil {
+		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 {
 	log := logger.With(logging.GetFields(ctx)...).Sugar()
 	keys := c.keysFor(key)
+	negative := c.opts.Negative > 0
 
-	// Record non-existence sentinel in the cache
-	err := c.clients[0].Set(ctx, keys.negative, 1, c.opts.Negative).Err()
+	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))
+		}
+	}()
+
+	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 +371,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{}

cache/cache_test.go

@@ -5,12 +5,18 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
+	"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 {
@@ -104,6 +110,7 @@ func (m mockWrapper) ExpectCacheFillWithLockErr(key string, err error) {
 }
 
 func (m mockWrapper) ExpectCacheFillNegative(key string) {
+	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")
 }
 
@@ -621,7 +628,8 @@ func TestCacheWithShadowWriteClientNegative(t *testing.T) {
 	cacheMock.ExpectCacheFetchEmpty("elephant")
 	cacheMock.ExpectCacheFillNegative("elephant")
 
-	// Expectations for shadow client (negative write only)
+	// Expectations for shadow client (evict)
+	shadowMock.ExpectDel("cache:data:objects:elephant", "cache:fresh:objects:elephant").SetVal(0)
 	shadowMock.ExpectSet("cache:negative:objects:elephant", 1, negative).SetVal("OK")
 
 	// Trigger Get with fetcher returning ErrDoesNotExist
@@ -713,8 +721,8 @@ func TestCacheOperationsSucceedWhenShadowClientFails(t *testing.T) {
 	shadowMock.Regexp().ExpectSet(".*", 1, fresh).SetErr(expectedErr)
 	shadowMock.ExpectTxPipelineExec().SetErr(expectedErr)
 
-	// For cache.Get("missing", ...) -> setNegative
-	shadowMock.Regexp().ExpectSet(".*", 1, negative).SetErr(expectedErr)
+	// For cache.Get("missing", ...) -> evict stops at the failed delete
+	shadowMock.Regexp().ExpectDel(".*", ".*").SetErr(expectedErr)
 
 	// Setup primary client expectations upfront
 	cacheMock.ExpectCacheFill("elephant", testObj{Value: "shadow_failure_test"})
@@ -760,3 +768,236 @@ 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")
+}
+
+// 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 (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 {
+		for _, cmd := range cmds {
+			if h.rejects(cmd) {
+				h.hits.Add(1)
+				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)
+			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")
+}