fix(csc): count coalesced attempt in re-run metrics

Closes the TODO(convergence) in processCached that #3989 left for
whichever of #3989/#4002 merged second (codex on #3989, "Count the
failed coalesced attempt in retry metrics"). With #3989 on master and
merged here, processStartingAt's explicit-start plumbing is on this
branch, so the fix can land.

When a coalesced miss reached a session connection and then failed with
a session/transport error, processCached re-ran it through
processWithRetry with the same startAttempt it was called with, so the
re-run started from zero: the failed coalesced attempt disappeared from
state.attempts (the OTel operation-duration metric), from the error
callback's retry count, and from the retry budget — the re-run got
MaxRetries+1 fresh attempts on top of the one already spent, and an
operation that made two attempts reported one.

The bail branch now bumps startAttempt by one when fetch reported a
serving connection. served is non-nil exactly when the request reached
a session connection: the writer attributes every request of a batch to
its conn before the write, and a pre-queue shed or a session that never
acquired a conn leaves it nil. That is the same "one attempt on that
conn" rule the success branch already applies, so the two branches now
agree. processWithRetry seeds both its reported count and its
retry-budget position from startAttempt and clamps it to MaxRetries, so
the re-run itself can never be disabled.

Test: TestProcessCached_CoalescerBailCarriesAttempt drives processCached
through a coalescer whose "session" is a goroutine that dequeues the
request, attributes it to a pipe connection and settles it with a
session error; the re-run then fails at once on a pooler whose Get
always errors (non-retryable), so processWithRetry runs exactly one
iteration. With MaxRetries=1 the seed is observable: 2 attempts when
the coalesced attempt reached a conn, 1 for a pre-queue wire-budget
shed. Verified against the pre-fix code: the session-conn case reports
1 there and fails; the shed control passes on both.

csc_integration.go

@@ -1514,11 +1514,21 @@ func (c *baseClient) processCached(ctx context.Context, cmd Cmder, state *proces
 			// nothing, so connection blips and wire-budget sheds left the key uncached
 			// and later readers missed).
 			//
-			// TODO(convergence): the re-run starts its retry metrics at attempt 0, so
-			// the coalesced attempt is missing from retry_attempts. Fix by starting the
-			// re-run at attempt 1 via the explicit-start plumbing (processStartingAt)
-			// once the full-duplex autopipeline branch, which introduces it, is on this
-			// branch. See #3989 review thread on coalescer attempt counting.
+			// Carry the coalesced attempt into the re-run's accounting. served is
+			// non-nil exactly when the request reached a session connection (the
+			// writer attributes every request of a batch to its conn before the
+			// write; a pre-queue shed or a session that never acquired a conn leaves
+			// it nil), which is the same "one attempt on that conn" rule the success
+			// branch below applies. processWithRetry seeds both its reported attempt
+			// count and its retry-budget position from startAttempt, so the failed
+			// coalesced attempt shows up in retry_attempts and the error callback, and
+			// the re-run does not get MaxRetries+1 fresh attempts on top of the one
+			// already spent (codex on #3989; needs the explicit-start plumbing from the
+			// full-duplex autopipeline branch, now merged). processWithRetry clamps to
+			// MaxRetries, so this can never disable the re-run itself.
+			if served != nil {
+				startAttempt++
+			}
 			token, shouldFetch = c.csc.Reserve(key, nsRedisKeys)
 			if !shouldFetch {
 				// Another waiter won the re-Reserve race and is fetching. WAIT on it

csc_test.go

@@ -876,6 +876,79 @@ func TestProcessCached_NilHitIsTerminal(t *testing.T) {
 	}
 }
 
+// TestProcessCached_CoalescerBailCarriesAttempt pins the convergence follow-up
+// from codex on #3989 ("Count the failed coalesced attempt in retry metrics"):
+// when a coalesced miss reached a session connection and then failed with a
+// session/transport error, the pooled re-run must start at attempt 1, so the
+// coalesced attempt is counted in state.attempts (the OTel duration metric)
+// and consumes one unit of retry budget — instead of processWithRetry
+// restarting from zero and reporting one attempt for an operation that made
+// two. A bail that never reached a session (a pre-queue shed) carries nothing.
+//
+// Deterministic: the "session" is a goroutine that dequeues the request,
+// attributes it to a connection (as the writer does before the write) and
+// settles it with a session error; the re-run then fails at once on a pooler
+// whose Get always errors (non-retryable), so processWithRetry runs exactly
+// one iteration. MaxRetries=1 keeps the seed visible (processWithRetry clamps
+// startAttempt to MaxRetries): 2 attempts when the coalesced attempt reached a
+// conn, 1 otherwise.
+func TestProcessCached_CoalescerBailCarriesAttempt(t *testing.T) {
+	newClient := func() (*baseClient, *cscMissCoalescer) {
+		cache := NewLocalCache(CacheConfig{MaxEntries: 16})
+		c := &baseClient{
+			opt:          &Options{Protocol: 3, MaxRetries: 1},
+			csc:          cache,
+			cscKeyPrefix: cscNamespacePrefix(0, ""),
+			connPool:     &erroringPooler{},
+		}
+		mc := &cscMissCoalescer{c: c, ch: make(chan *cscMissReq, 1), stop: make(chan struct{})}
+		c.cscMissCoalescer.Store(mc)
+		return c, mc
+	}
+	run := func(t *testing.T, c *baseClient) int {
+		t.Helper()
+		ctx := context.Background()
+		cmd := NewStringCmd(ctx, "get", "k")
+		var state processState
+		done := make(chan error, 1)
+		go func() { done <- c.processCached(ctx, cmd, &state, 0) }()
+		select {
+		case err := <-done:
+			if err == nil {
+				t.Fatal("processCached succeeded; the erroring pooler must fail the re-run")
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatal("processCached did not return")
+		}
+		return state.attempts
+	}
+
+	t.Run("reached_session_conn_counts", func(t *testing.T) {
+		c, mc := newClient()
+		server, client := net.Pipe()
+		t.Cleanup(func() { server.Close(); client.Close() })
+		cn := pool.NewConn(client)
+		go func() {
+			req := <-mc.ch
+			req.servedBy = cn // the writer attributes a batch to its conn before the write
+			mc.settleErr(req, errors.New("csc test: session read failed"))
+		}()
+		if got := run(t, c); got != 2 {
+			t.Fatalf("state.attempts = %d; want 2 (the coalesced attempt on the session conn "+
+				"plus the pooled re-run)", got)
+		}
+	})
+
+	t.Run("pre_queue_shed_carries_nothing", func(t *testing.T) {
+		c, mc := newClient()
+		// Saturate the wire budget so fetch sheds before queueing: served == nil.
+		mc.wireBytes.Store(cscMissWireBudgetBytes)
+		if got := run(t, c); got != 1 {
+			t.Fatalf("state.attempts = %d; want 1 (a shed never reached a connection)", got)
+		}
+	})
+}
+
 func TestProcessCached_RecordsCacheHitDuration(t *testing.T) {
 	cache := NewLocalCache(CacheConfig{MaxEntries: 16})
 	client := NewClient(&Options{