feat(csc): full-duplex reader-miss coalescing + invalidation batching

.github/workflows/govulncheck.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v7
         with:
-          go-version: "1.26.x"
+          go-version: "stable"
           cache: true
 
       - name: Install govulncheck

csc_coalesce_options_test.go

@@ -0,0 +1,61 @@
+package redis
+
+import (
+	"testing"
+	"time"
+)
+
+// TestCSCCoalesceModeRejectsPinnedPublicly pins that the buggy "pinned" PROTOTYPE
+// engine (no idle invalidation drain; can serve stale) is NOT selectable from the
+// public ClientSideCacheCoalesceMode option — it falls back to "workers" — while
+// the internal benchmark hook can still force it.
+func TestCSCCoalesceModeRejectsPinnedPublicly(t *testing.T) {
+	if got := cscCoalesceMode(&Options{ClientSideCacheCoalesceMode: "pinned"}); got != "workers" {
+		t.Fatalf("public \"pinned\" => %q, want \"workers\"", got)
+	}
+	if got := cscCoalesceMode(&Options{ClientSideCacheCoalesceMode: "fullduplex"}); got != "fullduplex" {
+		t.Fatalf("\"fullduplex\" => %q, want \"fullduplex\"", got)
+	}
+	if got := cscCoalesceMode(&Options{ClientSideCacheCoalesceMode: ""}); got != "workers" {
+		t.Fatalf("\"\" => %q, want \"workers\"", got)
+	}
+	if got := cscCoalesceMode(nil); got != "workers" {
+		t.Fatalf("nil opt => %q, want \"workers\"", got)
+	}
+
+	cscForcePinned = true
+	defer func() { cscForcePinned = false }()
+	if got := cscCoalesceMode(&Options{ClientSideCacheCoalesceMode: "pinned"}); got != "pinned" {
+		t.Fatalf("forced \"pinned\" => %q, want \"pinned\" (benchmark hook)", got)
+	}
+}
+
+// TestUniversalOptionsSimpleCopiesCSCCoalesce guards that the new CSC miss-
+// coalescing / invalidation-batching knobs reach a standalone Client through
+// UniversalOptions.Simple() (they were previously Options-only, so UniversalClient
+// users could not enable them).
+func TestUniversalOptionsSimpleCopiesCSCCoalesce(t *testing.T) {
+	u := &UniversalOptions{
+		ClientSideCacheRefreshOnInvalidate:     true,
+		ClientSideCacheCoalesceMisses:          true,
+		ClientSideCacheCoalesceMode:            "fullduplex",
+		ClientSideCacheCoalesceWorkers:         5,
+		ClientSideCacheInvalidationBatchWindow: 7 * time.Millisecond,
+	}
+	o := u.Simple()
+	if !o.ClientSideCacheRefreshOnInvalidate {
+		t.Error("Simple() dropped ClientSideCacheRefreshOnInvalidate")
+	}
+	if !o.ClientSideCacheCoalesceMisses {
+		t.Error("Simple() dropped ClientSideCacheCoalesceMisses")
+	}
+	if o.ClientSideCacheCoalesceMode != "fullduplex" {
+		t.Errorf("Simple() ClientSideCacheCoalesceMode = %q, want \"fullduplex\"", o.ClientSideCacheCoalesceMode)
+	}
+	if o.ClientSideCacheCoalesceWorkers != 5 {
+		t.Errorf("Simple() ClientSideCacheCoalesceWorkers = %d, want 5", o.ClientSideCacheCoalesceWorkers)
+	}
+	if o.ClientSideCacheInvalidationBatchWindow != 7*time.Millisecond {
+		t.Errorf("Simple() ClientSideCacheInvalidationBatchWindow = %v, want 7ms", o.ClientSideCacheInvalidationBatchWindow)
+	}
+}

csc_integration.go

@@ -82,6 +82,73 @@ type invalidateHandler struct {
 	// refresh, when set, receives evicted-but-hot entries for immediate refetch.
 	// Feeding it must never block the invalidation-delivery path.
 	refresh *cscRefreshQueue
+
+	// batcher offloads invalidation cache-deletes to a windowed background
+	// goroutine (Options.ClientSideCacheInvalidationBatchWindow). Lazily started
+	// (ensureBatcher) and nil when disabled; guarded by mu. Stopped and cleared
+	// when the last user releases (releaseLocked), so its goroutine does not live
+	// past the binding re-arming its timer forever; a later re-acquire starts a
+	// fresh one (picking up the successor's window).
+	batcher *cscInvalBatcher
+
+	// invalBatchWindow is the coalescing window for the batcher above, threaded
+	// from the owning client's Options at attach time. 0 (default) deletes inline.
+	// Read under mu alongside cache/keyPrefix/refresh.
+	invalBatchWindow time.Duration
+}
+
+// setInvalBatchWindow records the invalidation-batch coalescing window from the
+// owning client's Options. Normally set before any push can arrive (attach
+// time). When a second client binds to the same shared handler with a different
+// window, an already-running batcher was created with the previous window (its
+// window is fixed at creation), so it would keep honoring the old cadence and the
+// new client's staleness bound would not hold. Stop and drop that batcher so the
+// next invalidation lazily starts a fresh one with the new window via
+// ensureBatcher; stop() flushes what it holds, so no queued delete is lost.
+func (h *invalidateHandler) setInvalBatchWindow(w time.Duration) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if h.invalBatchWindow == w {
+		return
+	}
+	h.invalBatchWindow = w
+	if h.batcher != nil {
+		h.batcher.stop()
+		h.batcher = nil
+	}
+}
+
+// ensureBatcher lazily starts the windowed invalidation batcher. The common
+// case (already started) is a shared RLock; only first-start takes the write
+// lock, so the hot invalidation path stays cheap.
+func (h *invalidateHandler) ensureBatcher(w time.Duration) *cscInvalBatcher {
+	h.mu.RLock()
+	b := h.batcher
+	h.mu.RUnlock()
+	if b != nil {
+		return b
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	// Do not start a batcher for a released binding. releaseLocked stops+nils the
+	// batcher under this same lock when users hits 0, so a push racing that last
+	// release must NOT resurrect a goroutine that nothing would ever stop (once
+	// users is 0, release() no longer runs). The caller falls back to the inline
+	// delete path when this returns nil.
+	if h.users == 0 {
+		return nil
+	}
+	if h.batcher == nil {
+		h.batcher = &cscInvalBatcher{
+			h:      h,
+			window: w,
+			ch:     make(chan string, 8192),
+			stopCh: make(chan struct{}),
+			dropCh: make(chan struct{}, 1),
+		}
+		go h.batcher.run()
+	}
+	return h.batcher
 }
 
 func (h *invalidateHandler) setRefreshQueue(q *cscRefreshQueue) {
@@ -97,6 +164,8 @@ func (h *invalidateHandler) HandlePushNotification(
 ) error {
 	h.mu.RLock()
 	cache, keyPrefix, refresh := h.cache, h.keyPrefix, h.refresh
+	window := h.invalBatchWindow
+	batcher := h.batcher
 	h.mu.RUnlock()
 	if cache == nil || len(notification) < 2 {
 		return nil
@@ -105,7 +174,39 @@ func (h *invalidateHandler) HandlePushNotification(
 	switch payload := notification[1].(type) {
 	case nil:
 		cache.Flush()
+		// A full flush (FLUSHDB/FLUSHALL) removed every entry, so any per-key
+		// deletes the batcher still has queued are redundant — applying them after
+		// the flush would evict entries a reader repopulated post-flush, an extra
+		// miss for up to the window. Drop the batcher's pending queue.
+		if batcher != nil {
+			batcher.drop()
+		}
 	case []interface{}:
+		// Offload path: enqueue keys to the windowed background batcher instead of
+		// deleting inline, so invalidation work does not steal time from the
+		// coalescer's miss-reply reader (the low-concurrency churn p99 tail).
+		if window > 0 {
+			if _, ok := cache.(*LocalCache); ok {
+				// nil when the binding was just released (users==0): fall through
+				// to the inline delete path below rather than enqueue on a nil
+				// batcher (which would panic).
+				if b := h.ensureBatcher(window); b != nil {
+					for _, k := range payload {
+						var name string
+						switch v := k.(type) {
+						case string:
+							name = v
+						case []byte:
+							name = string(v)
+						default:
+							continue
+						}
+						b.enqueue(cscNamespacedKey(keyPrefix, name))
+					}
+					return nil
+				}
+			}
+		}
 		var hot []cscRefreshTarget
 		lc, canRefresh := cache.(*LocalCache)
 		canRefresh = canRefresh && refresh != nil
@@ -146,6 +247,14 @@ func (h *invalidateHandler) releaseLocked() {
 	if h.users == 0 {
 		h.cache = nil
 		h.keyPrefix = ""
+		// Stop the windowed batcher so its goroutine does not outlive the binding
+		// (re-arming its timer forever). stop() only closes a channel — it never
+		// touches h.mu and does not wait — so it is safe under the lock. A later
+		// re-acquire starts a fresh batcher via ensureBatcher.
+		if h.batcher != nil {
+			h.batcher.stop()
+			h.batcher = nil
+		}
 	}
 }
 
@@ -297,6 +406,12 @@ func (c *baseClient) attachSharedTrackingCSC(ctx context.Context, cache Cache) {
 		internal.Logger.Printf(ctx, "csc: failed to register invalidate handler: %v", err)
 		return
 	}
+	// Thread the invalidation-batch window from Options before any push can
+	// arrive, so the batcher (if enabled) sees the configured window on the very
+	// first invalidation rather than a zero default.
+	if ih := lookupInvalidateHandler(c.pushProcessor); ih != nil {
+		ih.setInvalBatchWindow(c.opt.ClientSideCacheInvalidationBatchWindow)
+	}
 	c.csc = cache
 	c.registerConnEvictHook(cache, reg)
 	c.startBackgroundDrainer()
@@ -803,6 +918,18 @@ func applyCachedReply(cmd Cmder, raw []byte) error {
 	return cmd.readReply(proto.NewReaderSize(bytes.NewReader(raw), len(raw)+1))
 }
 
+// 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
+// value or Nil is cacheable, a top-level RESP error is not. It reads the frame
+// generically, so it can only diverge from a concrete cmd's readReply on a
+// well-formed reply of an unexpected shape — which the next reader re-parses and
+// drops (see processCached), so a rare mis-cache self-heals.
+func classifyCachedReply(raw []byte) error {
+	_, err := proto.NewReaderSize(bytes.NewReader(raw), len(raw)+1).ReadReply()
+	return err
+}
+
 // isCacheableReplyResult reports whether a fully read Redis reply can be
 // cached. redis.Nil is a normal negative lookup, not a transport/protocol
 // failure; tracking will invalidate it if the key is later created.
@@ -894,13 +1021,21 @@ func (c *baseClient) processCached(ctx context.Context, cmd Cmder, state *proces
 			c.csc.DeleteByCacheKey(key)
 		}
 		// Original fetcher cancelled or its value was invalidated; try to take
-		// over so later waiters still benefit from the cache.
+		// over so later waiters still benefit from the cache. This is the 2x-RTT
+		// path under churn: we waited a round trip and still must fetch ourselves.
 		token, shouldFetch = c.csc.Reserve(key, nsRedisKeys)
 	}
 
 	// Reader-miss coalescing: hand the reserved miss to the batcher (no-op when off).
 	if shouldFetch && c.cscMissCoalescer != nil {
-		return c.cscMissCoalescer.fetch(ctx, cmd, key, token)
+		err := c.cscMissCoalescer.fetch(ctx, cmd, key, token)
+		if err == errCSCRetryUncached {
+			// The coalescer bowed out because CSC serving was disabled mid-miss; the
+			// command itself is fine and the reservation was already cancelled. Run it
+			// uncached on the normal path rather than surfacing a spurious ErrClosed.
+			return c.processWithRetry(ctx, cmd, nil, state)
+		}
+		return err
 	}
 
 	var fc cscFetchCapture