From fb5cbc9f3a002d49e6b33a903e2b02ef2255baad Mon Sep 17 00:00:00 2001 From: Kuzan Date: Tue, 14 Apr 2026 10:26:51 +0800 Subject: [PATCH] [Fix]: cm handle DEAD-rejoin and STANDBY lifecycle to prevent false reshard triggers --- src/cluster_manager/cm_hb_monitor.cc | 43 ++++++--- src/cluster_manager/cm_node_manager.cc | 34 ++++++- src/cluster_manager/cm_rpc_handler.cc | 13 +-- src/common/base/common_types.h | 3 +- .../test_cm_deferred_reshard.cc | 93 +++++++++++++++++++ 5 files changed, 163 insertions(+), 23 deletions(-) diff --git a/src/cluster_manager/cm_hb_monitor.cc b/src/cluster_manager/cm_hb_monitor.cc index ddb4fde..d5c1d15 100644 --- a/src/cluster_manager/cm_hb_monitor.cc +++ b/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); diff --git a/src/cluster_manager/cm_node_manager.cc b/src/cluster_manager/cm_node_manager.cc index ced2568..05172e5 100644 --- a/src/cluster_manager/cm_node_manager.cc +++ b/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; } } diff --git a/src/cluster_manager/cm_rpc_handler.cc b/src/cluster_manager/cm_rpc_handler.cc index ac105d4..18d8c0f 100644 --- a/src/cluster_manager/cm_rpc_handler.cc +++ b/src/cluster_manager/cm_rpc_handler.cc @@ -141,12 +141,13 @@ void NewNodeHandshakeHandler::Work(const std::shared_ptr 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: diff --git a/src/common/base/common_types.h b/src/common/base/common_types.h index 5f37d87..248c9b8 100644 --- a/src/common/base/common_types.h +++ b/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, diff --git a/tests/cluster_manager/test_cm_deferred_reshard.cc b/tests/cluster_manager/test_cm_deferred_reshard.cc index d0629d7..1850d87 100644 --- a/tests/cluster_manager/test_cm_deferred_reshard.cc +++ b/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> ds_list; + for (uint32_t i = 0; i < kDSNum; ++i) { + ds_list.push_back(std::make_unique( + 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