From a5d400f1e2eb087fe834e4b4e6606c36ac111d19 Mon Sep 17 00:00:00 2001 From: yong_ee Date: Thu, 15 Oct 2026 21:24:19 +0800 Subject: [PATCH 1/5] [Store] Fix master RPC reconnect stalls --- .../include/rpc_client_io_context.h | 7 ++++++ mooncake-store/include/master_client.h | 22 +++++++++++++++- mooncake-store/src/client_service.cpp | 3 ++- mooncake-store/src/master_client.cpp | 10 +++++--- mooncake-store/tests/rpc_timeout_test.cpp | 25 ++++++++++++++++++- 5 files changed, 60 insertions(+), 7 deletions(-) diff --git a/mooncake-common/include/rpc_client_io_context.h b/mooncake-common/include/rpc_client_io_context.h index b9842f0c99..e308c2f1b5 100644 --- a/mooncake-common/include/rpc_client_io_context.h +++ b/mooncake-common/include/rpc_client_io_context.h @@ -53,6 +53,13 @@ class RpcClientPool { return client_pool_; } + std::shared_ptr RecreateClientPool(std::string_view address) { + std::lock_guard lock(mutex_); + client_pool_ = ClientPool::create(address, config_, io_context_pool_); + address_ = address; + return client_pool_; + } + std::shared_ptr GetClientPool() const { std::shared_lock lock(mutex_); return client_pool_; diff --git a/mooncake-store/include/master_client.h b/mooncake-store/include/master_client.h index 6834234047..23b0b3cf38 100644 --- a/mooncake-store/include/master_client.h +++ b/mooncake-store/include/master_client.h @@ -72,6 +72,25 @@ inline void ApplyRpcTimeoutEnvOverrides(ClientConfig& client_config) { inline RpcClientPool::PoolConfig MakeMasterRpcClientPoolConfig() { RpcClientPool::PoolConfig config; + // Keep one retry for transient connection failures while avoiding the + // long default reconnect delay. The heartbeat thread still owns retries + // after the pool's bounded attempts are exhausted. + config.connect_retry_count = 1; + config.reconnect_wait_time = std::chrono::milliseconds(0); + config.client_config.connect_timeout_duration = std::chrono::seconds(1); + config.client_config.request_timeout_duration = std::chrono::seconds(5); + if (const char* retry_count = std::getenv("MC_RPC_CONNECT_RETRY_COUNT")) { + const auto parsed = std::atoll(retry_count); + config.connect_retry_count = + parsed > 0 ? static_cast( + parsed) + : 0; + } + if (const char* wait_ms = std::getenv("MC_RPC_RECONNECT_WAIT_MS")) { + const auto parsed = std::atoll(wait_ms); + config.reconnect_wait_time = + std::chrono::milliseconds(parsed > 0 ? parsed : 0); + } const char* value = std::getenv("MC_RPC_PROTOCOL"); if (value && std::string_view(value) == "rdma") { MaybeEnableRdmaSocketConfig(config.client_config.socket_config); @@ -108,7 +127,8 @@ class MasterClient { * @return ErrorCode indicating success/failure */ [[nodiscard]] ErrorCode Connect( - const std::string& master_addr = kDefaultMasterAddress); + const std::string& master_addr = kDefaultMasterAddress, + bool force_recreate = false); /** * @brief Checks if an object exists diff --git a/mooncake-store/src/client_service.cpp b/mooncake-store/src/client_service.cpp index d958240380..0e2887c183 100644 --- a/mooncake-store/src/client_service.cpp +++ b/mooncake-store/src/client_service.cpp @@ -5365,7 +5365,8 @@ void Client::StorageHeartbeatThreadMain() { LOG(ERROR) << "Failed to ping master for " << ping_fail_count << " times (non-HA); reconnecting to " << current_master_address; - auto err = master_client_.Connect(current_master_address); + auto err = master_client_.Connect(current_master_address, + /*force_recreate=*/true); if (err != ErrorCode::OK) { LOG(ERROR) << "Reconnect failed to " << current_master_address << ": " << toString(err); diff --git a/mooncake-store/src/master_client.cpp b/mooncake-store/src/master_client.cpp index 2e189801b4..4da9d25479 100644 --- a/mooncake-store/src/master_client.cpp +++ b/mooncake-store/src/master_client.cpp @@ -440,13 +440,15 @@ std::vector> MasterClient::invoke_batch_rpc( MasterClient::~MasterClient() = default; -ErrorCode MasterClient::Connect(const std::string& master_addr) { +ErrorCode MasterClient::Connect(const std::string& master_addr, + bool force_recreate) { ScopedVLogTimer timer(1, "MasterClient::Connect"); - timer.LogRequest("master_addr=", master_addr); + timer.LogRequest("master_addr=", master_addr, + ", force_recreate=", force_recreate); MutexLocker lock(&connect_mutex_); - if (client_addr_param_ != master_addr) { - client_accessor_.GetOrCreateClientPool(master_addr); + if (force_recreate || client_addr_param_ != master_addr) { + client_accessor_.RecreateClientPool(master_addr); client_addr_param_ = master_addr; } // The client pool does not have native connection check method, so we need diff --git a/mooncake-store/tests/rpc_timeout_test.cpp b/mooncake-store/tests/rpc_timeout_test.cpp index b2e31f46de..68bfcea508 100644 --- a/mooncake-store/tests/rpc_timeout_test.cpp +++ b/mooncake-store/tests/rpc_timeout_test.cpp @@ -134,12 +134,31 @@ TEST(RpcTimeoutTest, TimeoutEnvOverridesAreOptIn) { ::unsetenv("MC_RPC_TIMEOUT_MS"); ::unsetenv("MC_RPC_CONNECT_TIMEOUT_MS"); + ::unsetenv("MC_RPC_CONNECT_RETRY_COUNT"); + ::unsetenv("MC_RPC_RECONNECT_WAIT_MS"); StubClientConfig defaults; detail::ApplyRpcTimeoutEnvOverrides(defaults); EXPECT_EQ(defaults.request_timeout_duration, std::chrono::seconds(30)); EXPECT_EQ(defaults.connect_timeout_duration, std::chrono::seconds(30)); + // The master pool uses a bounded connection attempt and request deadline. + auto master_config = detail::MakeMasterRpcClientPoolConfig(); + EXPECT_EQ(master_config.connect_retry_count, 1u); + EXPECT_EQ(master_config.reconnect_wait_time, + std::chrono::milliseconds(0)); + EXPECT_EQ(master_config.client_config.connect_timeout_duration, + std::chrono::seconds(1)); + EXPECT_EQ(master_config.client_config.request_timeout_duration, + std::chrono::seconds(5)); + + ASSERT_EQ(::setenv("MC_RPC_CONNECT_RETRY_COUNT", "3", 1), 0); + ASSERT_EQ(::setenv("MC_RPC_RECONNECT_WAIT_MS", "250", 1), 0); + master_config = detail::MakeMasterRpcClientPoolConfig(); + EXPECT_EQ(master_config.connect_retry_count, 3u); + EXPECT_EQ(master_config.reconnect_wait_time, + std::chrono::milliseconds(250)); + ASSERT_EQ(::setenv("MC_RPC_TIMEOUT_MS", "1500", /*overwrite=*/1), 0); ASSERT_EQ(::setenv("MC_RPC_CONNECT_TIMEOUT_MS", "1000", /*overwrite=*/1), 0); @@ -152,12 +171,16 @@ TEST(RpcTimeoutTest, TimeoutEnvOverridesAreOptIn) { std::chrono::milliseconds(1000)); // The master pool is built from the same helper, so it sees them too. - auto master_config = detail::MakeMasterRpcClientPoolConfig(); + master_config = detail::MakeMasterRpcClientPoolConfig(); + EXPECT_EQ(master_config.client_config.request_timeout_duration, + std::chrono::milliseconds(1500)); EXPECT_EQ(master_config.client_config.connect_timeout_duration, std::chrono::milliseconds(1000)); ::unsetenv("MC_RPC_TIMEOUT_MS"); ::unsetenv("MC_RPC_CONNECT_TIMEOUT_MS"); + ::unsetenv("MC_RPC_CONNECT_RETRY_COUNT"); + ::unsetenv("MC_RPC_RECONNECT_WAIT_MS"); } } // namespace mooncake From b4c2572b86a7ad01a9e08960f3ffb167897b16ef Mon Sep 17 00:00:00 2001 From: yong_ee Date: Fri, 16 Oct 2026 14:54:15 +0800 Subject: [PATCH 2/5] [#3743] [Store] Split HA master RPC control pools --- mooncake-store/include/client_service.h | 1 + mooncake-store/include/master_client.h | 45 +++++++++++++++--- mooncake-store/src/client_service.cpp | 18 ++++++-- mooncake-store/src/master_client.cpp | 61 +++++++++++++++++++++---- 4 files changed, 106 insertions(+), 19 deletions(-) diff --git a/mooncake-store/include/client_service.h b/mooncake-store/include/client_service.h index de32eb898a..ff82863dd5 100644 --- a/mooncake-store/include/client_service.h +++ b/mooncake-store/include/client_service.h @@ -811,6 +811,7 @@ class Client { * @brief Internal helper functions for initialization and data transfer */ ErrorCode ConnectToMaster(const std::string& master_server_entry); + void EnterHaRuntimeMode(); ErrorCode InitTransferEngine( const std::string& local_hostname, const std::string& metadata_connstring, const std::string& protocol, diff --git a/mooncake-store/include/master_client.h b/mooncake-store/include/master_client.h index 23b0b3cf38..1231f956a0 100644 --- a/mooncake-store/include/master_client.h +++ b/mooncake-store/include/master_client.h @@ -4,6 +4,7 @@ #pragma once +#include #include #include #include @@ -70,14 +71,23 @@ inline void ApplyRpcTimeoutEnvOverrides(ClientConfig& client_config) { } } -inline RpcClientPool::PoolConfig MakeMasterRpcClientPoolConfig() { +inline RpcClientPool::PoolConfig MakeMasterRpcClientPoolConfig( + bool ha_enabled = false) { RpcClientPool::PoolConfig config; - // Keep one retry for transient connection failures while avoiding the - // long default reconnect delay. The heartbeat thread still owns retries - // after the pool's bounded attempts are exhausted. - config.connect_retry_count = 1; - config.reconnect_wait_time = std::chrono::milliseconds(0); - config.client_config.connect_timeout_duration = std::chrono::seconds(1); + if (ha_enabled) { + config.connect_retry_count = 0; + config.reconnect_wait_time = std::chrono::milliseconds(0); + config.client_config.connect_timeout_duration = + std::chrono::seconds(1); + } else { + // Keep one retry for transient connection failures while avoiding the + // long default reconnect delay. The heartbeat thread still owns + // retries after the pool's bounded attempts are exhausted. + config.connect_retry_count = 1; + config.reconnect_wait_time = std::chrono::milliseconds(0); + config.client_config.connect_timeout_duration = + std::chrono::seconds(1); + } config.client_config.request_timeout_duration = std::chrono::seconds(5); if (const char* retry_count = std::getenv("MC_RPC_CONNECT_RETRY_COUNT")) { const auto parsed = std::atoll(retry_count); @@ -111,11 +121,19 @@ class MasterClient { std::string tenant_id = "default") : client_accessor_(GetStoreRpcClientIoContextPool(), detail::MakeMasterRpcClientPoolConfig()), + ha_control_client_accessor_( + GetStoreRpcClientIoContextPool(), + detail::MakeMasterRpcClientPoolConfig(/*ha_enabled=*/true)), + ha_probe_client_accessor_( + GetStoreRpcClientIoContextPool(), + detail::MakeMasterRpcClientPoolConfig(/*ha_enabled=*/true)), client_id_(client_id), tenant_id_(std::move(tenant_id)), metrics_(metrics) {} ~MasterClient(); + void EnableHaConnectionPolicy(); + const std::string& tenant_id() const { return tenant_id_.value(); } MasterClient(const MasterClient&) = delete; @@ -714,6 +732,16 @@ class MasterClient { [[nodiscard]] tl::expected invoke_rpc( Args&&... args); + template + [[nodiscard]] tl::expected invoke_rpc_with_pool( + RpcClientPool& client_accessor, Args&&... args); + + template + [[nodiscard]] tl::expected + invoke_rpc_with_client_pool( + const std::shared_ptr& client_pool, + Args&&... args); + /** * @brief Generic RPC invocation helper for batch operations * @tparam ServiceMethod Pointer to WrappedMasterService member function @@ -728,6 +756,8 @@ class MasterClient { invoke_batch_rpc(size_t input_size, Args&&... args); RpcClientPool client_accessor_; + RpcClientPool ha_control_client_accessor_; + RpcClientPool ha_probe_client_accessor_; // The client identification. const UUID client_id_; @@ -741,6 +771,7 @@ class MasterClient { mutable Mutex connect_mutex_; // The address which is passed to the coro_rpc_client std::string client_addr_param_ GUARDED_BY(connect_mutex_); + std::atomic ha_connection_policy_enabled_{false}; }; } // namespace mooncake diff --git a/mooncake-store/src/client_service.cpp b/mooncake-store/src/client_service.cpp index 0e2887c183..ec49c20e0c 100644 --- a/mooncake-store/src/client_service.cpp +++ b/mooncake-store/src/client_service.cpp @@ -710,10 +710,6 @@ ErrorCode Client::ConnectToMaster(const std::string& master_server_entry) { leader_coordinator_ = std::move(coordinator.value()); direct_master_address_.clear(); - leader_monitor_running_ = true; - leader_monitor_thread_ = - std::thread([this]() { this->LeaderMonitorThreadMain(); }); - return ErrorCode::OK; } else { auto err = master_client_.Connect(master_server_entry); @@ -730,6 +726,18 @@ ErrorCode Client::ConnectToMaster(const std::string& master_server_entry) { } } +void Client::EnterHaRuntimeMode() { + if (!leader_coordinator_) { + return; + } + + master_client_.EnableHaConnectionPolicy(); + if (!leader_monitor_running_.exchange(true)) { + leader_monitor_thread_ = + std::thread([this]() { this->LeaderMonitorThreadMain(); }); + } +} + ErrorCode Client::SwitchLeader(const ha::MasterView& target_view) { std::lock_guard lock(leader_switch_mutex_); @@ -1145,6 +1153,8 @@ std::optional> Client::Create( } } + client->EnterHaRuntimeMode(); + // this only performs RPC calls if (protocol == "rpc_only") { LOG(INFO) << "Use rpc only. Skip initializing transfer engine."; diff --git a/mooncake-store/src/master_client.cpp b/mooncake-store/src/master_client.cpp index 4da9d25479..2eec064f4c 100644 --- a/mooncake-store/src/master_client.cpp +++ b/mooncake-store/src/master_client.cpp @@ -345,9 +345,9 @@ struct RpcNameTraits<&WrappedMasterService::PollRemoveAll> { }; template -tl::expected MasterClient::invoke_rpc(Args&&... args) { - auto pool = client_accessor_.GetClientPool(); - +tl::expected MasterClient::invoke_rpc_with_client_pool( + const std::shared_ptr& client_pool, + Args&&... args) { // Increment RPC counter if (metrics_) { metrics_->rpc_count.inc({RpcNameTraits::value}); @@ -356,7 +356,7 @@ tl::expected MasterClient::invoke_rpc(Args&&... args) { auto start_time = std::chrono::steady_clock::now(); return async_simple::coro::syncAwait( [&]() -> async_simple::coro::Lazy> { - auto ret = co_await pool->send_request( + auto ret = co_await client_pool->send_request( [&](coro_io::client_reuse_hint, coro_rpc::coro_rpc_client& client) { return client.send_request( @@ -387,6 +387,19 @@ tl::expected MasterClient::invoke_rpc(Args&&... args) { }()); } +template +tl::expected MasterClient::invoke_rpc_with_pool( + RpcClientPool& client_accessor, Args&&... args) { + return invoke_rpc_with_client_pool( + client_accessor.GetClientPool(), std::forward(args)...); +} + +template +tl::expected MasterClient::invoke_rpc(Args&&... args) { + return invoke_rpc_with_pool( + client_accessor_, std::forward(args)...); +} + template std::vector> MasterClient::invoke_batch_rpc( size_t input_size, Args&&... args) { @@ -440,6 +453,14 @@ std::vector> MasterClient::invoke_batch_rpc( MasterClient::~MasterClient() = default; +void MasterClient::EnableHaConnectionPolicy() { + MutexLocker lock(&connect_mutex_); + if (!client_addr_param_.empty()) { + ha_control_client_accessor_.GetOrCreateClientPool(client_addr_param_); + } + ha_connection_policy_enabled_.store(true, std::memory_order_release); +} + ErrorCode MasterClient::Connect(const std::string& master_addr, bool force_recreate) { ScopedVLogTimer timer(1, "MasterClient::Connect"); @@ -447,14 +468,23 @@ ErrorCode MasterClient::Connect(const std::string& master_addr, ", force_recreate=", force_recreate); MutexLocker lock(&connect_mutex_); - if (force_recreate || client_addr_param_ != master_addr) { + const bool use_ha_control_pool = + ha_connection_policy_enabled_.load(std::memory_order_acquire); + if (use_ha_control_pool) { + ha_probe_client_accessor_.GetOrCreateClientPool(master_addr); + } else if (force_recreate || client_addr_param_ != master_addr) { client_accessor_.RecreateClientPool(master_addr); - client_addr_param_ = master_addr; + } else { + client_accessor_.GetOrCreateClientPool(master_addr); } // The client pool does not have native connection check method, so we need // to use custom ServiceReady API. auto result = - invoke_rpc<&WrappedMasterService::ServiceReady, std::string>(); + use_ha_control_pool + ? invoke_rpc_with_pool<&WrappedMasterService::ServiceReady, + std::string>(ha_probe_client_accessor_) + : invoke_rpc_with_pool<&WrappedMasterService::ServiceReady, + std::string>(client_accessor_); if (!result.has_value()) { timer.LogResponse("error_code=", result.error()); return result.error(); @@ -468,6 +498,12 @@ ErrorCode MasterClient::Connect(const std::string& master_addr, timer.LogResponse("error_code=", ErrorCode::INVALID_VERSION); return ErrorCode::INVALID_VERSION; } + if (use_ha_control_pool) { + // Only move foreground traffic after the control-plane probe is ready. + client_accessor_.GetOrCreateClientPool(master_addr); + ha_control_client_accessor_.GetOrCreateClientPool(master_addr); + } + client_addr_param_ = master_addr; timer.LogResponse("error_code=", ErrorCode::OK); return ErrorCode::OK; } @@ -916,8 +952,17 @@ tl::expected MasterClient::Ping() { ScopedVLogTimer timer(1, "MasterClient::Ping"); timer.LogRequest("client_id=", client_id_); + std::shared_ptr ping_pool; + { + MutexLocker lock(&connect_mutex_); + ping_pool = + ha_connection_policy_enabled_.load(std::memory_order_acquire) + ? ha_control_client_accessor_.GetClientPool() + : client_accessor_.GetClientPool(); + } auto result = - invoke_rpc<&WrappedMasterService::Ping, PingResponse>(client_id_); + invoke_rpc_with_client_pool<&WrappedMasterService::Ping, PingResponse>( + ping_pool, client_id_); timer.LogResponseExpected(result); return result; } From e0a40c332a83bab974c03c86f23d44968374976b Mon Sep 17 00:00:00 2001 From: yong_ee Date: Fri, 16 Oct 2026 14:54:15 +0800 Subject: [PATCH 3/5] [#3943] [Store] Protect HA RPC pool lifetimes --- .../include/rpc_client_io_context.h | 121 ++++++++++++++++-- mooncake-store/include/master_client.h | 1 + mooncake-store/src/master_client.cpp | 17 ++- 3 files changed, 128 insertions(+), 11 deletions(-) diff --git a/mooncake-common/include/rpc_client_io_context.h b/mooncake-common/include/rpc_client_io_context.h index e308c2f1b5..6e1e3f0282 100644 --- a/mooncake-common/include/rpc_client_io_context.h +++ b/mooncake-common/include/rpc_client_io_context.h @@ -1,14 +1,17 @@ #pragma once -#include +#include #include -#include +#include +#include #include #include #include #include #include +#include #include +#include #include #include @@ -19,16 +22,116 @@ namespace mooncake { std::shared_ptr CreateRpcClientIoContextPool( uint32_t thread_count); +class RpcDrainGuard { + private: + struct State { + std::atomic stopping{false}; + std::atomic inflight{0}; + std::mutex mutex; + std::condition_variable cv; + }; + + public: + class ScopedCall { + public: + explicit ScopedCall(const RpcDrainGuard& guard) + : state_(guard.state_), ok_(try_enter(*state_)) {} + ~ScopedCall() { + if (ok_) { + leave(*state_); + } + } + ScopedCall(const ScopedCall&) = delete; + ScopedCall& operator=(const ScopedCall&) = delete; + bool ok() const { return ok_; } + + private: + static bool try_enter(State& state) { + if (state.stopping.load(std::memory_order_acquire)) { + return false; + } + state.inflight.fetch_add(1, std::memory_order_acq_rel); + if (state.stopping.load(std::memory_order_acquire)) { + leave(state); + return false; + } + return true; + } + + static void leave(State& state) { + if (state.inflight.fetch_sub(1, std::memory_order_acq_rel) == 1) { + std::lock_guard lock(state.mutex); + state.cv.notify_all(); + } + } + + std::shared_ptr state_; + bool ok_; + }; + + bool drain_for(std::chrono::milliseconds timeout) { + state_->stopping.store(true, std::memory_order_release); + std::unique_lock lock(state_->mutex); + return state_->cv.wait_for(lock, timeout, [&] { + return state_->inflight.load(std::memory_order_acquire) == 0; + }); + } + + private: + std::shared_ptr state_ = std::make_shared(); +}; + template coro_io::io_context_pool& GetRpcClientIoContextPool(uint32_t thread_count) { static const auto io_pool = CreateRpcClientIoContextPool(thread_count); return *io_pool; } +namespace detail { + +using MasterClientPool = + coro_io::client_pool; + +inline void KeepMasterPoolAlive(std::shared_ptr pool) { + static std::mutex keep_mutex; + static std::vector> retired_pools; + std::lock_guard lock(keep_mutex); + retired_pools.push_back(std::move(pool)); +} + +inline std::shared_ptr SharedPoolRegistryImpl( + std::string_view address, MasterClientPool::pool_config config, + coro_io::io_context_pool& io_context_pool) { + static std::mutex registry_mutex; + static std::unordered_map> + registry; + std::lock_guard lock(registry_mutex); + + std::string key(address); + auto append = [&key](auto value) { + key.push_back('|'); + key.append(std::to_string(value)); + }; + append(config.max_connection); + append(config.connect_retry_count); + append(config.reconnect_wait_time.count()); + append(config.client_config.connect_timeout_duration.count()); + append(config.client_config.request_timeout_duration.count()); + append(static_cast(config.client_config.socket_config.index())); + + auto& pool = registry[key]; + if (!pool) { + pool = MasterClientPool::create(std::string(address), std::move(config), + io_context_pool); + } + return pool; +} + +} // namespace detail + /** - * A replaceable client pool for callers that communicate with one target at a - * time. Requests retain a shared_ptr to the old pool while they are in flight; - * after an address switch the old pool is destroyed when those requests end. + * A replaceable client pool. Pools are retained process-wide because ylt's + * background reconnect coroutines may outlive the active holder. */ class RpcClientPool { public: @@ -38,7 +141,6 @@ class RpcClientPool { explicit RpcClientPool(coro_io::io_context_pool& io_context_pool, PoolConfig config = {}) : io_context_pool_(io_context_pool), config_(std::move(config)) { - // Address replacement supersedes background recovery of the old host. config_.host_alive_detect_duration = std::chrono::seconds(0); } @@ -46,8 +148,8 @@ class RpcClientPool { std::string_view address) { std::lock_guard lock(mutex_); if (!client_pool_ || address_ != address) { - client_pool_ = - ClientPool::create(address, config_, io_context_pool_); + client_pool_ = detail::SharedPoolRegistryImpl( + address, config_, io_context_pool_); address_ = address; } return client_pool_; @@ -55,6 +157,9 @@ class RpcClientPool { std::shared_ptr RecreateClientPool(std::string_view address) { std::lock_guard lock(mutex_); + if (client_pool_) { + detail::KeepMasterPoolAlive(client_pool_); + } client_pool_ = ClientPool::create(address, config_, io_context_pool_); address_ = address; return client_pool_; diff --git a/mooncake-store/include/master_client.h b/mooncake-store/include/master_client.h index 1231f956a0..3692e24eac 100644 --- a/mooncake-store/include/master_client.h +++ b/mooncake-store/include/master_client.h @@ -758,6 +758,7 @@ class MasterClient { RpcClientPool client_accessor_; RpcClientPool ha_control_client_accessor_; RpcClientPool ha_probe_client_accessor_; + RpcDrainGuard rpc_drain_; // The client identification. const UUID client_id_; diff --git a/mooncake-store/src/master_client.cpp b/mooncake-store/src/master_client.cpp index 2eec064f4c..13faa2e4c0 100644 --- a/mooncake-store/src/master_client.cpp +++ b/mooncake-store/src/master_client.cpp @@ -348,6 +348,10 @@ template tl::expected MasterClient::invoke_rpc_with_client_pool( const std::shared_ptr& client_pool, Args&&... args) { + RpcDrainGuard::ScopedCall inflight(rpc_drain_); + if (!inflight.ok()) { + return tl::make_unexpected(ErrorCode::RPC_FAIL); + } // Increment RPC counter if (metrics_) { metrics_->rpc_count.inc({RpcNameTraits::value}); @@ -403,9 +407,12 @@ tl::expected MasterClient::invoke_rpc(Args&&... args) { template std::vector> MasterClient::invoke_batch_rpc( size_t input_size, Args&&... args) { + RpcDrainGuard::ScopedCall inflight(rpc_drain_); + if (!inflight.ok()) { + return std::vector>( + input_size, tl::make_unexpected(ErrorCode::RPC_FAIL)); + } auto pool = client_accessor_.GetClientPool(); - - // Increment RPC counter if (metrics_) { metrics_->rpc_count.inc({RpcNameTraits::value}); } @@ -451,7 +458,11 @@ std::vector> MasterClient::invoke_batch_rpc( }()); } -MasterClient::~MasterClient() = default; +MasterClient::~MasterClient() { + if (!rpc_drain_.drain_for(std::chrono::seconds(30))) { + LOG(ERROR) << "MasterClient teardown: RPCs still in flight after 30s"; + } +} void MasterClient::EnableHaConnectionPolicy() { MutexLocker lock(&connect_mutex_); From 8164de5bc1bec7e6f2a6dfac25a2f2d52aea6b12 Mon Sep 17 00:00:00 2001 From: yong_ee Date: Fri, 16 Oct 2026 14:54:16 +0800 Subject: [PATCH 4/5] [#3743] [Store] Publish HA master after RPC readiness --- mooncake-common/etcd/etcd_wrapper.go | 59 +++++++++++++++++ mooncake-store/include/etcd_helper.h | 3 + .../backends/etcd/etcd_leader_coordinator.h | 3 + .../ha/leadership/leader_coordinator.h | 19 ++++++ mooncake-store/include/master_client.h | 29 ++++----- mooncake-store/src/client_service.cpp | 21 ++++--- mooncake-store/src/etcd_helper.cpp | 63 ++++++++++++++++++- .../backends/etcd/etcd_leader_coordinator.cpp | 39 +++++++++++- .../leadership/master_service_supervisor.cpp | 14 +++++ mooncake-store/tests/rpc_timeout_test.cpp | 29 +++++++-- 10 files changed, 246 insertions(+), 33 deletions(-) diff --git a/mooncake-common/etcd/etcd_wrapper.go b/mooncake-common/etcd/etcd_wrapper.go index 4ac718072c..75303de186 100644 --- a/mooncake-common/etcd/etcd_wrapper.go +++ b/mooncake-common/etcd/etcd_wrapper.go @@ -791,6 +791,65 @@ func EtcdStoreBatchCreateWrapper(keys **C.char, values **C.char, count C.int, er return 0 } +//export EtcdStoreTxnCompareAndPutWithLeaseWrapper +func EtcdStoreTxnCompareAndPutWithLeaseWrapper(compareKeys **C.char, compareKeySizes *C.int, compareKinds *C.int, compareValues **C.char, compareValueSizes *C.int, compareCount C.int, putKeys **C.char, putKeySizes *C.int, putValues **C.char, putValueSizes *C.int, putCount C.int, leaseId int64, errMsg **C.char) int { + cli := getStoreClient() + if cli == nil { + *errMsg = C.CString("etcd client not initialized") + return -1 + } + + cmpN := int(compareCount) + putN := int(putCount) + cmps := make([]clientv3.Cmp, 0, cmpN) + if cmpN > 0 { + compareKeyPtrs := (*[1 << 28]*C.char)(unsafe.Pointer(compareKeys))[:cmpN:cmpN] + compareKeySizeList := (*[1 << 28]C.int)(unsafe.Pointer(compareKeySizes))[:cmpN:cmpN] + compareKindList := (*[1 << 28]C.int)(unsafe.Pointer(compareKinds))[:cmpN:cmpN] + compareValuePtrs := (*[1 << 28]*C.char)(unsafe.Pointer(compareValues))[:cmpN:cmpN] + compareValueSizeList := (*[1 << 28]C.int)(unsafe.Pointer(compareValueSizes))[:cmpN:cmpN] + for i := 0; i < cmpN; i++ { + k := C.GoStringN(compareKeyPtrs[i], compareKeySizeList[i]) + switch int(compareKindList[i]) { + case 0: + v := C.GoStringN(compareValuePtrs[i], compareValueSizeList[i]) + cmps = append(cmps, clientv3.Compare(clientv3.Value(k), "=", v)) + case 1: + cmps = append(cmps, clientv3.Compare(clientv3.CreateRevision(k), "=", 0)) + default: + *errMsg = C.CString("unsupported compare kind") + return -1 + } + } + } + + ops := make([]clientv3.Op, 0, putN) + if putN > 0 { + putKeyPtrs := (*[1 << 28]*C.char)(unsafe.Pointer(putKeys))[:putN:putN] + putKeySizeList := (*[1 << 28]C.int)(unsafe.Pointer(putKeySizes))[:putN:putN] + putValuePtrs := (*[1 << 28]*C.char)(unsafe.Pointer(putValues))[:putN:putN] + putValueSizeList := (*[1 << 28]C.int)(unsafe.Pointer(putValueSizes))[:putN:putN] + for i := 0; i < putN; i++ { + k := C.GoStringN(putKeyPtrs[i], putKeySizeList[i]) + v := C.GoStringN(putValuePtrs[i], putValueSizeList[i]) + ops = append(ops, clientv3.OpPut(k, v, clientv3.WithLease(clientv3.LeaseID(leaseId)))) + } + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + resp, err := cli.Txn(ctx).If(cmps...).Then(ops...).Commit() + if err != nil { + *errMsg = C.CString(err.Error()) + return -1 + } + if !resp.Succeeded { + *errMsg = C.CString("etcd transaction compare failed") + return -2 + } + return 0 +} + //export EtcdStoreTxnCompareAndPutWrapper func EtcdStoreTxnCompareAndPutWrapper(compareKeys **C.char, compareKeySizes *C.int, compareKinds *C.int, compareValues **C.char, compareValueSizes *C.int, compareCount C.int, putKeys **C.char, putKeySizes *C.int, putValues **C.char, putValueSizes *C.int, putCount C.int, errMsg **C.char) int { cli := getStoreClient() diff --git a/mooncake-store/include/etcd_helper.h b/mooncake-store/include/etcd_helper.h index b0d851c5b6..880314085f 100644 --- a/mooncake-store/include/etcd_helper.h +++ b/mooncake-store/include/etcd_helper.h @@ -95,6 +95,9 @@ class EtcdHelper { */ static ErrorCode TxnCompareAndPut(const std::vector& compares, const std::vector& puts); + static ErrorCode TxnCompareAndPutWithLease( + const std::vector& compares, + const std::vector& puts, EtcdLeaseId lease_id); /* * @brief Grant a lease from the etcd. diff --git a/mooncake-store/include/ha/leadership/backends/etcd/etcd_leader_coordinator.h b/mooncake-store/include/ha/leadership/backends/etcd/etcd_leader_coordinator.h index 30e5c8e0b7..a22f9ae05f 100644 --- a/mooncake-store/include/ha/leadership/backends/etcd/etcd_leader_coordinator.h +++ b/mooncake-store/include/ha/leadership/backends/etcd/etcd_leader_coordinator.h @@ -26,6 +26,9 @@ class EtcdLeaderCoordinator final : public LeaderCoordinator { tl::expected TryAcquireLeadership( const std::string& leader_address) override; + ErrorCode PublishServiceReady( + const LeadershipSession& session) override; + tl::expected RenewLeadership( const LeadershipSession& session) override; diff --git a/mooncake-store/include/ha/leadership/leader_coordinator.h b/mooncake-store/include/ha/leadership/leader_coordinator.h index 923ca55ee0..e9180de043 100644 --- a/mooncake-store/include/ha/leadership/leader_coordinator.h +++ b/mooncake-store/include/ha/leadership/leader_coordinator.h @@ -28,6 +28,11 @@ class LeaderCoordinator { virtual tl::expected TryAcquireLeadership(const std::string& leader_address) = 0; + virtual ErrorCode PublishServiceReady( + const LeadershipSession& /*session*/) { + return ErrorCode::OK; + } + virtual tl::expected RenewLeadership( const LeadershipSession& session) = 0; @@ -42,5 +47,19 @@ class LeaderCoordinator { virtual ErrorCode ReleaseLeadership(const LeadershipSession& session) = 0; }; +inline tl::expected, ErrorCode> +ReadCurrentViewOrWaitForReady(LeaderCoordinator& coordinator, + std::chrono::milliseconds timeout) { + auto current_view = coordinator.ReadCurrentView(); + if (!current_view || current_view->has_value()) { + return current_view; + } + auto ready_view = coordinator.WaitForViewChange(std::nullopt, timeout); + if (!ready_view) { + return tl::make_unexpected(ready_view.error()); + } + return ready_view->current_view; +} + } // namespace ha } // namespace mooncake diff --git a/mooncake-store/include/master_client.h b/mooncake-store/include/master_client.h index 3692e24eac..40c5a558d2 100644 --- a/mooncake-store/include/master_client.h +++ b/mooncake-store/include/master_client.h @@ -74,21 +74,17 @@ inline void ApplyRpcTimeoutEnvOverrides(ClientConfig& client_config) { inline RpcClientPool::PoolConfig MakeMasterRpcClientPoolConfig( bool ha_enabled = false) { RpcClientPool::PoolConfig config; - if (ha_enabled) { - config.connect_retry_count = 0; - config.reconnect_wait_time = std::chrono::milliseconds(0); - config.client_config.connect_timeout_duration = - std::chrono::seconds(1); - } else { - // Keep one retry for transient connection failures while avoiding the - // long default reconnect delay. The heartbeat thread still owns - // retries after the pool's bounded attempts are exhausted. - config.connect_retry_count = 1; - config.reconnect_wait_time = std::chrono::milliseconds(0); - config.client_config.connect_timeout_duration = - std::chrono::seconds(1); - } - config.client_config.request_timeout_duration = std::chrono::seconds(5); + // The outer heartbeat/reconnect loops own recovery in both HA and + // non-HA modes. Avoid pool-level retries and waits so every Master RPC + // returns connection failures to its caller promptly. + config.connect_retry_count = 0; + config.reconnect_wait_time = std::chrono::milliseconds(0); + config.client_config.connect_timeout_duration = + std::chrono::seconds(1); + // Keep the parameter to preserve the HA/non-HA policy construction API; + // both modes currently use the same bounded connection policy. + (void)ha_enabled; + const char* value = std::getenv("MC_RPC_PROTOCOL"); if (const char* retry_count = std::getenv("MC_RPC_CONNECT_RETRY_COUNT")) { const auto parsed = std::atoll(retry_count); config.connect_retry_count = @@ -101,11 +97,12 @@ inline RpcClientPool::PoolConfig MakeMasterRpcClientPoolConfig( config.reconnect_wait_time = std::chrono::milliseconds(parsed > 0 ? parsed : 0); } - const char* value = std::getenv("MC_RPC_PROTOCOL"); if (value && std::string_view(value) == "rdma") { MaybeEnableRdmaSocketConfig(config.client_config.socket_config); } + // Unset MC_RPC_* variables keep the configured pool defaults; explicit + // values override those defaults for both HA and non-HA pools. ApplyRpcTimeoutEnvOverrides(config.client_config); return config; } diff --git a/mooncake-store/src/client_service.cpp b/mooncake-store/src/client_service.cpp index ec49c20e0c..ac9136ea72 100644 --- a/mooncake-store/src/client_service.cpp +++ b/mooncake-store/src/client_service.cpp @@ -689,19 +689,20 @@ ErrorCode Client::ConnectToMaster(const std::string& master_server_entry) { return coordinator.error(); } - auto current_view = coordinator.value()->ReadCurrentView(); - if (!current_view) { - LOG(ERROR) << "Failed to read current master view: " - << toString(current_view.error()); - return current_view.error(); - } - if (!current_view.value().has_value()) { - LOG(ERROR) << "No master is available in HA backend"; + auto master_view = ha::ReadCurrentViewOrWaitForReady( + *coordinator.value(), std::chrono::seconds(30)); + if (!master_view) { + LOG(ERROR) << "Failed to discover a ready master view: " + << toString(master_view.error()); + return master_view.error(); + } + if (!master_view->has_value()) { + LOG(ERROR) << "No master became ready within 30 seconds"; return ErrorCode::UNAVAILABLE_IN_CURRENT_STATUS; } - const auto& master_view = current_view.value().value(); - auto err = SwitchLeader(master_view); + const auto& current_master_view = master_view->value(); + auto err = SwitchLeader(current_master_view); if (err != ErrorCode::OK) { LOG(ERROR) << "Failed to connect to master"; return err; diff --git a/mooncake-store/src/etcd_helper.cpp b/mooncake-store/src/etcd_helper.cpp index e49611c026..98a3156888 100644 --- a/mooncake-store/src/etcd_helper.cpp +++ b/mooncake-store/src/etcd_helper.cpp @@ -217,6 +217,54 @@ ErrorCode EtcdHelper::TxnCompareAndPut(const std::vector& compares, return ErrorCode::OK; } +ErrorCode EtcdHelper::TxnCompareAndPutWithLease( + const std::vector& compares, + const std::vector& puts, EtcdLeaseId lease_id) { + std::vector compare_keys; + std::vector compare_key_sizes; + std::vector compare_kinds; + std::vector compare_values; + std::vector compare_value_sizes; + for (const auto& compare : compares) { + compare_keys.push_back(const_cast(compare.key.data())); + compare_key_sizes.push_back(static_cast(compare.key.size())); + compare_kinds.push_back(static_cast(compare.kind)); + compare_values.push_back( + const_cast(compare.expected_value.data())); + compare_value_sizes.push_back( + static_cast(compare.expected_value.size())); + } + std::vector put_keys; + std::vector put_key_sizes; + std::vector put_values; + std::vector put_value_sizes; + for (const auto& put : puts) { + put_keys.push_back(const_cast(put.key.data())); + put_key_sizes.push_back(static_cast(put.key.size())); + put_values.push_back(const_cast(put.value.data())); + put_value_sizes.push_back(static_cast(put.value.size())); + } + char* err_msg = nullptr; + int ret = EtcdStoreTxnCompareAndPutWithLeaseWrapper( + compare_keys.data(), compare_key_sizes.data(), compare_kinds.data(), + compare_values.data(), compare_value_sizes.data(), + static_cast(compares.size()), put_keys.data(), + put_key_sizes.data(), put_values.data(), put_value_sizes.data(), + static_cast(puts.size()), static_cast(lease_id), + &err_msg); + if (ret == -2) { + if (err_msg != nullptr) free(err_msg); + return ErrorCode::ETCD_TRANSACTION_FAIL; + } + if (ret != 0) { + LOG(ERROR) << "TxnCompareAndPutWithLease failed: " + << (err_msg == nullptr ? "" : err_msg); + if (err_msg != nullptr) free(err_msg); + return ErrorCode::ETCD_OPERATION_ERROR; + } + return ErrorCode::OK; +} + ErrorCode EtcdHelper::GrantLease(int64_t lease_ttl, EtcdLeaseId& lease_id) { char* err_msg = nullptr; GoInt64 go_lease_id = 0; @@ -602,10 +650,21 @@ ErrorCode EtcdHelper::Create(const char* key, const size_t key_size, return ErrorCode::ETCD_OPERATION_ERROR; } -ErrorCode EtcdHelper::TxnCompareAndPut(const std::vector& compares, - const std::vector& puts) { +ErrorCode EtcdHelper::TxnCompareAndPut( + const std::vector& compares, + const std::vector& puts) { + (void)compares; + (void)puts; + LOG(FATAL) << "Etcd is not enabled in compilation"; + return ErrorCode::ETCD_OPERATION_ERROR; +} + +ErrorCode EtcdHelper::TxnCompareAndPutWithLease( + const std::vector& compares, + const std::vector& puts, EtcdLeaseId lease_id) { (void)compares; (void)puts; + (void)lease_id; LOG(FATAL) << "Etcd is not enabled in compilation"; return ErrorCode::ETCD_OPERATION_ERROR; } diff --git a/mooncake-store/src/ha/leadership/backends/etcd/etcd_leader_coordinator.cpp b/mooncake-store/src/ha/leadership/backends/etcd/etcd_leader_coordinator.cpp index 26bc7b23e7..39d0882236 100644 --- a/mooncake-store/src/ha/leadership/backends/etcd/etcd_leader_coordinator.cpp +++ b/mooncake-store/src/ha/leadership/backends/etcd/etcd_leader_coordinator.cpp @@ -23,6 +23,16 @@ namespace etcd { namespace { constexpr int kKeepAliveReadyTimeoutMs = 1000; +constexpr std::string_view kServiceWarmingPrefix = + "__mooncake_service_warming__:"; + +std::string MakeServiceWarmingValue(const OwnerToken& owner_token) { + return std::string(kServiceWarmingPrefix) + owner_token; +} + +bool IsServiceWarmingValue(const std::string& value) { + return value.rfind(kServiceWarmingPrefix, 0) == 0; +} // Fallback poll interval used only when a watch cannot be armed (e.g. the watch // RPC fails, or the remaining timeout is too small to be worth a watch). The // steady-state path blocks on an etcd watch and does not poll. @@ -185,6 +195,10 @@ EtcdLeaderCoordinator::ReadCurrentView() { return tl::make_unexpected(err); } + if (IsServiceWarmingValue(leader_address)) { + return std::optional{std::nullopt}; + } + return std::optional{ MasterView{.leader_address = std::move(leader_address), .view_version = view_version}}; @@ -205,9 +219,11 @@ EtcdLeaderCoordinator::TryAcquireLeadership(const std::string& leader_address) { } ViewVersionId view_version = 0; + const auto owner_token = MakeOwnerToken(lease_id); + const auto warming_value = MakeServiceWarmingValue(owner_token); err = EtcdHelper::CreateWithLease( master_view_key_.c_str(), master_view_key_.size(), - leader_address.c_str(), leader_address.size(), lease_id, view_version); + warming_value.c_str(), warming_value.size(), lease_id, view_version); if (err == ErrorCode::ETCD_TRANSACTION_FAIL) { auto revoke_err = EtcdHelper::RevokeLease(lease_id); if (revoke_err != ErrorCode::OK) { @@ -259,6 +275,27 @@ EtcdLeaderCoordinator::TryAcquireLeadership(const std::string& leader_address) { }; } +ErrorCode EtcdLeaderCoordinator::PublishServiceReady( + const LeadershipSession& session) { + auto err = EnsureConnected(); + if (err != ErrorCode::OK) { + return err; + } + const auto owner_token = session.owner_token; + const auto warming_value = MakeServiceWarmingValue(owner_token); + auto lease_id = ParseLeaseId(session.owner_token); + if (!lease_id) { + return lease_id.error(); + } + return EtcdHelper::TxnCompareAndPutWithLease( + {{.key = master_view_key_, + .kind = EtcdHelper::TxnCompareKind::kValueEquals, + .expected_value = warming_value}}, + {{.key = master_view_key_, + .value = session.view.leader_address}}, + lease_id.value()); +} + tl::expected EtcdLeaderCoordinator::RenewLeadership( const LeadershipSession& session) { auto err = EnsureConnected(); diff --git a/mooncake-store/src/ha/leadership/master_service_supervisor.cpp b/mooncake-store/src/ha/leadership/master_service_supervisor.cpp index fb61077341..fefc8c08b9 100644 --- a/mooncake-store/src/ha/leadership/master_service_supervisor.cpp +++ b/mooncake-store/src/ha/leadership/master_service_supervisor.cpp @@ -487,6 +487,20 @@ int RunSupervisorLoop(const HABackendSpec& spec, return -1; } + if (leader_coordinator.PublishServiceReady(*leadership_session) != + ErrorCode::OK) { + LOG(ERROR) << "Failed to publish master service readiness"; + StopLeadershipMonitor(leadership_monitor_handle); + server.stop(); + std::move(ec).get(); + DeactivateServingState(admin_server, label_reconciler); + EnterStandbyMode(admin_server, *standby_controller, + accept_standby_runtime_updates, + leadership_session->view); + leader_coordinator.ReleaseLeadership(*leadership_session); + continue; + } + if (!serve_shutdown_requested.load(std::memory_order_acquire)) { ActivateServingState(admin_server, wrapped_master_service, label_reconciler); diff --git a/mooncake-store/tests/rpc_timeout_test.cpp b/mooncake-store/tests/rpc_timeout_test.cpp index 68bfcea508..8f51ff36da 100644 --- a/mooncake-store/tests/rpc_timeout_test.cpp +++ b/mooncake-store/tests/rpc_timeout_test.cpp @@ -142,15 +142,24 @@ TEST(RpcTimeoutTest, TimeoutEnvOverridesAreOptIn) { EXPECT_EQ(defaults.request_timeout_duration, std::chrono::seconds(30)); EXPECT_EQ(defaults.connect_timeout_duration, std::chrono::seconds(30)); - // The master pool uses a bounded connection attempt and request deadline. + // Both modes use the bounded Master connection policy. auto master_config = detail::MakeMasterRpcClientPoolConfig(); - EXPECT_EQ(master_config.connect_retry_count, 1u); + EXPECT_EQ(master_config.connect_retry_count, 0u); EXPECT_EQ(master_config.reconnect_wait_time, std::chrono::milliseconds(0)); EXPECT_EQ(master_config.client_config.connect_timeout_duration, std::chrono::seconds(1)); EXPECT_EQ(master_config.client_config.request_timeout_duration, - std::chrono::seconds(5)); + std::chrono::seconds(30)); + + auto ha_config = detail::MakeMasterRpcClientPoolConfig(/*ha_enabled=*/true); + EXPECT_EQ(ha_config.connect_retry_count, 0u); + EXPECT_EQ(ha_config.reconnect_wait_time, + std::chrono::milliseconds(0)); + EXPECT_EQ(ha_config.client_config.connect_timeout_duration, + std::chrono::seconds(1)); + EXPECT_EQ(ha_config.client_config.request_timeout_duration, + std::chrono::seconds(30)); ASSERT_EQ(::setenv("MC_RPC_CONNECT_RETRY_COUNT", "3", 1), 0); ASSERT_EQ(::setenv("MC_RPC_RECONNECT_WAIT_MS", "250", 1), 0); @@ -159,6 +168,12 @@ TEST(RpcTimeoutTest, TimeoutEnvOverridesAreOptIn) { EXPECT_EQ(master_config.reconnect_wait_time, std::chrono::milliseconds(250)); + auto ha_overridden_config = + detail::MakeMasterRpcClientPoolConfig(/*ha_enabled=*/true); + EXPECT_EQ(ha_overridden_config.connect_retry_count, 3u); + EXPECT_EQ(ha_overridden_config.reconnect_wait_time, + std::chrono::milliseconds(250)); + ASSERT_EQ(::setenv("MC_RPC_TIMEOUT_MS", "1500", /*overwrite=*/1), 0); ASSERT_EQ(::setenv("MC_RPC_CONNECT_TIMEOUT_MS", "1000", /*overwrite=*/1), 0); @@ -170,12 +185,18 @@ TEST(RpcTimeoutTest, TimeoutEnvOverridesAreOptIn) { EXPECT_EQ(overridden.connect_timeout_duration, std::chrono::milliseconds(1000)); - // The master pool is built from the same helper, so it sees them too. + // The helper applies timeout overrides to the selected Master pool. master_config = detail::MakeMasterRpcClientPoolConfig(); EXPECT_EQ(master_config.client_config.request_timeout_duration, std::chrono::milliseconds(1500)); EXPECT_EQ(master_config.client_config.connect_timeout_duration, std::chrono::milliseconds(1000)); + ha_overridden_config = + detail::MakeMasterRpcClientPoolConfig(/*ha_enabled=*/true); + EXPECT_EQ(ha_overridden_config.client_config.request_timeout_duration, + std::chrono::milliseconds(1500)); + EXPECT_EQ(ha_overridden_config.client_config.connect_timeout_duration, + std::chrono::milliseconds(1000)); ::unsetenv("MC_RPC_TIMEOUT_MS"); ::unsetenv("MC_RPC_CONNECT_TIMEOUT_MS"); From f3687044ed1233d7b9534a57c8bb0b30cfafe581 Mon Sep 17 00:00:00 2001 From: lihenga <1438677126@qq.com> Date: Sat, 17 Oct 2026 13:59:10 +0800 Subject: [PATCH 5/5] test: align RPC pool lifetime expectation with validated integration --- mooncake-common/tests/rpc_client_io_context_test.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/mooncake-common/tests/rpc_client_io_context_test.cpp b/mooncake-common/tests/rpc_client_io_context_test.cpp index c3b18c7127..32392bb9ce 100644 --- a/mooncake-common/tests/rpc_client_io_context_test.cpp +++ b/mooncake-common/tests/rpc_client_io_context_test.cpp @@ -51,7 +51,8 @@ TEST(RpcClientIoContextPoolTest, ReplacesPoolWhenTargetChanges) { auto second = pools.GetOrCreateClientPool("127.0.0.1:10002"); EXPECT_NE(first, second); first.reset(); - EXPECT_TRUE(old_pool.expired()); + // The registry retains retired pools for background reconnect coroutines. + EXPECT_FALSE(old_pool.expired()); EXPECT_EQ(pools.GetClientPool(), second); }