adapter: Fence Redis reads before snapshots

adapter/redis_lists.go

@@ -709,11 +709,14 @@ 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)
+	readTS, readPin, proxied, ok, err := r.fenceRangeListReadGroups(ctx, key, startRaw, endRaw)
+	if err != nil {
+		return nil, err
+	} else if ok {
+		return proxied, nil
 	}
+	defer readPin.Release()
 
-	readTS := r.readTS()
 	typ, err := r.keyTypeAt(ctx, key, readTS)
 	if err != nil {
 		return nil, err
@@ -725,14 +728,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,6 +744,23 @@ 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.redisTxnReadFenceKeys(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)
 

adapter/redis_proxy_leader.go

@@ -50,6 +50,18 @@ func (r *RedisServer) proxyTransactionToLeader(conn redcon.Conn, queue []redcon.
 	if !ok {
 		return
 	}
+	r.proxyTransactionToLeaderAddr(conn, queue, leaderAddr)
+}
+
+func (r *RedisServer) proxyTransactionToLeaderForKey(conn redcon.Conn, routingKey []byte, queue []redcon.Command) {
+	leaderAddr, ok := r.resolveLeaderRedisAddrForKey(conn, routingKey)
+	if !ok {
+		return
+	}
+	r.proxyTransactionToLeaderAddr(conn, queue, leaderAddr)
+}
+
+func (r *RedisServer) proxyTransactionToLeaderAddr(conn redcon.Conn, queue []redcon.Command, leaderAddr string) {
 	cli := r.getOrCreateLeaderClient(leaderAddr)
 
 	ctx, cancel := context.WithTimeout(r.handlerContext(), redisDispatchTimeout)
@@ -81,6 +93,20 @@ func (r *RedisServer) resolveLeaderRedisAddr(conn redcon.Conn) (string, bool) {
 	return leaderAddr, true
 }
 
+func (r *RedisServer) resolveLeaderRedisAddrForKey(conn redcon.Conn, key []byte) (string, bool) {
+	leader := r.coordinator.RaftLeaderForKey(key)
+	if leader == "" {
+		writeRedisError(conn, ErrLeaderNotFound)
+		return "", false
+	}
+	leaderAddr, ok := r.leaderRedis[leader]
+	if !ok || leaderAddr == "" {
+		conn.WriteError(fmt.Sprintf("ERR leader redis address unknown for raft address %s", leader))
+		return "", false
+	}
+	return leaderAddr, true
+}
+
 // execTxPipeline sends queue as a single TxPipelined batch and returns the
 // per-command result handles together with any pipeline-level error.
 func (r *RedisServer) execTxPipeline(ctx context.Context, cli *redis.Client, queue []redcon.Command) ([]*redis.Cmd, error) {
@@ -161,8 +187,8 @@ func writeProxyCmdsResult(conn redcon.Conn, cmds []*redis.Cmd) {
 	}
 }
 
-func (r *RedisServer) proxyLRange(key []byte, startRaw, endRaw []byte) ([]string, error) {
-	leader := r.coordinator.RaftLeaderForKey(key)
+func (r *RedisServer) proxyLRange(key, routingKey []byte, startRaw, endRaw []byte) ([]string, error) {
+	leader := r.coordinator.RaftLeaderForKey(routingKey)
 	if leader == "" {
 		return nil, ErrLeaderNotFound
 	}