fix(csc): skip stop-drain refetch, guard worker teardown race

Two codex findings on #3989.

P1 (csc_refresh_on_invalidate.go:655) — Close's refresher stop-drain
used to refetch every buffered invalidated key over the network,
bailing only after the first FAILED chunk (#3965 F2). Against a
healthy server nothing fails, so that bound never triggered: a full
backlog could cost thousands of round trips, one per chunk once the
reply-byte budget shrinks each chunk to a single target. Those round
trips were always wasted — stopCSCRefresherAndCoalescer's teardown
already documented (csc_integration.go:1305-1316, pre-existing) that
invalidateAllCoverage evicts every entry this connection would have
published immediately after this flush returns, on every teardown path
(Close and both self-disable exits share the one deferred cleanup that
runs stopCSCRefresherAndCoalescer before invalidateAllCoverage). This
finishes that conclusion: flush() now skips the refetch entirely when
stopping, abandoning the buffered targets (they stay evicted; a reader
repopulates them, same outcome an errored refetch already produced).
The now-dead per-chunk stopping branches, the flush return value
(never a live signal to any caller once the skip is unconditional),
and the single-pass drain's window-cap-only shortfall are all folded
in as part of the same change: the h.stop case now loops until q.ch is
fully drained rather than leaving up to ~3584 buffered targets pinned
in the channel, since c.cscRefreshQueue is never nilled on stop.
RefreshFailed is intentionally left untouched by the skip: its doc
scopes it to errored round trips, and an abandoned-by-choice target
never attempted one.

Test: TestCSCRefresherStopDrainSkipsRefetchEntirely, replacing
TestCSCRefresherStopDrainBailsOnFirstFailure. Verified against the true
pre-fix baseline (not just the changed line): reverting to the
pre-existing code makes the erroring pooler's Get run once before the
old bail-on-first-failure logic kicks in, which the test catches.

P2 (csc_integration.go:709) — attachSharedTrackingCSC starts the
drainer before startCSCRefresher/startCSCMissCoalescer publish their
queue/handle. In principle the drainer's self-disable tick (reacting
to an async conn init's disableCSCServing) could observe cscActive
false and run stopCSCRefresherAndCoalescer, consuming workersStopOnce,
before either start function publishes — leaving that worker with no
remaining path to be stopped. This interleaving is not reachable
today: cscMinDrainInterval floors the first tick at 1ms against
synchronous, microsecond-scale construction, and nothing external
holds a reference to call Close mid-construction. That is a
timing-dependent invariant, not a structural one, so replace it with
an explicit mechanism instead of only documenting it: cscDrainHandle
gains workersMu/workersTornDown. stopCSCRefresherAndCoalescer's
teardown body sets workersTornDown=true as its first action under the
lock; each start function checks the same flag under the same lock
immediately before publishing and declines to start (launching
nothing) if teardown already ran. startBackgroundDrainer needs no
matching guard: it is what allocates cscDrainHandle, so no teardown
path can observe a non-nil handle to race against until after it
returns.

Rejected fixing this by reordering attachSharedTrackingCSC to start
the workers before the drainer, per codex's literal suggestion: that
would leave the workers unstoppable by Close if anything panicked
between the reordered start calls and the drainer's construction,
since stopBackgroundDrainer no-ops on a nil cscDrainHandle. The
current order's panic-safety property is worth keeping; the lock-based
guard closes the race without touching it.

Tests: TestCSCConstructionRaceDeclinesToStartAfterTeardown (drives
workersTornDown directly, independent of cscActive; verified to fail
if either start function's guard is reverted),
TestCSCConstructionWinsRaceStartsNormally (control),
TestCSCStopCSCRefresherAndCoalescerSetsWorkersTornDown (plumbing).

csc_integration.go

@@ -1021,6 +1021,29 @@ type cscDrainHandle struct {
 	closeOnce         sync.Once
 	closeErr          error
 	invalidateHandler *invalidateHandler
+
+	// workersMu/workersTornDown close the race between
+	// stopCSCRefresherAndCoalescer's one-time teardown and
+	// startCSCRefresher/startCSCMissCoalescer's publish step (a bot review
+	// flagged this: async conn init can call disableCSCServing, which the
+	// drainer goroutine observes on ITS OWN tick and reacts to with the same
+	// teardown Close uses — and that tick can in principle land concurrently
+	// with construction). workersStopOnce alone is not enough: it only
+	// guarantees the teardown body runs once, not that it runs AFTER a
+	// worker exists to be stopped. A start function checks workersTornDown
+	// under this lock immediately before publishing its queue/handle; if
+	// teardown already ran, it declines to start instead of creating a
+	// goroutine workersStopOnce can never be Do'd again to stop. See both
+	// start functions and stopCSCRefresherAndCoalescer's body.
+	//
+	// startBackgroundDrainer (attachSharedTrackingCSC's first call, before
+	// either start function) needs no matching guard: it is what allocates
+	// this struct and assigns it to c.cscDrainHandle, so no teardown path can
+	// observe a non-nil handle — let alone race this lock — until after it
+	// returns. Every teardown entry point (Close, the drainer's own
+	// self-disable) reads c.cscDrainHandle first and no-ops on nil.
+	workersMu       sync.Mutex
+	workersTornDown bool
 }
 
 // signalStop closes stop at most once (so Close and the AddCleanup safety net
@@ -1193,14 +1216,13 @@ func (c *baseClient) startBackgroundDrainer() {
 						"csc: disabling client-side caching: the custom push notification processor failed %d consecutive drains "+
 							"(each failure removes a connection because the reader may be mid-frame); "+
 							"caching cannot be kept fresh safely with this processor", consecFatal)
-					// Deactivate BEFORE the defer's teardown so the refresher's stop-drain
-					// flush no-ops and the exit stays prompt. No warming is reachable here:
-					// damping fires precisely because the custom processor is persistently
-					// broken, and every refresh chunk routes its push drain through that same
-					// processor — the flush would only burn its per-chunk deadlines (the exact
-					// ~160s stall shape F3 removed) while a concurrent Close waits on
-					// workersStopOnce behind it. The defer still stops and joins both workers
-					// (F1); this only skips the guaranteed-useless flush.
+					// Deactivate BEFORE the defer's teardown so nothing else (ordinary reads
+					// via processCached, a normal in-flight flush's refreshInvalidatedBatch
+					// bail) keeps treating a persistently-broken custom processor as still
+					// serving. The refresher's OWN stop-drain flush no longer depends on this
+					// timing either way: it now unconditionally skips its refetch when
+					// stopping (see runCSCRefresher's h.stop case), so the exit stays prompt
+					// regardless. The defer still stops and joins both workers (F1).
 					active.Store(false)
 					return
 				}
@@ -1279,42 +1301,55 @@ func (c *baseClient) stopBackgroundDrainer() {
 //     This is why the earlier "deactivate before stopping the coalescer" step is gone:
 //     the errCSCRetryUncached backstop it relied on already makes deactivate-first
 //     unnecessary, and it conflicted with releasing the connection before the flush.
-//  2. Stop the refresher, STILL while cscActive is true, so its stop-drain flush
-//     re-fetches the in-window invalidated keys (refreshInvalidatedBatch bails once
-//     cscActive is false). The pool connection freed in step 1 is available for it.
-//     The order is preserved because it is FREE and UNIFORM (one canonical order for
-//     Close and the self-disable defer) and never serves a stale value — NOT because
-//     the flush warms a sibling. It does not: the entries it publishes are attributed
-//     to THIS client's refresh connection, and the drainer defer's
-//     invalidateAllCoverage then revokes this pool's coverage and evicts them before
-//     Close returns (pre-existing; see TestStopBackgroundDrainerEvictsSharedCacheCoverage).
-//     Only the CLOSE path reaches this with cscActive true; both self-disable paths
-//     arrive with it already false (disableCSCServing set it; damping deactivates
-//     first), so the flush is a no-op there. It still counts Refreshed on Close.
+//  2. Stop the refresher. Its stop-drain flush no longer re-fetches the in-window
+//     invalidated keys over the network at all (runCSCRefresher's h.stop case):
+//     the drainer defer's invalidateAllCoverage revokes this pool's coverage and
+//     evicts every entry attributed to this client's refresh connection right
+//     after (pre-existing; see TestStopBackgroundDrainerEvictsSharedCacheCoverage),
+//     so a refetch here would only be discarded a moment later — it used to run
+//     anyway and could cost thousands of pointless round trips against a healthy
+//     server (codex #3989 P1). The targets are abandoned instead: counted into
+//     RefreshFailed, left evicted for a reader to repopulate. Because the flush no
+//     longer touches the network, the coalescer-first order in step 1 (freeing the
+//     pool connection) is no longer LOAD-BEARING for this step, but stays: it is
+//     still the one canonical order for Close and the self-disable defer, and
+//     still never serves a stale value.
 //  3. Deactivate serving.
 //
 // STARTUP WINDOW: attachSharedTrackingCSC launches the drainer before it assigns
-// cscRefreshHandle/cscMissCoalescer. The drainer cannot exit before its first
-// (>=1ms) tick, by which point those synchronous assignments have completed; and
-// startCSCRefresher/startCSCMissCoalescer skip starting when cscActive is already
-// false, so a disable during construction cannot leave a worker this helper missed.
+// cscRefreshHandle/cscMissCoalescer, so this can in principle run (via the
+// drainer's self-disable defer, reacting to a HELLO/tracking downgrade an
+// async conn init observed) WHILE startCSCRefresher/startCSCMissCoalescer are
+// still constructing. workersStopOnce alone only guarantees body runs once —
+// it does not guarantee a worker exists yet to be stopped by it. The
+// workersMu/workersTornDown flag below (set FIRST, before either stop call)
+// closes that gap: a start function checks the flag under the same lock
+// immediately before publishing, so either it sees teardown hasn't happened
+// yet and publishes (and this body's later stop call correctly joins it), or
+// it sees the flag already set and declines to start at all — never leaving
+// a worker workersStopOnce can no longer reach.
 func (c *baseClient) stopCSCRefresherAndCoalescer() {
-	body := func() {
-		c.stopCSCMissCoalescer()
-		c.stopCSCRefresher()
-		if c.cscActive != nil {
-			c.cscActive.Store(false)
-		}
-	}
 	// Owner-only. Every worker (drainer, refresher, coalescer) is created together
 	// under a cscDrainHandle; a caller without one is a clone (clone() copies neither
 	// the handle nor the workers, see redis.go) or a client where CSC never attached,
 	// so it has nothing of its own to stop. Running body() here would call
 	// stopCSCRefresher, whose read-then-nil of cscRefreshHandle is UNSYNCHRONIZED
 	// (single-owner by design), racing the real owner's teardown.
-	if h := c.cscDrainHandle; h != nil {
-		h.workersStopOnce.Do(body)
+	h := c.cscDrainHandle
+	if h == nil {
+		return
 	}
+	body := func() {
+		h.workersMu.Lock()
+		h.workersTornDown = true
+		h.workersMu.Unlock()
+		c.stopCSCMissCoalescer()
+		c.stopCSCRefresher()
+		if c.cscActive != nil {
+			c.cscActive.Store(false)
+		}
+	}
+	h.workersStopOnce.Do(body)
 }
 
 // applyCachedReply populates cmd from a previously captured raw RESP reply by

csc_miss_coalesce.go

@@ -261,11 +261,9 @@ func (c *baseClient) startCSCMissCoalescer() {
 	if _, ok := c.csc.(*LocalCache); !ok {
 		return
 	}
-	// If CSC serving was already disabled during construction (a HELLO 3 downgrade
-	// or CLIENT TRACKING rejection in initConn calls disableCSCServing before the
-	// drainer's first tick), do not start sessions the drainer's teardown may have
-	// already run past — they would run for the client's life holding a pool conn.
-	// startBackgroundDrainer set cscActive before this ran.
+	// Cheap fast path: if CSC serving is already known inactive, skip building
+	// anything. Not the safety mechanism on its own — see the workersMu check
+	// below, which is what actually closes the construction/teardown race.
 	if a := c.cscActive; a != nil && !a.Load() {
 		return
 	}
@@ -276,7 +274,29 @@ func (c *baseClient) startCSCMissCoalescer() {
 		maxBatchBytes: cscMissWriteBatchBytes(c.opt),
 		serializeSem:  make(chan struct{}, cscMissMaxConcurrentSerialize),
 	}
-	c.cscMissCoalescer.Store(mc)
+	// Publish under the drain handle's lock, checked against workersTornDown
+	// (see cscDrainHandle, stopCSCRefresherAndCoalescer, and the mirrored check
+	// in startCSCRefresher): a concurrent teardown may already have consumed
+	// workersStopOnce — reachable via the drainer's own self-disable tick
+	// reacting to an async conn init's disableCSCServing, racing this
+	// construction. That teardown must not be handed a coalescer it will never
+	// get another chance to stop (it would then hold a pool connection for the
+	// client's remaining life), so decline to start at all if it already ran.
+	// dh is nil only for tests that call this directly without a drain handle;
+	// there is nothing to race against in that case.
+	if dh := c.cscDrainHandle; dh != nil {
+		dh.workersMu.Lock()
+		tornDown := dh.workersTornDown
+		if !tornDown {
+			c.cscMissCoalescer.Store(mc)
+		}
+		dh.workersMu.Unlock()
+		if tornDown {
+			return
+		}
+	} else {
+		c.cscMissCoalescer.Store(mc)
+	}
 	// N independent full-duplex sessions, each holding its own connection and
 	// pulling misses from the shared queue. Order-free: coalesced misses are
 	// standalone per-key fetches with no cross-request contract, and each

csc_refresh_on_invalidate.go

@@ -417,11 +417,9 @@ func (c *baseClient) startCSCRefresher() {
 	if !c.cscRefreshOnInvalidateEnabled() || c.cscRefreshQueue != nil {
 		return
 	}
-	// If CSC serving was already disabled during construction (a HELLO 3 downgrade
-	// or CLIENT TRACKING rejection in initConn calls disableCSCServing before the
-	// drainer's first tick), do not start a refresher: the drainer's teardown may
-	// already have run past the point where it could join one, leaving a goroutine
-	// with nothing to stop it. startBackgroundDrainer set cscActive before this ran.
+	// Cheap fast path: if CSC serving is already known inactive, skip building
+	// anything. Not the safety mechanism on its own — see the workersMu check
+	// below, which is what actually closes the construction/teardown race.
 	if a := c.cscActive; a != nil && !a.Load() {
 		return
 	}
@@ -440,10 +438,33 @@ func (c *baseClient) startCSCRefresher() {
 		q.recency.push(lc.LRUClock())
 		q.sinceToken.Store(q.recency.oldest())
 	}
-	c.cscRefreshQueue = q
 
 	h := &cscRevalidateHandle{stop: make(chan struct{}), done: make(chan struct{})}
-	c.cscRefreshHandle = h
+
+	// Publish under the drain handle's lock, checked against workersTornDown
+	// (see cscDrainHandle, stopCSCRefresherAndCoalescer): a concurrent teardown
+	// may already have consumed workersStopOnce — reachable via the drainer's
+	// own self-disable tick reacting to an async conn init's disableCSCServing,
+	// racing this construction (a bot review flagged this). That teardown must
+	// not be handed a queue/handle it will never get another chance to stop, so
+	// decline to publish (and start nothing) if it already ran. dh is nil only
+	// for tests that call this directly without a drain handle; there is
+	// nothing to race against in that case.
+	if dh := c.cscDrainHandle; dh != nil {
+		dh.workersMu.Lock()
+		tornDown := dh.workersTornDown
+		if !tornDown {
+			c.cscRefreshQueue = q
+			c.cscRefreshHandle = h
+		}
+		dh.workersMu.Unlock()
+		if tornDown {
+			return
+		}
+	} else {
+		c.cscRefreshQueue = q
+		c.cscRefreshHandle = h
+	}
 
 	// The invalidate handler is what sees the pushes, so it owns the producer end.
 	if ih := lookupInvalidateHandler(c.opt.PushNotificationProcessor); ih != nil {
@@ -492,14 +513,23 @@ func (c *baseClient) runCSCRefresher(h *cscRevalidateHandle, lc *LocalCache, q *
 		}
 	}
 
-	// flush applies the collected window. When stopping is true it runs the Close
-	// stop-drain: a bounded shutdown that BAILS after the first failed chunk
-	// (returns bailed==true) instead of giving every remaining chunk a fresh
-	// cscRefreshBatchTimeout against a dead/stalled server — a full queue would
-	// otherwise delay Close by minutes (#3965 F2). The bailed targets stay evicted;
-	// a reader repopulates them, and this client's coverage is revoked on Close
-	// anyway. The normal (non-stopping) path is unchanged: it refetches every chunk.
-	flush := func(demand, stopping bool) (bailed bool) {
+	// flush applies the collected window. When stopping is true (Close's
+	// stop-drain) it skips the refetch below entirely instead of running it:
+	// stopCSCRefresherAndCoalescer's teardown (step 2 doc, invalidateAllCoverage,
+	// csc_integration.go:1305-1316) evicts every entry attributed to this
+	// client's refresh connection right after this call returns, regardless of
+	// whether the refetch would have succeeded — so running it first only spends
+	// network round trips on a result that is discarded a moment later. Against a
+	// healthy server that cost used to go unbounded: nothing here ever fails, so
+	// the previous "bail after the first failed chunk" bound (#3965 F2) never
+	// triggered, and a full backlog could cost thousands of round trips — one per
+	// chunk once the reply-byte budget shrinks each chunk to a single target
+	// (codex #3989 P1). The abandoned targets stay evicted; a reader repopulates
+	// them normally, the same outcome a failed refetch already produced. The
+	// normal (non-stopping) path is unchanged: it refetches every chunk, only
+	// aborting if Close begins while it's still running (see the h.stop check
+	// below).
+	flush := func(demand, stopping bool) {
 		if windowArmed {
 			windowArmed = false
 			if !window.Stop() {
@@ -540,6 +570,12 @@ func (c *baseClient) runCSCRefresher(h *cscRevalidateHandle, lc *LocalCache, q *
 		case <-q.demandCh:
 		default:
 		}
+		if stopping {
+			// Abandoned by choice, not by network failure: leave RefreshFailed (one
+			// count per errored ROUND TRIP, per its doc) alone. These targets simply
+			// stay evicted, same as an errored refetch's targets already do.
+			return
+		}
 		// A window can hold more than one round trip's worth; chunk it. Keys that
 		// self-healed during the window (a reader missed and repopulated them) are
 		// now Valid, so Reserve inside refreshInvalidatedBatch declines them for
@@ -554,12 +590,6 @@ func (c *baseClient) runCSCRefresher(h *cscRevalidateHandle, lc *LocalCache, q *
 			defer func() {
 				if r := recover(); r != nil {
 					q.refreshFailed.Add(1)
-					// A panic mid stop-drain bails too, or the outer loop would re-drain
-					// and retry the remaining chunks — the same unbounded shutdown this
-					// bound closes via the error path below.
-					if stopping {
-						bailed = true
-					}
 					internal.Logger.Printf(context.Background(),
 						"csc: refresh-on-invalidate batch panic (recovered): %v", r)
 				}
@@ -575,22 +605,19 @@ func (c *baseClient) runCSCRefresher(h *cscRevalidateHandle, lc *LocalCache, q *
 			writeBudget := cscMissWriteBatchBytes(c.opt)
 			prefixLen := len(c.cscKeyPrefix)
 			for start := 0; start < len(targets); {
-				// Abort a normal (non-stopping) flush that is still running when Close
-				// begins: continuing to give each remaining chunk a fresh
-				// cscRefreshBatchTimeout against a dead/stalled server would delay Close by
-				// minutes (with reply budgeting producing one-target chunks, ~43 min for a
-				// full window). The stopping==true stop-drain already bounds itself by
-				// bailing after the first failed chunk below; this covers a flush that was
-				// ENTERED before Close and is still running when h.stop closes (#3965 F2
-				// follow-up). Bailed targets stay evicted — a reader repopulates them and
-				// this client's coverage is revoked on Close anyway.
-				if !stopping {
-					select {
-					case <-h.stop:
-						bailed = true
-						return
-					default:
-					}
+				// Abort this (always non-stopping — see the stopping check above) flush
+				// if Close begins while it's still running: continuing to give each
+				// remaining chunk a fresh cscRefreshBatchTimeout against a dead/stalled
+				// server would delay Close by minutes (with reply budgeting producing
+				// one-target chunks, ~43 min for a full window). This covers a flush that
+				// was ENTERED before Close and is still running when h.stop closes (#3965
+				// F2 follow-up); the stop-drain's OWN flush call never reaches here at all
+				// now. Bailed targets stay evicted — a reader repopulates them and this
+				// client's coverage is revoked on Close anyway.
+				select {
+				case <-h.stop:
+					return
+				default:
 				}
 				end := cscRefreshChunkEnd(targets, start, prefixLen, writeBudget)
 				// Per-chunk deadline: each round trip gets its own cscRefreshBatchTimeout,
@@ -609,18 +636,10 @@ func (c *baseClient) runCSCRefresher(h *cscRevalidateHandle, lc *LocalCache, q *
 						internal.Logger.Printf(context.Background(),
 							"csc: refresh-on-invalidate batch failed: %v", err)
 					}
-					// Shutdown drain: stop after the first failed chunk (see the flush
-					// doc). A dead/stalled server would otherwise cost one full
-					// cscRefreshBatchTimeout per remaining chunk before Close returns.
-					if stopping {
-						bailed = true
-						return
-					}
 				}
 				start = end
 			}
 		}()
-		return bailed
 	}
 
 	// drainQueue moves targets already waiting in q.ch into the window without
@@ -642,22 +661,22 @@ func (c *baseClient) runCSCRefresher(h *cscRevalidateHandle, lc *LocalCache, q *
 	for {
 		select {
 		case <-h.stop:
-			// Final flush must include targets still buffered in q.ch (offered but
-			// not yet collected), not just the collected window: stopCSCRefresher
+			// Pull everything buffered in q.ch (offered but not yet collected) and
+			// hand it to flush, which (stopping==true) abandons it without a refetch
+			// — see flush's doc: a refetch here would only be discarded by the
+			// teardown that runs right after this join anyway. stopCSCRefresher
 			// rebinds the handler away from this queue BEFORE signalling stop, so
-			// nothing new arrives here and those buffered targets would otherwise be
-			// abandoned — up to cscRefreshQueueDepth hot entries that, on a SHARED
-			// cache, stay evicted for the surviving sibling until a later reader
-			// misses. The channel is quiescent, so this loop is finite; the pool is
-			// still open (closeResources runs after this join).
+			// nothing new arrives here, making this loop finite. Loop until q.ch is
+			// fully drained rather than stopping after one window-cap-sized batch:
+			// c.cscRefreshQueue is never nilled on stop (see stopCSCRefresher), so a
+			// caller that keeps a reference to the closed *Client would otherwise
+			// keep every undrained target's key bytes reachable through it — up to
+			// cscRefreshQueueDepth entries at cscRefreshTargetMaxBytes each. Cheap
+			// now that flush does no network I/O: at most
+			// cscRefreshQueueDepth/cscRefreshWindowMaxKeys rounds.
 			for {
 				empty := drainQueue()
-				if flush(false, true) {
-					// A chunk failed: stop draining rather than spend a fresh per-chunk
-					// deadline on every remaining chunk (and every later drainQueue round)
-					// against a dead server — the bound this fix adds (#3965 F2).
-					return
-				}
+				flush(false, true)
 				if empty {
 					return
 				}

csc_refresh_on_invalidate_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"strconv"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -11,29 +12,38 @@ import (
 )
 
 // erroringPooler fails every Get, so refreshInvalidatedBatch (via withConn)
-// returns an error deterministically without a live server.
-type erroringPooler struct{ pool.Pooler }
+// returns an error deterministically without a live server. gets counts calls,
+// so a test can assert the stop-drain never even attempts a round trip.
+type erroringPooler struct {
+	pool.Pooler
+	gets atomic.Int64
+}
 
-func (*erroringPooler) Get(context.Context) (*pool.Conn, error) {
+func (p *erroringPooler) Get(context.Context) (*pool.Conn, error) {
+	p.gets.Add(1)
 	return nil, errors.New("csc test: pool get always fails")
 }
 
-// TestCSCRefresherStopDrainBailsOnFirstFailure pins #3965 F2: the Close stop-drain
-// must BAIL after the first failed refresh chunk rather than give each of many
-// chunks a fresh cscRefreshBatchTimeout against a dead/stalled server (which could
-// delay Close by minutes). With several chunks queued and every refresh failing,
-// exactly ONE chunk is attempted.
-func TestCSCRefresherStopDrainBailsOnFirstFailure(t *testing.T) {
+// TestCSCRefresherStopDrainSkipsRefetchEntirely pins #3989 P1: the Close
+// stop-drain must not spend ANY network round trip on the buffered backlog — a
+// refetch here is always discarded a moment later by the teardown that follows
+// (invalidateAllCoverage evicts every entry this connection would publish), so
+// attempting it only delays Close. This supersedes #3965 F2's weaker "bail
+// after the first failed chunk" bound, which never triggered against a HEALTHY
+// server: with several chunks queued and Get never even called, the drain must
+// abandon every target without touching the pool.
+func TestCSCRefresherStopDrainSkipsRefetchEntirely(t *testing.T) {
 	// No cscRefreshWindow mutation: runCSCRefresher runs synchronously below with
 	// stop already signalled, so the drain completes in microseconds against the
 	// failing pool — the default window/recency timers never fire, and mutating the
 	// shared global would risk a -race read against another test's live refresher.
 	lc := NewLocalCache(CacheConfig{MaxEntries: 10000})
+	pooler := &erroringPooler{}
 	c := &baseClient{
 		opt:          &Options{},
 		csc:          lc,
 		cscKeyPrefix: "p:",
-		connPool:     &erroringPooler{},
+		connPool:     pooler,
 	}
 	q := &cscRefreshQueue{
 		ch:       make(chan cscRefreshTarget, 1024),
@@ -43,8 +53,8 @@ func TestCSCRefresherStopDrainBailsOnFirstFailure(t *testing.T) {
 	h := &cscRevalidateHandle{stop: make(chan struct{}), done: make(chan struct{})}
 
 	// More than one chunk's worth (cscRefreshBatchMax) but within one drainQueue
-	// round (< cscRefreshWindowMaxKeys) so the whole backlog is flushed once; a
-	// buggy drain would then attempt every chunk.
+	// round (< cscRefreshWindowMaxKeys) so the whole backlog is flushed at once; a
+	// regression that still refetches would then attempt multiple chunks.
 	const chunks = 3
 	const targets = chunks * cscRefreshBatchMax
 	for i := 0; i < targets; i++ {
@@ -53,13 +63,21 @@ func TestCSCRefresherStopDrainBailsOnFirstFailure(t *testing.T) {
 	}
 
 	// Signal stop up front, then run the refresher synchronously: it takes the
-	// stop-drain path, flushes the backlog, and returns.
+	// stop-drain path, abandons the backlog, and returns.
 	close(h.stop)
 	c.runCSCRefresher(h, lc, q)
 
-	if got := q.refreshFailed.Load(); got != 1 {
-		t.Fatalf("refreshFailed = %d; want 1 — the stop-drain must bail after the first "+
-			"failed chunk, not attempt all %d chunks (#3965 F2)", got, chunks)
+	if got := pooler.gets.Load(); got != 0 {
+		t.Fatalf("pool.Get calls = %d; want 0 — the stop-drain must not attempt any "+
+			"refetch at all, not even the first chunk (#3989 P1)", got)
+	}
+	if got := q.refreshed.Load(); got != 0 {
+		t.Fatalf("refreshed = %d; want 0 — no refetch ran", got)
+	}
+	if got := q.refreshFailed.Load(); got != 0 {
+		t.Fatalf("refreshFailed = %d; want 0 — these targets were abandoned by choice, "+
+			"not by a failed round trip, so the round-trip-failure metric must stay "+
+			"untouched (its doc: \"counts refresh round trips that errored\")", got)
 	}
 }
 

csc_test.go

@@ -3863,6 +3863,119 @@ func TestCSCSelfDisableStopsRefresherAndCoalescer(t *testing.T) {
 	}
 }
 
+// TestCSCConstructionRaceDeclinesToStartAfterTeardown pins the fix for a bot
+// review finding: attachSharedTrackingCSC starts the drainer BEFORE assigning
+// cscRefreshHandle/cscMissCoalescer, so stopCSCRefresherAndCoalescer's
+// one-time workersStopOnce body can in principle run (via the drainer's own
+// self-disable defer reacting to an async conn init's disableCSCServing)
+// WHILE startCSCRefresher/startCSCMissCoalescer are still constructing.
+// workersStopOnce alone only guarantees the teardown body runs once — not
+// that it runs after a worker exists for it to stop. Forcing teardown to run
+// FIRST (deterministic, no goroutine timing needed) must make both start
+// functions decline to publish anything, rather than create a goroutine
+// nothing will ever stop again.
+func TestCSCConstructionRaceDeclinesToStartAfterTeardown(t *testing.T) {
+	lc := NewLocalCache(CacheConfig{MaxEntries: 16})
+	dh := &cscDrainHandle{stop: make(chan struct{}), done: make(chan struct{})}
+	active := &atomic.Bool{}
+	active.Store(true)
+	c := &baseClient{
+		opt: &Options{
+			ClientSideCacheRefreshOnInvalidate: true,
+			ClientSideCacheCoalesceMisses:      true,
+		},
+		csc:            lc,
+		cscKeyPrefix:   "p:",
+		cscDrainHandle: dh,
+		cscActive:      active,
+	}
+
+	// Set ONLY workersTornDown, bypassing the rest of stopCSCRefresherAndCoalescer's
+	// body — cscActive stays TRUE. This isolates the new check: if it were removed,
+	// both start functions' pre-existing "cscActive already false" fast path would
+	// not fire either (cscActive is true here), so a pass can only be explained by
+	// the workersMu/workersTornDown check actually running.
+	dh.workersMu.Lock()
+	dh.workersTornDown = true
+	dh.workersMu.Unlock()
+
+	// A worker that starts AFTER losing the race must decline entirely: no
+	// queue/handle/coalescer published, no goroutine leaked past Close().
+	c.startCSCRefresher()
+	if c.cscRefreshQueue != nil || c.cscRefreshHandle != nil {
+		t.Fatal("startCSCRefresher published a queue/handle after teardown already ran " +
+			"(workersStopOnce can never be Do'd again to stop it — leaked goroutine)")
+	}
+	c.startCSCMissCoalescer()
+	if c.cscMissCoalescer.Load() != nil {
+		t.Fatal("startCSCMissCoalescer published a coalescer after teardown already ran " +
+			"(workersStopOnce can never be Do'd again to stop it — leaked goroutine holding a pool conn)")
+	}
+}
+
+// TestCSCStopCSCRefresherAndCoalescerSetsWorkersTornDown pins the plumbing
+// TestCSCConstructionRaceDeclinesToStartAfterTeardown assumes: the real
+// teardown entry point actually sets the flag the start functions check, not
+// just a hand-set test double.
+func TestCSCStopCSCRefresherAndCoalescerSetsWorkersTornDown(t *testing.T) {
+	dh := &cscDrainHandle{stop: make(chan struct{}), done: make(chan struct{})}
+	active := &atomic.Bool{}
+	active.Store(true)
+	c := &baseClient{cscDrainHandle: dh, cscActive: active}
+
+	c.stopCSCRefresherAndCoalescer()
+
+	dh.workersMu.Lock()
+	tornDown := dh.workersTornDown
+	dh.workersMu.Unlock()
+	if !tornDown {
+		t.Fatal("stopCSCRefresherAndCoalescer did not set workersTornDown")
+	}
+	if active.Load() {
+		t.Fatal("stopCSCRefresherAndCoalescer did not deactivate cscActive")
+	}
+}
+
+// TestCSCConstructionWinsRaceStartsNormally is the control for
+// TestCSCConstructionRaceDeclinesToStartAfterTeardown: absent a concurrent
+// teardown, both start functions must publish and launch exactly as before —
+// the fix only changes behavior on the losing side of the race.
+//
+// startCSCMissCoalescer launches real fullDuplexLoop goroutines against a
+// baseClient with a nil connPool. That's safe only because those goroutines
+// select on mc.stop/mc.ch before ever touching the pool, and this test's mc.ch
+// never receives — stopCSCRefresherAndCoalescer below signals stop first. If a
+// future change pre-feeds cscMissCoalescer.ch here, this becomes a nil-pointer
+// panic on the pool, not a race-detector finding.
+func TestCSCConstructionWinsRaceStartsNormally(t *testing.T) {
+	lc := NewLocalCache(CacheConfig{MaxEntries: 16})
+	dh := &cscDrainHandle{stop: make(chan struct{}), done: make(chan struct{})}
+	active := &atomic.Bool{}
+	active.Store(true)
+	c := &baseClient{
+		opt: &Options{
+			ClientSideCacheRefreshOnInvalidate: true,
+			ClientSideCacheCoalesceMisses:      true,
+		},
+		csc:            lc,
+		cscKeyPrefix:   "p:",
+		cscDrainHandle: dh,
+		cscActive:      active,
+	}
+
+	c.startCSCRefresher()
+	if c.cscRefreshQueue == nil || c.cscRefreshHandle == nil {
+		t.Fatal("startCSCRefresher did not publish when no teardown raced it")
+	}
+	c.startCSCMissCoalescer()
+	if c.cscMissCoalescer.Load() == nil {
+		t.Fatal("startCSCMissCoalescer did not publish when no teardown raced it")
+	}
+
+	// Clean up what was started so the test does not leak goroutines.
+	c.stopCSCRefresherAndCoalescer()
+}
+
 // TestCSCTeardownReleasesCoalescerBeforeRefresher pins F3's ordering: teardown must
 // stop the coalescer (releasing its held pool connection) BEFORE it signals and
 // joins the refresher, whose stop-drain flush needs a main-pool connection. It also