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

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,59 @@ 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. Set before any push can arrive (attach time).
+func (h *invalidateHandler) setInvalBatchWindow(w time.Duration) {
+	h.mu.Lock()
+	h.invalBatchWindow = w
+	h.mu.Unlock()
+}
+
+// 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{}),
+		}
+		go h.batcher.run()
+	}
+	return h.batcher
 }
 
 func (h *invalidateHandler) setRefreshQueue(q *cscRefreshQueue) {
@@ -97,6 +150,7 @@ func (h *invalidateHandler) HandlePushNotification(
 ) error {
 	h.mu.RLock()
 	cache, keyPrefix, refresh := h.cache, h.keyPrefix, h.refresh
+	window := h.invalBatchWindow
 	h.mu.RUnlock()
 	if cache == nil || len(notification) < 2 {
 		return nil
@@ -106,6 +160,31 @@ func (h *invalidateHandler) HandlePushNotification(
 	case nil:
 		cache.Flush()
 	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 +225,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 +384,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 +896,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 +999,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

csc_inval_batch.go

@@ -0,0 +1,112 @@
+package redis
+
+import (
+	"sync"
+	"time"
+)
+
+// Windowed background invalidation batcher.
+//
+// Normally the invalidate handler deletes cache entries INLINE, on whatever
+// goroutine read the RESP3 "invalidate" push — which, for the coalescer's
+// reply reader, means invalidation work steals time from reading miss replies
+// and inflates the low-concurrency churn p99 tail.
+//
+// When Options.ClientSideCacheInvalidationBatchWindow>0 the handler instead
+// ENQUEUES invalidated keys (cheap, off the read path) and a single background
+// goroutine applies the deletes in batches once per window. Deferring an
+// invalidation's application by <= window means a reader may see the
+// pre-invalidation value for <= window, which is exactly what MaxStaleness=window
+// already licenses; set the window <= MaxStaleness to stay within contract (a
+// nonzero window with MaxStaleness=0 is an explicit strictness relaxation).
+
+const cscInvalBatchMax = 4096 // size-cap flush regardless of the timer
+
+type cscInvalBatcher struct {
+	h        *invalidateHandler
+	window   time.Duration
+	ch       chan string
+	stopCh   chan struct{}
+	stopOnce sync.Once
+}
+
+// stop signals run() to flush and exit; idempotent. It only closes a channel —
+// it never touches h.mu and does not wait for the goroutine — so it is safe to
+// call while holding the handler lock (see releaseLocked).
+func (b *cscInvalBatcher) stop() {
+	b.stopOnce.Do(func() { close(b.stopCh) })
+}
+
+// enqueue hands a namespaced key to the batcher without blocking the caller. On
+// a full queue it applies the delete inline so an invalidation is never dropped.
+func (b *cscInvalBatcher) enqueue(nsKey string) {
+	select {
+	case b.ch <- nsKey:
+	default:
+		b.apply([]string{nsKey})
+	}
+}
+
+// apply deletes the given namespaced keys and feeds any evicted-hot entries to
+// the refresher. Reads the handler binding under its lock, same as the inline path.
+func (b *cscInvalBatcher) apply(keys []string) {
+	h := b.h
+	h.mu.RLock()
+	cache, refresh := h.cache, h.refresh
+	h.mu.RUnlock()
+	if cache == nil {
+		return
+	}
+	lc, canRefresh := cache.(*LocalCache)
+	canRefresh = canRefresh && refresh != nil
+	var hot []cscRefreshTarget
+	for _, k := range keys {
+		if !canRefresh {
+			cache.DeleteByRedisKey(k)
+			continue
+		}
+		hot = lc.deleteByRedisKeyCollectingHot(k, refresh.sinceToken.Load(), hot[:0])
+		for i := range hot {
+			refresh.offer(hot[i])
+		}
+	}
+}
+
+func (b *cscInvalBatcher) run() {
+	t := time.NewTimer(b.window)
+	defer t.Stop()
+	pending := make([]string, 0, 256)
+	seen := make(map[string]struct{}, 256)
+	flush := func() {
+		if len(pending) == 0 {
+			return
+		}
+		b.apply(pending)
+		pending = pending[:0]
+		for k := range seen {
+			delete(seen, k)
+		}
+	}
+	for {
+		select {
+		case <-b.stopCh:
+			// Last user released the binding: flush what is pending (a no-op once
+			// the cache is nil) and exit so the goroutine does not live on
+			// re-arming the timer forever.
+			flush()
+			return
+		case k := <-b.ch:
+			if _, dup := seen[k]; !dup {
+				seen[k] = struct{}{}
+				pending = append(pending, k)
+			}
+			if len(pending) >= cscInvalBatchMax {
+				flush()
+				t.Reset(b.window)
+			}
+		case <-t.C:
+			flush()
+			t.Reset(b.window)
+		}
+	}
+}