fix(proxy): retry secondary writes on "read timestamp has been compacted"

EVALSHA replays to the ElasticKV secondary can surface
FailedPrecondition / "read timestamp has been compacted" when the
script's startTS falls behind a peer node's MinRetainedTS (the local
readPin only protects the node that picked the timestamp). Each retry
re-sends the command so the secondary re-selects a fresh read snapshot.

Author: claude

proxy/dualwrite.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"strings"
 	"sync"
 	"time"
 
@@ -25,8 +26,30 @@ const (
 	// contention bounded; this is only tolerable in modes where the script write
 	// is targeting the non-authoritative backend.
 	maxScriptWriteGoroutines = 64
+
+	// maxCompactedRetries caps retries when the secondary returns
+	// "read timestamp has been compacted". Each attempt re-sends the command so
+	// the secondary re-selects a fresh read snapshot; a small bound is enough
+	// because the compaction waterline advances slowly relative to SecondaryTimeout.
+	maxCompactedRetries = 3
+	// compactedRetryInitialBackoff is the first delay before retrying a secondary
+	// command that failed with a compacted-read error.
+	compactedRetryInitialBackoff = 10 * time.Millisecond
 )
 
+// readTSCompactedMarker is the substring produced by
+// store.ErrReadTSCompacted as it flows through gRPC (wrapped as
+// FailedPrecondition) and Lua PCall. Matching on substring is necessary
+// because both layers erase the typed error.
+const readTSCompactedMarker = "read timestamp has been compacted"
+
+func isReadTSCompactedError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(err.Error(), readTSCompactedMarker)
+}
+
 // DualWriter routes commands to primary and secondary backends based on mode.
 type DualWriter struct {
 	primary   Backend
@@ -225,14 +248,7 @@ func (d *DualWriter) writeSecondary(cmd string, iArgs []any) {
 	defer cancel()
 
 	start := time.Now()
-	result := d.secondary.Do(sCtx, iArgs...)
-	_, sErr := result.Result()
-	if isNoScriptError(sErr) {
-		if fallbackArgs, ok := d.evalFallbackArgs(cmd, iArgs); ok {
-			result = d.secondary.Do(sCtx, fallbackArgs...)
-			_, sErr = result.Result()
-		}
-	}
+	sErr := d.executeSecondary(sCtx, cmd, iArgs)
 	d.metrics.CommandDuration.WithLabelValues(cmd, d.secondary.Name()).Observe(time.Since(start).Seconds())
 
 	if sErr != nil && !errors.Is(sErr, redis.Nil) {
@@ -248,6 +264,38 @@ func (d *DualWriter) writeSecondary(cmd string, iArgs []any) {
 	d.metrics.CommandTotal.WithLabelValues(cmd, d.secondary.Name(), "ok").Inc()
 }
 
+// executeSecondary sends the command to the secondary, handling the NOSCRIPT
+// → EVAL fallback and transparently retrying when the secondary reports that
+// 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.
+func (d *DualWriter) executeSecondary(sCtx context.Context, cmd string, iArgs []any) error {
+	backoff := compactedRetryInitialBackoff
+	var sErr error
+	for attempt := 0; ; attempt++ {
+		result := d.secondary.Do(sCtx, iArgs...)
+		_, sErr = result.Result()
+		if isNoScriptError(sErr) {
+			if fallbackArgs, ok := d.evalFallbackArgs(cmd, iArgs); ok {
+				result = d.secondary.Do(sCtx, fallbackArgs...)
+				_, sErr = result.Result()
+			}
+		}
+		if !isReadTSCompactedError(sErr) {
+			return sErr
+		}
+		if attempt >= maxCompactedRetries {
+			return sErr
+		}
+		select {
+		case <-sCtx.Done():
+			return sErr
+		case <-time.After(backoff):
+		}
+		backoff *= 2
+	}
+}
+
 // goWrite launches fn in a bounded write goroutine.
 func (d *DualWriter) goWrite(fn func()) {
 	d.goAsyncWithSem(d.writeSem, fn)

proxy/proxy_test.go

@@ -920,6 +920,60 @@ func TestDualWriter_Script_EvalSHARO_FallsBackToEvalRO(t *testing.T) {
 	assert.InDelta(t, 0, testutil.ToFloat64(metrics.SecondaryWriteErrors), 0.001)
 }
 
+func TestDualWriter_writeSecondary_RetriesReadTSCompacted(t *testing.T) {
+	// Simulate ElasticKV returning "read timestamp has been compacted"
+	// (wrapped as gRPC FailedPrecondition and surfaced through Lua PCall).
+	// The proxy must transparently retry so the async EVALSHA replay
+	// eventually succeeds with a fresher snapshot timestamp.
+	primary := newMockBackend("primary")
+	primary.doFunc = makeCmd("OK", nil)
+
+	secondary := newMockBackend("secondary")
+	compactedErr := testRedisErr("<string>:71: rpc error: code = FailedPrecondition desc = rpc error: code = FailedPrecondition desc = read timestamp has been compacted stack traceback: \t[G]: in function 'rcall'")
+	var calls int
+	secondary.doFunc = func(ctx context.Context, args ...any) *redis.Cmd {
+		calls++
+		cmd := redis.NewCmd(ctx, args...)
+		if calls < 3 {
+			cmd.SetErr(compactedErr)
+			return cmd
+		}
+		cmd.SetVal("OK")
+		return cmd
+	}
+
+	metrics := newTestMetrics()
+	d := NewDualWriter(primary, secondary, ProxyConfig{Mode: ModeDualWrite, SecondaryTimeout: time.Second}, metrics, newTestSentry(), testLogger)
+
+	d.writeSecondary("EVALSHA", []any{[]byte("EVALSHA"), []byte("deadbeef"), []byte("0")})
+
+	assert.Equal(t, 3, calls, "secondary must be retried until it succeeds")
+	assert.InDelta(t, 0, testutil.ToFloat64(metrics.SecondaryWriteErrors), 0.001,
+		"a retried success must not count as a secondary write error")
+}
+
+func TestDualWriter_writeSecondary_ReadTSCompactedRetriesAreBounded(t *testing.T) {
+	// When the compacted error is persistent, the retry loop must stop after
+	// maxCompactedRetries+1 attempts so the secondary goroutine returns
+	// instead of burning a scriptSem slot indefinitely.
+	primary := newMockBackend("primary")
+	primary.doFunc = makeCmd("OK", nil)
+
+	secondary := newMockBackend("secondary")
+	compactedErr := testRedisErr("rpc error: code = FailedPrecondition desc = read timestamp has been compacted")
+	secondary.doFunc = makeCmd(nil, compactedErr)
+
+	metrics := newTestMetrics()
+	d := NewDualWriter(primary, secondary, ProxyConfig{Mode: ModeDualWrite, SecondaryTimeout: time.Second}, metrics, newTestSentry(), testLogger)
+
+	d.writeSecondary("EVALSHA", []any{[]byte("EVALSHA"), []byte("deadbeef"), []byte("0")})
+
+	assert.Equal(t, maxCompactedRetries+1, secondary.CallCount(),
+		"secondary must stop after maxCompactedRetries+1 attempts")
+	assert.InDelta(t, 1, testutil.ToFloat64(metrics.SecondaryWriteErrors), 0.001,
+		"a persistent compacted error must still be reported as a secondary write error")
+}
+
 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.