fix(cluster): freeze slot writes until topology update after migration (#3147)

src/cluster/cluster.cc

@@ -124,19 +124,17 @@ Status Cluster::SetSlotRanges(const std::vector<SlotRange> &slot_ranges, const s
       to_assign_node->slots[slot] = true;
       slots_nodes_[slot] = to_assign_node;
 
-      // Clear data of migrated slot or record of imported slot
+      // Clear data of migrated slot
       if (old_node == myself_ && old_node != to_assign_node) {
-        // If slot is migrated from this node
         if (migrated_slots_.count(slot) > 0) {
           auto s = srv_->slot_migrator->ClearKeysOfSlotRange(ctx, kDefaultNamespace, SlotRange::GetPoint(slot));
           if (!s.ok()) {
             ERROR("failed to clear data of migrated slot: {}", s.ToString());
           }
           migrated_slots_.erase(slot);
-        }
-        // If slot is imported into this node
-        if (imported_slots_.count(slot) > 0) {
-          imported_slots_.erase(slot);
+          if (migrated_slots_.empty()) {
+            srv_->slot_migrator->ReleaseForbiddenSlotRange();
+          }
         }
       }
     }
@@ -206,7 +204,7 @@ Status Cluster::SetClusterNodes(const std::string &nodes_str, int64_t version, b
     return s.Prefixed("failed to set master-replica replication");
   }
 
-  // Clear data of migrated slots
+  // Clear data of migrated slots and drop stale forbidden state.
   if (!migrated_slots_.empty()) {
     engine::Context ctx(srv_->storage);
     for (const auto &[slot, _] : migrated_slots_) {
@@ -217,10 +215,10 @@ Status Cluster::SetClusterNodes(const std::string &nodes_str, int64_t version, b
         }
       }
     }
+    srv_->slot_migrator->ReleaseForbiddenSlotRange();
   }
-  // Clear migrated and imported slot info
+  // Clear migrated slot info
   migrated_slots_.clear();
-  imported_slots_.clear();
 
   return Status::OK();
 }
@@ -278,6 +276,12 @@ Status Cluster::SetMasterSlaveRepl() {
 
 bool Cluster::IsNotMaster() { return myself_ == nullptr || myself_->role != kClusterMaster || srv_->IsSlave(); }
 
+void Cluster::ClearImportingSlotRange() {
+  if (myself_) {
+    myself_->importing_slot_range = {-1, -1};
+  }
+}
+
 Status Cluster::SetSlotRangeMigrated(const SlotRange &slot_range, const std::string &ip_port) {
   if (!slot_range.IsValid()) {
     return {Status::NotOK, errSlotRangeInvalid};
@@ -293,19 +297,6 @@ Status Cluster::SetSlotRangeMigrated(const SlotRange &slot_range, const std::str
   return Status::OK();
 }
 
-Status Cluster::SetSlotRangeImported(const SlotRange &slot_range) {
-  if (!slot_range.IsValid()) {
-    return {Status::NotOK, errSlotRangeInvalid};
-  }
-
-  // It is called by command 'cluster import'. When executing the command, the
-  // exclusive lock has been locked. Therefore, it can't be locked again.
-  for (auto slot = slot_range.start; slot <= slot_range.end; slot++) {
-    imported_slots_.insert(slot);
-  }
-  return Status::OK();
-}
-
 Status Cluster::MigrateSlotRange(const SlotRange &slot_range, const std::string &dst_node_id,
                                  SyncMigrateContext *blocking_ctx) {
   if (nodes_.find(dst_node_id) == nodes_.end()) {
@@ -389,11 +380,13 @@ Status Cluster::ImportSlotRange(redis::Connection *conn, const SlotRange &slot_r
     case kImportSuccess:
       s = srv_->slot_import->Success(slot_range);
       if (!s.IsOK()) return s;
+      ClearImportingSlotRange();
       INFO("[import] Mark the importing slot(s) {} as succeed", slot_range.String());
       break;
     case kImportFailed:
       s = srv_->slot_import->Fail(slot_range);
       if (!s.IsOK()) return s;
+      ClearImportingSlotRange();
       INFO("[import] Mark the importing slot(s) {} as failed", slot_range.String());
       break;
     default:
@@ -914,10 +907,10 @@ Status Cluster::CanExecByMySelf(const redis::CommandAttributes *attributes, cons
 
   if (myself_ && myself_ == slots_nodes_[slot]) {
     // We use central controller to manage the topology of the cluster.
-    // Server can't change the topology directly, so we record the migrated slots
-    // to move the requests of the migrated slots to the destination node.
-    if (migrated_slots_.count(slot) > 0) {  // I'm not serving the migrated slot
-      return {Status::RedisMoved, fmt::format("{} {}", slot, migrated_slots_[slot])};
+    // Server can't change the topology directly, so reject writes to migrated slots
+    // until the controller confirms their owner with a topology update.
+    if (migrated_slots_.count(slot) > 0 && (flags & redis::kCmdWrite)) {
+      return {Status::RedisTryAgain, "Slot migration finished, waiting for topology update"};
     }
     // To keep data consistency, slot will be forbidden write while sending the last incremental data.
     // During this phase, the requests of the migrating slot has to be rejected.
@@ -937,13 +930,6 @@ Status Cluster::CanExecByMySelf(const redis::CommandAttributes *attributes, cons
     return Status::OK();  // I'm serving the importing connection or asking connection
   }
 
-  if (myself_ && imported_slots_.count(slot)) {
-    // After the slot is migrated, new requests of the migrated slot will be moved to
-    // the destination server. Before the central controller change the topology, the destination
-    // server should record the imported slots to accept new data of the imported slots.
-    return Status::OK();  // I'm serving the imported slot
-  }
-
   if (myself_ && myself_->role == kClusterSlave && !(flags & redis::kCmdWrite) &&
       nodes_.find(myself_->master_id) != nodes_.end() && nodes_[myself_->master_id] == slots_nodes_[slot] &&
       conn->IsFlagEnabled(redis::Connection::kReadOnly)) {
@@ -984,7 +970,6 @@ Status Cluster::Reset() {
     n = nullptr;
   }
   migrated_slots_.clear();
-  imported_slots_.clear();
 
   // The migrator's forbidden slot range persists past a successful migration
   // and is only harmless while slots_nodes_[slot] no longer points at us.

src/cluster/cluster.h

@@ -24,7 +24,6 @@
 #include <bitset>
 #include <map>
 #include <memory>
-#include <set>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -77,7 +76,7 @@ class Cluster {
   Status SetNodeId(const std::string &node_id);
   Status SetSlotRanges(const std::vector<SlotRange> &slot_ranges, const std::string &node_id, int64_t version);
   Status SetSlotRangeMigrated(const SlotRange &slot_range, const std::string &ip_port);
-  Status SetSlotRangeImported(const SlotRange &slot_range);
+  void ClearImportingSlotRange();
   Status GetSlotsInfo(std::vector<SlotInfo> *slot_infos);
   Status GetClusterInfo(std::string *cluster_infos);
   int64_t GetVersion() const { return version_; }
@@ -116,5 +115,4 @@ class Cluster {
   std::shared_ptr<ClusterNode> slots_nodes_[kClusterSlots];
 
   std::map<int, std::string> migrated_slots_;
-  std::set<int> imported_slots_;
 };

src/cluster/slot_import.cc

@@ -57,12 +57,6 @@ Status SlotImport::Success(const SlotRange &slot_range) {
     return {Status::NotOK, fmt::format("mismatch slot, importing slot(s): {}, but got: {}", import_slot_range_.String(),
                                        slot_range.String())};
   }
-
-  Status s = srv_->cluster->SetSlotRangeImported(import_slot_range_);
-  if (!s.IsOK()) {
-    return {Status::NotOK, fmt::format("unable to set imported status: {}", slot_range.String())};
-  }
-
   import_status_ = kImportSuccess;
   return Status::OK();
 }
@@ -109,6 +103,9 @@ Status SlotImport::StopForLinkError() {
   }
 
   import_status_ = kImportFailed;
+  if (srv_->cluster) {
+    srv_->cluster->ClearImportingSlotRange();
+  }
   return Status::OK();
 }
 

tests/gocase/integration/slotimport/slotimport_test.go

@@ -129,7 +129,8 @@ func TestImportedServer(t *testing.T) {
 		require.NotContains(t, clusterNodes, fmt.Sprintf("[%d-<-%s]", slotNum, srvAID))
 
 		time.Sleep(50 * time.Millisecond)
-		require.Equal(t, "slot1", rdbB.Get(ctx, slotKey).Val())
+		// After import success, ordinary clients must not be served until topology is updated.
+		util.ErrorRegexp(t, rdbB.Get(ctx, slotKey).Err(), fmt.Sprintf("MOVED %d.*%d.*", slotNum, srvA.Port()))
 	})
 
 	t.Run("IMPORT - slot state 'error'", func(t *testing.T) {

tests/gocase/integration/slotmigrate/slotmigrate_test.go

@@ -153,6 +153,8 @@ func TestSlotMigrateDestServerKilledAgain(t *testing.T) {
 		time.Sleep(500 * time.Millisecond)
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", slot, id1).Val())
 		waitForMigrateState(t, rdb0, slot, SlotMigrationStateSuccess)
+		assignSlotToNode(t, ctx, rdb0, slot, id1)
+		assignSlotToNode(t, ctx, rdb1, slot, id1)
 		require.Equal(t, "slot0", rdb1.Get(ctx, "").Val())
 		require.Equal(t, "", rdb1.Get(ctx, util.SlotTable[slot]).Val())
 		require.NoError(t, rdb1.Del(ctx, util.SlotTable[slot]).Err())
@@ -170,6 +172,8 @@ func TestSlotMigrateDestServerKilledAgain(t *testing.T) {
 		require.NoError(t, rdb0.Set(ctx, k2, "\0000\0001", 0).Err())
 		time.Sleep(time.Second)
 		waitForImportState(t, rdb1, slot, SlotImportStateSuccess)
+		assignSlotToNode(t, ctx, rdb0, slot, id1)
+		assignSlotToNode(t, ctx, rdb1, slot, id1)
 		require.EqualValues(t, cnt, rdb1.LLen(ctx, k1).Val())
 		require.Equal(t, "\0000\0001", rdb1.LPop(ctx, k1).Val())
 		require.Equal(t, "\0000\0001", rdb1.Get(ctx, k2).Val())
@@ -306,6 +310,12 @@ func TestSlotMigrateDisablePersistClusterNodes(t *testing.T) {
 	}
 	require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", slot, id1).Val())
 	waitForMigrateState(t, rdb0, slot, SlotMigrationStateSuccess)
+	// Source still serves reads of frozen data; writes wait for topology update.
+	require.EqualValues(t, cnt, rdb0.LLen(ctx, util.SlotTable[slot]).Val())
+	require.ErrorContains(t, rdb0.Set(ctx, util.SlotTable[slot], "source-value", 0).Err(), "TRYAGAIN")
+	util.ErrorRegexp(t, rdb1.Get(ctx, util.SlotTable[slot]).Err(), fmt.Sprintf("MOVED %d.*%d.*", slot, srv0.Port()))
+	assignSlotToNode(t, ctx, rdb0, slot, id1)
+	assignSlotToNode(t, ctx, rdb1, slot, id1)
 	require.EqualValues(t, cnt, rdb1.LLen(ctx, util.SlotTable[slot]).Val())
 
 	k := fmt.Sprintf("{%s}_1", util.SlotTable[slot])
@@ -346,6 +356,8 @@ func TestSlotMigrateNewNodeAndAuth(t *testing.T) {
 		}
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", slot, id1).Val())
 		waitForMigrateState(t, rdb0, slot, SlotMigrationStateSuccess)
+		assignSlotToNode(t, ctx, rdb0, slot, id1)
+		assignSlotToNode(t, ctx, rdb1, slot, id1)
 		require.EqualValues(t, cnt, rdb1.LLen(ctx, util.SlotTable[slot]).Val())
 
 		k := fmt.Sprintf("{%s}_1", util.SlotTable[slot])
@@ -376,6 +388,8 @@ func TestSlotMigrateNewNodeAndAuth(t *testing.T) {
 		require.NoError(t, rdb0.ConfigSet(ctx, "requirepass", "password").Err())
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", slot, id1).Val())
 		waitForMigrateState(t, rdb0, slot, SlotMigrationStateSuccess)
+		assignSlotToNode(t, ctx, rdb0, slot, id1)
+		assignSlotToNode(t, ctx, rdb1, slot, id1)
 		require.EqualValues(t, 1, rdb1.Exists(ctx, util.SlotTable[slot]).Val())
 		require.EqualValues(t, cnt, rdb1.LLen(ctx, util.SlotTable[slot]).Val())
 	})
@@ -540,6 +554,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 		otherSlot := 2
 		require.ErrorContains(t, rdb0.Do(ctx, "clusterx", "migrate", otherSlot, id1).Err(), "There is already a migrating job")
 		waitForMigrateState(t, rdb0, slot, SlotMigrationStateSuccess)
+		assignSlotToNodeOnClients(t, ctx, slot, id1, rdb0, rdb1)
 		require.EqualValues(t, cnt, rdb1.LLen(ctx, util.SlotTable[slot]).Val())
 	})
 
@@ -625,6 +640,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 		} else {
 			require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", testSlot, id1, "sync").Val())
 		}
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 
 		// check destination data
 		// type string
@@ -704,6 +720,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 		require.NoError(t, rdb0.XDel(ctx, keys["stream"], "1-0").Err())
 		require.NoError(t, rdb0.Do(ctx, "XSETID", keys["stream"], "1001-0", "MAXDELETEDID", "2-0").Err())
 		waitForMigrateStateInDuration(t, rdb0, testSlot, SlotMigrationStateSuccess, time.Minute)
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 
 		streamInfo = rdb1.XInfoStream(ctx, keys["stream"]).Val()
 		require.EqualValues(t, "1001-0", streamInfo.LastGeneratedID)
@@ -741,6 +758,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", testSlot, id1).Val())
 		waitForMigrateState(t, rdb0, testSlot, SlotMigrationStateSuccess)
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 
 		require.ErrorContains(t, rdb0.Exists(ctx, key).Err(), "MOVED")
 
@@ -780,6 +798,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", testSlot, id1).Val())
 		waitForMigrateState(t, rdb0, testSlot, SlotMigrationStateSuccess)
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 
 		require.ErrorContains(t, rdb0.Exists(ctx, key).Err(), "MOVED")
 
@@ -902,6 +921,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 		siv := rdb0.Do(ctx, "SIRANGE", keys[9], 0, -1).Val()
 		waitForMigrateStateInDuration(t, rdb0, migratingSlot, SlotMigrationStateSuccess, time.Minute)
 		waitForImportState(t, rdb1, migratingSlot, SlotImportStateSuccess)
+		assignSlotToNodeOnClients(t, ctx, migratingSlot, id1, rdb0, rdb1)
 		// check if the data is consistent
 		// 1. type string
 		require.EqualValues(t, cnt, rdb1.LLen(ctx, keys[0]).Val())
@@ -964,14 +984,16 @@ func TestSlotMigrateDataType(t *testing.T) {
 		migrateIncrementalData(t)
 	})
 
-	t.Run("MIGRATE - Accessing slot is forbidden on source server but not on destination server", func(t *testing.T) {
+	t.Run("MIGRATE - Writes are blocked until topology changes", func(t *testing.T) {
 		testSlot += 1
 		require.NoError(t, rdb0.Set(ctx, util.SlotTable[testSlot], 3, 0).Err())
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", testSlot, id1).Val())
 		waitForMigrateState(t, rdb0, testSlot, SlotMigrationStateSuccess)
-		require.ErrorContains(t, rdb0.Set(ctx, util.SlotTable[testSlot], "source-value", 0).Err(), "MOVED")
-		require.ErrorContains(t, rdb0.Del(ctx, util.SlotTable[testSlot]).Err(), "MOVED")
-		require.ErrorContains(t, rdb0.Exists(ctx, util.SlotTable[testSlot]).Err(), "MOVED")
+		// Source still serves reads of frozen data; writes wait for topology update.
+		require.Equal(t, "3", rdb0.Get(ctx, util.SlotTable[testSlot]).Val())
+		require.ErrorContains(t, rdb0.Set(ctx, util.SlotTable[testSlot], "source-value", 0).Err(), "TRYAGAIN")
+		util.ErrorRegexp(t, rdb1.Get(ctx, util.SlotTable[testSlot]).Err(), fmt.Sprintf("MOVED %d.*%d.*", testSlot, srv0.Port()))
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 		require.NoError(t, rdb1.Set(ctx, util.SlotTable[testSlot], "destination-value", 0).Err())
 	})
 
@@ -986,6 +1008,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 		// write during migrating
 		require.EqualValues(t, cnt+1, rdb0.LPush(ctx, util.SlotTable[testSlot], cnt).Val())
 		waitForMigrateState(t, rdb0, testSlot, SlotMigrationStateSuccess)
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 		require.Equal(t, strconv.Itoa(cnt), rdb1.LPop(ctx, util.SlotTable[testSlot]).Val())
 	})
 
@@ -994,9 +1017,9 @@ func TestSlotMigrateDataType(t *testing.T) {
 		require.NoError(t, rdb0.Set(ctx, util.SlotTable[testSlot], "slot6", 0).Err())
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", testSlot, id1).Val())
 		waitForMigrateState(t, rdb0, testSlot, SlotMigrationStateSuccess)
-		require.Equal(t, "slot6", rdb1.Get(ctx, util.SlotTable[testSlot]).Val())
+		require.Contains(t, rdb1.Keys(ctx, "*").Val(), util.SlotTable[testSlot])
 		require.Contains(t, rdb0.Keys(ctx, "*").Val(), util.SlotTable[testSlot])
-		require.NoError(t, rdb0.Do(ctx, "clusterx", "setslot", testSlot, "node", id1, "2").Err())
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 		require.NotContains(t, rdb0.Keys(ctx, "*").Val(), util.SlotTable[testSlot])
 	})
 
@@ -1030,6 +1053,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 		}
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", testSlot, id1).Val())
 		waitForMigrateState(t, rdb0, testSlot, SlotMigrationStateSuccess)
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 		require.EqualValues(t, cnt, rdb1.LLen(ctx, util.SlotTable[testSlot]).Val())
 		// write the migrated slot to source server
 		k := fmt.Sprintf("{%s}_1", util.SlotTable[testSlot])
@@ -1069,6 +1093,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 			require.NoError(t, rdb0.LMove(ctx, srcListName, dstListName, "RIGHT", "LEFT").Err())
 		}
 		waitForMigrateStateInDuration(t, rdb0, testSlot, SlotMigrationStateSuccess, time.Minute)
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 
 		require.ErrorContains(t, rdb0.RPush(ctx, srcListName, "element1000").Err(), "MOVED")
 		require.Equal(t, int64(10), rdb1.LLen(ctx, dstListName).Val())
@@ -1110,6 +1135,7 @@ func TestSlotMigrateDataType(t *testing.T) {
 			require.NoError(t, rdb0.LMove(ctx, srcListName, srcListName, "RIGHT", "LEFT").Err())
 		}
 		waitForMigrateStateInDuration(t, rdb0, testSlot, SlotMigrationStateSuccess, time.Minute)
+		assignSlotToNodeOnClients(t, ctx, testSlot, id1, rdb0, rdb1)
 
 		require.ErrorContains(t, rdb0.RPush(ctx, srcListName, "element1000").Err(), "MOVED")
 		require.Equal(t, int64(srcLen), rdb1.LLen(ctx, srcListName).Val())
@@ -1197,6 +1223,7 @@ func TestSlotMigrateCuckooFilter(t *testing.T) {
 		}
 
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", slot, id1, "sync").Val())
+		assignSlotToNodeOnClients(t, ctx, slot, id1, rdb0, rdb1)
 		require.Equal(t, "MBbloomCF", rdb1.Type(ctx, key).Val())
 		require.Equal(t, int64(1), rdb1.Do(ctx, "cf.add", key, "new-item").Val())
 	})
@@ -1250,6 +1277,25 @@ func waitForImportState(t testing.TB, client *redis.Client, n int, state SlotImp
 	}, 10*time.Second, 100*time.Millisecond)
 }
 
+func assignSlotToNode(t testing.TB, ctx context.Context, client *redis.Client, slot int, nodeID string) {
+	t.Helper()
+	assignSlotRangeToNode(t, ctx, client, slot, nodeID)
+}
+
+func assignSlotToNodeOnClients(t testing.TB, ctx context.Context, slot int, nodeID string, clients ...*redis.Client) {
+	t.Helper()
+	for _, client := range clients {
+		assignSlotToNode(t, ctx, client, slot, nodeID)
+	}
+}
+
+func assignSlotRangeToNode(t testing.TB, ctx context.Context, client *redis.Client, slotRange any, nodeID string) {
+	t.Helper()
+	version, err := client.Do(ctx, "clusterx", "version").Int()
+	require.NoError(t, err)
+	require.NoError(t, client.Do(ctx, "clusterx", "setslot", slotRange, "node", nodeID, version+1).Err())
+}
+
 func migrateSlotRangeAndSetSlot(t *testing.T, ctx context.Context, source *redis.Client, dest *redis.Client, destID string, slotRange string) {
 	require.Equal(t, "OK", source.Do(ctx, "clusterx", "migrate", slotRange, destID).Val())
 	waitForMigrateSlotRangeState(t, source, slotRange, SlotMigrationStateSuccess)
@@ -1359,7 +1405,7 @@ func TestSlotRangeMigrate(t *testing.T) {
 		require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate", "112-113", id1).Val())
 		waitForMigrateSlotRangeState(t, rdb0, "112-113", SlotMigrationStateSuccess)
 		for slot := 112; slot <= 118; slot++ {
-			require.ErrorContains(t, rdb0.LPush(ctx, util.SlotTable[slot], 10).Err(), "MOVED")
+			require.ErrorContains(t, rdb0.LPush(ctx, util.SlotTable[slot], 10).Err(), "TRYAGAIN")
 		}
 
 		// overlap
@@ -1461,7 +1507,7 @@ func TestClusterResetPreservesForbiddenSlot(t *testing.T) {
 	require.NoError(t, rdb0.FlushDB(ctx).Err())
 
 	// CLUSTER RESET HARD — this clears cluster topology, node id, migrated_slots_,
-	// imported_slots_, etc., but does NOT reset the migrator's forbidden_slot_range_.
+	// etc., but does NOT reset the migrator's forbidden_slot_range_.
 	require.Equal(t, "OK", rdb0.Do(ctx, "cluster", "reset", "hard").Val())
 
 	// Recreate the cluster; srv0 now owns slot 3300 again.