proxy: reduce secondary replay mismatches

Author: bootjp

proxy/backend.go

@@ -14,6 +14,7 @@ const (
 	defaultDialTimeout  = 5 * time.Second
 	defaultReadTimeout  = 3 * time.Second
 	defaultWriteTimeout = 3 * time.Second
+	respProtocolV2      = 2
 )
 
 // Backend abstracts a Redis-protocol endpoint (real Redis or ElasticKV).
@@ -72,6 +73,7 @@ func NewRedisBackendWithOptions(addr string, name string, opts BackendOptions) *
 			Addr:         addr,
 			DB:           opts.DB,
 			Password:     opts.Password,
+			Protocol:     respProtocolV2,
 			PoolSize:     opts.PoolSize,
 			DialTimeout:  opts.DialTimeout,
 			ReadTimeout:  opts.ReadTimeout,

proxy/dualwrite.go

@@ -32,9 +32,11 @@ type DualWriter struct {
 	writeSem  chan struct{} // bounds concurrent secondary write goroutines
 	shadowSem chan struct{} // bounds concurrent shadow read goroutines
 
-	wg     sync.WaitGroup
-	mu     sync.Mutex // protects closed; held briefly to make wg.Add atomic with close check
-	closed bool
+	wg       sync.WaitGroup
+	mu       sync.Mutex // protects closed; held briefly to make wg.Add atomic with close check
+	closed   bool
+	scriptMu sync.RWMutex
+	scripts  map[string]string
 }
 
 // NewDualWriter creates a DualWriter with the given backends.
@@ -48,6 +50,7 @@ func NewDualWriter(primary, secondary Backend, cfg ProxyConfig, metrics *ProxyMe
 		logger:    logger,
 		writeSem:  make(chan struct{}, maxWriteGoroutines),
 		shadowSem: make(chan struct{}, maxShadowGoroutines),
+		scripts:   make(map[string]string),
 	}
 
 	if cfg.Mode == ModeDualWriteShadow || cfg.Mode == ModeElasticKVPrimary {
@@ -190,6 +193,7 @@ func (d *DualWriter) Script(ctx context.Context, cmd string, args [][]byte) (any
 	result := d.primary.Do(ctx, iArgs...)
 	resp, err := result.Result()
 	d.metrics.CommandDuration.WithLabelValues(cmd, d.primary.Name()).Observe(time.Since(start).Seconds())
+	d.rememberScript(cmd, args)
 
 	if err != nil && !errors.Is(err, redis.Nil) {
 		d.metrics.CommandTotal.WithLabelValues(cmd, d.primary.Name(), "error").Inc()
@@ -211,6 +215,12 @@ func (d *DualWriter) writeSecondary(cmd string, iArgs []any) {
 	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()
+		}
+	}
 	d.metrics.CommandDuration.WithLabelValues(cmd, d.secondary.Name()).Observe(time.Since(start).Seconds())
 
 	if sErr != nil && !errors.Is(sErr, redis.Nil) {

proxy/proxy_test.go

@@ -768,6 +768,15 @@ func TestDefaultBackendOptions(t *testing.T) {
 	assert.Equal(t, 5*time.Second, opts.DialTimeout)
 }
 
+func TestNewRedisBackend_UsesRESP2(t *testing.T) {
+	backend := NewRedisBackend("127.0.0.1:6379", "test")
+	t.Cleanup(func() {
+		assert.NoError(t, backend.Close())
+	})
+
+	assert.Equal(t, respProtocolV2, backend.client.Options().Protocol)
+}
+
 // ========== Pipeline error handling tests ==========
 
 func TestPipeline_TransportError(t *testing.T) {
@@ -781,6 +790,45 @@ func TestPipeline_TransportError(t *testing.T) {
 	assert.Len(t, results, 3)
 }
 
+func TestDualWriter_Script_CachesEvalForEvalSHAFallback(t *testing.T) {
+	primary := newMockBackend("primary")
+	primary.doFunc = makeCmd("OK", nil)
+
+	secondary := newMockBackend("secondary")
+	script := "return ARGV[1]"
+	sha := scriptSHA(script)
+	var calls int
+	secondary.doFunc = func(ctx context.Context, args ...any) *redis.Cmd {
+		calls++
+		cmd := redis.NewCmd(ctx, args...)
+		switch calls {
+		case 1:
+			assert.Equal(t, []byte("EVALSHA"), args[0])
+			assert.Equal(t, []byte(sha), args[1])
+			cmd.SetErr(testRedisErr("NOSCRIPT No matching script. Please use EVAL."))
+		case 2:
+			assert.Equal(t, []byte("EVAL"), args[0])
+			assert.Equal(t, []byte(script), args[1])
+			cmd.SetVal("OK")
+		default:
+			t.Fatalf("unexpected secondary call %d", calls)
+		}
+		return cmd
+	}
+
+	metrics := newTestMetrics()
+	d := NewDualWriter(primary, secondary, ProxyConfig{Mode: ModeRedisOnly, SecondaryTimeout: time.Second}, metrics, newTestSentry(), testLogger)
+
+	_, err := d.Script(context.Background(), "EVAL", [][]byte{[]byte("EVAL"), []byte(script), []byte("0"), []byte("value")})
+	assert.NoError(t, err)
+
+	d.cfg.Mode = ModeDualWrite
+	d.writeSecondary("EVALSHA", []any{[]byte("EVALSHA"), []byte(sha), []byte("0"), []byte("value")})
+
+	assert.Equal(t, 2, calls)
+	assert.InDelta(t, 0, testutil.ToFloat64(metrics.SecondaryWriteErrors), 0.001)
+}
+
 // ========== writeRedisValue tests ==========
 
 // testRedisErr satisfies the redis.Error interface for testing.

proxy/script_cache.go

@@ -0,0 +1,114 @@
+package proxy
+
+import (
+	"crypto/sha1" // #nosec G505 -- Redis EVALSHA specifies SHA1 script digests.
+	"encoding/hex"
+	"errors"
+	"strings"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	cmdEval    = "EVAL"
+	cmdEvalSHA = "EVALSHA"
+	cmdScript  = "SCRIPT"
+
+	minScriptSubcommandArgs = 2
+	scriptLoadArgIndex      = 2
+	minEvalSHAArgs          = 2
+)
+
+func (d *DualWriter) rememberScript(cmd string, args [][]byte) {
+	upper := strings.ToUpper(cmd)
+
+	switch upper {
+	case cmdEval, "EVAL_RO":
+		if len(args) > 1 {
+			d.storeScript(string(args[1]))
+		}
+	case cmdScript:
+		if len(args) < minScriptSubcommandArgs {
+			return
+		}
+		switch strings.ToUpper(string(args[1])) {
+		case "LOAD":
+			if len(args) > scriptLoadArgIndex {
+				d.storeScript(string(args[scriptLoadArgIndex]))
+			}
+		case "FLUSH":
+			d.clearScripts()
+		}
+	}
+}
+
+func (d *DualWriter) storeScript(script string) {
+	sha := scriptSHA(script)
+
+	d.scriptMu.Lock()
+	defer d.scriptMu.Unlock()
+	d.scripts[sha] = script
+}
+
+func (d *DualWriter) clearScripts() {
+	d.scriptMu.Lock()
+	defer d.scriptMu.Unlock()
+	clear(d.scripts)
+}
+
+func (d *DualWriter) lookupScript(sha string) (string, bool) {
+	d.scriptMu.RLock()
+	defer d.scriptMu.RUnlock()
+	script, ok := d.scripts[strings.ToLower(sha)]
+	return script, ok
+}
+
+func (d *DualWriter) evalFallbackArgs(cmd string, iArgs []any) ([]any, bool) {
+	upper := strings.ToUpper(cmd)
+	if upper != cmdEvalSHA && upper != "EVALSHA_RO" {
+		return nil, false
+	}
+	if len(iArgs) < minEvalSHAArgs {
+		return nil, false
+	}
+
+	sha := stringArg(iArgs[1])
+	script, ok := d.lookupScript(sha)
+	if !ok {
+		return nil, false
+	}
+
+	fallback := make([]any, len(iArgs))
+	fallback[0] = []byte(cmdEval)
+	fallback[1] = []byte(script)
+	copy(fallback[2:], iArgs[2:])
+	return fallback, true
+}
+
+func isNoScriptError(err error) bool {
+	if err == nil {
+		return false
+	}
+	var redisErr redis.Error
+	if errors.As(err, &redisErr) {
+		return strings.HasPrefix(redisErr.Error(), "NOSCRIPT ")
+	}
+	return strings.HasPrefix(err.Error(), "NOSCRIPT ")
+}
+
+func scriptSHA(script string) string {
+	// #nosec G401 -- Redis EVALSHA uses SHA1 digests by protocol.
+	sum := sha1.Sum([]byte(script))
+	return hex.EncodeToString(sum[:])
+}
+
+func stringArg(arg any) string {
+	switch v := arg.(type) {
+	case []byte:
+		return strings.ToLower(string(v))
+	case string:
+		return strings.ToLower(v)
+	default:
+		return strings.ToLower(string(argsToBytes([]any{arg})[0]))
+	}
+}