go-redis/csc_refresh_support_test.go

package redis

import (
	"context"
	"testing"
)

func TestCollectHotAndDeleteNoHorizonRefreshesEverything(t *testing.T) {
	ctx := context.Background()
	lc := NewLocalCache(CacheConfig{MaxEntries: 64})

	tok, fetch := lc.Reserve("ck:old", []string{"rk:old"})
	if tok == 0 || !fetch {
		t.Fatalf("Reserve(ck:old) = (%d, %v)", tok, fetch)
	}
	if !lc.fulfill("ck:old", tok, 0, []byte("v")) {
		t.Fatal("fulfill(ck:old) failed")
	}

	// A horizon taken long after the entry's last read would normally exclude
	// it (see TestCollectHotAndDeleteRecencyFilter). cscInvalNoHorizon must
	// override that and collect it anyway.
	for i := 0; i < 100; i++ {
		lruSequence.Add(1)
	}

	got := lc.deleteByRedisKeyCollectingHot("rk:old", cscInvalNoHorizon, ^uint64(0), nil)
	if len(got) != 1 || got[0].cacheKey != "ck:old" {
		t.Fatalf("cscInvalNoHorizon did not collect the entry as a refetch target: %v", got)
	}
	if _, ok := lc.Get(ctx, "ck:old"); ok {
		t.Fatal("entry was not deleted by the invalidation")
	}
}

func TestCollectHotFetchOrderGuard(t *testing.T) {
	ctx := context.Background()
	lc := NewLocalCache(CacheConfig{MaxEntries: 64})
	mkValid := func(cacheKey string, redisKeys ...string) {
		lc.DeleteByCacheKey(cacheKey)
		tok, fetch := lc.Reserve(cacheKey, redisKeys)
		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)
		}
	}

	// Entry fetched, then an invalidation observed: snapshot the fetch clock now.
	mkValid("ck", "rk")
	snapStale := cscFetchSeq.Load()

	// The entry is refetched AFTER the invalidation was observed (a refresh or a
	// concurrent read miss): its fetchSeq now exceeds the stale snapshot.
	mkValid("ck", "rk")

	// Stale invalidation applied: guard keeps the refetched entry and does not offer
	// it for refetch.
	if got := lc.deleteByRedisKeyCollectingHot("rk", 0, snapStale, nil); len(got) != 0 {
		t.Fatalf("refetched entry offered as a refetch target: %v", got)
	}
	if _, ok := lc.Get(ctx, "ck"); !ok {
		t.Fatal("stale invalidation evicted an entry refetched after it (fetch-order guard failed)")
	}

	// A GENUINE later invalidation (observed now, after the refetch) must evict.
	snapFresh := cscFetchSeq.Load()
	lc.deleteByRedisKeyCollectingHot("rk", 0, snapFresh, nil)
	if _, ok := lc.Get(ctx, "ck"); ok {
		t.Fatal("genuine post-refetch invalidation did not evict the entry")
	}
}

// TestCollectHotMultiKeyRefreshSurvives pins the multi-key variant of #3965 that
// dedup (seen) structurally cannot fix: one cached command depends on several
// Redis keys, so a refresh triggered by one key's invalidation republishes the
// entry, and another key's (stale) invalidation must not re-evict it. seen is keyed
// by Redis key, so k1 and k2 are different dedup entries — only the per-entry
// fetch-order guard can reason about "was THIS entry refetched after me".
func TestCollectHotMultiKeyRefreshSurvives(t *testing.T) {
	ctx := context.Background()
	lc := NewLocalCache(CacheConfig{MaxEntries: 64})
	mkValid := func(cacheKey string, redisKeys ...string) {
		lc.DeleteByCacheKey(cacheKey)
		tok, fetch := lc.Reserve(cacheKey, redisKeys)
		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)
		}
	}

	// Two cached commands sharing k2: C1 = MGET k1 k2, C2 = GET k2.
	mkValid("mget:k1k2", "k1", "k2")
	mkValid("get:k2", "k2")

	// Both k1 and k2 change; the client observes both invalidations now (same push,
	// so one snapshot covers both items).
	snap := cscFetchSeq.Load()

	// k1's invalidation applies first: it deletes C1 (depends on k1).
	if _, ok := lc.Get(ctx, "mget:k1k2"); !ok {
		t.Fatal("precondition: C1 should be present")
	}
	lc.deleteByRedisKeyCollectingHot("k1", 0, snap, nil)
	if _, ok := lc.Get(ctx, "mget:k1k2"); ok {
		t.Fatal("k1 invalidation should have deleted C1")
	}

	// The refresh refetches C1 (MGET k1 k2) — issued after both writes, so the reply
	// already reflects the new k2. fetchSeq now exceeds snap.
	mkValid("mget:k1k2", "k1", "k2")

	// k2's invalidation applies second, fanning out to BOTH C1 and C2 via byRedisKey.
	// C1 was refetched after snap -> guard keeps it; C2 predates snap -> evicted.
	lc.deleteByRedisKeyCollectingHot("k2", 0, snap, nil)

	if _, ok := lc.Get(ctx, "mget:k1k2"); !ok {
		t.Fatal("k2's stale invalidation evicted the refetched multi-key entry C1 " +
			"(guard must keep an MGET result refetched after the write)")
	}
	if _, ok := lc.Get(ctx, "get:k2"); ok {
		t.Fatal("k2's invalidation should have evicted C2 (GET k2 was not refetched)")
	}
}

// TestCSCRefreshReplyCacheable pins the refresh cacheability decision on the FULL
// reply, not raw[0]. A RESP3 attribute-prefixed error leads with RespAttr ('|'),
// so a first-byte error check would publish an attributed error as a fresh entry
// (a false success that bumps Refreshed and forces the next reader to evict and
// refetch). A nil reply IS cacheable; an empty reply never is.
func TestCSCRefreshReplyCacheable(t *testing.T) {
	cases := []struct {
		name      string
		raw       string
		cacheable bool
	}{
		{"plain value", "$2\r\nhi\r\n", true},
		{"resp3 nil", "_\r\n", true},
		{"resp2 nil", "$-1\r\n", true},
		{"top-level error", "-WRONGTYPE nope\r\n", false},
		{"blob error", "!11\r\nERR bad arg\r\n", false},
		// The bug: the error frame is prefixed by a RESP3 attribute, so raw[0]=='|'.
		{"attribute-prefixed error", "|1\r\n$3\r\nttl\r\n:100\r\n-WRONGTYPE nope\r\n", false},
		{"attribute-prefixed value", "|1\r\n$3\r\nttl\r\n:100\r\n$2\r\nhi\r\n", true},
		{"empty", "", false},
	}
	for _, tc := range cases {
		if got := cscRefreshReplyCacheable([]byte(tc.raw)); got != tc.cacheable {
			t.Errorf("%s: cscRefreshReplyCacheable = %v, want %v", tc.name, got, tc.cacheable)
		}
	}
}

// TestCSCRefreshChunkEnd pins the refresh round-trip boundaries: count cap,
// request-byte budget (key wire form), and the expected-REPLY-byte budget —
// the reply side is what jams a flow-controlled path, so large evicted values
// must shrink the chunk. The first target always goes.
func TestCSCRefreshChunkEnd(t *testing.T) {
	mk := func(key string, val int) cscRefreshTarget {
		return cscRefreshTarget{cacheKey: key, valBytes: val}
	}

	// Reply budget splits the chunk: three 15KiB values fit two per chunk
	// under the 32KiB reply budget (request bytes tiny).
	big := []cscRefreshTarget{mk("get:a", 15<<10), mk("get:b", 15<<10), mk("get:c", 15<<10)}
	if got := cscRefreshChunkEnd(big, 0, 0, 1<<20); got != 2 {
		t.Fatalf("reply budget: end = %d, want 2 (two 15KiB replies fit, third exceeds 32KiB)", got)
	}
	if got := cscRefreshChunkEnd(big, 2, 0, 1<<20); got != 3 {
		t.Fatalf("tail: end = %d, want 3", got)
	}

	// A single oversized value still goes (progress guarantee).
	huge := []cscRefreshTarget{mk("get:x", 1<<20), mk("get:y", 1)}
	if got := cscRefreshChunkEnd(huge, 0, 0, 1<<20); got != 1 {
		t.Fatalf("oversized first: end = %d, want 1 (always include the first target)", got)
	}

	// Request-byte budget still enforced: long keys, zero-size values.
	longKey := make([]byte, 100)
	for i := range longKey {
		longKey[i] = 'k'
	}
	keys := []cscRefreshTarget{mk(string(longKey), 0), mk(string(longKey), 0), mk(string(longKey), 0)}
	if got := cscRefreshChunkEnd(keys, 0, 0, 150); got != 1 {
		t.Fatalf("request budget: end = %d, want 1 (second 100B key exceeds 150B budget)", got)
	}

	// Count cap: many tiny targets stop at cscRefreshBatchMax.
	many := make([]cscRefreshTarget, cscRefreshBatchMax+10)
	for i := range many {
		many[i] = mk("get:k", 1)
	}
	if got := cscRefreshChunkEnd(many, 0, 0, 1<<20); got != cscRefreshBatchMax {
		t.Fatalf("count cap: end = %d, want %d", got, cscRefreshBatchMax)
	}
}

// TestCSCRefreshDemandGenerationIgnoresStaleSignal pins the demand generation gate:
// a demand nudge is stamped with the window generation it belongs to, and the
// refresher ignores a nudge whose generation was retired — so a nudge for an
// already-flushed window cannot flush the NEXT one early (which would defeat the
// coalescing window and inflate DemandFlushes). Pins signalDemand's stamping and the
// demandIsCurrent truth table (the goroutine race lives inside runCSCRefresher).
func TestCSCRefreshDemandGenerationIgnoresStaleSignal(t *testing.T) {
	if !cscDemandRefresh {
		t.Skip("demand refresh disabled")
	}
	q := &cscRefreshQueue{demandCh: make(chan uint64, 1)}

	q.pendingSet.Store("ck", struct{}{})
	q.signalDemand("ck")
	var stale uint64
	select {
	case stale = <-q.demandCh:
	default:
		t.Fatal("signalDemand did not stamp a nudge for a pending key")
	}
	if !q.demandIsCurrent(stale) {
		t.Fatal("a nudge for the CURRENT window must be honored")
	}

	q.pendingSet.Delete("ck")
	q.demandGen.Add(1) // refresher clears window N and retires its generation

	if q.demandIsCurrent(stale) {
		t.Fatal("a nudge from a retired generation must be ignored")
	}

	q.pendingSet.Store("ck2", struct{}{})
	q.signalDemand("ck2")
	select {
	case g := <-q.demandCh:
		if !q.demandIsCurrent(g) {
			t.Fatal("a nudge for the new current window must be honored")
		}
	default:
		t.Fatal("signalDemand did not stamp a nudge for the new window's key")
	}
}

// TestCollectHotGuardKeepsInProgress pins that the fetch-order guard covers
// in-progress reservations, not only Valid entries (#3989). A reservation issued
// AFTER the invalidation was observed (fetchSeq > snapshot) cannot hold the
// pre-write value, so it must survive — removing it would fail its racing Fulfill
// and wake every waiter as a miss. A reservation issued at or before the observe is
// still removed (it may predate the write).
func TestCollectHotGuardKeepsInProgress(t *testing.T) {
	ctx := context.Background()
	lc := NewLocalCache(CacheConfig{MaxEntries: 64})

	// In-progress reservation issued BEFORE the invalidation is observed.
	tokOld, fetchOld := lc.Reserve("ck:old", []string{"rk"})
	if tokOld == 0 || !fetchOld {
		t.Fatalf("Reserve(ck:old) = (%d, %v)", tokOld, fetchOld)
	}
	snap := cscFetchSeq.Load()
	// In-progress reservation issued AFTER the observe.
	tokNew, fetchNew := lc.Reserve("ck:new", []string{"rk"})
	if tokNew == 0 || !fetchNew {
		t.Fatalf("Reserve(ck:new) = (%d, %v)", tokNew, fetchNew)
	}

	lc.deleteByRedisKeyCollectingHot("rk", 0, snap, nil)

	// Pre-observe reservation removed: its racing Fulfill now fails.
	if lc.fulfill("ck:old", tokOld, 0, []byte("v")) {
		t.Fatal("pre-observe in-progress reservation should have been removed (Fulfill must fail)")
	}
	// Post-observe reservation kept: its Fulfill still publishes.
	if !lc.fulfill("ck:new", tokNew, 0, []byte("v")) {
		t.Fatal("post-observe in-progress reservation was wrongly removed (Fulfill failed)")
	}
	if _, ok := lc.Get(ctx, "ck:new"); !ok {
		t.Fatal("post-observe in-progress entry did not survive the invalidation")
	}
}