fix(csc): apply invalidation batches in one shard pass (#4033)

* fix(csc): apply invalidation batches in one shard pass

ClientSideCacheInvalidationBatchWindow applied only a fraction of the
invalidations it received under load. apply() looped per key calling
deleteByRedisKeyCollectingHot, and that walks all 16 shards taking each
write lock, so one worker doing 16*N contended acquisitions fell
permanently behind the arriving stream. The rest stayed in the spill
buffer, and the cache went on serving values the server had invalidated
while reporting a ~100% hit rate.

Measured on a 20k/sec invalidation stream: 7,729 of 68,846 enqueued
invalidations applied (15.5%), with nothing dropped by policy
(discarded=0, epochSkip=0).

Invert the loops: take each shard's lock once per batch and handle every
key of the batch under it, so a batch costs 16 acquisitions instead of
16 per key. Per-key semantics (fetch-order guard, recency horizon,
target collection) are unchanged; only the locking granularity differs.
Invalidations applied goes to 100%.

* docs(csc): note that cached reads are not monotonic

The invalidation batch window already documents staleness -- a reader can
see the pre-invalidation value for up to the window. It does not document
the ordering consequence: a caller reading through a non-cached path can
observe a newer value than a concurrent in-flight fetch is about to
publish, and a later cached read then returns the older one.

This is inherent to asynchronous invalidation, not specific to the window.
At publish time the client cannot know its in-flight value is already
stale, because the only signal is a push that has not arrived yet; a guard
that refuses to publish values superseded by an observed invalidation was
tried and never fires. The window widens the exposure, because a published
value stays readable until its invalidation is APPLIED rather than
observed.

Observed roughly once per 44M reads with a 5ms window, and not at all with
the window off.

* fix(csc): keep the read bit on batched refreshes

collectHotAndDeleteBatch built the refresh target without the entry's
second-chance bit, which collectHotAndDelete copies for
stageRefreshAccess. With the invalidation batch window and refresh on,
a refreshed hot key was republished with its bit clear and was the next
eviction victim in a warm shard. Copy it, as the single-key path does.

csc_inval_batch.go

@@ -360,39 +360,40 @@ func (b *cscInvalBatcher) apply(items []cscInvalItem) {
 	cur := b.epoch.Load()
 	lc, canRefresh := cache.(*LocalCache)
 	canRefresh = canRefresh && refresh != nil
-	var hot []cscRefreshTarget
+	// Group the batch and apply it in ONE pass over the shards: each shard
+	// lock is taken once for the whole batch rather than once per key. The
+	// per-key semantics (fetch-order guard, recency horizon, target
+	// collection) are unchanged.
+	keys := make([]string, 0, len(items))
+	sinces := make([]int64, 0, len(items))
+	snaps := make([]uint64, 0, len(items))
 	for _, it := range items {
 		// Stale epoch: enqueued before a full cache Flush that superseded it
 		// (see drop()); applying it would only evict a post-flush repopulation.
 		if it.epoch != cur {
 			continue
 		}
-		k := it.key
-		if !canRefresh {
-			if lc != nil {
-				// *LocalCache with refresh OFF: still honor the fetch-order guard, or a
-				// delayed invalidation would delete a value refetched during the batch
-				// window (fetchSeq > fetchSnap) — a spurious miss that can also cancel an
-				// in-progress fetch and wake waiters as duplicates. Use the sequence-aware
-				// deletion path and discard any collected refresh targets (refresh is off).
-				_ = lc.deleteByRedisKeyCollectingHot(k, cscInvalNoHorizon, it.fetchSnap, nil)
-			} else {
-				// Non-*LocalCache: no per-entry fetch sequence to compare; plain delete.
-				cache.DeleteByRedisKey(k)
-			}
-			continue
-		}
-		// Use the horizon snapshotted at ENQUEUE, not a live load: a batch-window
-		// delay advances the live horizon and would chill keys that were hot when
-		// the invalidation arrived (see cscInvalItem.sinceToken). An item enqueued
-		// with no refresh binding carries cscInvalNoHorizon; the binding that
-		// exists NOW (a client attached in between) supplies the live horizon, which
-		// is a real recency test — never horizon 0, which would mark every entry hot.
 		since := it.sinceToken
-		if since == cscInvalNoHorizon {
+		if canRefresh && since == cscInvalNoHorizon {
 			since = refresh.sinceToken.Load()
 		}
-		hot = lc.deleteByRedisKeyCollectingHot(k, since, it.fetchSnap, hot[:0])
+		keys = append(keys, it.key)
+		sinces = append(sinces, since)
+		snaps = append(snaps, it.fetchSnap)
+	}
+	if len(keys) == 0 {
+		return
+	}
+	if lc == nil {
+		// Non-*LocalCache: no per-entry fetch sequence to compare; plain deletes.
+		for _, k := range keys {
+			cache.DeleteByRedisKey(k)
+		}
+		return
+	}
+	var hot []cscRefreshTarget
+	hot, _ = lc.deleteManyByRedisKeyCollectingHot(keys, sinces, snaps, hot)
+	if canRefresh {
 		for i := range hot {
 			refresh.offer(hot[i])
 		}

csc_inval_batch_accounting_test.go

@@ -0,0 +1,105 @@
+package redis
+
+import (
+	"context"
+	"testing"
+)
+
+// The batched invalidation apply must report deletions exactly as the
+// single-key path does, including the no-op count.
+//
+// The first version of the batch path derived no-ops from the batch total --
+// "nothing in this batch matched" -- which silently undercounted the common
+// case where some keys match and others do not. DeletionStats is public, so
+// that is a user-visible metric regression, and it is invisible in a
+// throughput benchmark.
+func TestDeleteManyByRedisKeyDeletionAccounting(t *testing.T) {
+	newCache := func() *LocalCache {
+		return NewLocalCache(CacheConfig{
+			MaxEntries: 128,
+		})
+	}
+	ctx := context.Background()
+
+	// Seed one entry under redis key "present".
+	seed := func(c *LocalCache, cacheKey, redisKey string) {
+		tok, ok := c.Reserve(cacheKey, []string{redisKey})
+		if !ok || tok == 0 {
+			t.Fatalf("Reserve(%q) failed", cacheKey)
+		}
+		if !c.fulfill(cacheKey, tok, 0, []byte("v")) {
+			t.Fatalf("Fulfill(%q) failed", cacheKey)
+		}
+		if _, hit := c.Get(ctx, cacheKey); !hit {
+			t.Fatalf("expected %q cached", cacheKey)
+		}
+	}
+
+	t.Run("mixed batch counts one no-op per unmatched key", func(t *testing.T) {
+		c := newCache()
+		seed(c, "ck1", "present")
+
+		// One key matches, four do not.
+		keys := []string{"present", "absent1", "absent2", "absent3", "absent4"}
+		since := make([]int64, len(keys))
+		snaps := make([]uint64, len(keys))
+		for i := range snaps {
+			snaps[i] = ^uint64(0)
+		}
+		_, removed := c.deleteManyByRedisKeyCollectingHot(keys, since, snaps, nil)
+		if removed != 1 {
+			t.Fatalf("removed = %d, want 1", removed)
+		}
+		del, noop := c.DeletionStats()
+		if del != uint64(len(keys)) {
+			t.Errorf("deletions = %d, want %d", del, len(keys))
+		}
+		// Four keys matched nothing. Deriving this from the batch total
+		// would report 0 here.
+		if noop != 4 {
+			t.Errorf("noop = %d, want 4 (one per unmatched key)", noop)
+		}
+	})
+
+	t.Run("matches the single-key path", func(t *testing.T) {
+		batch := newCache()
+		single := newCache()
+		seed(batch, "ck1", "present")
+		seed(single, "ck1", "present")
+
+		keys := []string{"present", "absent1", "absent2"}
+		since := make([]int64, len(keys))
+		snaps := make([]uint64, len(keys))
+		for i := range snaps {
+			snaps[i] = ^uint64(0)
+		}
+		batch.deleteManyByRedisKeyCollectingHot(keys, since, snaps, nil)
+		for _, k := range keys {
+			single.DeleteByRedisKey(k)
+		}
+
+		bd, bn := batch.DeletionStats()
+		sd, sn := single.DeletionStats()
+		if bd != sd || bn != sn {
+			t.Errorf("batch (del=%d noop=%d) != single (del=%d noop=%d)", bd, bn, sd, sn)
+		}
+	})
+
+	t.Run("whole batch unmatched", func(t *testing.T) {
+		c := newCache()
+		keys := []string{"a", "b", "c"}
+		since := make([]int64, len(keys))
+		snaps := make([]uint64, len(keys))
+		for i := range snaps {
+			snaps[i] = ^uint64(0)
+		}
+		_, removed := c.deleteManyByRedisKeyCollectingHot(keys, since, snaps, nil)
+		if removed != 0 {
+			t.Fatalf("removed = %d, want 0", removed)
+		}
+		del, noop := c.DeletionStats()
+		if del != 3 || noop != 3 {
+			t.Errorf("deletions=%d noop=%d, want 3/3", del, noop)
+		}
+	})
+}

local_cache.go

@@ -320,6 +320,97 @@ type cacheShard struct {
 	staleTimeout   time.Duration
 }
 
+// collectHotAndDeleteBatch applies a whole invalidation batch under ONE
+// acquisition of this shard's lock.
+//
+// The per-key work is identical to collectHotAndDelete; only the locking
+// granularity changes. The caller loops shards on the outside and keys on the
+// inside, so a batch of N costs 16 lock acquisitions instead of 16*N -- which
+// is what let the single batcher worker fall permanently behind a 20k/sec
+// invalidation stream and leave the cache serving stale entries.
+//
+// keys and the parallel guards are indexed together: sinceTokens[i] and
+// fetchSnaps[i] belong to keys[i].
+//
+// matched[i] is set when this shard removed something for keys[i]. The caller
+// ORs it across shards, because one redis key can appear in several shards: a
+// multi-key entry is filed under its CACHE key's shard, so each of its redis
+// keys is indexed wherever that entry lives. Per-key match tracking is what
+// keeps the no-op deletion count equal to the single-key path's.
+func (s *cacheShard) collectHotAndDeleteBatch(keys []string, sinceTokens []int64, fetchSnaps []uint64, dst []cscRefreshTarget, matched []bool) ([]cscRefreshTarget, int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	removed := 0
+	for i, redisKey := range keys {
+		cacheKeys, ok := s.byRedisKey[redisKey]
+		if !ok {
+			continue
+		}
+		sinceToken, fetchSnap := sinceTokens[i], fetchSnaps[i]
+		toRemove := make([]string, 0, len(cacheKeys))
+		for cacheKey := range cacheKeys {
+			toRemove = append(toRemove, cacheKey)
+		}
+		for _, cacheKey := range toRemove {
+			entry, exists := s.entries[cacheKey]
+			if exists && entry.fetchSeq > fetchSnap {
+				continue
+			}
+			if exists && entry.state == cacheEntryValid && entry.lastAccessNs.Load() > sinceToken {
+				ks := make([]string, len(entry.redisKeys))
+				copy(ks, entry.redisKeys)
+				dst = append(dst, cscRefreshTarget{
+					cacheKey:  cacheKey,
+					redisKeys: ks,
+					accessNs:  entry.lastAccessNs.Load(),
+					// The second-chance bit, as collectHotAndDelete keeps it:
+					// the refresh republishes the entry with it.
+					read:     entry.readSinceSweep.Load(),
+					valBytes: len(entry.value),
+				})
+			}
+			if s.removeEntryLocked(cacheKey) {
+				matched[i] = true
+				removed++
+			}
+		}
+	}
+	return dst, removed
+}
+
+// deleteManyByRedisKeyCollectingHot is the cache-level batch entry point: one
+// pass over the shards, every key of the batch handled under each shard's
+// single lock.
+func (c *LocalCache) deleteManyByRedisKeyCollectingHot(keys []string, sinceTokens []int64, fetchSnaps []uint64, dst []cscRefreshTarget) ([]cscRefreshTarget, int) {
+	if len(keys) == 0 {
+		return dst, 0
+	}
+	removed := 0
+	matched := make([]bool, len(keys))
+	for i := range c.shards {
+		var n int
+		dst, n = c.shards[i].collectHotAndDeleteBatch(keys, sinceTokens, fetchSnaps, dst, matched)
+		removed += n
+	}
+	// Applied-delete accounting, per KEY, exactly as the single-key path does
+	// it: every key is one deletion, and a key that removed nothing is one
+	// no-op. Deriving the no-op count from the batch total instead would
+	// undercount -- a batch where one key matched and nine did not would
+	// record zero no-ops rather than nine.
+	c.deletions.Add(uint64(len(keys)))
+	noop := 0
+	for _, ok := range matched {
+		if !ok {
+			noop++
+		}
+	}
+	if noop > 0 {
+		c.deletionsNoop.Add(uint64(noop))
+	}
+	return dst, removed
+}
+
 // shardFor returns the shard responsible for cacheKey.
 func (c *LocalCache) shardFor(cacheKey string) *cacheShard {
 	if c.shardCount == 1 {

local_cache_batch_read_bit_test.go

@@ -0,0 +1,39 @@
+package redis
+
+import (
+	"context"
+	"testing"
+)
+
+// The batched invalidation path must hand the refresher the same target the
+// single-key path does, second-chance bit included: stageRefreshAccess
+// republishes the entry with it. The batch collector left `read` false, so
+// with the batch window and refresh on, a refreshed hot key came back with its
+// bit clear and was the next eviction victim in a warm shard.
+func TestBatchedInvalidationKeepsReadBit(t *testing.T) {
+	ctx := context.Background()
+	for _, batched := range []bool{false, true} {
+		// A full shard, so the read is recorded even under the pressure gate.
+		lc := NewLocalCache(CacheConfig{MaxEntries: 1})
+		const key, rk = "get:k", "rk"
+		tok, _ := lc.Reserve(key, []string{rk})
+		if !lc.fulfill(key, tok, 0, []byte("v")) {
+			t.Fatal("seed fulfill failed")
+		}
+		if _, ok := lc.Get(ctx, key); !ok {
+			t.Fatal("seeded key missing")
+		}
+		var targets []cscRefreshTarget
+		if batched {
+			targets, _ = lc.deleteManyByRedisKeyCollectingHot([]string{rk}, []int64{cscInvalNoHorizon}, []uint64{^uint64(0)}, nil)
+		} else {
+			targets = lc.deleteByRedisKeyCollectingHot(rk, cscInvalNoHorizon, ^uint64(0), nil)
+		}
+		if len(targets) != 1 {
+			t.Fatalf("batched=%v: %d targets, want 1", batched, len(targets))
+		}
+		if !targets[0].read {
+			t.Fatalf("batched=%v: the target lost the read bit the entry had", batched)
+		}
+	}
+}

options.go

@@ -445,6 +445,11 @@ type Options struct {
 	// If nil, maintnotifications are in "auto" mode and will be enabled if the server supports it.
 	MaintNotificationsConfig *maintnotifications.Config
 
+	// Client-side caching is eventually consistent. A cached value may be
+	// stale until its invalidation arrives, and reads are not guaranteed to
+	// be monotonic: a value observed through an uncached path can be newer
+	// than one a subsequent cached read returns. Applications that require
+	// monotonic reads must not rely on the cache for ordering.
 	// ClientSideCacheConfig enables client-side caching when non-nil. Together
 	// with ClientSideCache it is the on/off switch for the feature: leave both
 	// nil to disable CSC, set either one to enable it. If ClientSideCache is also set, it
@@ -546,6 +551,15 @@ type Options struct {
 	// larger than the cache MaxStaleness: deferring a delete by up to the window
 	// lets a reader see the pre-invalidation value for up to that long.
 	//
+	// Deferring a delete also defers ORDERING. A caller that reads the key
+	// through a non-cached path -- for example when a miss is shed to the
+	// pooled path -- can observe a newer value than the one a concurrent
+	// in-flight fetch is about to publish, and a later cached read can then
+	// return the older value. Client-side caching is eventually consistent
+	// and does not guarantee monotonic reads; the window widens that envelope,
+	// because a published value stays readable until its invalidation is
+	// APPLIED rather than observed.
+	//
 	// Requires the built-in cache (ClientSideCacheConfig, or ClientSideCache set
 	// to a *LocalCache), like ClientSideCacheCoalesceMisses: the batcher's
 	// hot-entry refresh integration is LocalCache-specific. With a custom Cache