adapter: Fence Redis reads before snapshots

adapter/redis_compat_helpers.go

@@ -884,11 +884,11 @@ func (r *RedisServer) snapshotGetAt(key []byte, readTS uint64) ([]byte, error) {
 }
 
 func (r *RedisServer) doGetAt(key []byte, readTS uint64, verify bool) ([]byte, error) {
-	// Leadership is partitioned by the logical user key, so strip the internal
-	// prefix before asking the coordinator.
+	// Leadership is partitioned by the logical user key, but route through a
+	// Redis wrapper so list-internal-shaped user keys remain literal.
 	routingKey := key
 	if userKey := extractRedisInternalUserKey(key); userKey != nil {
-		routingKey = userKey
+		routingKey = redisUserRouteKey(userKey)
 	}
 	if r.coordinator.IsLeaderForKey(routingKey) {
 		if verify {

adapter/redis_compat_types.go

@@ -142,6 +142,7 @@ const (
 	redisHLLPrefix    = "!redis|hll|"
 	redisStreamPrefix = "!redis|stream|"
 	redisTTLPrefix    = "!redis|ttl|"
+	redisRoutePrefix  = "!redis|route|"
 )
 
 var redisInternalPrefixes = []string{
@@ -250,6 +251,12 @@ func redisTTLKey(userKey []byte) []byte {
 	return append([]byte(redisTTLPrefix), userKey...)
 }
 
+// redisUserRouteKey routes a literal Redis user key without recursively
+// decoding bytes that happen to look like another internal storage key.
+func redisUserRouteKey(userKey []byte) []byte {
+	return append([]byte(redisRoutePrefix), userKey...)
+}
+
 func redisExactSetStorageKey(kind string, userKey []byte) []byte {
 	switch kind {
 	case "set":

adapter/redis_delta_compactor.go

@@ -214,7 +214,7 @@ func (c *DeltaCompactor) compactUrgentKey(ctx context.Context, req urgentCompact
 	}()
 	// Use per-key leadership so that in sharded deployments this node compacts
 	// keys for the shards it leads, not just those of the default Raft group.
-	if !c.coord.IsLeaderForKey(req.userKey) {
+	if !c.coord.IsLeaderForKey(redisUserRouteKey(req.userKey)) {
 		return
 	}
 	h := c.handlerByTypeName(req.typeName)
@@ -574,7 +574,7 @@ func (c *DeltaCompactor) buildBatchElems(ctx context.Context, h collectionDeltaH
 		// In sharded deployments IsLeaderForKey returns false for keys whose
 		// shard this node does not lead. Skip those to avoid dispatching a
 		// transaction that the responsible leader will reject.
-		if !c.coord.IsLeaderForKey(userKey) {
+		if !c.coord.IsLeaderForKey(redisUserRouteKey(userKey)) {
 			continue
 		}
 		elems, buildErr := h.buildElems(ctx, userKey, deltaKVs, readTS)

adapter/redis_error_prefix_test.go

@@ -124,7 +124,7 @@ func TestWriteRedisError(t *testing.T) {
 		// instead of :prefix :leader. This covers proxyDBSize / proxyDel /
 		// proxyFlushDatabase / proxyFlushLegacy in redis_proxy.go +
 		// proxyKeys / proxyLRange / proxyRPush / proxyLPush /
-		// leaderClientForKey / resolveLeaderRedisAddr in redis.go — all
+		// leaderClientForRedisUserKey / resolveLeaderRedisAddr in redis.go — all
 		// errors.Newf'd with the same "ERR leader redis address unknown
 		// for %s" prefix.
 		{"leader-address-unknown config-gap is already ERR-prefixed",
@@ -178,6 +178,52 @@ func TestHandleProxyTxnError(t *testing.T) {
 
 }
 
+func TestHandleProxyTxnTerminalExecError(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name string
+		err  error
+		want string
+	}{
+		{
+			name: "ambiguous route change typed",
+			err:  errors.WithStack(errRedisExecRouteChangedAfterAmbiguousAttempt),
+			want: errRedisExecRouteChangedAfterAmbiguousAttempt.Error(),
+		},
+		{
+			name: "ambiguous route change decoded RESP",
+			err:  errors.New(errRedisExecRouteChangedAfterAmbiguousAttempt.Error()),
+			want: errRedisExecRouteChangedAfterAmbiguousAttempt.Error(),
+		},
+		{
+			name: "split leader typed",
+			err:  errors.WithStack(errRedisExecSplitShardLeaders),
+			want: errRedisExecSplitShardLeaders.Error(),
+		},
+		{
+			name: "split leader decoded RESP",
+			err:  errors.New(errRedisExecSplitShardLeaders.Error()),
+			want: errRedisExecSplitShardLeaders.Error(),
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			c := &captureConn{}
+			handled := handleProxyTxnError(c, tc.err)
+			if !handled {
+				t.Fatal("handleProxyTxnError returned false")
+			}
+			if c.lastErr != tc.want {
+				t.Fatalf("last error = %q", c.lastErr)
+			}
+			if c.wroteArray {
+				t.Fatalf("unexpected array reply %d", c.lastArray)
+			}
+		})
+	}
+}
+
 func TestHandleProxyTxnHeavyCommandBusyError(t *testing.T) {
 	t.Parallel()
 	c := &captureConn{}
@@ -222,6 +268,51 @@ func TestHandleProxyTxnCommandError(t *testing.T) {
 		}
 	})
 
+	t.Run("ambiguous route change is promoted to top-level EXEC error", func(t *testing.T) {
+		t.Parallel()
+
+		for _, tc := range []struct {
+			name string
+			err  error
+			want string
+		}{
+			{
+				name: "ambiguous route change typed",
+				err:  errors.WithStack(errRedisExecRouteChangedAfterAmbiguousAttempt),
+				want: errRedisExecRouteChangedAfterAmbiguousAttempt.Error(),
+			},
+			{
+				name: "ambiguous route change decoded RESP",
+				err:  errors.New(errRedisExecRouteChangedAfterAmbiguousAttempt.Error()),
+				want: errRedisExecRouteChangedAfterAmbiguousAttempt.Error(),
+			},
+			{
+				name: "split leader typed",
+				err:  errors.WithStack(errRedisExecSplitShardLeaders),
+				want: errRedisExecSplitShardLeaders.Error(),
+			},
+			{
+				name: "split leader decoded RESP",
+				err:  errors.New(errRedisExecSplitShardLeaders.Error()),
+				want: errRedisExecSplitShardLeaders.Error(),
+			},
+		} {
+			t.Run(tc.name, func(t *testing.T) {
+				t.Parallel()
+				cmd := redis.NewCmd(context.Background(), "SET", "k", "v")
+				cmd.SetErr(tc.err)
+				c := &captureConn{}
+				handled := handleProxyTxnCommandError(c, []*redis.Cmd{cmd})
+				if !handled {
+					t.Fatal("handleProxyTxnCommandError returned false")
+				}
+				if c.lastErr != tc.want {
+					t.Fatalf("last error = %q", c.lastErr)
+				}
+			})
+		}
+	})
+
 }
 
 func TestHandleProxyTxnCommandHeavyCommandBusyError(t *testing.T) {

adapter/redis_lists.go

@@ -265,7 +265,7 @@ func (r *RedisServer) dispatchListPushReuse(ctx context.Context, key []byte, pen
 	// (non-retryable errors escape to the client; pending is then
 	// discarded with the goroutine, so the update is wasted and the
 	// stale value would be misleading if some future caller reads it).
-	if isRetryableRedisTxnErr(dispErr) {
+	if isReusableRedisTxnErr(dispErr) {
 		pending.commitTS = commitTS
 	}
 	return 0, false, errors.WithStack(dispErr)
@@ -440,7 +440,7 @@ func (r *RedisServer) listPushCoreWithDedup(ctx context.Context, key []byte, val
 		// retryRedisWrite's retry predicate; ambiguous errors that escape
 		// to the client are a separate problem space (cross-request
 		// idempotency cache) and out of scope for this design.
-		if isRetryableRedisTxnErr(dispErr) {
+		if isReusableRedisTxnErr(dispErr) {
 			pending = &reusableListPush{
 				ops:      ops,
 				startTS:  startTS,
@@ -709,11 +709,40 @@ func (r *RedisServer) fetchListRange(ctx context.Context, key []byte, meta store
 }
 
 func (r *RedisServer) rangeList(ctx context.Context, key []byte, startRaw, endRaw []byte) ([]string, error) {
-	if !r.coordinator.IsLeaderForKey(key) {
-		return r.proxyLRange(key, startRaw, endRaw)
+	var out []string
+	err := r.retryRedisWrite(ctx, func() error {
+		routeVersion := r.redisReadFenceRouteVersion()
+		readTS, readPin, proxied, ok, err := r.fenceRangeListReadGroups(ctx, key, startRaw, endRaw)
+		if err != nil {
+			return err
+		} else if ok {
+			if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil {
+				return err
+			}
+			out = proxied
+			return nil
+		}
+		defer readPin.Release()
+		if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil {
+			return err
+		}
+		next, err := r.rangeListAt(ctx, key, startRaw, endRaw, readTS)
+		if err != nil {
+			return err
+		}
+		if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil {
+			return err
+		}
+		out = next
+		return nil
+	})
+	if err != nil {
+		return nil, err
 	}
+	return out, nil
+}
 
-	readTS := r.readTS()
+func (r *RedisServer) rangeListAt(ctx context.Context, key []byte, startRaw, endRaw []byte, readTS uint64) ([]string, error) {
 	typ, err := r.keyTypeAt(ctx, key, readTS)
 	if err != nil {
 		return nil, err
@@ -725,14 +754,6 @@ func (r *RedisServer) rangeList(ctx context.Context, key []byte, startRaw, endRa
 		return nil, wrongTypeError()
 	}
 
-	// PR #749 follow-up: pass the per-call dispatch ctx so a stalled
-	// VerifyLeaderForKey honours the caller's deadline rather than the
-	// long-lived handlerContext + verifyLeaderEngineCtx fallback. Same
-	// shape as keys() / FLUSHDB.
-	if err := r.coordinator.VerifyLeaderForKey(ctx, key); err != nil {
-		return nil, errors.WithStack(err)
-	}
-
 	meta, exists, err := r.resolveListMeta(ctx, key, readTS)
 	if err != nil {
 		return nil, err
@@ -749,12 +770,29 @@ func (r *RedisServer) rangeList(ctx context.Context, key []byte, startRaw, endRa
 	return r.fetchListRange(ctx, key, meta, int64(s), int64(e), readTS)
 }
 
+func (r *RedisServer) fenceRangeListReadGroups(ctx context.Context, key []byte, startRaw, endRaw []byte) (uint64, *kv.ActiveTimestampToken, []string, bool, error) {
+	groupKeys := r.redisReadFenceGroupKeys(r.redisTxnReadFenceKeysForRanges(key, redisListReadFenceRanges(key)))
+	proxyKey, ok, err := r.readFenceProxyKey(groupKeys)
+	if err != nil {
+		return 0, nil, nil, false, err
+	}
+	if ok {
+		proxied, err := r.proxyLRange(key, proxyKey, startRaw, endRaw)
+		return 0, nil, proxied, true, err
+	}
+	readTS, readPin, err := r.redisReadFencedTimestamp(ctx, groupKeys, r.readTS)
+	if err != nil {
+		return 0, nil, nil, false, err
+	}
+	return readTS, readPin, nil, false, nil
+}
+
 type listPushFunc func(ctx context.Context, key []byte, values [][]byte) (int64, error)
 type listProxyFunc func(key []byte, values [][]byte) (int64, error)
 
 func (r *RedisServer) listPushCmd(conn redcon.Conn, cmd redcon.Command, pushFn listPushFunc, proxyFn listProxyFunc) {
 	key := cmd.Args[1]
-	if !r.coordinator.IsLeaderForKey(key) {
+	if !r.coordinator.IsLeaderForKey(redisUserRouteKey(key)) {
 		length, err := proxyFn(key, cmd.Args[2:])
 		if err != nil {
 			writeRedisError(conn, err)

adapter/redis_proxy.go

@@ -33,7 +33,7 @@ func (r *RedisServer) proxyDel(keys [][]byte) (int64, error) {
 	// Group keys by leader Redis address.
 	byAddr := make(map[string][]string)
 	for _, k := range keys {
-		leader := r.coordinator.RaftLeaderForKey(k)
+		leader := r.coordinator.RaftLeaderForKey(redisUserRouteKey(k))
 		if leader == "" {
 			return 0, ErrLeaderNotFound
 		}