adapter: Fence Redis reads before snapshots

adapter/redis_lists.go

@@ -713,6 +713,14 @@ func (r *RedisServer) rangeList(ctx context.Context, key []byte, startRaw, endRa
 		return r.proxyLRange(key, startRaw, endRaw)
 	}
 
+	// 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)
+	}
+
 	readTS := r.readTS()
 	typ, err := r.keyTypeAt(ctx, key, readTS)
 	if err != nil {
@@ -725,14 +733,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

adapter/redis_txn.go

@@ -38,6 +38,33 @@ var txnApplyHandlers = map[string]txnCommandHandler{
 	cmdPExpire: (*txnContext).applyExpireMilliseconds,
 }
 
+func (r *RedisServer) verifyQueuedCommandLeaders(ctx context.Context, queue []redcon.Command) error {
+	if r.coordinator == nil {
+		return nil
+	}
+	seen := make(map[string]struct{}, len(queue))
+	for _, cmd := range queue {
+		if len(cmd.Args) == 0 {
+			continue
+		}
+		meta, ok := redisCommandTable[strings.ToUpper(string(cmd.Args[0]))]
+		if !ok {
+			continue
+		}
+		for _, key := range redisCommandGetKeys(meta, cmd.Args) {
+			keyID := string(key)
+			if _, ok := seen[keyID]; ok {
+				continue
+			}
+			seen[keyID] = struct{}{}
+			if err := r.coordinator.VerifyLeaderForKey(ctx, key); err != nil {
+				return errors.WithStack(err)
+			}
+		}
+	}
+	return nil
+}
+
 // MULTI/EXEC/DISCARD handling
 func (r *RedisServer) multi(conn redcon.Conn, _ redcon.Command) {
 	state := getConnState(conn)
@@ -2379,6 +2406,10 @@ func (r *RedisServer) runTransactionDirect(queue []redcon.Command) ([]redisResul
 
 	var results []redisResult
 	err := r.retryRedisWrite(dispatchCtx, func() error {
+		if err := r.verifyQueuedCommandLeaders(dispatchCtx, queue); err != nil {
+			return err
+		}
+
 		startTS := r.txnStartTS()
 		readPin := r.pinReadTS(startTS)
 		defer readPin.Release()
@@ -2593,6 +2624,10 @@ func (r *RedisServer) runTransactionWithDedup(queue []redcon.Command) ([]redisRe
 // from runTransactionWithDedup to keep that loop under the cyclop
 // budget; the dedup rationale lives there.
 func (r *RedisServer) firstExecAttempt(dispatchCtx context.Context, queue []redcon.Command) ([]redisResult, *reusableExecTxn, error) {
+	if err := r.verifyQueuedCommandLeaders(dispatchCtx, queue); err != nil {
+		return nil, nil, err
+	}
+
 	startTS := r.txnStartTS()
 	readPin := r.pinReadTS(startTS)
 	defer readPin.Release()

adapter/redis_txn_test.go

@@ -42,6 +42,34 @@ func newRedisTxnTestContext(server *RedisServer) *txnContext {
 	}
 }
 
+type verifyHookCoordinator struct {
+	*localAdapterCoordinator
+	verifyForKey func(context.Context, []byte) error
+	verifyCalls  int
+}
+
+func newVerifyHookCoordinator(st store.MVCCStore) *verifyHookCoordinator {
+	return &verifyHookCoordinator{localAdapterCoordinator: newLocalAdapterCoordinator(st)}
+}
+
+func (c *verifyHookCoordinator) VerifyLeaderForKey(ctx context.Context, key []byte) error {
+	c.verifyCalls++
+	if c.verifyForKey != nil {
+		return c.verifyForKey(ctx, key)
+	}
+	return c.localAdapterCoordinator.VerifyLeaderForKey(ctx, key)
+}
+
+func seedRedisListAt(t *testing.T, st store.MVCCStore, key []byte, ts uint64, values ...string) {
+	t.Helper()
+	metaBytes, err := store.MarshalListMeta(store.ListMeta{Len: int64(len(values))})
+	require.NoError(t, err)
+	require.NoError(t, st.PutAt(context.Background(), store.ListMetaKey(key), metaBytes, ts, 0))
+	for i, value := range values {
+		require.NoError(t, st.PutAt(context.Background(), listItemKey(key, int64(i)), []byte(value), ts, 0))
+	}
+}
+
 func elemKeysContain(elems []*kv.Elem[kv.OP], want []byte) bool {
 	for _, elem := range elems {
 		if elem != nil && string(elem.Key) == string(want) {
@@ -117,6 +145,75 @@ func requireReadKeysMatch(t *testing.T, got [][]byte, want [][]byte) {
 	require.Equal(t, wantSet, gotSet)
 }
 
+func TestRedisRangeListVerifiesLeaderBeforeSnapshot(t *testing.T) {
+	t.Parallel()
+
+	st := store.NewMVCCStore()
+	key := []byte("list:leader-fence-lrange")
+	coord := newVerifyHookCoordinator(st)
+	server := NewRedisServer(nil, "", st, coord, nil, nil)
+
+	seeded := false
+	coord.verifyForKey = func(_ context.Context, got []byte) error {
+		require.Equal(t, key, got)
+		if !seeded {
+			seedRedisListAt(t, st, key, 10, "v1")
+			seeded = true
+		}
+		return nil
+	}
+
+	got, err := server.rangeList(context.Background(), key, []byte("0"), []byte("-1"))
+	require.NoError(t, err)
+	require.Equal(t, []string{"v1"}, got)
+	require.Equal(t, 1, coord.verifyCalls)
+}
+
+func TestRedisExecVerifiesLeaderBeforeSnapshot(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name  string
+		dedup bool
+	}{
+		{name: "direct", dedup: false},
+		{name: "dedup", dedup: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			st := store.NewMVCCStore()
+			key := []byte("list:leader-fence-exec:" + tc.name)
+			coord := newVerifyHookCoordinator(st)
+			server := &RedisServer{
+				store:            st,
+				coordinator:      coord,
+				scriptCache:      map[string]string{},
+				onePhaseTxnDedup: tc.dedup,
+			}
+
+			seeded := false
+			coord.verifyForKey = func(_ context.Context, got []byte) error {
+				require.Equal(t, key, got)
+				if !seeded {
+					seedRedisListAt(t, st, key, 10, "v1")
+					seeded = true
+				}
+				return nil
+			}
+
+			results, err := server.runTransaction([]redcon.Command{{
+				Args: [][]byte{[]byte(cmdLRange), key, []byte("0"), []byte("-1")},
+			}})
+			require.NoError(t, err)
+			require.Len(t, results, 1)
+			require.Equal(t, resultArray, results[0].typ)
+			require.Equal(t, []string{"v1"}, results[0].arr)
+			require.Equal(t, 1, coord.verifyCalls)
+		})
+	}
+}
+
 // TestRedisTxnValidateReadSet_ConcurrentRPushTriggersConflict verifies that a
 // concurrent RPUSH to a list triggers an OCC read-write conflict for a MULTI
 // transaction that read the list via LRANGE.  Without the boundary key tracking