redis/go-redis
1
0
Fork 0
mirror of https://github.com/redis/go-redis.git synced 2026-10-22 19:49:00 +03:00
Code Activity

perf(csc)!: mark reads with a second-chance bit (#4034)

Browse Source 
* feat(csc)!: always refresh invalidated entries

ClientSideCacheRefreshRecencyWindow bounded refresh to entries read
within a window. It was the only consumer that needed a per-READ
recency timestamp on the cache entry, and that timestamp is the
dominant cost of a cache hit.

Removing the option lets the read path stop stamping entries, which the
following commit does. Refresh now covers every invalidated entry.

BREAKING CHANGE: Options.ClientSideCacheRefreshRecencyWindow is removed.
Refresh-on-invalidate always refreshes every invalidated entry.

* perf(csc): mark reads with a second-chance bit

A cache hit stamped entry.lastAccessNs. The read already touches the
entry's cache line, but a store dirties it, so the line must be held
exclusive and costs a writeback when evicted from the CPU cache. With
many concurrent readers over a working set larger than L2, that store
dominates the cost of a hit.

A read now sets a readSinceSweep bit instead. A load that finds it set
leaves the line SHARED, so every core can hold it at once and only the
first read after an eviction sweep writes. Eviction consumes the bit as
a second chance; ordering still uses lastAccessNs, written on insert.

Measured at 256 concurrent callers, get=90/set=10, 99.94% hit rate:
263,801 -> 328,842 reads/s (+24.7%), CPU 27.0 -> 18.9 us/op, hit rate
unchanged. Removing the recency update entirely measured +40.5%, so
this recovers about three fifths of the headroom while keeping
eviction honest.

* perf(csc): decode simple cached replies without a reader

Serving a hit allocated a proto.Reader (and its bufio buffer) per hit
and re-parsed the cached bytes. applyCachedFast decodes the simple
single-frame replies -- nil, status, bulk string -- straight into the
Cmder, and getShared serves the bytes without the defensive copy where
the caller provably cannot retain them.

* fix(csc): do not evict the entry just fulfilled

fulfill publishes an entry with its second-chance bit clear and then
evicts. In a warm memory-capped shard, where every resident entry has
been read, that entry was the only clear-bit candidate and was evicted
on the spot; and since a clear candidate always existed, the sweep that
resets the others never ran, so the shard admitted nothing new. The
entry just fulfilled is now skipped in its own eviction pass.

A refresh republish also restores the invalidated entry's read bit with
its access token, so a refreshed hot key is not the first victim.

* fix(csc): decline negative bulk lengths other than -1

The fast cached-reply decoder took any negative bulk length as a RESP2
null. The reader rejects lengths below -1, and that rejection is what
makes processCached drop a bad entry and refetch; the fast path turned
it into a false miss. It now handles only -1 and declines the rest.

* fix(csc): reject bare LF in cached status replies

The fast cached-reply decoder rejected a status frame with an interior
CR but took one with a bare LF, such as "+OK\nMORE\r\n", as a valid
string. The reader stops at the first LF and rejects the frame, and
that rejection is what makes processCached drop the entry and refetch.
The fast path now declines an interior CR or LF.

* fix(csc): keep refresh recency in the publish lock

A refresh republish published the entry with its second-chance bit
clear and restored the bit in a second lock. On a warm shard the
republished entry was then the only clear-bit victim, so an insert in
that gap evicted the key the refresh had just fetched. The refresher
now stages the kept recency on its reservation (stageRefreshAccess),
and fulfill applies it in the same lock as the publish.

* docs(csc)!: refresh-on-invalidate refreshes all

ClientSideCacheRefreshOnInvalidate still said it re-fetches recently
read keys. Since the recency window was removed, it re-fetches every
cached entry of an invalidated key, so on a write-heavy keyspace the
refetch traffic covers the whole resident cache.

BREAKING CHANGE: Options.ClientSideCacheRefreshRecencyWindow and
UniversalOptions.ClientSideCacheRefreshRecencyWindow are removed. They
shipped only in v9.23.0-beta.1; delete the field where it is set.

* docs(csc): note reads are not monotonic per caller

A read that loses the reservation twice reads the server directly.
Another caller's fetch, reserved earlier and still running, can store
an older value afterwards, and a later hit returns it until the key's
invalidation arrives. This is within the staleness a hit already has,
so document it rather than drop the older fetch, which would cost its
waiters a miss each under the churn that leads here.
This commit is contained in:
Nedyalko Dyakov
2026-10-01 14:36:58 +03:00
committed by GitHub
parent a8c360cc45
commit 58cc03e024
16 changed files with 725 additions and 463 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
README.md
View File

@@ -330,7 +330,14 @@ supported because they use dedicated connections.
Invalidations are processed asynchronously. `DrainInterval` controls how often
idle connections are checked for them, while `MaxStaleness` can provide an
optional upper bound on an entry's lifetime. See the
optional upper bound on an entry's lifetime.
Until a key's invalidation arrives, a cache hit can return a value older than
the one on the server. Reads are also not monotonic for one caller. A read that
waits twice for another caller's fetch of the same key reads the server
directly instead. The other fetch ran on the server earlier, can store its
older value afterwards, and a later hit then returns that older value until the
invalidation arrives. See the
[client-side caching example](./example/client-side-caching) for a working
demonstration.

172
csc_applyfast_test.go Normal file
View File

@@ -0,0 +1,172 @@
package redis
import (
"bytes"
"context"
"fmt"
"strings"
"testing"
"github.com/redis/go-redis/v9/internal/proto"
)
// applyCachedFast decodes the simple reply shapes without allocating a
// proto.Reader. It is only safe if it agrees with the generic path on every
// input, including the ones it declines: a fast path that silently decodes
// something differently from the reader would corrupt a cached read.
//
// This drives both paths over the same payloads and compares value and error.
func TestApplyCachedFastMatchesGenericReader(t *testing.T) {
ctx := context.Background()
big := strings.Repeat("x", 600)
cases := []struct {
name string
raw string
}{
{"empty bulk", "$0\r\n\r\n"},
{"short bulk", "$5\r\nhello\r\n"},
{"bulk with CR inside", "$4\r\na\rb\r\n"},
{"bulk with LF inside", "$4\r\na\nb\r\n"},
{"bulk with CRLF inside", "$6\r\na\r\nbc\r\n"},
{"binary bulk", "$3\r\n\x00\x01\x02\r\n"},
{"large bulk", fmt.Sprintf("$%d\r\n%s\r\n", len(big), big)},
{"resp2 null bulk", "$-1\r\n"},
{"resp3 null", "_\r\n"},
{"status ok", "+OK\r\n"},
{"status with space", "+string\r\n"},
{"empty status", "+\r\n"},
// Shapes the fast path must DECLINE rather than misread.
{"integer", ":42\r\n"},
{"array", "*2\r\n$1\r\na\r\n$1\r\nb\r\n"},
{"resp3 double", ",3.14\r\n"},
{"resp3 bool", "#t\r\n"},
{"verbatim", "=9\r\ntxt:hello\r\n"},
{"error", "-ERR nope\r\n"},
// Malformed / truncated: must not panic, must not disagree.
{"truncated bulk", "$5\r\nhel\r\n"},
{"bulk length lies long", "$99\r\nhello\r\n"},
{"no terminator", "$5\r\nhello"},
{"bad length", "$abc\r\nhello\r\n"},
// A bare LF ends the reader's line early, so these are invalid.
{"status with bare LF", "+OK\nMORE\r\n"},
{"status that is only a bare LF", "+\n\r\n"},
// Only -1 is the RESP2 null; the reader rejects any other negative.
{"negative bulk length", "$-2\r\n"},
{"very negative bulk length", "$-100\r\n"},
{"just crlf", "\r\n"},
{"empty", ""},
}
for _, tc := range cases {
for _, kind := range []string{"string", "status"} {
t.Run(tc.name+"/"+kind, func(t *testing.T) {
raw := []byte(tc.raw)
mk := func() Cmder {
if kind == "string" {
return NewStringCmd(ctx, "get", "k")
}
return NewStatusCmd(ctx, "type", "k")
}
// Generic path, verbatim from applyCachedReply's fallback.
gen := mk()
genErr := gen.readReply(proto.NewReaderSize(bytes.NewReader(raw), len(raw)+1))
// Fast path via applyCachedReply (falls back internally when it
// declines, so this is what production actually runs).
fast := mk()
fastErr := applyCachedReply(fast, raw)
sameErr := (genErr == nil) == (fastErr == nil)
if sameErr && genErr != nil {
// Both non-nil: Nil must match Nil exactly; other errors
// only have to agree on being errors, since the generic
// reader's messages are its own.
sameErr = (genErr == Nil) == (fastErr == Nil)
}
if !sameErr {
t.Fatalf("error mismatch: generic=%v fast=%v", genErr, fastErr)
}
var genVal, fastVal string
switch c := gen.(type) {
case *StringCmd:
genVal = c.Val()
fastVal = fast.(*StringCmd).Val()
case *StatusCmd:
genVal = c.Val()
fastVal = fast.(*StatusCmd).Val()
}
if genVal != fastVal {
t.Fatalf("value mismatch: generic=%q fast=%q", genVal, fastVal)
}
})
}
}
}
// The fast path must not alias the cache entry's bytes into the command: the
// slice it is handed belongs to the cache (see LocalCache.getShared), so a
// decoded value that shared its backing array would change under the caller
// the next time that entry was refetched.
func TestApplyCachedFastCopiesTheValue(t *testing.T) {
ctx := context.Background()
raw := []byte("$5\r\nhello\r\n")
cmd := NewStringCmd(ctx, "get", "k")
if err := applyCachedReply(cmd, raw); err != nil {
t.Fatalf("applyCachedReply: %v", err)
}
if got := cmd.Val(); got != "hello" {
t.Fatalf("value = %q, want hello", got)
}
// Scribble over the source the way a refetch writing into a reused buffer
// would. The decoded string must be unaffected.
for i := range raw {
raw[i] = 'Z'
}
if got := cmd.Val(); got != "hello" {
t.Fatalf("decoded value aliases the source: became %q after overwrite", got)
}
}
// getShared hands out the entry's own slice. Prove it really is shared (that
// is the point -- no copy) and that Get still copies, so the public contract
// is unchanged for third-party callers.
func TestGetSharedAliasesAndGetCopies(t *testing.T) {
ctx := context.Background()
lc := NewLocalCache(CacheConfig{MaxEntries: 8})
tok, fetch := lc.Reserve("ck", []string{"rk"})
if tok == 0 || !fetch {
t.Fatal("Reserve did not hand out a token")
}
if !lc.fulfill("ck", tok, 0, []byte("payload")) {
t.Fatal("fulfill failed")
}
a, ok := lc.getShared(ctx, "ck")
if !ok {
t.Fatal("getShared missed")
}
b, ok := lc.getShared(ctx, "ck")
if !ok {
t.Fatal("second getShared missed")
}
if &a[0] != &b[0] {
t.Fatal("getShared copied; it must alias the entry")
}
c, ok := lc.Get(ctx, "ck")
if !ok {
t.Fatal("Get missed")
}
if &c[0] == &a[0] {
t.Fatal("Get aliased the entry; the public contract must keep copying")
}
if string(c) != "payload" {
t.Fatalf("Get returned %q", c)
}
}

4
csc_coalesce_options_test.go
View File

@@ -12,7 +12,6 @@ import (
func TestUniversalOptionsSimpleCopiesCSCCoalesce(t *testing.T) {
u := &UniversalOptions{
ClientSideCacheRefreshOnInvalidate: true,
ClientSideCacheRefreshRecencyWindow: 3 * time.Millisecond,
ClientSideCacheCoalesceMisses: true,
ClientSideCacheInvalidationBatchWindow: 7 * time.Millisecond,
}
@@ -20,9 +19,6 @@ func TestUniversalOptionsSimpleCopiesCSCCoalesce(t *testing.T) {
if !o.ClientSideCacheRefreshOnInvalidate {
t.Error("Simple() dropped ClientSideCacheRefreshOnInvalidate")
}
if o.ClientSideCacheRefreshRecencyWindow != 3*time.Millisecond {
t.Errorf("Simple() ClientSideCacheRefreshRecencyWindow = %v, want 3ms", o.ClientSideCacheRefreshRecencyWindow)
}
if !o.ClientSideCacheCoalesceMisses {
t.Error("Simple() dropped ClientSideCacheCoalesceMisses")
}

124
csc_integration.go
View File

@@ -14,6 +14,7 @@ import (
"github.com/redis/go-redis/v9/internal"
"github.com/redis/go-redis/v9/internal/pool"
"github.com/redis/go-redis/v9/internal/proto"
"github.com/redis/go-redis/v9/internal/util"
"github.com/redis/go-redis/v9/push"
)
@@ -1352,12 +1353,120 @@ func (c *baseClient) stopCSCRefresherAndCoalescer() {
h.workersStopOnce.Do(body)
}
// cscGet reads a cached value, skipping the defensive copy when the cache is
// the built-in one.
//
// LocalCache.Get copies because Cache is a public interface whose []byte
// return a third-party implementation's caller may retain. The built-in read
// path does not retain it: every caller hands the bytes straight to
// applyCachedReply, which decodes and drops them. Pairing the built-in cache
// with the built-in read path is therefore the one place the copy is provably
// unnecessary -- the same "only the built-in *LocalCache" gate miss coalescing
// already uses.
//
// The returned slice must not be retained or mutated.
func (c *baseClient) cscGet(ctx context.Context, key string) ([]byte, bool) {
if lc, ok := c.csc.(*LocalCache); ok {
return lc.getShared(ctx, key)
}
return c.csc.Get(ctx, key)
}
// applyCachedReply populates cmd from a previously captured raw RESP reply by
// replaying it through the command's own readReply.
//
// The generic path allocates a bytes.Reader and a proto.Reader (with its bufio
// buffer) per hit, then re-parses. applyCachedFast skips both for the two
// reply shapes that dominate the cacheable command set.
func applyCachedReply(cmd Cmder, raw []byte) error {
if err, done := applyCachedFast(cmd, raw); done {
return err
}
return cmd.readReply(proto.NewReaderSize(bytes.NewReader(raw), len(raw)+1))
}
// applyCachedFast decodes the simple single-frame replies directly into the
// command, returning done=false for anything it does not fully recognise so
// the caller falls back to the generic reader.
//
// Only shapes whose decoded value is an immutable Go string are handled:
// a blob string into *StringCmd (get, getrange, hget, ...) and a status into
// *StatusCmd (type). Container replies -- mget, hgetall, smembers, zrange and
// the rest of the allow-list -- keep re-parsing, deliberately: their decoded
// form is a slice or map, and handing the same one to two callers would let
// either mutate what the other reads, and the cached entry with it. For those,
// re-parsing IS the defensive copy.
//
// raw must not be retained: it aliases the cache entry's value (see
// LocalCache.getShared). Everything produced here is a fresh string.
func applyCachedFast(cmd Cmder, raw []byte) (err error, done bool) {
if len(raw) < 3 || raw[len(raw)-2] != '\r' || raw[len(raw)-1] != '\n' {
return nil, false
}
switch raw[0] {
case proto.RespNil: // RESP3 null: "_\r\n"
if len(raw) != 3 {
return nil, false
}
switch c := cmd.(type) {
case *StringCmd:
c.SetVal("")
return Nil, true
case *StatusCmd:
c.SetVal("")
return Nil, true
}
return nil, false
case proto.RespStatus: // "+OK\r\n"
c, ok := cmd.(*StatusCmd)
if !ok {
return nil, false
}
// Single frame only: a CR or a bare LF before the terminator would mean
// more than one line, which this path does not handle. The reader stops
// at the first LF and rejects the frame, and that rejection is what
// makes processCached drop the bad entry and refetch.
if bytes.ContainsAny(raw[:len(raw)-2], "\r\n") {
return nil, false
}
c.SetVal(string(raw[1 : len(raw)-2]))
return nil, true
case proto.RespString: // "$5\r\nhello\r\n", or "$-1\r\n" for RESP2 null
c, ok := cmd.(*StringCmd)
if !ok {
return nil, false
}
i := bytes.IndexByte(raw, '\r')
if i < 1 || i+1 >= len(raw) || raw[i+1] != '\n' {
return nil, false
}
n, perr := util.ParseInt(raw[1:i], 10, 64)
if perr != nil {
return nil, false
}
if n < 0 {
// Only -1 is the RESP2 null bulk string. The reader rejects any
// other negative length, so decline it: the generic path then
// fails and processCached drops the bad entry and refetches.
if n != -1 || i+2 != len(raw) {
return nil, false
}
c.SetVal("")
return Nil, true
}
// Payload must be exactly n bytes followed by the trailing CRLF.
start := i + 2
if int64(len(raw)) != int64(start)+n+2 {
return nil, false
}
c.SetVal(string(raw[start : start+int(n)]))
return nil, true
}
return nil, false
}
// classifyCachedReply reports the same error applyCachedReply would, without a
// caller command to populate. The miss coalescer uses it on the abandoned path
// (the caller returned and owns its Cmder again) to decide cache-vs-cancel: a
@@ -1453,7 +1562,7 @@ func (c *baseClient) processCached(ctx context.Context, cmd Cmder, state *proces
}
// Serve hits straight from the cache.
if data, ok := c.csc.Get(ctx, key); ok {
if data, ok := c.cscGet(ctx, key); ok {
if err := ctx.Err(); err != nil {
return err
}
@@ -1475,7 +1584,7 @@ func (c *baseClient) processCached(ctx context.Context, cmd Cmder, state *proces
token, shouldFetch := c.csc.Reserve(key, nsRedisKeys)
if !shouldFetch {
// Another goroutine is fetching; Get below waits until it completes.
if data, ok := c.csc.Get(ctx, key); ok {
if data, ok := c.cscGet(ctx, key); ok {
if err := ctx.Err(); err != nil {
return err
}
@@ -1567,7 +1676,7 @@ func (c *baseClient) processCached(ctx context.Context, cmd Cmder, state *proces
// one hot key into a pool stampede mid-recovery — defeating the
// coalescing this path exists for. Same shape (and 2x-RTT churn
// tradeoff) as the first-Reserve loser path above.
if data, ok := c.csc.Get(ctx, key); ok {
if data, ok := c.cscGet(ctx, key); ok {
if err := ctx.Err(); err != nil {
return err
}
@@ -1594,6 +1703,15 @@ func (c *baseClient) processCached(ctx context.Context, cmd Cmder, state *proces
}
}
// With shouldFetch still false, this caller lost the reservation twice and
// reads the server directly, uncached. That read can return a newer value
// than a fetch another caller reserved earlier and is still running. That
// fetch may store its older value afterwards, and a later hit returns it
// until the key's invalidation arrives, so reads are not monotonic per
// caller. This stays within the staleness every hit already has until its
// invalidation arrives. Dropping the older fetch would cost each of its
// waiters a miss, under the churn that leads here, so it is documented
// instead (README, client-side caching).
var fc cscFetchCapture
var capture *cscFetchCapture
if shouldFetch {

11
csc_inval_batch.go
View File

@@ -89,12 +89,11 @@ type cscInvalItem struct {
}
// cscInvalNoHorizon marks an invalidation enqueued while no refresh binding
// existed (see cscInvalItem.sinceToken). It doubles as the LIVE horizon a
// cscRefreshQueue holds when Options.ClientSideCacheRefreshRecencyWindow is
// unset (refresh-everything mode, see startCSCRefresher): -1 is less than any
// real lastAccessNs token, so deleteByRedisKeyCollectingHot's `> sinceToken`
// check marks every Valid entry hot regardless of recency, in either case.
// The two meanings never conflict because they resolve to the same effect.
// existed (see cscInvalItem.sinceToken). It is also the horizon every
// cscRefreshQueue holds, since refresh-on-invalidate refreshes every entry
// (see startCSCRefresher): -1 is less than any real lastAccessNs token, so
// deleteByRedisKeyCollectingHot's `> sinceToken` check marks every Valid entry
// hot regardless of recency, in either case.
const cscInvalNoHorizon = -1
type cscInvalBatcher struct {

43
csc_inval_batch_test.go
View File

@@ -119,14 +119,15 @@ func TestInvalBatcherEnqueueAtStampsSuppliedFetchSnap(t *testing.T) {
}
}
// TestInvalBatcherRefreshAttachUsesLiveHorizon pins the recency gate across a
// refresh ATTACH: an item enqueued while the batcher had no refresh binding must
// not be applied as "horizon 0" once a binding appears before apply (setRefreshQueue
// repoints the batcher, then stop-drains it). Horizon 0 would mark every valid
// entry hot and refetch cold keys — the loop the horizon exists to prevent. The
// item carries cscInvalNoHorizon and apply falls back to the live horizon, so a
// COLD entry is deleted but NOT offered, while a hot one still is.
func TestInvalBatcherRefreshAttachUsesLiveHorizon(t *testing.T) {
// 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})
@@ -159,7 +160,7 @@ func TestInvalBatcherRefreshAttachUsesLiveHorizon(t *testing.T) {
// 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(lc.LRUClock())
q.sinceToken.Store(cscInvalNoHorizon)
b.refresh.Store(q)
// Hot control: read after the horizon, enqueued with a real snapshot.
@@ -183,13 +184,21 @@ func TestInvalBatcherRefreshAttachUsesLiveHorizon(t *testing.T) {
if _, ok := lc.Get(ctx, "ck:hot"); ok {
t.Fatal("hot entry not deleted")
}
// ... but only the hot one is offered for refetch. Without the sentinel the
// cold item would apply with horizon 0 and be offered too (len == 2).
if n := len(q.ch); n != 1 {
t.Fatalf("refresh targets offered = %d, want 1 (hot only); cold key must not be refetched", n)
// ... 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)
}
if got := <-q.ch; got.cacheKey != "ck:hot" {
t.Fatalf("offered target = %q, want ck:hot", got.cacheKey)
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)
}
}
@@ -544,7 +553,7 @@ func TestInvalBatcherSizeCapGuardKeepsRefreshed(t *testing.T) {
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(lc.LRUClock()) // seed the horizon so the entry reads as hot
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
@@ -626,7 +635,7 @@ func TestInvalBatcherDedupKeepsLatestFetchSnap(t *testing.T) {
// 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(lc.LRUClock())
q.sinceToken.Store(cscInvalNoHorizon)
b.refresh.Store(q)
mkValid("ck:hot", "rk") // original entry

80
csc_refresh_on_invalidate.go
View File

@@ -108,9 +108,6 @@ const (
// share this constant.
cscRefreshBatchTimeout = 5 * time.Second
// cscRefreshRecencyTick is how often the "recently read" horizon advances.
cscRefreshRecencyTick = 200 * time.Millisecond
cscRefreshWarnEvery = 30 * time.Second
)
@@ -187,14 +184,17 @@ type cscRefreshTarget struct {
// invalidated), so this narrows the jam window rather than closing it.
valBytes int
// accessNs is the evicted entry's reader-access token (lastAccessNs), captured
// under the shard lock in collectHotAndDelete. The refresh republish restores it
// after fulfill so a background refresh does NOT count as a fresh reader access:
// fulfill stamps a new token, which would keep the key above the refresh horizon
// forever and make each invalidation refresh it again even after all readers stop
// (a self-sustaining refetch loop, contrary to the cold-key guard). Restoring the
// original token means only a real reader read (get, which re-stamps) keeps a key
// eligible for refresh. See restoreAccessToken.
// under the shard lock in collectHotAndDelete. The refresh republish keeps it
// so a background refresh does NOT count as a fresh reader access: a new token
// would keep the key above the refresh horizon forever and make each
// invalidation refresh it again even after all readers stop (a self-sustaining
// refetch loop, contrary to the cold-key guard). Keeping the original token
// means only real reader activity keeps a key eligible for refresh. See
// stageRefreshAccess.
accessNs int64
// read is the evicted entry's second-chance bit, restored with accessNs so
// the refreshed entry keeps the eviction standing the old one had.
read bool
}
// cscRefreshQueue carries invalidated-but-hot keys from the drainer to the
@@ -228,16 +228,17 @@ type cscRefreshQueue struct {
// invalidation — a feedback loop that generates its own work. Same guard the
// time-based sweeper uses for the same reason.
//
// Holds cscInvalNoHorizon (-1) forever when
// Options.ClientSideCacheRefreshRecencyWindow is unset (the default): every
// Valid entry is then "hot" by construction, and the feedback-loop tradeoff
// above is an accepted, documented default (see the option's doc comment) —
// bound it with the option to restore this guard's original intent.
// Pinned at cscInvalNoHorizon (-1): every Valid entry is "hot" by
// construction, so an invalidation always schedules a refetch. The
// feedback-loop tradeoff above is therefore accepted unconditionally —
// refreshing re-registers the key with the server's tracking table, so a
// write-heavy, rarely-read key keeps refreshing itself off write traffic
// for as long as its entry survives capacity eviction.
//
// The field is kept rather than dropped because the sentinel has a SECOND
// meaning that still applies: an invalidation enqueued while no refresh
// binding existed carries it too (see cscInvalItem.sinceToken).
sinceToken atomic.Int64
// recency holds the multi-tick history backing sinceToken when
// ClientSideCacheRefreshRecencyWindow is set; nil in the default
// refresh-everything mode, where sinceToken is fixed and this is unused.
recency *cscRecencyRing
enqueued atomic.Uint64
dropped atomic.Uint64
@@ -428,16 +429,10 @@ func (c *baseClient) startCSCRefresher() {
ch: make(chan cscRefreshTarget, cscRefreshQueueDepth),
demandCh: make(chan uint64, 1),
}
// Default (window <= 0): refresh every invalidated Valid entry regardless of
// recency. cscInvalNoHorizon (-1) is less than any real lastAccessNs token,
// so it marks everything hot; sinceToken then never changes (see the
// recency-tick case in runCSCRefresher, which no-ops when q.recency is nil).
// Refresh every invalidated Valid entry, regardless of how recently it was
// read. cscInvalNoHorizon (-1) is less than any real lastAccessNs token, so
// it marks everything hot, and it never changes.
q.sinceToken.Store(cscInvalNoHorizon)
if w := c.opt.ClientSideCacheRefreshRecencyWindow; w > 0 {
q.recency = newCscRecencyRing(cscRefreshWindowTicks(w))
q.recency.push(lc.LRUClock())
q.sinceToken.Store(q.recency.oldest())
}
h := &cscRevalidateHandle{stop: make(chan struct{}), done: make(chan struct{})}
@@ -504,9 +499,6 @@ func (c *baseClient) startCSCRefresher() {
func (c *baseClient) runCSCRefresher(h *cscRevalidateHandle, lc *LocalCache, q *cscRefreshQueue) {
defer close(h.done)
recency := time.NewTicker(cscRefreshRecencyTick)
defer recency.Stop()
// The collection window. It starts (timer armed) when the first key of a
// batch is collected and is NOT re-armed on later arrivals — a sliding reset
// would never fire under continuous churn, leaving only the size cap. Flushed
@@ -704,16 +696,6 @@ func (c *baseClient) runCSCRefresher(h *cscRevalidateHandle, lc *LocalCache, q *
}
}
case <-recency.C:
// Advance the horizon: entries not read within the configured window
// stop being worth refetching. Default (q.recency nil, window
// unset): sinceToken is fixed at cscInvalNoHorizon — refresh
// everything, forever — so there is nothing to advance.
if q.recency != nil {
q.recency.push(lc.LRUClock())
q.sinceToken.Store(q.recency.oldest())
}
case t := <-q.ch:
collect(t)
// Opportunistically take whatever else is already waiting so one wake
@@ -931,16 +913,16 @@ func (c *baseClient) refreshInvalidatedBatch(ctx context.Context, targets []cscR
key: kept[i].cacheKey,
token: kept[i].token,
}
// Keep the invalidated entry's recency: a background refresh is not a
// reader read, so it must not renew the key's refresh eligibility (see
// stageRefreshAccess). It is staged on the reservation so the publish
// applies it atomically. Refresh runs only with the built-in
// *LocalCache; the hook path fulfills that same cache.
if lc, ok := c.csc.(*LocalCache); ok {
lc.stageRefreshAccess(kept[i].cacheKey, kept[i].token, kept[i].accessNs, kept[i].read)
}
if c.fulfillCached(kept[i].cacheKey, kept[i].token, fc) {
published++
// Undo the fresh access token fulfill stamped: a background refresh is
// not a reader read, so it must not renew the key's refresh eligibility,
// or each invalidation would keep refreshing it after all readers stop
// (see restoreAccessToken). Refresh runs only with the built-in
// *LocalCache; the hook path fulfills that same cache.
if lc, ok := c.csc.(*LocalCache); ok {
lc.restoreAccessToken(kept[i].cacheKey, kept[i].accessNs)
}
}
// fulfillCached cancels on its own failure paths, so the token is
// settled either way.

71
csc_refresh_on_invalidate_test.go
View File

@@ -49,7 +49,7 @@ func TestCSCRefresherStopDrainSkipsRefetchEntirely(t *testing.T) {
ch: make(chan cscRefreshTarget, 1024),
demandCh: make(chan uint64, 1),
}
q.sinceToken.Store(lc.LRUClock())
q.sinceToken.Store(cscInvalNoHorizon)
h := &cscRevalidateHandle{stop: make(chan struct{}), done: make(chan struct{})}
// More than one chunk's worth (cscRefreshBatchMax) but within one drainQueue
@@ -115,7 +115,7 @@ func TestCSCRefresherNormalFlushAbortsOnClose(t *testing.T) {
ch: make(chan cscRefreshTarget, cscRefreshWindowMaxKeys+16),
demandCh: make(chan uint64, 1),
}
q.sinceToken.Store(lc.LRUClock())
q.sinceToken.Store(cscInvalNoHorizon)
// A full window (4 * cscRefreshBatchMax) triggers a NORMAL flush(false,false) from the
// q.ch case — this is NOT the stop-drain path (h.stop is not pre-closed).
@@ -139,73 +139,6 @@ func TestCSCRefresherNormalFlushAbortsOnClose(t *testing.T) {
}
}
// TestStartCSCRefresherRecencyWindowWiring pins the construction-time wiring
// between Options.ClientSideCacheRefreshRecencyWindow and the live
// cscRefreshQueue: default (window unset) must leave q.recency nil and pin
// sinceToken at the unbounded sentinel (refresh everything); a configured
// window must build a ring sized by cscRefreshWindowTicks and seed
// sinceToken from it (a real token, not the sentinel). The ring's own
// push/oldest arithmetic and the window->ticks rounding are separately
// pinned (TestCscRecencyRingOldest, TestCscRefreshWindowTicks); this only
// covers startCSCRefresher actually wiring them together.
func TestStartCSCRefresherRecencyWindowWiring(t *testing.T) {
t.Run("default_is_unbounded", func(t *testing.T) {
lc := NewLocalCache(CacheConfig{MaxEntries: 1000})
c := &baseClient{
opt: &Options{ClientSideCacheRefreshOnInvalidate: true},
csc: lc,
cscKeyPrefix: "p:",
}
c.startCSCRefresher()
defer c.stopCSCRefresher()
q := c.cscRefreshQueue
if q == nil {
t.Fatal("startCSCRefresher did not build a refresh queue")
}
if q.recency != nil {
t.Fatal("default (window unset) built a recency ring; want nil (unbounded mode)")
}
if got := q.sinceToken.Load(); got != cscInvalNoHorizon {
t.Fatalf("default sinceToken = %d, want cscInvalNoHorizon (%d)", got, cscInvalNoHorizon)
}
})
t.Run("configured_window_builds_ring", func(t *testing.T) {
lc := NewLocalCache(CacheConfig{MaxEntries: 1000})
c := &baseClient{
opt: &Options{
ClientSideCacheRefreshOnInvalidate: true,
ClientSideCacheRefreshRecencyWindow: 2 * cscRefreshRecencyTick,
},
csc: lc,
cscKeyPrefix: "p:",
}
c.startCSCRefresher()
defer c.stopCSCRefresher()
q := c.cscRefreshQueue
if q == nil {
t.Fatal("startCSCRefresher did not build a refresh queue")
}
if q.recency == nil {
t.Fatal("configured window left q.recency nil; want a ring")
}
if got, want := len(q.recency.buf), cscRefreshWindowTicks(2*cscRefreshRecencyTick); got != want {
t.Fatalf("ring size = %d, want %d (cscRefreshWindowTicks(2*tick))", got, want)
}
if got := q.sinceToken.Load(); got == cscInvalNoHorizon {
t.Fatal("configured window initialized sinceToken to the unbounded sentinel")
}
})
}
// TestCloneSharesRefreshQueueForDemand pins #3965 F4: clone() (WithTimeout/
// WithContext) must SHARE the owner's refresh queue so a derived client's
// processCached can signal demand on it. Without the share the clone's field is
// nil and signalDemand no-ops, so a clone's miss waits the full window instead of
// nudging the owner's in-window batch. Lifecycle stays owner-only: the clone only
// signals (nil-safe, non-blocking) and never stops the shared refresher.
func TestCloneSharesRefreshQueueForDemand(t *testing.T) {
parent := &baseClient{
opt: &Options{},

52
csc_refresh_restore_test.go Normal file
View File

@@ -0,0 +1,52 @@
package redis
import (
"context"
"testing"
)
// TestRefreshRepublishSurvivesConcurrentInsert pins that a refresh republish
// keeps the invalidated entry's recency in the same lock as the publish. It
// used to publish with the second-chance bit clear and restore the bit in a
// second lock. On a warm shard the republished entry was then the only
// clear-bit victim, so an insert in that gap evicted the key the refresh had
// just fetched.
func TestRefreshRepublishSurvivesConcurrentInsert(t *testing.T) {
ctx := context.Background()
lc := NewLocalCache(CacheConfig{MaxEntries: 2})
fill := func(key, rk, val string) {
t.Helper()
tok, _ := lc.Reserve(key, []string{rk})
if !lc.fulfill(key, tok, 0, []byte(val)) {
t.Fatalf("fill %s failed", key)
}
}
// b is older than a; both are read, so the shard is warm.
fill("get:b", "b", "vb")
fill("get:a", "a", "va")
for _, k := range []string{"get:a", "get:b"} {
if _, ok := lc.Get(ctx, k); !ok {
t.Fatalf("%s missing", k)
}
}
// Refresh a: collect its target, reserve, stage the kept recency, publish.
targets := lc.deleteByRedisKeyCollectingHot("a", cscInvalNoHorizon, ^uint64(0), nil)
if len(targets) != 1 || !targets[0].read {
t.Fatalf("want 1 hot target carrying the read bit, got %+v", targets)
}
tok, _ := lc.Reserve("get:a", []string{"a"})
lc.stageRefreshAccess("get:a", tok, targets[0].accessNs, targets[0].read)
if !lc.fulfill("get:a", tok, 0, []byte("va2")) {
t.Fatal("republish fulfill failed")
}
// An insert right after the publish. Both residents were read, so the
// sweep consumes their second chances and evicts the older one, b.
if tok, _ := lc.Reserve("get:c", []string{"c"}); tok == 0 {
t.Fatal("reserve c failed")
}
if _, ok := lc.Get(ctx, "get:a"); !ok {
t.Fatal("the refreshed key was evicted by the next insert")
}
}

81
csc_refresh_support.go
View File

@@ -2,7 +2,6 @@ package redis
import (
"sync"
"time"
)
// Support shims for refresh-on-invalidate and reader-miss coalescing, kept in
@@ -21,83 +20,6 @@ func (h *cscRevalidateHandle) signalStop() {
h.stopOnce.Do(func() { close(h.stop) })
}
// LRUClock returns the current global recency token; the refresher uses it as
// the "recently read" horizon.
func (c *LocalCache) LRUClock() int64 { return lruSequence.Load() }
// cscRecencyRing holds the last N per-tick LRUClock() snapshots, giving
// startCSCRefresher's sinceToken horizon a bounded multi-tick lookback
// instead of the single most-recent tick, so
// Options.ClientSideCacheRefreshRecencyWindow can span more than one
// cscRefreshRecencyTick. oldest() is the horizon: once the ring has filled,
// an entry read since oldest() was captured is guaranteed to fall within the
// configured window, regardless of where an invalidation lands relative to
// the tick phase (see the ring-sizing comment in startCSCRefresher). Owned
// solely by the refresher goroutine — push/oldest are not called
// concurrently, so no lock.
type cscRecencyRing struct {
buf []int64
pos int
n int
}
// cscRecencyRingMaxSize caps the ring allocation newCscRecencyRing builds. At
// the 200ms tick this is ~58 hours of coverage — a pathological
// ClientSideCacheRefreshRecencyWindow (a duration typo, or math.MaxInt64)
// degrades to that cap instead of an oversized or panicking make([]int64, N).
const cscRecencyRingMaxSize = 1 << 20
// newCscRecencyRing builds a ring of the given size, clamped to
// [1, cscRecencyRingMaxSize].
func newCscRecencyRing(size int) *cscRecencyRing {
if size < 1 {
size = 1
}
if size > cscRecencyRingMaxSize {
size = cscRecencyRingMaxSize
}
return &cscRecencyRing{buf: make([]int64, size)}
}
// cscRefreshWindowTicks converts a requested
// Options.ClientSideCacheRefreshRecencyWindow into a cscRecencyRing size.
// Ring size N gives a GUARANTEED-minimum lookback of (N-1)*cscRefreshRecencyTick
// once the ring has filled (oldest() is then N-1 ticks behind the latest
// push) — the tick phase relative to an arbitrary invalidation can cost up to
// one full tick of coverage, so N-1 must itself already be >= ceil(w/tick)
// for the enforced window to never fall short of w. Hence N = ceil(w/tick)+1,
// giving an enforced window in [w, w+tick). Extracted pure so the rounding is
// unit-tested; w<=0 (unbounded mode) never reaches this function.
func cscRefreshWindowTicks(w time.Duration) int {
ticks := int(w / cscRefreshRecencyTick)
if w%cscRefreshRecencyTick != 0 {
ticks++
}
return ticks + 1
}
// push records the latest per-tick snapshot, evicting the oldest once full.
func (r *cscRecencyRing) push(v int64) {
r.buf[r.pos] = v
r.pos = (r.pos + 1) % len(r.buf)
if r.n < len(r.buf) {
r.n++
}
}
// oldest returns the least-recent snapshot currently held. Before the ring
// fills (just after the refresher starts), it returns the earliest snapshot
// taken so far, so every access since the refresher started counts as hot
// until the configured window has had time to mature. Once full, r.pos
// always points at the slot about to be overwritten next — the oldest value
// held — a standard circular-buffer property.
func (r *cscRecencyRing) oldest() int64 {
if r.n < len(r.buf) {
return r.buf[0]
}
return r.buf[r.pos]
}
// InvalidationStats reports INCOMING invalidation pushes: the count of keys named
// in server invalidation messages, tallied at the handler before dedup/batching.
func (c *LocalCache) InvalidationStats() (invalidations uint64) {
@@ -172,9 +94,10 @@ func (s *cacheShard) collectHotAndDelete(redisKey string, sinceToken int64, fetc
cacheKey: cacheKey,
redisKeys: keys,
// Preserve the reader-access token so the refresh republish can restore it
// and not renew demand (see cscRefreshTarget.accessNs / restoreAccessToken).
// and not renew demand (see cscRefreshTarget.accessNs / stageRefreshAccess).
// Re-load is safe: the shard lock is held, so no writer intervenes.
accessNs: entry.lastAccessNs.Load(),
read: entry.readSinceSweep.Load(),
// The dying entry's payload size approximates the refetch REPLY size —
// the refresher chunks round trips by expected reply bytes with it (see
// cscRefreshChunkEnd). Known for free here, under the same shard lock.

152
csc_refresh_support_test.go
View File

@@ -3,69 +3,8 @@ package redis
import (
"context"
"testing"
"time"
)
// TestCollectHotAndDeleteRecencyFilter pins the recency gate that decides which
// invalidated entries the refresher re-fetches. Every invalidated entry is
// deleted, but only those that were VALID and read since the horizon are
// returned as refetch targets. Getting the comparison or the horizon wrong
// would silently refetch every invalidated key — extra round trips, no test
// failure — so the cold case is asserted explicitly alongside the hot one.
func TestCollectHotAndDeleteRecencyFilter(t *testing.T) {
ctx := context.Background()
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; want a fresh reservation", cacheKey, tok, fetch)
}
if !lc.fulfill(cacheKey, tok, 0, []byte("v")) {
t.Fatalf("fulfill(%q) failed", cacheKey)
}
}
// Cold entry is created first; the horizon is snapshotted after it, so its
// recency token is at or below the horizon.
mkValid("ck:cold", "rk:cold")
horizon := lruSequence.Load()
// Hot entry is created after the horizon snapshot, so its recency token is
// strictly above the horizon regardless of any concurrent LRU activity.
mkValid("ck:hot", "rk:hot")
// Cold: NOT collected (read no more recently than the horizon), but still deleted.
// snap = max: the fetch-order guard never skips, so this isolates the hot/cold
// recency behavior (guard interplay is covered separately).
if got := lc.deleteByRedisKeyCollectingHot("rk:cold", horizon, ^uint64(0), nil); len(got) != 0 {
t.Fatalf("cold entry collected as a refetch target: %v", got)
}
if _, ok := lc.Get(ctx, "ck:cold"); ok {
t.Fatal("cold entry was not deleted by the invalidation")
}
// Hot: collected (read after the horizon) and deleted.
got := lc.deleteByRedisKeyCollectingHot("rk:hot", horizon, ^uint64(0), nil)
if len(got) != 1 || got[0].cacheKey != "ck:hot" {
t.Fatalf("hot entry not collected as the sole refetch target: %v", got)
}
if _, ok := lc.Get(ctx, "ck:hot"); ok {
t.Fatal("hot entry was not deleted by the invalidation")
}
// Unknown redis key: nothing collected, no panic on the missing index.
if got := lc.deleteByRedisKeyCollectingHot("rk:nonexistent", 0, ^uint64(0), nil); len(got) != 0 {
t.Fatalf("unknown redis key produced refetch targets: %v", got)
}
}
// TestCollectHotAndDeleteNoHorizonRefreshesEverything pins the default
// refresh-everything mode (Options.ClientSideCacheRefreshRecencyWindow
// unset): cscInvalNoHorizon passed as sinceToken must collect a Valid entry
// as a refetch target REGARDLESS of how long ago it was last read — the
// mirror image of TestCollectHotAndDeleteRecencyFilter's cold case, which is
// what a real (non-sentinel) horizon excludes.
func TestCollectHotAndDeleteNoHorizonRefreshesEverything(t *testing.T) {
ctx := context.Background()
lc := NewLocalCache(CacheConfig{MaxEntries: 64})
@@ -94,97 +33,6 @@ func TestCollectHotAndDeleteNoHorizonRefreshesEverything(t *testing.T) {
}
}
// TestCscRecencyRingOldest pins the ring's circular-buffer arithmetic:
// oldest() must return the earliest-still-held snapshot both before the ring
// fills (ramp-up) and after it wraps repeatedly.
func TestCscRecencyRingOldest(t *testing.T) {
r := newCscRecencyRing(3)
// Ramp-up: not yet full, oldest is always the first push.
r.push(10)
if got := r.oldest(); got != 10 {
t.Fatalf("after 1 push: oldest = %d, want 10", got)
}
r.push(20)
if got := r.oldest(); got != 10 {
t.Fatalf("after 2 pushes: oldest = %d, want 10 (ring not full yet)", got)
}
// Full: oldest is the least-recent of the last 3.
r.push(30)
if got := r.oldest(); got != 10 {
t.Fatalf("after 3 pushes (just filled): oldest = %d, want 10", got)
}
// Wraps: each push evicts the true oldest.
r.push(40) // holds {20,30,40}
if got := r.oldest(); got != 20 {
t.Fatalf("after 4 pushes: oldest = %d, want 20", got)
}
r.push(50) // holds {30,40,50}
if got := r.oldest(); got != 30 {
t.Fatalf("after 5 pushes: oldest = %d, want 30", got)
}
// Size 1 degenerates to "just the latest push" — today's pre-option behavior.
one := newCscRecencyRing(1)
one.push(7)
if got := one.oldest(); got != 7 {
t.Fatalf("size-1 ring: oldest = %d, want 7", got)
}
one.push(9)
if got := one.oldest(); got != 9 {
t.Fatalf("size-1 ring after second push: oldest = %d, want 9", got)
}
// size < 1 clamps to 1 rather than panicking on a zero-length buffer.
zero := newCscRecencyRing(0)
zero.push(5)
if got := zero.oldest(); got != 5 {
t.Fatalf("size-0 ring (clamped to 1): oldest = %d, want 5", got)
}
}
// TestCscRefreshWindowTicks pins the window->ring-size rounding: the enforced
// window (ticks-1)*tick must never fall short of the requested duration,
// regardless of how it divides the tick, and must match exactly on an exact
// multiple (no off-by-one slack beyond the documented [w, w+tick) range).
func TestCscRefreshWindowTicks(t *testing.T) {
tick := cscRefreshRecencyTick
cases := []struct {
name string
w time.Duration
want int
}{
{"exact one tick", tick, 2},
{"exact two ticks", 2 * tick, 3},
{"sub-tick rounds up to one tick's guarantee", tick / 4, 2},
{"just over one tick", tick + time.Millisecond, 3},
{"just under two ticks", 2*tick - time.Millisecond, 3},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := cscRefreshWindowTicks(tc.w); got != tc.want {
t.Fatalf("cscRefreshWindowTicks(%v) = %d, want %d", tc.w, got, tc.want)
}
// The guarantee itself: enforced minimum lookback (ticks-1)*tick must
// cover the requested window.
enforced := time.Duration(tc.want-1) * tick
if enforced < tc.w {
t.Fatalf("cscRefreshWindowTicks(%v): enforced minimum %v < requested %v", tc.w, enforced, tc.w)
}
})
}
}
// TestCollectHotFetchOrderGuard is the discriminating test for the fetch-order
// guard (#3965). A stale invalidation snapshots the fetch clock at OBSERVE time; an
// entry refetched AFTER that (fetchSeq > snapshot) already reflects the write and
// must survive — evicting it would be a spurious miss and would undo a refresh. A
// GENUINE later invalidation, observed after the refetch, must still evict. This is
// the case that a fulfill-time stamp would get wrong: the invalidation and the
// reply race on different connections, so a stale reply can fulfill after the
// invalidate is observed; only stamping at fetch ISSUE (Reserve) is sound.
func TestCollectHotFetchOrderGuard(t *testing.T) {
ctx := context.Background()
lc := NewLocalCache(CacheConfig{MaxEntries: 64})

48
csc_test.go
View File

@@ -4589,9 +4589,45 @@ func TestInvalidateHandlerFullFlushNonComparableCacheNoPanic(t *testing.T) {
}
}
// A refresh republish keeps the invalidated entry's second-chance bit. It used
// to republish with the bit clear, so a hot key that had just been refreshed was
// the first eviction victim and could vanish before its next reader.
func TestRefreshRepublishKeepsSecondChance(t *testing.T) {
ctx := context.Background()
// A full shard, so the read is recorded even where reads are only marked
// under eviction pressure.
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 { // a reader: sets the bit
t.Fatal("seeded key missing")
}
targets := lc.deleteByRedisKeyCollectingHot(rk, cscInvalNoHorizon, ^uint64(0), nil)
if len(targets) != 1 || !targets[0].read {
t.Fatalf("want 1 hot target carrying the read bit, got %+v", targets)
}
tok2, _ := lc.Reserve(key, []string{rk})
lc.stageRefreshAccess(key, tok2, targets[0].accessNs, targets[0].read)
if !lc.fulfill(key, tok2, 0, []byte("v2")) {
t.Fatal("republish fulfill failed")
}
s := lc.shardFor(key)
s.mu.RLock()
e := s.entries[key]
s.mu.RUnlock()
if e == nil || !e.readSinceSweep.Load() {
t.Fatal("the refreshed entry lost the second chance its read had earned")
}
}
// A refresh republish must NOT renew a key's reader-access recency, or every
// invalidation would keep refreshing a key nobody reads anymore (a self-sustaining
// refetch loop). After the republish + restoreAccessToken, the entry must be COLD
// refetch loop). After a staged republish, the entry must be COLD
// relative to the horizon of its last real read.
func TestRefreshRepublishDoesNotRenewDemand(t *testing.T) {
lc := NewLocalCache(CacheConfig{MaxEntries: 16})
@@ -4604,20 +4640,20 @@ func TestRefreshRepublishDoesNotRenewDemand(t *testing.T) {
}
// One refresh cycle: collect the hot target (captures the reader-access token and
// deletes the entry), republish a fresh value, restore the captured token.
targets := lc.deleteByRedisKeyCollectingHot(rk, lc.LRUClock()-1, ^uint64(0), nil)
// deletes the entry), republish a fresh value keeping the captured token.
targets := lc.deleteByRedisKeyCollectingHot(rk, cscInvalNoHorizon, ^uint64(0), nil)
if len(targets) != 1 {
t.Fatalf("want 1 hot target, got %d", len(targets))
}
tok2, _ := lc.Reserve(key, []string{rk})
lc.stageRefreshAccess(key, tok2, targets[0].accessNs, targets[0].read)
if !lc.fulfill(key, tok2, 0, []byte("v2")) {
t.Fatal("republish fulfill failed")
}
lc.restoreAccessToken(key, targets[0].accessNs)
// No reader touched the key since. At the horizon of its last real read the entry
// must be COLD (lastAccessNs == that token, not > it). Without the restore the
// republish would leave a newer token here and the entry would still be collected
// must be COLD (lastAccessNs == that token, not > it). Without the staged token
// the republish would leave a newer token here and the entry would still be collected
// — the self-sustaining loop.
if hot := lc.deleteByRedisKeyCollectingHot(rk, targets[0].accessNs, ^uint64(0), nil); len(hot) != 0 {
t.Fatalf("refreshed-but-unread entry still hot after restore; got %d targets", len(hot))

205
local_cache.go
View File

@@ -31,11 +31,41 @@ type cacheEntry struct {
waitCh chan struct{}
waitClosed bool
// lastAccessNs is a recency token for LRU eviction: a global atomic counter
// bumped on every access, stored atomically so the read path can mark a
// touch under the shard's RLock without upgrading to a write lock.
// lastAccessNs orders entries for eviction: a global atomic counter bumped
// when the entry is created, reserved or fulfilled — all of which happen
// under the shard write lock, on the MISS path. It is stored atomically
// because the read path reads it under the shard's RLock.
//
// It is NOT bumped on a read any more; a read sets readSinceSweep instead.
lastAccessNs atomic.Int64
// readSinceSweep records that a READ has touched this entry since the last
// eviction sweep. It is the second-chance bit for eviction: a victim is
// chosen from the entries with the bit CLEAR, and the sweep clears the bits
// when every candidate has been read.
//
// It exists because writing lastAccessNs on every hit was the dominant cost
// of a cache hit under concurrency. ~40k resident entries at a few hundred
// bytes each is a working set far larger than L2; the read already touches
// the entry's cache line, but a STORE dirties it, and a dirty line costs a
// writeback when it is evicted from the CPU cache. Measured on a 14-core
// host at 256 concurrent readers, get=90/set=10, 99.94% hit rate:
// 263,801 -> 328,842 reads/s (+24.7%) and 27.0 -> 18.9 CPU us/op, with the
// hit rate unchanged. Removing the recency update altogether measured
// +40.5%, so this recovers about three fifths of the available headroom
// while keeping eviction honest.
//
// A load that finds the bit already set leaves the line SHARED, so every
// core can hold it at once; only the first read after a sweep writes.
readSinceSweep atomic.Bool
// refreshKeep marks an IN_PROGRESS refresh reservation: fulfill publishes
// it with refreshAccessNs and refreshRead instead of a fresh recency. See
// stageRefreshAccess. Written and read under the shard write lock.
refreshKeep bool
refreshRead bool
refreshAccessNs int64
// validAt retains time.Now's monotonic component for the MaxStaleness
// backstop, so wall-clock corrections cannot extend an entry's lifetime.
// Written under Lock (Set/Fulfill), read under RLock (get).
@@ -328,10 +358,34 @@ func defaultCacheSizer(cacheKey string, redisKeys []string, value []byte) int64
// Get returns a copy of a cached value, waiting for an in-progress fetch when
// necessary.
func (c *LocalCache) Get(ctx context.Context, cacheKey string) ([]byte, bool) {
value, ok := c.get(ctx, cacheKey, true)
return value, ok
}
// getShared is Get without the defensive copy. The returned slice ALIASES the
// cache entry, so the caller must neither mutate nor retain it; it is valid
// only until the caller returns.
//
// Safe because a published value is immutable: cacheShard.get never writes
// through the slice, and a refetch REPLACES entry.value wholesale under the
// shard write lock rather than editing it in place, so an old slice a reader
// already holds keeps its contents.
//
// Not on the Cache interface, and deliberately so: Get's []byte return means a
// third-party implementation's caller may legitimately retain what it gets
// back, so the copy has to stay there. Only the built-in cache paired with the
// built-in read path (processCached, which parses the bytes and drops them)
// can skip it -- the same "only the built-in *LocalCache" gate miss coalescing
// uses.
func (c *LocalCache) getShared(ctx context.Context, cacheKey string) ([]byte, bool) {
return c.get(ctx, cacheKey, false)
}
func (c *LocalCache) get(ctx context.Context, cacheKey string, clone bool) ([]byte, bool) {
if ctx == nil {
ctx = context.Background()
}
value, ok := c.shardFor(cacheKey).get(ctx, cacheKey)
value, ok := c.shardFor(cacheKey).get(ctx, cacheKey, clone)
if ok {
c.hits.Add(1)
} else {
@@ -343,7 +397,9 @@ func (c *LocalCache) Get(ctx context.Context, cacheKey string) ([]byte, bool) {
// get is the read-side hot path. Holds only the shard's read lock; updates
// the LRU recency timestamp via atomic store on the entry — no write-lock
// upgrade is needed.
func (s *cacheShard) get(ctx context.Context, cacheKey string) ([]byte, bool) {
// get is the read-side hot path. clone=false returns a slice ALIASING the
// entry (see LocalCache.getShared for why that is safe and who may use it).
func (s *cacheShard) get(ctx context.Context, cacheKey string, clone bool) ([]byte, bool) {
for {
s.mu.RLock()
entry, ok := s.entries[cacheKey]
@@ -400,11 +456,18 @@ func (s *cacheShard) get(ctx context.Context, cacheKey string) ([]byte, bool) {
return nil, false
}
value := cloneBytes(entry.value)
// Record access timestamp without upgrading the lock. Last writer
// wins; cross-goroutine ordering of timestamps is fine for
// approximate-LRU semantics.
entry.lastAccessNs.Store(nextLRUToken())
value := entry.value
if clone {
value = cloneBytes(value)
}
// Mark the second-chance bit instead of stamping a recency token. The
// load short-circuits every read after the first since the last sweep,
// which is the overwhelming majority, and leaves the entry's cache line
// SHARED rather than taking it exclusively per hit. See
// cacheEntry.readSinceSweep for the measurements.
if !entry.readSinceSweep.Load() {
entry.readSinceSweep.Store(true)
}
s.mu.RUnlock()
return value, true
}
@@ -439,8 +502,11 @@ func (c *LocalCache) Reserve(cacheKey string, redisKeys []string) (token uint64,
switch entry.state {
case cacheEntryValid:
// Existing-VALID hit: record access; caller will re-Get to
// retrieve.
// retrieve. This IS an access, so it also earns the second chance
// the read path grants -- it is under the write lock and off the
// hot path, so stamping the token here costs nothing.
entry.lastAccessNs.Store(nextLRUToken())
entry.readSinceSweep.Store(true)
return 0, false
case cacheEntryInProgress:
if time.Since(entry.reservedAt) < s.staleTimeout {
@@ -520,33 +586,54 @@ func (c *LocalCache) fulfill(cacheKey string, token, ownerConnID uint64, value [
entry.state = cacheEntryValid
entry.validAt = time.Now()
entry.token = 0
entry.lastAccessNs.Store(nextLRUToken())
if entry.refreshKeep {
// A refresh republish keeps the old entry's standing, set here in the
// same lock as the publish so the eviction pass below, and any insert
// after it, already see it.
entry.lastAccessNs.Store(entry.refreshAccessNs)
entry.readSinceSweep.Store(entry.refreshRead)
entry.refreshKeep = false
} else {
entry.lastAccessNs.Store(nextLRUToken())
}
if ownerConnID != 0 {
entry.ownerConnID = ownerConnID
s.indexConnLocked(ownerConnID, cacheKey)
}
s.closeWaitersLocked(entry)
s.evictIfNeededLocked()
// The entry just fulfilled is not a candidate in its own eviction pass. It
// is born with its second-chance bit clear, so in a warm shard (every
// resident entry read) it would be the only clear-bit candidate and be
// evicted right here: a memory-capped shard, where the value only goes
// over the cap now, could then admit nothing, and since a clear candidate
// always existed the sweep that resets the others never ran. Excluding it
// lets that sweep run and take the oldest entry instead.
s.evictIfNeededLocked(entry)
current, stillExists := s.entries[cacheKey]
return stillExists && current == entry && entry.state == cacheEntryValid
}
// restoreAccessToken resets a Valid entry's reader-access recency (lastAccessNs)
// to accessNs, undoing the fresh token fulfill stamps. The refresh republish calls
// this so a background refresh does NOT count as a reader access: otherwise the
// refreshed key stays above the refresh horizon and every later invalidation
// refreshes it again even after all readers stop — a self-sustaining refetch loop
// contrary to the cold-key guard. A reader that get()s between the fulfill and this
// call has its newer token overwritten, which only UNDER-refreshes that key (the
// next reader refetches), never serves stale. No-op if the entry is gone or not
// Valid.
func (c *LocalCache) restoreAccessToken(cacheKey string, accessNs int64) {
// stageRefreshAccess sets the recency a refresh republish keeps on the
// IN_PROGRESS reservation token: lastAccessNs becomes accessNs and the
// second-chance bit becomes read, both taken from the invalidated entry.
// fulfill applies them in the same lock as the publish.
//
// The refresh does not count as a reader access: a fresh token would keep the
// key above the refresh horizon, so every later invalidation would refresh it
// again after all readers stop -- a self-sustaining refetch loop. Keeping the
// bit stops a hot key from being the first eviction victim only because the
// refresh republished it. Setting both before the publish, not after it,
// closes the gap in which an insert could evict the republished entry while
// its bit was still clear. No-op when the reservation is gone.
func (c *LocalCache) stageRefreshAccess(cacheKey string, token uint64, accessNs int64, read bool) {
s := c.shardFor(cacheKey)
s.mu.Lock()
defer s.mu.Unlock()
if entry, ok := s.entries[cacheKey]; ok && entry.state == cacheEntryValid {
entry.lastAccessNs.Store(accessNs)
if entry, ok := s.entries[cacheKey]; ok && entry.state == cacheEntryInProgress && entry.token == token {
entry.refreshKeep = true
entry.refreshAccessNs = accessNs
entry.refreshRead = read
}
}
@@ -785,11 +872,14 @@ func (s *cacheShard) overCapacityLocked() bool {
// well-sized caches) until under capacity. Used by Set/Fulfill: it prefers a
// Valid victim but falls back to the oldest IN_PROGRESS placeholder to keep the
// hard cap (that placeholder's Fulfill then fails and its waiters refetch).
func (s *cacheShard) evictIfNeededLocked() {
//
// keep, when non-nil, is never chosen: it is the entry the caller just
// published and must not evict in the same pass.
func (s *cacheShard) evictIfNeededLocked(keep *cacheEntry) {
for s.overCapacityLocked() {
victim := s.oldestLocked(cacheEntryValid)
victim := s.oldestLocked(cacheEntryValid, keep)
if victim == nil {
victim = s.oldestLocked(cacheEntryInProgress)
victim = s.oldestLocked(cacheEntryInProgress, keep)
}
if victim == nil {
return
@@ -803,7 +893,7 @@ func (s *cacheShard) evictIfNeededLocked() {
// peer's in-flight fetch.
func (s *cacheShard) evictValidLocked() {
for s.overCapacityLocked() {
victim := s.oldestLocked(cacheEntryValid)
victim := s.oldestLocked(cacheEntryValid, nil)
if victim == nil {
return
}
@@ -811,21 +901,58 @@ func (s *cacheShard) evictValidLocked() {
}
}
// oldestLocked returns the entry in the given state with the smallest
// lastAccessNs (the least-recently-used), or nil when none exists.
func (s *cacheShard) oldestLocked(state cacheEntryState) *cacheEntry {
var victim *cacheEntry
var oldestNs int64 = math.MaxInt64
// oldestLocked returns the eviction victim in the given state, or nil when no
// entry is in that state. keep, when non-nil, is skipped entirely: it is
// neither a victim nor part of the sweep.
//
// Second chance. A read no longer stamps a recency token (that store was the
// dominant cost of a cache hit; see cacheEntry.readSinceSweep), so recency is
// carried by the readSinceSweep bit and ordering by lastAccessNs, which is the
// token assigned when the entry was created, reserved or fulfilled.
//
// The victim is the oldest entry whose bit is CLEAR — never read since the last
// sweep. When every candidate has been read, that is the sweep: clear all their
// bits, give them a fresh chance, and fall back to the oldest by token. So a
// read protects an entry from exactly one eviction pass, which is what
// approximate LRU asks for.
//
// Ordering by lastAccessNs rather than by map order is what keeps the choice
// DETERMINISTIC. Go randomises map iteration, so picking "any entry with a
// clear bit" would evict a different key run to run, which callers (and the
// LRU tests) reasonably do not expect.
func (s *cacheShard) oldestLocked(state cacheEntryState, keep *cacheEntry) *cacheEntry {
var victim, fallback *cacheEntry
var oldestNs, oldestAny int64 = math.MaxInt64, math.MaxInt64
swept := false
for _, e := range s.entries {
if e.state != state {
if e.state != state || e == keep {
continue
}
if ns := e.lastAccessNs.Load(); ns < oldestNs {
oldestNs = ns
victim = e
ns := e.lastAccessNs.Load()
if ns < oldestAny {
oldestAny, fallback = ns, e
}
if e.readSinceSweep.Load() {
swept = true
continue
}
if ns < oldestNs {
oldestNs, victim = ns, e
}
}
return victim
if victim != nil {
return victim
}
if swept {
// Every candidate had been read: consume their second chances so the
// next pass can distinguish them again.
for _, e := range s.entries {
if e.state == state && e != keep {
e.readSinceSweep.Store(false)
}
}
}
return fallback
}
func cloneBytes(src []byte) []byte {

89
local_cache_test.go
View File

@@ -771,3 +771,92 @@ func TestLocalCache_ReserveHonorsCapWithoutEvictingPeers(t *testing.T) {
}
}
}
// TestLocalCache_SecondChanceIsConsumed pins the half of the second-chance
// contract the existing eviction tests do not reach: a read buys an entry
// exactly ONE eviction pass, not permanent immunity.
//
// Reads no longer stamp a recency token (that store was the dominant cost of a
// cache hit under concurrency; see cacheEntry.readSinceSweep). They set a bit,
// and eviction picks the oldest entry whose bit is CLEAR. If nothing ever
// cleared those bits, a cache whose whole working set had been read once would
// have no clear-bit candidate and would fall back to insertion order forever,
// silently losing LRU behaviour. oldestLocked clears them when every candidate
// has been read, which is what this test drives.
// A memory-capped shard whose resident entries have all been read must still
// admit a new key. The placeholder costs nothing, so Reserve does not evict;
// the value only goes over the cap in fulfill. The fresh entry used to be born
// with its second-chance bit clear, which made it the only clear-bit candidate:
// it was evicted on the spot and a warm shard could not admit anything. A fresh
// entry is now born referenced, as a page is in second-chance eviction, so the
// sweep takes the oldest entry instead.
func TestLocalCache_WarmMemoryCappedShardAdmitsNewKeys(t *testing.T) {
ctx := context.Background()
cache := NewLocalCache(CacheConfig{
MaxMemoryBytes: 6,
Sizer: func(_ string, _ []string, value []byte) int64 {
return int64(len(value))
},
})
for _, k := range []string{"a", "b"} {
if !cache.set(k, []string{k}, []byte(k+k)) { // 2 bytes each
t.Fatalf("failed to set %s", k)
}
}
for round, k := range []string{"c", "d", "e", "f"} {
// Read everything resident, so no entry has a clear bit.
for _, r := range []string{"a", "b", "c", "d", "e", "f"} {
_, _ = cache.Get(ctx, r)
}
if !cache.set(k, []string{k}, []byte(k+k+k)) { // 3 bytes: over the cap
t.Fatalf("round %d: a warm memory-capped shard rejected new key %q", round, k)
}
if _, ok := cache.Get(ctx, k); !ok {
t.Fatalf("round %d: %q was admitted but is not cached", round, k)
}
}
}
func TestLocalCache_SecondChanceIsConsumed(t *testing.T) {
ctx := context.Background()
cache := NewLocalCache(CacheConfig{MaxEntries: 3})
for _, k := range []string{"a", "b", "c"} {
if !cache.set(k, []string{k}, []byte(k)) {
t.Fatalf("failed to set %s", k)
}
}
// Every resident entry has now been read, so every second-chance bit is set
// and there is no clear-bit victim available.
for _, k := range []string{"a", "b", "c"} {
if _, ok := cache.Get(ctx, k); !ok {
t.Fatalf("%s should exist", k)
}
}
// First insert over capacity: no candidate has a clear bit, so this pass
// sweeps the bits and evicts the oldest by token -- "a", inserted first.
if !cache.set("d", []string{"d"}, []byte("d")) {
t.Fatal("failed to set d")
}
if _, ok := cache.Get(ctx, "a"); ok {
t.Fatal("a should be evicted: oldest entry once every second chance was consumed")
}
// The sweep cleared b and c. Reading only c re-arms c alone, so the next
// insert must take b -- proving the bits really were cleared (otherwise b
// would still look protected and the newest entry would be taken instead).
if _, ok := cache.Get(ctx, "c"); !ok {
t.Fatal("c should still be cached")
}
if !cache.set("e", []string{"e"}, []byte("e")) {
t.Fatal("failed to set e")
}
if _, ok := cache.Get(ctx, "b"); ok {
t.Fatal("b should be evicted: its second chance was consumed by the sweep and not renewed")
}
for _, k := range []string{"c", "d", "e"} {
if _, ok := cache.Get(ctx, k); !ok {
t.Fatalf("%s should remain cached", k)
}
}
}

35
options.go
View File

@@ -495,8 +495,12 @@ type Options struct {
// Experimental: this API may change in a minor release.
ClientSideCacheStrategy CSCStrategy
// ClientSideCacheRefreshOnInvalidate re-fetches recently-read keys as soon as
// their invalidation arrives, instead of waiting for a reader to miss.
// ClientSideCacheRefreshOnInvalidate re-fetches every cached entry of an
// invalidated key as soon as its invalidation arrives, instead of waiting for
// a reader to miss. It does not look at how recently the entry was read, so
// each invalidation of a cached key costs one background read: on a
// write-heavy keyspace that is refetch traffic across the whole resident
// cache, not only its hot part.
//
// Requires the built-in cache (ClientSideCacheConfig, or ClientSideCache set
// to a *LocalCache), like the other CSC knobs: the refresher's hot-entry
@@ -506,33 +510,6 @@ type Options struct {
// Experimental: this API may change in a minor release.
ClientSideCacheRefreshOnInvalidate bool
// ClientSideCacheRefreshRecencyWindow bounds ClientSideCacheRefreshOnInvalidate
// to keys read within this window before their invalidation arrived. 0
// (default) refreshes every invalidated Valid entry, regardless of how long
// ago it was last read. A positive value only refreshes entries whose last
// read falls inside the window; an entry outside it is just evicted (an
// ordinary miss on the next read, same as refresh being off).
//
// Tradeoff: refreshing re-registers the key with the server's tracking
// table, which manufactures the NEXT invalidation on the next write — so
// the default (refresh everything) turns a write-heavy, rarely-read key
// into a self-sustaining refresh loop driven by write traffic, not read
// traffic, for as long as its entry survives capacity eviction. Set a
// window to bound that blast radius to keys actually being read.
//
// Recency is tracked at the existing 200ms tick resolution
// (cscRefreshRecencyTick): a nonzero window is rounded up to the tick
// boundary that guarantees AT LEAST the requested window is covered
// regardless of where an invalidation lands relative to the tick phase, so
// the enforced window is somewhere in [window, window+200ms). It also takes
// up to one tick to reach full accuracy right after the refresher starts.
//
// Ignored unless ClientSideCacheRefreshOnInvalidate is set, and requires
// the built-in cache like that option does.
//
// Experimental: this API may change in a minor release.
ClientSideCacheRefreshRecencyWindow time.Duration
// ClientSideCacheCoalesceMisses coalesces concurrent cache misses so they
// stream on a held tracked full-duplex connection instead of each taking a
// pool connection: a lone miss is written immediately (no batching delay —

12
universal.go
View File

@@ -204,18 +204,13 @@ type UniversalOptions struct {
// Experimental: this API may change in a minor release.
ClientSideCacheStrategy CSCStrategy
// ClientSideCacheRefreshOnInvalidate re-fetches recently-read keys as soon as
// their invalidation arrives. See Options.ClientSideCacheRefreshOnInvalidate.
// ClientSideCacheRefreshOnInvalidate re-fetches every cached entry of an
// invalidated key as soon as its invalidation arrives, whether or not it was
// read recently. See Options.ClientSideCacheRefreshOnInvalidate.
//
// Experimental: this API may change in a minor release.
ClientSideCacheRefreshOnInvalidate bool
// ClientSideCacheRefreshRecencyWindow bounds ClientSideCacheRefreshOnInvalidate
// to recently-read keys. See Options.ClientSideCacheRefreshRecencyWindow.
//
// Experimental: this API may change in a minor release.
ClientSideCacheRefreshRecencyWindow time.Duration
// ClientSideCacheCoalesceMisses coalesces concurrent cache misses onto a held
// full-duplex connection. See Options.ClientSideCacheCoalesceMisses.
//
@@ -435,7 +430,6 @@ func (o *UniversalOptions) Simple() *Options {
ClientSideCacheStrategy: o.ClientSideCacheStrategy,
ClientSideCacheRefreshOnInvalidate: o.ClientSideCacheRefreshOnInvalidate,
ClientSideCacheRefreshRecencyWindow: o.ClientSideCacheRefreshRecencyWindow,
ClientSideCacheCoalesceMisses: o.ClientSideCacheCoalesceMisses,
ClientSideCacheInvalidationBatchWindow: o.ClientSideCacheInvalidationBatchWindow,
}