go-redis/csc_inval_batch_test.go

package redis

import (
	"context"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// enqueue is the single-key convenience form of enqueueAt, used only by tests:
// it snapshots the current fetch order per call. Production code always goes
// through enqueueAt with a fetchSnap taken once at push-observe time (see the
// enqueueAt doc), so this wrapper lives here rather than in the package to keep
// it out of the shipped surface.
func (b *cscInvalBatcher) enqueue(nsKey string) {
	b.enqueueAt(nsKey, cscFetchSeq.Load())
}

// countingCache implements just enough of Cache for the batcher's apply path
// (refresh==nil + a non-*LocalCache cache => apply calls DeleteByRedisKey once
// per unique key). Other Cache methods are never reached; the embedded nil Cache
// would panic if one were, which is the intended tripwire.
type countingCache struct {
	Cache
	deletes atomic.Int64
	flushes atomic.Int64
	mu      sync.Mutex
	keys    []string
}

func (c *countingCache) DeleteByRedisKey(k string) int {
	c.deletes.Add(1)
	c.mu.Lock()
	c.keys = append(c.keys, k)
	c.mu.Unlock()
	return 1
}

func (c *countingCache) Flush() int { c.flushes.Add(1); return 0 }

func newTestBatcher(cache Cache, chCap int, window time.Duration) *cscInvalBatcher {
	return &cscInvalBatcher{
		window: window,
		cache:  cache,
		ch:     make(chan cscInvalItem, chCap),
		wake:   make(chan struct{}, 1),
		stopCh: make(chan struct{}),
		done:   make(chan struct{}),
	}
}

// TestInvalBatcherStopJoinIsSynchronous pins the stop+join teardown contract:
// once join returns, the stop-drain has fully applied — every enqueued delete is
// visible, with no Eventually. Without the join the drain ran asynchronously
// after Close returned (straggler goroutine; a late apply could evict what a
// successor client on a shared cache just repopulated).
func TestInvalBatcherStopJoinIsSynchronous(t *testing.T) {
	cache := &countingCache{}
	b := newTestBatcher(cache, 8, time.Hour) // window never fires; only the stop-drain applies
	go b.run()

	b.enqueue("k1")
	b.enqueue("k2")
	b.stop()
	b.join()

	if got := cache.deletes.Load(); got != 2 {
		t.Fatalf("deletes after stop+join = %d, want 2 (drain must be synchronous)", got)
	}
}

// TestInvalBatcherEnqueueSnapshotsSinceToken pins the #16 horizon snapshot across
// the atomic.Pointer conversion of the batcher's refresh field: enqueue must read
// the CURRENT refresh binding's sinceToken and stamp it on the item, so a
// batch-window delay cannot chill a key that was hot when the invalidation
// arrived.
func TestInvalBatcherEnqueueSnapshotsSinceToken(t *testing.T) {
	b := newTestBatcher(&countingCache{}, 4, time.Hour) // worker not started: item stays on ch
	q := &cscRefreshQueue{}
	q.sinceToken.Store(4242)
	b.refresh.Store(q)

	b.enqueue("k")

	select {
	case it := <-b.ch:
		if it.sinceToken != 4242 {
			t.Fatalf("enqueued item sinceToken = %d, want 4242 (snapshot at enqueue)", it.sinceToken)
		}
	default:
		t.Fatal("enqueue did not land on ch")
	}
}

// TestInvalBatcherEnqueueAtStampsSuppliedFetchSnap pins the per-push fetch-order
// snapshot: enqueueAt must stamp the CALLER's fetchSnap, not a fresh cscFetchSeq
// load. The push handler snapshots cscFetchSeq once per notification and reuses it
// for every key (csc_integration.go), so a fetch reserved after the push was
// observed cannot slip in as "not newer" for a later key in the same push.
func TestInvalBatcherEnqueueAtStampsSuppliedFetchSnap(t *testing.T) {
	b := newTestBatcher(&countingCache{}, 4, time.Hour) // worker not started: item stays on ch

	prev := cscFetchSeq.Load()
	cscFetchSeq.Store(999) // the value a stale per-key load would pick up
	defer cscFetchSeq.Store(prev)

	b.enqueueAt("k", 4242)

	select {
	case it := <-b.ch:
		if it.fetchSnap != 4242 {
			t.Fatalf("enqueued item fetchSnap = %d, want 4242 (caller snapshot, not a fresh load)", it.fetchSnap)
		}
	default:
		t.Fatal("enqueueAt did not land on ch")
	}
}

// TestInvalBatcherRefreshAttachAppliesUnderNewBinding pins the refresh ATTACH
// path: an item enqueued while the batcher had no refresh binding must still be
// applied once a binding appears before apply (setRefreshQueue repoints the
// batcher, then stop-drains it), rather than being dropped on the floor.
//
// It used to also pin a recency gate — the cold entry was deleted but NOT
// offered for refetch. That gate is gone: every invalidated Valid entry is now
// a refetch target, so both keys are offered and the assertion below says so.
func TestInvalBatcherRefreshAttachAppliesUnderNewBinding(t *testing.T) {
	lc := NewLocalCache(CacheConfig{MaxEntries: 64})
	mkValid := func(cacheKey, redisKey string) {
		tok, fetch := lc.Reserve(cacheKey, []string{redisKey})
		if tok == 0 || !fetch {
			t.Fatalf("Reserve(%q) = token %d, shouldFetch %v", cacheKey, tok, fetch)
		}
		if !lc.fulfill(cacheKey, tok, 0, []byte("v")) {
			t.Fatalf("fulfill(%q) failed", cacheKey)
		}
	}

	// Cold entry: populated BEFORE the attaching client's horizon exists.
	mkValid("ck:cold", "rk:cold")

	// Batcher built with refresh OFF; worker not started so items stay on ch.
	b := newTestBatcher(lc, 4, time.Hour)
	b.enqueue("rk:cold")
	var cold cscInvalItem
	select {
	case cold = <-b.ch:
	default:
		t.Fatal("enqueue did not land on ch")
	}
	if cold.sinceToken != cscInvalNoHorizon {
		t.Fatalf("item enqueued with refresh off: sinceToken = %d, want cscInvalNoHorizon (%d)",
			cold.sinceToken, cscInvalNoHorizon)
	}

	// A refresh-enabled client attaches: its horizon is seeded NOW, after the cold
	// entry, and the batcher is repointed at its queue (as setRefreshQueue does
	// before the stop-drain).
	q := &cscRefreshQueue{ch: make(chan cscRefreshTarget, 8)}
	q.sinceToken.Store(cscInvalNoHorizon)
	b.refresh.Store(q)

	// Hot control: read after the horizon, enqueued with a real snapshot.
	mkValid("ck:hot", "rk:hot")
	b.enqueue("rk:hot")
	var hot cscInvalItem
	select {
	case hot = <-b.ch:
	default:
		t.Fatal("enqueue did not land on ch")
	}

	// The stop-drain applies both under the NEW binding.
	b.apply([]cscInvalItem{cold, hot})

	// Both deleted ...
	ctx := context.Background()
	if _, ok := lc.Get(ctx, "ck:cold"); ok {
		t.Fatal("cold entry not deleted")
	}
	if _, ok := lc.Get(ctx, "ck:hot"); ok {
		t.Fatal("hot entry not deleted")
	}
	// ... and BOTH are offered for refetch. There is no recency horizon any
	// more: every invalidated Valid entry is a refetch target, so the cold
	// entry is offered exactly like the hot one. This assertion used to be
	// "hot only", which is what the bounded window bought; the remaining point
	// of this test is that items enqueued BEFORE the attach still apply under
	// the newly bound queue rather than being dropped.
	if n := len(q.ch); n != 2 {
		t.Fatalf("refresh targets offered = %d, want 2 (both, refresh-everything)", n)
	}
	got := map[string]bool{}
	for len(q.ch) > 0 {
		got[(<-q.ch).cacheKey] = true
	}
	if !got["ck:hot"] || !got["ck:cold"] {
		t.Fatalf("offered targets = %v, want both ck:hot and ck:cold", got)
	}
}

// A full ch must send overflow to spill, never apply a delete inline on the
// caller (which, on the coalescer's reply reader, would stall miss replies).
func TestInvalBatcherOverflowSpillsNotInline(t *testing.T) {
	cache := &countingCache{}
	// Worker NOT started: ch never drains, so enqueues past its cap must overflow.
	b := newTestBatcher(cache, 2, time.Hour)

	b.enqueue("k1") // -> ch
	b.enqueue("k2") // -> ch (now full)
	b.enqueue("k3") // ch full -> spill
	b.enqueue("k4") // ch full -> spill

	if got := cache.deletes.Load(); got != 0 {
		t.Fatalf("overflow applied %d deletes inline on the caller; want 0 (must spill)", got)
	}
	if got := b.spilled.Load(); got != 2 {
		t.Fatalf("spilled counter = %d, want 2", got)
	}
	b.spillMu.Lock()
	n := len(b.spill)
	b.spillMu.Unlock()
	if n != 2 {
		t.Fatalf("spill holds %d items, want 2", n)
	}
}

// The worker drains BOTH ch and spill through the shared seen/pending dedup, so a
// burst of duplicate keys collapses to one delete per unique key.
func TestInvalBatcherWorkerDrainsSpillDedup(t *testing.T) {
	cache := &countingCache{}
	// Large window so only the stop-drain flushes (deterministic single apply).
	b := newTestBatcher(cache, 2, time.Hour)

	const dups = 200
	for i := 0; i < dups; i++ {
		b.enqueue("hot") // fills ch (cap 2) then overflows to spill
	}
	b.enqueue("cold")

	if b.spilled.Load() == 0 {
		t.Fatal("setup: expected overflow to spill")
	}
	if got := cache.deletes.Load(); got != 0 {
		t.Fatalf("setup: %d inline deletes before worker start; want 0", got)
	}

	go b.run()
	b.stop() // stop-drain: drains spill + ch, flushes once, exits

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) && cache.deletes.Load() < 2 {
		time.Sleep(time.Millisecond)
	}
	if got := cache.deletes.Load(); got != 2 {
		t.Fatalf("applied %d deletes; want 2 (dedup must collapse %d 'hot' + 1 'cold')", got, dups)
	}
}

// Past the spill hard cap, enqueue stops growing spill and asks the worker to
// full-Flush the cache instead (bounded-memory fallback for an invalidation
// flood on a keyset larger than the local cache).
func TestInvalBatcherSpillCapTriggersFlush(t *testing.T) {
	cache := &countingCache{}
	b := newTestBatcher(cache, 2, time.Hour)

	// Worker not running: fill ch, then overflow past the cap with distinct keys.
	for i := 0; i < cscInvalSpillMax+16; i++ {
		b.enqueue("k" + strconv.Itoa(i))
	}

	if !b.flushReq.Load() {
		t.Fatal("spill exceeded the cap but flushReq was not set")
	}
	b.spillMu.Lock()
	n := len(b.spill)
	b.spillMu.Unlock()
	if n >= cscInvalSpillMax {
		t.Fatalf("spill = %d, want < cap %d (cap must clear the backlog)", n, cscInvalSpillMax)
	}

	// The worker must consume flushReq and Flush the whole cache exactly.
	go b.run()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) && cache.flushes.Load() == 0 {
		time.Sleep(time.Millisecond)
	}
	b.stop()
	if cache.flushes.Load() == 0 {
		t.Fatal("worker did not full-Flush the cache after the spill cap was hit")
	}
}

// Past the spill RETAINED-BYTE bound, enqueue takes the same full-Flush fallback as
// the item cap — a distinct-large-key flood can pin GBs while the item count stays
// far below cscInvalSpillMax, so the byte bound must trip first.
func TestInvalBatcherSpillByteCapTriggersFlush(t *testing.T) {
	cache := &countingCache{}
	b := newTestBatcher(cache, 2, time.Hour)

	// 1 MiB per key: a dozen distinct keys past ch cross cscInvalSpillMaxBytes (8 MiB)
	// while the item count stays a few, far under the 65536 item cap.
	bigKey := strings.Repeat("x", 1<<20)
	const n = 2 + 12 // ch cap 2, then 12 overflow to spill
	for i := 0; i < n; i++ {
		b.enqueue(bigKey + strconv.Itoa(i))
	}

	if !b.flushReq.Load() {
		t.Fatal("large-key spill crossed the byte bound but flushReq was not set")
	}
	b.spillMu.Lock()
	items := len(b.spill)
	b.spillMu.Unlock()
	if items >= cscInvalSpillMax {
		t.Fatalf("spill held %d items (>= item cap %d); the BYTE bound should have tripped first",
			items, cscInvalSpillMax)
	}

	// The worker must consume flushReq and full-Flush the cache.
	go b.run()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) && cache.flushes.Load() == 0 {
		time.Sleep(time.Millisecond)
	}
	b.stop()
	if cache.flushes.Load() == 0 {
		t.Fatal("worker did not full-Flush the cache after the spill byte bound was hit")
	}
}

// The worker's PENDING batch is size-capped by retained key BYTES, not only by item
// count: a stream of large keys (each under the admission limit) that stays far below
// the 4096-item cap must still flush once pending bytes cross the byte cap, so pending
// memory cannot grow to GBs across one window (fS2RM). The byte cap applies the batch
// (per-key deletes), it does not full-Flush.
func TestInvalBatcherPendingByteCapFlushes(t *testing.T) {
	cache := &countingCache{}
	b := newTestBatcher(cache, 64, time.Hour) // window never fires; only the byte cap can flush
	b.batchMaxBytes = 4096                    // small pending byte cap; item cap stays at 4096
	go b.run()

	// ~1 KiB keys: a dozen distinct keys cross the 4 KiB byte cap several times while
	// the item count stays a dozen, far under the 4096-item cap.
	bigKey := strings.Repeat("y", 1<<10)
	for i := 0; i < 12; i++ {
		b.enqueue(bigKey + strconv.Itoa(i))
	}

	// The byte cap must apply the batch BEFORE any stop-drain and WITHOUT the (1h)
	// window firing. Capture deletes before stop, then tear down.
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) && cache.deletes.Load() == 0 {
		time.Sleep(time.Millisecond)
	}
	got := cache.deletes.Load()
	b.stop()
	b.join()
	if got == 0 {
		t.Fatal("pending byte cap did not flush: deletes=0 before stop (only the item cap/window would have)")
	}
	if f := cache.flushes.Load(); f != 0 {
		t.Fatalf("byte cap should apply the batch, not full-Flush; flushes=%d", f)
	}
}

// TestClearRefreshQueueDetachesBatcherWithoutJoin pins the Group-A contract: the
// handler paths that tear a batcher down (here clearRefreshQueue; also
// setRefreshQueue/releaseLocked/setInvalBatchWindow) must DETACH and SIGNAL the
// batcher under h.mu but NOT join it there — join waits on the stop-drain, which
// would stall a sibling on the hot path and must never block the GC finalizer. The
// batcher's run() is never started, so its done channel never closes: the old code
// (b.join() under the lock) would hang forever, the fixed code returns the detached
// batcher for the caller to join OUTSIDE the lock.
func TestClearRefreshQueueDetachesBatcherWithoutJoin(t *testing.T) {
	cache := NewLocalCache(CacheConfig{MaxEntries: 8})
	h := &invalidateHandler{}
	if err := h.bindTo(cache, "p:"); err != nil {
		t.Fatalf("bindTo: %v", err)
	}
	q := &cscRefreshQueue{}
	h.setRefreshQueue(q) // active binding (no batcher yet, returns nil)

	// A batcher whose run() was NEVER started: done never closes, so join() blocks.
	b := newTestBatcher(cache, 1, time.Hour)
	h.mu.Lock()
	h.batcher = b
	h.mu.Unlock()

	var got *cscInvalBatcher
	done := make(chan struct{})
	go func() {
		got = h.clearRefreshQueue(q)
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(3 * time.Second):
		t.Fatal("clearRefreshQueue blocked: it joined the batcher under h.mu " +
			"(a never-started worker); the fix must detach+signal only")
	}
	if got != b {
		t.Fatalf("clearRefreshQueue returned %p, want the detached batcher %p", got, b)
	}
	if !b.stopped {
		t.Fatal("clearRefreshQueue did not signal the batcher to stop")
	}
	if h.batcher != nil {
		t.Fatal("clearRefreshQueue did not detach the batcher from the handler")
	}
}

// TestInvalBatcherChByteGateTripsFlush pins finding #2: the fast-path channel is
// bounded by item COUNT, not memory, so a burst of large keys must trip the
// full-Flush fallback via the retained ch-byte gate — not silently retain up to
// one channel's worth of large keys. Worker NOT started, ch cap large, so items
// stay in ch and only the byte gate can fire.
func TestInvalBatcherChByteGateTripsFlush(t *testing.T) {
	b := newTestBatcher(&countingCache{}, 1024, time.Hour)

	keySize := cscInvalSpillMaxBytes / 4 // four fill the 8 MiB cap exactly
	bigKey := strings.Repeat("k", keySize)
	// Four admit to ch (4 * cap/4 == cap, still within bound); the fifth would cross
	// the byte cap, so it must divert to the full-Flush fallback instead of ch.
	for i := 0; i < 5; i++ {
		b.enqueue(bigKey + strconv.Itoa(i)) // distinct keys: no dedup collapse
	}

	if !b.flushReq.Load() {
		t.Fatal("ch retained > byte cap but the full-Flush fallback was not requested " +
			"(ch admits by item count only; the byte gate must trip)")
	}
	if got := b.chBytes.Load(); got > int64(cscInvalSpillMaxBytes) {
		t.Fatalf("chBytes = %d exceeds the cap %d; the gate admitted an over-budget key",
			got, cscInvalSpillMaxBytes)
	}
}

// TestInvalBatcherOversizedKeyTripsFlush: a single key larger than the whole byte
// cap trips the full-Flush fallback on its own, never admitted to ch.
func TestInvalBatcherOversizedKeyTripsFlush(t *testing.T) {
	b := newTestBatcher(&countingCache{}, 1024, time.Hour)
	b.enqueue(strings.Repeat("x", cscInvalSpillMaxBytes+1))
	if !b.flushReq.Load() {
		t.Fatal("an oversized single key was admitted instead of tripping the full-Flush fallback")
	}
	if got := b.chBytes.Load(); got != 0 {
		t.Fatalf("chBytes = %d; the oversized key must not be admitted to ch", got)
	}
}

// TestInvalBatcherChByteReservationIsAtomic pins the atomic ch-byte reservation:
// the gate must RESERVE before the send (add-and-rollback), not Load-then-Add.
// Concurrent producers each sending a large distinct key could otherwise each read
// a stale below-cap Load, all send, and all Add — overshooting cscInvalSpillMaxBytes
// while the counter briefly read ~0. Worker NOT started and ch cap far exceeds the
// producer count, so ONLY the byte reservation (not the item-count bound or a
// draining worker) can keep chBytes under the cap.
func TestInvalBatcherChByteReservationIsAtomic(t *testing.T) {
	b := newTestBatcher(&countingCache{}, 4096, time.Hour)

	const producers = 64
	bigKey := strings.Repeat("x", 1<<20) // 1 MiB; ~8 fit the 8 MiB cap, the rest divert

	start := make(chan struct{})
	var wg sync.WaitGroup
	for i := 0; i < producers; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			<-start
			b.enqueue(bigKey + strconv.Itoa(i)) // distinct keys: no dedup collapse
		}(i)
	}
	close(start) // release all producers at once to maximize the reservation race
	wg.Wait()

	if got := b.chBytes.Load(); got > int64(cscInvalSpillMaxBytes) {
		t.Fatalf("chBytes = %d exceeds the cap %d after %d concurrent enqueues; the byte "+
			"reservation overshot (non-atomic Load-then-Add admits over-cap bytes)",
			got, cscInvalSpillMaxBytes, producers)
	}
}

// TestInvalBatcherNoRefreshGuardKeepsRefetched pins r3936994566: the batcher must apply
// the fetch-order guard even when refresh-on-invalidate is OFF (a *LocalCache with no
// refresh binding). A delayed invalidation whose fetchSnap predates a refetch must not
// evict the newer value (fetchSeq > fetchSnap). Before the fix the !canRefresh branch did
// an unconditional DeleteByRedisKey and served the refetch as a spurious miss.
func TestInvalBatcherNoRefreshGuardKeepsRefetched(t *testing.T) {
	ctx := context.Background()
	lc := NewLocalCache(CacheConfig{MaxEntries: 1024})
	mkValid := func(cacheKey, redisKey string) {
		lc.DeleteByCacheKey(cacheKey)
		tok, fetch := lc.Reserve(cacheKey, []string{redisKey})
		if tok == 0 || !fetch {
			t.Fatalf("Reserve(%q) = (%d, %v)", cacheKey, tok, fetch)
		}
		if !lc.fulfill(cacheKey, tok, 0, []byte("v")) {
			t.Fatalf("fulfill(%q) failed", cacheKey)
		}
	}

	// Refresh is OFF: no b.refresh binding -> apply takes the !canRefresh branch.
	b := newTestBatcher(lc, 64, time.Hour)

	mkValid("ck:hot", "rk") // original entry
	go b.run()

	// Invalidation snapshotted BEFORE the refetch below (its fetchSnap predates it).
	b.enqueue("rk")
	// Refetch after the invalidation was observed: the new entry's fetchSeq exceeds the
	// duplicate's snapshot, so the guard must keep it.
	mkValid("ck:hot", "rk")

	b.stop()
	b.join()

	if _, ok := lc.Get(ctx, "ck:hot"); !ok {
		t.Fatal("refresh-off apply evicted a value refetched after the invalidation; the " +
			"fetch-order guard must apply even without refresh-on-invalidate (r3936994566)")
	}
}

// TestInvalBatcherSizeCapGuardKeepsRefreshed pins #3965 F1 under the fetch-order
// guard. The size-cap flush now CLEARS the per-epoch dedup (seen), so a key can be
// applied on both sides of the cap — but a stale duplicate that straddles a
// refetch must not evict the fresh entry. The duplicate is enqueued (its fetch
// snapshot frozen) BEFORE the simulated refetch, so at apply the entry's fetchSeq
// exceeds the duplicate's snapshot and the delete is skipped. The DeletionStats
// noop delta proves the second occurrence was actually applied (seen was cleared,
// not deduped) and the guard — not dedup — is what saved the entry.
func TestInvalBatcherSizeCapGuardKeepsRefreshed(t *testing.T) {
	const batchMax = 8 // small cap so the size-cap flush fires deterministically

	ctx := context.Background()
	lc := NewLocalCache(CacheConfig{MaxEntries: 1024})
	mkValid := func(cacheKey, redisKey string) {
		lc.DeleteByCacheKey(cacheKey)
		tok, fetch := lc.Reserve(cacheKey, []string{redisKey})
		if tok == 0 || !fetch {
			t.Fatalf("Reserve(%q) = (%d, %v)", cacheKey, tok, fetch)
		}
		if !lc.fulfill(cacheKey, tok, 0, []byte("v")) {
			t.Fatalf("fulfill(%q) failed", cacheKey)
		}
	}

	// window never fires: only the size-cap flush and the stop-drain apply.
	b := newTestBatcher(lc, batchMax*2, time.Hour)
	b.batchMax = batchMax // per-batcher, set before the worker starts (no global race)
	q := &cscRefreshQueue{ch: make(chan cscRefreshTarget, 64)}
	q.sinceToken.Store(cscInvalNoHorizon) // seed the horizon so the entry reads as hot
	b.refresh.Store(q)

	mkValid("ck:hot", "rk:hot") // the target entry, read after the horizon

	go b.run()

	// Enqueue the target FIRST, then fillers so pending reaches the cap and the
	// size-cap flush applies the target in the first half of the window (deleting it
	// and clearing seen).
	b.enqueue("rk:hot")
	for i := 0; i < batchMax-1; i++ {
		b.enqueue("rk:filler" + strconv.Itoa(i))
	}

	// Wait for the size-cap flush to FULLY apply batch #1 — all 8 keys, so the
	// deletion counters are settled before we snapshot them (polling Get would race
	// apply mid-loop while fillers still bump the counters).
	deadline := time.Now().Add(2 * time.Second)
	for {
		if del, _ := lc.DeletionStats(); del >= batchMax {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("size-cap flush did not apply batch #1")
		}
		time.Sleep(time.Millisecond)
	}
	if _, ok := lc.Get(ctx, "ck:hot"); ok {
		t.Fatal("apply #1 did not delete the target")
	}

	// The SAME key arrives again, post-flush, with its fetch snapshot frozen NOW —
	// BEFORE the refetch below. This is the stale straddling duplicate: it predates
	// the refresh republish, so the guard must skip it at apply time.
	delBefore, noopBefore := lc.DeletionStats()
	b.enqueue("rk:hot")

	// Simulate the refresh republish AFTER the duplicate was enqueued: the refetch
	// reserves a new entry whose fetchSeq exceeds the duplicate's snapshot.
	mkValid("ck:hot", "rk:hot")

	b.stop()
	b.join()

	if _, ok := lc.Get(ctx, "ck:hot"); !ok {
		t.Fatal("the stale straddling duplicate evicted the refreshed entry; the " +
			"fetch-order guard must keep an entry refetched after the invalidation (#3965 F1)")
	}
	// The duplicate was applied (seen cleared) but skipped by the guard: applied
	// deletes advanced by 1 and that delete removed nothing (noop), which is the
	// guard-skip signature — not a dedup drop (which would advance neither).
	delAfter, noopAfter := lc.DeletionStats()
	if delAfter-delBefore != 1 || noopAfter-noopBefore != 1 {
		t.Fatalf("duplicate apply delta = (deletions %d, noop %d), want (1, 1): the "+
			"second occurrence must be applied and skipped by the guard, not deduped away",
			delAfter-delBefore, noopAfter-noopBefore)
	}
}

// TestInvalBatcherDedupKeepsLatestFetchSnap pins #3989: when two writes to the same
// key land in one batching window with a refetch between them, the second write's
// invalidation is deduped, but its NEWER fetch snapshot must be carried onto the
// retained item — otherwise the apply-time guard uses the first write's older snapshot
// and preserves the entry refetched BETWEEN the writes, serving it stale.
func TestInvalBatcherDedupKeepsLatestFetchSnap(t *testing.T) {
	ctx := context.Background()
	lc := NewLocalCache(CacheConfig{MaxEntries: 1024})
	mkValid := func(cacheKey, redisKey string) {
		lc.DeleteByCacheKey(cacheKey)
		tok, fetch := lc.Reserve(cacheKey, []string{redisKey})
		if tok == 0 || !fetch {
			t.Fatalf("Reserve(%q) = (%d, %v)", cacheKey, tok, fetch)
		}
		if !lc.fulfill(cacheKey, tok, 0, []byte("v")) {
			t.Fatalf("fulfill(%q) failed", cacheKey)
		}
	}

	// window never fires; only the stop-drain applies (which dedups ch in order).
	b := newTestBatcher(lc, 64, time.Hour)
	q := &cscRefreshQueue{ch: make(chan cscRefreshTarget, 64)}
	q.sinceToken.Store(cscInvalNoHorizon)
	b.refresh.Store(q)

	mkValid("ck:hot", "rk") // original entry
	go b.run()

	// Write 1's invalidation: snapshot BEFORE the refetch below.
	b.enqueue("rk")
	// Refetch between the two writes: the entry's fetchSeq now exceeds occ1's snapshot.
	mkValid("ck:hot", "rk")
	// Write 2's invalidation: same key, same window -> deduped, but carries a NEWER
	// snapshot that must overwrite occ1's so the guard evicts the between-writes entry.
	b.enqueue("rk")

	b.stop()
	b.join()

	if _, ok := lc.Get(ctx, "ck:hot"); ok {
		t.Fatal("entry refetched between two same-window writes was preserved; dedup must " +
			"keep the LATEST fetch snapshot so the guard evicts it (#3989)")
	}
}