refactor(proxy): address review feedback on retry and leader discovery

- proxy/dualwrite.go: preserve NOSCRIPT→EVAL fallback across compacted
  retries so each attempt after the first skips the known-missing
  EVALSHA; add jitter + bounded exponential backoff; use time.NewTimer
  so the timer is released on ctx cancellation; tag the two Result()
  reads with the same nolint:wrapcheck used elsewhere.
- proxy/leader_aware_backend.go: derive a cancellable context from
  stopCh so Close() interrupts in-flight INFO probes instead of
  waiting for refreshTimeout, check ctx.Err() between candidates, and
  log per-client Close errors.
- proxy/leader_aware_backend_test.go: replace net.Listen with
  net.ListenConfig.Listen (noctx linter).
- adapter/redis_info_test.go: gofmt.
- proxy/proxy_test.go: regression test that the NOSCRIPT resolution is
  reused across compacted retries.

Author: claude

adapter/redis_info_test.go

@@ -21,12 +21,12 @@ type infoTestCoordinator struct {
 func (c *infoTestCoordinator) Dispatch(context.Context, *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
 	return &kv.CoordinateResponse{}, nil
 }
-func (c *infoTestCoordinator) IsLeader() bool                              { return c.isLeader }
-func (c *infoTestCoordinator) VerifyLeader() error                         { return nil }
-func (c *infoTestCoordinator) RaftLeader() raft.ServerAddress              { return c.raftLeader }
-func (c *infoTestCoordinator) IsLeaderForKey([]byte) bool                  { return c.isLeader }
-func (c *infoTestCoordinator) VerifyLeaderForKey([]byte) error             { return nil }
-func (c *infoTestCoordinator) RaftLeaderForKey([]byte) raft.ServerAddress  { return c.raftLeader }
+func (c *infoTestCoordinator) IsLeader() bool                             { return c.isLeader }
+func (c *infoTestCoordinator) VerifyLeader() error                        { return nil }
+func (c *infoTestCoordinator) RaftLeader() raft.ServerAddress             { return c.raftLeader }
+func (c *infoTestCoordinator) IsLeaderForKey([]byte) bool                 { return c.isLeader }
+func (c *infoTestCoordinator) VerifyLeaderForKey([]byte) error            { return nil }
+func (c *infoTestCoordinator) RaftLeaderForKey([]byte) raft.ServerAddress { return c.raftLeader }
 func (c *infoTestCoordinator) Clock() *kv.HLC {
 	if c.clock == nil {
 		c.clock = kv.NewHLC()

proxy/dualwrite.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"math/rand/v2"
 	"strings"
 	"sync"
 	"time"
@@ -35,6 +36,10 @@ const (
 	// compactedRetryInitialBackoff is the first delay before retrying a secondary
 	// command that failed with a compacted-read error.
 	compactedRetryInitialBackoff = 10 * time.Millisecond
+	// compactedRetryMaxBackoff caps the jittered exponential backoff so it
+	// stays well within SecondaryTimeout even if the retry policy is ever
+	// widened.
+	compactedRetryMaxBackoff = 100 * time.Millisecond
 )
 
 // readTSCompactedMarker is the substring produced by
@@ -269,16 +274,22 @@ func (d *DualWriter) writeSecondary(cmd string, iArgs []any) {
 // the read snapshot has been compacted. A re-sent command causes the backend
 // to re-select a fresh read timestamp, which is the only way to recover once
 // the original startTS has fallen behind MinRetainedTS on a peer node.
+//
+// If the first attempt triggers a NOSCRIPT fallback, subsequent retry
+// attempts use the resolved EVAL args directly so we don't waste a
+// round-trip on the known-missing EVALSHA every iteration.
 func (d *DualWriter) executeSecondary(sCtx context.Context, cmd string, iArgs []any) error {
 	backoff := compactedRetryInitialBackoff
 	var sErr error
+	args := iArgs
 	for attempt := 0; ; attempt++ {
-		result := d.secondary.Do(sCtx, iArgs...)
-		_, sErr = result.Result()
+		result := d.secondary.Do(sCtx, args...)
+		_, sErr = result.Result() //nolint:wrapcheck // classification/error propagation preserves redis.Error types
 		if isNoScriptError(sErr) {
-			if fallbackArgs, ok := d.evalFallbackArgs(cmd, iArgs); ok {
-				result = d.secondary.Do(sCtx, fallbackArgs...)
-				_, sErr = result.Result()
+			if fallbackArgs, ok := d.evalFallbackArgs(cmd, args); ok {
+				args = fallbackArgs
+				result = d.secondary.Do(sCtx, args...)
+				_, sErr = result.Result() //nolint:wrapcheck // classification/error propagation preserves redis.Error types
 			}
 		}
 		if !isReadTSCompactedError(sErr) {
@@ -287,15 +298,49 @@ func (d *DualWriter) executeSecondary(sCtx context.Context, cmd string, iArgs []
 		if attempt >= maxCompactedRetries {
 			return sErr
 		}
-		select {
-		case <-sCtx.Done():
+		if !waitCompactedRetryBackoff(sCtx, backoff) {
 			return sErr
-		case <-time.After(backoff):
 		}
-		backoff *= 2
+		backoff = nextCompactedRetryBackoff(backoff)
 	}
 }
 
+// waitCompactedRetryBackoff sleeps for a jittered interval or returns early
+// when the context is cancelled. Returns false if the caller should abort
+// the retry loop (context done).
+//
+// Jitter is in [backoff, backoff + backoff/2) so that concurrent retries
+// caused by a single compaction waterline advancement do not re-hit the
+// secondary in lockstep. A NewTimer is used instead of time.After so the
+// timer is released promptly on ctx cancellation (avoiding a leak until
+// expiry when the async goroutine is shutting down).
+func waitCompactedRetryBackoff(ctx context.Context, backoff time.Duration) bool {
+	if backoff <= 0 {
+		return ctx.Err() == nil
+	}
+	jitter := time.Duration(0)
+	if half := int64(backoff / 2); half > 0 {
+		jitter = time.Duration(rand.Int64N(half)) //nolint:gosec // jitter for retry backoff, not security sensitive
+	}
+	timer := time.NewTimer(backoff + jitter)
+	defer timer.Stop()
+
+	select {
+	case <-ctx.Done():
+		return false
+	case <-timer.C:
+		return true
+	}
+}
+
+func nextCompactedRetryBackoff(current time.Duration) time.Duration {
+	next := current * 2
+	if next > compactedRetryMaxBackoff {
+		return compactedRetryMaxBackoff
+	}
+	return next
+}
+
 // goWrite launches fn in a bounded write goroutine.
 func (d *DualWriter) goWrite(fn func()) {
 	d.goAsyncWithSem(d.writeSem, fn)

proxy/leader_aware_backend.go

@@ -112,7 +112,17 @@ func normalizeSeeds(seeds []string) []string {
 func (b *LeaderAwareRedisBackend) refreshLoop() {
 	defer close(b.done)
 
-	b.refreshLeader(context.Background())
+	// Derive a cancellable context from stopCh so that an in-flight INFO
+	// probe is interrupted as soon as Close() is called; otherwise the
+	// refreshTimeout must elapse before the loop can exit.
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		<-b.stopCh
+		cancel()
+	}()
+
+	b.refreshLeader(ctx)
 
 	t := time.NewTicker(b.refreshInterval)
 	defer t.Stop()
@@ -121,9 +131,9 @@ func (b *LeaderAwareRedisBackend) refreshLoop() {
 		case <-b.stopCh:
 			return
 		case <-t.C:
-			b.refreshLeader(context.Background())
+			b.refreshLeader(ctx)
 		case <-b.refreshCh:
-			b.refreshLeader(context.Background())
+			b.refreshLeader(ctx)
 		}
 	}
 }
@@ -145,6 +155,9 @@ func (b *LeaderAwareRedisBackend) TriggerRefresh() {
 // healthy, so this converges in one probe during steady state.
 func (b *LeaderAwareRedisBackend) refreshLeader(ctx context.Context) {
 	for _, addr := range b.candidateAddrs() {
+		if ctx.Err() != nil {
+			return
+		}
 		leader, err := b.probeLeader(ctx, addr)
 		if err != nil {
 			b.logger.Debug("leader probe failed", "addr", addr, "err", err)
@@ -331,8 +344,11 @@ func (b *LeaderAwareRedisBackend) Close() error {
 
 	var firstErr error
 	for addr, cli := range clients {
-		if err := cli.Close(); err != nil && firstErr == nil {
-			firstErr = fmt.Errorf("close %s client %s: %w", b.name, addr, err)
+		if err := cli.Close(); err != nil {
+			b.logger.Warn("close leader-aware client failed", "backend", b.name, "addr", addr, "err", err)
+			if firstErr == nil {
+				firstErr = fmt.Errorf("close %s client %s: %w", b.name, addr, err)
+			}
 		}
 	}
 	return firstErr

proxy/leader_aware_backend_test.go

@@ -67,7 +67,8 @@ type fakeElasticKVNode struct {
 
 func newFakeElasticKVNode(t *testing.T, leader string) *fakeElasticKVNode {
 	t.Helper()
-	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	var lc net.ListenConfig
+	ln, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0")
 	require.NoError(t, err)
 	n := &fakeElasticKVNode{ln: ln, addr: ln.Addr().String()}
 	n.SetLeader(leader)

proxy/proxy_test.go

@@ -974,6 +974,51 @@ func TestDualWriter_writeSecondary_ReadTSCompactedRetriesAreBounded(t *testing.T
 		"a persistent compacted error must still be reported as a secondary write error")
 }
 
+func TestDualWriter_writeSecondary_RetriesDoNotRepeatNoScriptProbe(t *testing.T) {
+	// After the EVAL fallback resolves a NOSCRIPT, a compacted-retry must
+	// re-send the resolved EVAL form directly — never the known-missing
+	// EVALSHA — so we never burn an extra round-trip per retry attempt.
+	primary := newMockBackend("primary")
+	primary.doFunc = makeCmd("OK", nil)
+
+	script := "return ARGV[1]"
+	sha := scriptSHA(script)
+	compactedErr := testRedisErr("rpc error: code = FailedPrecondition desc = read timestamp has been compacted")
+
+	secondary := newMockBackend("secondary")
+	var evalshaCalls, evalCalls int
+	secondary.doFunc = func(ctx context.Context, args ...any) *redis.Cmd {
+		cmd := redis.NewCmd(ctx, args...)
+		switch string(args[0].([]byte)) {
+		case "EVALSHA":
+			evalshaCalls++
+			cmd.SetErr(testRedisErr("NOSCRIPT No matching script. Please use EVAL."))
+		case "EVAL":
+			evalCalls++
+			// First resolved EVAL fails with compacted; second succeeds.
+			if evalCalls == 1 {
+				cmd.SetErr(compactedErr)
+			} else {
+				cmd.SetVal("OK")
+			}
+		default:
+			t.Fatalf("unexpected secondary command %v", args[0])
+		}
+		return cmd
+	}
+
+	metrics := newTestMetrics()
+	d := NewDualWriter(primary, secondary, ProxyConfig{Mode: ModeRedisOnly, SecondaryTimeout: time.Second}, metrics, newTestSentry(), testLogger)
+	d.storeScript(script)
+
+	d.cfg.Mode = ModeDualWrite
+	d.writeSecondary("EVALSHA", []any{[]byte("EVALSHA"), []byte(sha), []byte("0"), []byte("value")})
+
+	assert.Equal(t, 1, evalshaCalls, "EVALSHA must be probed only once; the compacted retry must use the resolved EVAL form")
+	assert.Equal(t, 2, evalCalls, "EVAL must be retried after the compacted failure")
+	assert.InDelta(t, 0, testutil.ToFloat64(metrics.SecondaryWriteErrors), 0.001)
+}
+
 func TestDualWriter_Script_NoRememberOnPrimaryError(t *testing.T) {
 	// Verify that a failed SCRIPT FLUSH on the primary does NOT clear the proxy
 	// script cache, so that subsequent EVALSHA → EVAL fallbacks still work.