[Fix]: cm handle DEAD-rejoin and STANDBY lifecycle to prevent false reshard triggers

src/cluster_manager/cm_hb_monitor.cc

@@ -89,15 +89,17 @@ error_code_t ClusterManagerHBMonitor::OnRecvNodeHeartbeat(const std::string &log
 }
 
 void ClusterManagerHBMonitor::OnDeferredReshardResolved(const std::string &logical_node_id) {
-  // Reset heartbeat records for this logical_id so the new DS starts fresh
+  // Reset (or create) heartbeat records for this logical_id so the rejoining DS starts fresh.
+  // This must handle two cases:
+  //   1. Records exist (DEFERRED_RESHARD / IP_UPDATE): clear + push a fresh timestamp.
+  //   2. Records were deleted (DEAD path) or never created (scale-out): insert a fresh entry.
+  // In both cases we want exactly one fresh heartbeat timestamp so the node doesn't immediately
+  // time out on its next HB scan cycle.
   auto uomap_locked = ds_hb_records_.wlock();
-  auto it = uomap_locked->find(logical_node_id);
-  if (it != uomap_locked->end()) {
-    it->second.clear();
-    // Push a fresh heartbeat so the new DS doesn't immediately time out
-    simm::common::NodeHeartbeatTs hb_ts(std::chrono::steady_clock::now(), std::chrono::system_clock::now());
-    it->second.push_back(hb_ts);
-  }
+  auto &records = (*uomap_locked)[logical_node_id];  // operator[] inserts default if absent
+  records.clear();
+  simm::common::NodeHeartbeatTs hb_ts(std::chrono::steady_clock::now(), std::chrono::system_clock::now());
+  records.push_back(hb_ts);
   MLOG_INFO("Node handshake resolved: logical_id={}, heartbeat records reset", logical_node_id);
 }
 
@@ -173,16 +175,33 @@ void ClusterManagerHBMonitor::BgHBScanLoop() {
               cm_node_manager_ptr_->SetNodeStatus(
                   logical_id, NodeStatus::DEFERRED_RESHARD, now, NodeStatus::RUNNING);
 
-            } else if (entry.status == NodeStatus::DEAD) {
-              // Already dead, skip
+            } else if (entry.status == NodeStatus::DEAD || entry.status == NodeStatus::STANDBY) {
+              // DEAD: already handled, skip.
+              // STANDBY: rejoined after DEAD but holds no shards.
+              //   HB timeout → mark DEAD (no reshard needed, no shards to migrate).
+              if (entry.status == NodeStatus::STANDBY) {
+                MLOG_WARN("STANDBY node heartbeat timeout: logical_id={} (ip={}), marking DEAD",
+                          logical_id, entry.current_ip_port);
+                cm_node_manager_ptr_->SetNodeStatus(
+                    logical_id, NodeStatus::DEAD, {}, NodeStatus::STANDBY);
+                // No reshard: STANDBY held no shards.
+              }
             }
 
           } else if (entry_opt && !FLAGS_cm_deferred_reshard_enabled) {
             // logical_node_id available, deferred reshard disabled
             // Immediate DEAD + reshard
             auto& entry = *entry_opt;
-            if (entry.status == NodeStatus::DEAD) {
-              // Already dead, skip
+            if (entry.status == NodeStatus::DEAD || entry.status == NodeStatus::STANDBY) {
+              // DEAD: already handled, skip.
+              // STANDBY: holds no shards, nothing to reshard.
+              //   HB timeout → mark DEAD (cleanup only, no shard migration).
+              if (entry.status == NodeStatus::STANDBY) {
+                MLOG_WARN("STANDBY node heartbeat timeout: logical_id={} (ip={}), marking DEAD",
+                          logical_id, entry.current_ip_port);
+                cm_node_manager_ptr_->SetNodeStatus(
+                    logical_id, NodeStatus::DEAD, {}, NodeStatus::STANDBY);
+              }
             } else {
               // Pre-check: is reshard feasible?
               auto alive = cm_node_manager_ptr_->GetAllNodeAddress(true);

src/cluster_manager/cm_node_manager.cc

@@ -345,16 +345,42 @@ HandshakeResult ClusterManagerNodeManager::ProcessHandshake(
       return result;
     }
 
+    case NodeStatus::STANDBY: {
+      // Case 6: node was STANDBY (rejoined post-DEAD, holds no shards).
+      // Another handshake arrives (e.g. DS restarted again or retry).
+      // Keep it in STANDBY — nothing to give back, no routing table change needed.
+      // Update IP if it changed, to keep maps consistent.
+      if (entry.current_ip_port != new_ip_port) {
+        migrateNodeIp(logical_id, entry, new_ip_port);
+        node_status_map_.insert_or_assign(new_ip_port, NodeStatus::STANDBY);
+        logical_node_table_.assign(logical_id, entry);
+        MLOG_INFO("STANDBY node IP update: logical_id={} old_ip={} new_ip={}",
+                  logical_id, entry.current_ip_port, new_ip_port);
+      } else {
+        MLOG_INFO("STANDBY node re-handshake (same IP): logical_id={} ip={}", logical_id, new_ip_port);
+      }
+      result.action = HandshakeResult::Action::NEW_NODE;
+      result.shards_to_assign = {};
+      return result;
+    }
+
     case NodeStatus::DEAD:
     default: {
-      // Case 5: node was DEAD (reshard already happened), treat as new
+      // Case 5: node was DEAD (reshard already happened).
+      // Put node into STANDBY: it is online but holds no shards.
+      // - Not counted as "alive" for reshard feasibility checks.
+      // - Not eligible to receive shards in rebalance.
+      // - HB scan ignores STANDBY nodes (no deferred window, no reshard).
       migrateNodeIp(logical_id, entry, new_ip_port);
-      entry.status = NodeStatus::RUNNING;
+      // migrateNodeIp calls AddNode which sets node_status_map_ to RUNNING.
+      // Override to STANDBY so this node is excluded from alive counts and rebalance.
+      node_status_map_.insert_or_assign(new_ip_port, NodeStatus::STANDBY);
+      entry.status = NodeStatus::STANDBY;
       entry.deferred_reshard_since = {};
       logical_node_table_.assign(logical_id, entry);
       result.action = HandshakeResult::Action::NEW_NODE;
-      result.shards_to_assign = reported_shards;
-      MLOG_INFO("Node rejoin after DEAD: logical_id={} ip={}", logical_id, new_ip_port);
+      result.shards_to_assign = {};  // no shards to give back — reshard already happened
+      MLOG_INFO("Node rejoin after DEAD → STANDBY: logical_id={} ip={}", logical_id, new_ip_port);
       return result;
     }
   }

src/cluster_manager/cm_rpc_handler.cc

@@ -141,12 +141,13 @@ void NewNodeHandshakeHandler::Work(const std::shared_ptr<sicl::rpc::RpcContext>
         break;
       }
       case HandshakeResult::Action::NEW_NODE: {
-        // Post-grace-period scale-out: a brand-new logical_node_id not seen before.
-        // Shard assignment for scale-out is not yet supported; no shards assigned here.
-        // TODO: implement scale-out shard assignment
-        MLOG_WARN("New node registered post-grace-period (scale-out not yet supported): logical_id={} ip={}",
-                  logical_id,
-                  addr_str);
+        // Two sub-cases:
+        // (a) Rejoin after DEAD: logical_id is known but was DEAD (reshard already done).
+        //     Must reset HB records so the node doesn't immediately time out again.
+        // (b) Brand-new logical_id post-grace-period (scale-out): HB records don't exist yet,
+        //     OnDeferredReshardResolved is a no-op if key is absent, which is safe.
+        hb_monitor_->OnDeferredReshardResolved(logical_id);
+        MLOG_INFO("Node rejoin/scale-out post-grace-period: logical_id={} ip={}", logical_id, addr_str);
         break;
       }
       default:

src/common/base/common_types.h

@@ -17,7 +17,8 @@ namespace common {
   C(UNKNOWN)                  \
   C(RUNNING)                  \
   C(DEFERRED_RESHARD)         \
-  C(DEAD)
+  C(DEAD)                     \
+  C(STANDBY)
 
 #define NODESTATUS_ENUM(name) name,
 #define NODESTATUS_STRING(name) #name,

tests/cluster_manager/test_cm_deferred_reshard.cc

@@ -32,6 +32,7 @@ DECLARE_uint32(cm_heartbeat_timeout_inSecs);
 DECLARE_uint32(cm_heartbeat_bg_scan_interval_inSecs);
 DECLARE_bool(cm_deferred_reshard_enabled);
 DECLARE_uint32(cm_deferred_reshard_window_inSecs);
+DECLARE_uint32(dataserver_min_num);
 DECLARE_string(cm_log_file);
 
 namespace simm {
@@ -607,6 +608,98 @@ TEST(ProcessHandshakeUnitTest, ReregistrationInheritsShard) {
   MLOG_INFO("PASS: Routing table unchanged after re-registration (idempotent)");
 }
 
+// ═════════════════════════════════════════════════════════════════════════════
+// TEST 6: DS comes back after DEAD (rejoin → STANDBY), then goes silent again.
+//         CM should eventually mark it DEAD (no reshard, it held no shards).
+// ═════════════════════════════════════════════════════════════════════════════
+
+TEST_F(DeferredReshardTest, StandbyTimesOutToDeadWithoutReshard) {
+  // Use a very short deferred window so the first DS quickly reaches DEAD.
+  FLAGS_cm_deferred_reshard_window_inSecs = 3;
+  // Lower min_num so that killing 1-of-3 still allows reshard (2 >= 2).
+  auto saved_min_num = FLAGS_dataserver_min_num;
+  FLAGS_dataserver_min_num = 2;
+
+  constexpr uint32_t kDSNum = 3;
+  const std::string ip_prefix = "10.88.0.";
+  const int base_port = 54000;
+
+  folly::CPUThreadPoolExecutor executor(8);
+  auto cm = StartCM(executor, kDSNum, ip_prefix, base_port);
+
+  std::vector<std::unique_ptr<MockDeferredDS>> ds_list;
+  for (uint32_t i = 0; i < kDSNum; ++i) {
+    ds_list.push_back(std::make_unique<MockDeferredDS>(
+        ip_prefix + std::to_string(i + 1), base_port + i,
+        "standby-ds-" + std::to_string(i)));
+    ds_list.back()->Start();
+  }
+
+  // Wait for grace period + stable HB
+  std::this_thread::sleep_for(std::chrono::seconds(FLAGS_cm_cluster_init_grace_period_inSecs + 2));
+
+  // Kill DS-0: RUNNING → DEFERRED_RESHARD → DEAD (window=3s)
+  MLOG_INFO("=== Killing DS-0 ===");
+  ds_list[0]->Stop();
+
+  // Wait past HB timeout + deferred window
+  std::this_thread::sleep_for(std::chrono::seconds(
+      FLAGS_cm_heartbeat_timeout_inSecs +
+      FLAGS_cm_deferred_reshard_window_inSecs +
+      FLAGS_cm_heartbeat_bg_scan_interval_inSecs * 2 + 2));
+
+  auto e0 = cm->node_mgr->GetNodeEntry("standby-ds-0");
+  ASSERT_TRUE(e0.has_value());
+  EXPECT_EQ(e0->status, NodeStatus::DEAD)
+      << "DS-0 should be DEAD after deferred window expires";
+  MLOG_INFO("PASS: DS-0 is DEAD after deferred window");
+
+  // DS-0 comes back → STANDBY (reshard already happened)
+  MLOG_INFO("=== DS-0 rejoins → expecting STANDBY ===");
+  ds_list[0]->Start();
+  std::this_thread::sleep_for(std::chrono::seconds(2));
+
+  e0 = cm->node_mgr->GetNodeEntry("standby-ds-0");
+  ASSERT_TRUE(e0.has_value());
+  EXPECT_EQ(e0->status, NodeStatus::STANDBY)
+      << "DS-0 should be STANDBY after rejoining post-DEAD";
+
+  // Verify no shards were given back to DS-0
+  auto standby_shards = cm->shard_mgr->GetShardsOwnedByNode(ds_list[0]->GetAddr());
+  EXPECT_EQ(standby_shards.size(), 0u)
+      << "STANDBY node should own no shards";
+  MLOG_INFO("PASS: DS-0 is STANDBY with 0 shards");
+
+  // DS-0 goes silent again (STANDBY → DEAD, no reshard expected)
+  MLOG_INFO("=== DS-0 goes silent from STANDBY ===");
+  ds_list[0]->Stop();
+
+  // Wait past HB timeout for STANDBY to be marked DEAD
+  std::this_thread::sleep_for(std::chrono::seconds(
+      FLAGS_cm_heartbeat_timeout_inSecs +
+      FLAGS_cm_heartbeat_bg_scan_interval_inSecs + 2));
+
+  e0 = cm->node_mgr->GetNodeEntry("standby-ds-0");
+  ASSERT_TRUE(e0.has_value());
+  EXPECT_EQ(e0->status, NodeStatus::DEAD)
+      << "STANDBY node should become DEAD after HB timeout";
+  MLOG_INFO("PASS: STANDBY → DEAD after HB timeout (no reshard triggered)");
+
+  // Shards of DS-0 were already redistributed during the first DEAD transition;
+  // verify other nodes still hold shards (cluster is healthy).
+  auto ds1_shards = cm->shard_mgr->GetShardsOwnedByNode(ds_list[1]->GetAddr());
+  auto ds2_shards = cm->shard_mgr->GetShardsOwnedByNode(ds_list[2]->GetAddr());
+  EXPECT_GT(ds1_shards.size() + ds2_shards.size(), 0u)
+      << "Remaining nodes should still hold all shards";
+  MLOG_INFO("PASS: Cluster shards healthy on DS-1 ({}) and DS-2 ({})",
+            ds1_shards.size(), ds2_shards.size());
+
+  for (auto& ds : ds_list) ds->Stop();
+  FLAGS_dataserver_min_num = saved_min_num;
+  StopCM(cm);
+  executor.join();
+}
+
 }  // namespace cm
 }  // namespace simm