feat(cluster): disconnect replication connections to orphan slaves

src/cluster/cluster.cc

@@ -167,7 +167,7 @@ Status Cluster::SetClusterNodes(const std::string &nodes_str, int64_t version, b
 
   // Update version and cluster topology
   version_ = version;
-  nodes_ = nodes;
+  nodes_ = std::move(nodes);
   size_ = 0;
 
   // Update slots to nodes
@@ -252,6 +252,11 @@ Status Cluster::SetMasterSlaveRepl() {
       srv_->slot_migrator->SetStopMigrationFlag(false);
       LOG(INFO) << "Change server role to master, restart migration task";
     }
+
+    if (!is_slave) {
+      srv_->CleanupOrphanSlaves(version_, nodes_);
+    }
+
     return Status::OK();
   }
 
@@ -977,3 +982,13 @@ Status Cluster::Reset() {
   unlink(srv_->GetConfig()->NodesFilePath().data());
   return Status::OK();
 }
+
+// Note that if current version is lower than the given version,
+// it can't be determined whether the node is in the cluster, so just regard it as in the cluster.
+bool Cluster::IsInCluster(const std::string &node_id, int64_t version) const {
+  if (version < 0 || this->version_ < version) {
+    return true;
+  }
+
+  return nodes_.count(node_id) > 0;
+}

src/cluster/cluster.h

@@ -95,6 +95,7 @@ class Cluster {
   Status Reset();
 
   static bool SubCommandIsExecExclusive(const std::string &subcommand);
+  bool IsInCluster(const std::string &node_id, int64_t version) const;
 
  private:
   std::string getNodeIDBySlot(int slot) const;

src/cluster/cluster_defs.h

@@ -41,6 +41,7 @@ inline constexpr const char *errNoMasterNode = "The node isn't a master";
 inline constexpr const char *errClusterNoInitialized = "The cluster is not initialized";
 inline constexpr const char *errInvalidClusterNodeInfo = "Invalid cluster nodes info";
 inline constexpr const char *errInvalidImportState = "Invalid import state";
+inline constexpr const char *errNodeDecommissioned = "Node has been decommissioned";
 
 /// SlotRange is a range of cluster slots covering [start, end],
 /// where the valid values for start and end are [0, kClusterSlots).

src/cluster/replication.cc

@@ -58,7 +58,7 @@
 
 Status FeedSlaveThread::Start() {
   auto s = util::CreateThread("feed-replica", [this] {
-    sigset_t mask, omask;
+    sigset_t mask{}, omask{};
     sigemptyset(&mask);
     sigemptyset(&omask);
     sigaddset(&mask, SIGCHLD);
@@ -459,6 +459,12 @@ ReplicationThread::CBState ReplicationThread::replConfWriteCB(bufferevent *bev)
     data_to_send.emplace_back("ip-address");
     data_to_send.emplace_back(config->replica_announce_ip);
   }
+  if (!next_try_without_peer_id_ && config->cluster_enabled) {
+    data_to_send.emplace_back("peer-id");
+    data_to_send.emplace_back(srv_->cluster->GetMyId());
+    data_to_send.emplace_back("version");
+    data_to_send.emplace_back(std::to_string(srv_->cluster->GetVersion()));
+  }
   SendString(bev, redis::ArrayOfBulkStrings(data_to_send));
   repl_state_.store(kReplReplConf, std::memory_order_relaxed);
   LOG(INFO) << "[replication] replconf request was sent, waiting for response";
@@ -470,26 +476,40 @@ ReplicationThread::CBState ReplicationThread::replConfReadCB(bufferevent *bev) {
   UniqueEvbufReadln line(input, EVBUFFER_EOL_CRLF_STRICT);
   if (!line) return CBState::AGAIN;
 
+  auto resp = line.View();
+
   // on unknown option: first try without announce ip, if it fails again - do nothing (to prevent infinite loop)
-  if (isUnknownOption(line.View()) && !next_try_without_announce_ip_address_) {
-    next_try_without_announce_ip_address_ = true;
-    LOG(WARNING) << "The old version master, can't handle ip-address, "
-                 << "try without it again";
-    // Retry previous state, i.e. send replconf again
-    return CBState::PREV;
+  if (isUnknownOption(resp)) {
+    if (!next_try_without_peer_id_) {
+      next_try_without_peer_id_ = true;
+      LOG(WARNING) << "The old version master, can't handle peer-id, try without it again";
+      return CBState::PREV;
+    }
+    if (!next_try_without_announce_ip_address_) {
+      next_try_without_announce_ip_address_ = true;
+      LOG(WARNING) << "The old version master, can't handle ip-address, try without it again";
+      return CBState::PREV;
+    }
   }
-  if (line[0] == '-' && isRestoringError(line.View())) {
+
+  if (ResponseLineIsOK(resp)) {
+    LOG(INFO) << "[replication] replconf is ok, start psync";
+    return CBState::NEXT;
+  }
+
+  if (isRestoringError(resp)) {
     LOG(WARNING) << "The master was restoring the db, retry later";
     return CBState::RESTART;
   }
-  if (!ResponseLineIsOK(line.View())) {
-    LOG(WARNING) << "[replication] Failed to replconf: " << line.get() + 1;
-    //  backward compatible with old version that doesn't support replconf cmd
-    return CBState::NEXT;
-  } else {
-    LOG(INFO) << "[replication] replconf is ok, start psync";
-    return CBState::NEXT;
+
+  if (isNodeDecommissioned(resp)) {
+    LOG(ERROR) << "The master has fired the node, stop the replication";
+    return CBState::QUIT;
   }
+
+  LOG(WARNING) << "[replication] Failed to replconf: " << line.get() + 1;
+  //  backward compatible with old version that doesn't support replconf cmd
+  return CBState::NEXT;
 }
 
 ReplicationThread::CBState ReplicationThread::tryPSyncWriteCB(bufferevent *bev) {
@@ -772,6 +792,7 @@ Status ReplicationThread::parallelFetchFile(const std::string &dir,
   std::atomic<uint32_t> fetch_cnt = {0};
   std::atomic<uint32_t> skip_cnt = {0};
   std::vector<std::future<Status>> results;
+  results.reserve(concurrency);
   for (size_t tid = 0; tid < concurrency; ++tid) {
     results.push_back(
         std::async(std::launch::async, [this, dir, &files, tid, concurrency, &fetch_cnt, &skip_cnt]() -> Status {
@@ -1034,12 +1055,22 @@ Status ReplicationThread::parseWriteBatch(const rocksdb::WriteBatch &write_batch
 
 bool ReplicationThread::isRestoringError(std::string_view err) {
   // err doesn't contain the CRLF, so cannot use redis::Error here.
-  return err == RESP_PREFIX_ERROR + redis::StatusToRedisErrorMsg({Status::RedisLoading, redis::errRestoringBackup});
+  static const auto msg = fmt::format(RESP_PREFIX_ERROR "{}",
+                                      redis::StatusToRedisErrorMsg({Status::RedisLoading, redis::errRestoringBackup}));
+  return err == msg;
 }
 
 bool ReplicationThread::isWrongPsyncNum(std::string_view err) {
   // err doesn't contain the CRLF, so cannot use redis::Error here.
-  return err == RESP_PREFIX_ERROR + redis::StatusToRedisErrorMsg({Status::NotOK, redis::errWrongNumOfArguments});
+  static const auto msg =
+      fmt::format(RESP_PREFIX_ERROR "{}", redis::StatusToRedisErrorMsg({Status::NotOK, redis::errWrongNumOfArguments}));
+  return err == msg;
+}
+
+bool ReplicationThread::isNodeDecommissioned(std::string_view err) {
+  static const auto msg =
+      fmt::format(RESP_PREFIX_ERROR "{}", redis::StatusToRedisErrorMsg({Status::NotOK, errNodeDecommissioned}));
+  return err == msg;
 }
 
 bool ReplicationThread::isUnknownOption(std::string_view err) {

src/cluster/replication.h

@@ -26,6 +26,7 @@
 #include <deque>
 #include <memory>
 #include <string>
+#include <string_view>
 #include <thread>
 #include <tuple>
 #include <utility>
@@ -159,6 +160,7 @@ class ReplicationThread : private EventCallbackBase<ReplicationThread> {
   std::atomic<ReplState> repl_state_;
   std::atomic<int64_t> last_io_time_secs_ = 0;
   bool next_try_old_psync_ = false;
+  bool next_try_without_peer_id_ = false;
   bool next_try_without_announce_ip_address_ = false;
 
   std::function<bool()> pre_fullsync_cb_;
@@ -209,6 +211,7 @@ class ReplicationThread : private EventCallbackBase<ReplicationThread> {
   static bool isRestoringError(std::string_view err);
   static bool isWrongPsyncNum(std::string_view err);
   static bool isUnknownOption(std::string_view err);
+  static bool isNodeDecommissioned(std::string_view err);
 
   Status parseWriteBatch(const rocksdb::WriteBatch &write_batch);
 };

src/commands/cmd_cluster.cc

@@ -292,7 +292,7 @@ class CommandClusterX : public Commander {
     } else {
       return {Status::RedisExecErr, "Invalid cluster command options"};
     }
-    if (need_persist_nodes_info && srv->GetConfig()->persist_cluster_nodes_enabled) {
+    if (need_persist_nodes_info && srv->GetConfig()->persist_cluster_nodes_enabled && srv->cluster->GetVersion() >= 0) {
       return srv->cluster->DumpClusterNodes(srv->GetConfig()->NodesFilePath());
     }
     return Status::OK();