[Store] (CI run_tests_with_ssd failed / promotion-on-hit failed) eliminate race conditions (#2235)

* fix(transfer-engine): eliminate race condition in rePublishRpcMetaEntry

Remove the redundant storage_plugin_->remove() call before set().
All storage backends (HTTP PUT, Redis SET, Etcd put) have upsert
semantics, so remove-then-set creates a window where concurrent
get() returns empty / 404, causing transfer failures (-800).

Also change Json::UInt64 to Json::UInt for rpc_port to ensure
existing == desired comparison works correctly after JSON parse.

* fix: adjust eviction thread initial timing to prevent race in CI SSD tests

Start last_discard_time with an already-elapsed window so the first loop
iteration triggers DiscardExpiredProcessingReplicas immediately. Without
this, a task admitted shortly after thread startup can survive the first
reaper cycle and not be cleaned until ~2s later, causing promotion-on-hit
tests that sleep for 2s to flake.

Author: LujhCoconut

mooncake-store/src/master_service.cpp

@@ -3476,7 +3476,13 @@ auto MasterService::NotifyPromotionFailure(const UUID& client_id,
 void MasterService::EvictionThreadFunc() {
     VLOG(1) << "action=eviction_thread_started";
 
-    auto last_discard_time = std::chrono::system_clock::now();
+    // Start with an already-elapsed window so the first loop iteration
+    // (after kEvictionThreadSleepMs) triggers DiscardExpiredProcessingReplicas.
+    // Without this, a task admitted shortly after thread startup can survive
+    // the first reaper cycle and not be cleaned until ~2s later, causing
+    // promotion-on-hit tests that sleep for 2s to flake.
+    auto last_discard_time =
+        std::chrono::system_clock::now() - put_start_release_timeout_sec_;
     while (eviction_running_) {
         const auto now = std::chrono::system_clock::now();
         double used_ratio =

mooncake-transfer-engine/src/transfer_metadata.cpp

@@ -1128,7 +1128,7 @@ int TransferMetadata::addRpcMetaEntry(const std::string &server_name,
 
     Json::Value rpcMetaJSON;
     rpcMetaJSON["ip_or_host_name"] = desc.ip_or_host_name;
-    rpcMetaJSON["rpc_port"] = static_cast<Json::UInt64>(desc.rpc_port);
+    rpcMetaJSON["rpc_port"] = static_cast<Json::UInt>(desc.rpc_port);
     if (!storage_plugin_->set(rpc_meta_prefix_ + server_name, rpcMetaJSON)) {
         LOG(ERROR) << "Failed to set location of " << server_name;
         return ERR_METADATA;
@@ -1157,19 +1157,16 @@ int TransferMetadata::rePublishRpcMetaEntry(const std::string &server_name) {
     if (storage_plugin_->get(full_key, existing)) {
         Json::Value desired;
         desired["ip_or_host_name"] = local_rpc_meta_.ip_or_host_name;
-        desired["rpc_port"] =
-            static_cast<Json::UInt64>(local_rpc_meta_.rpc_port);
+        desired["rpc_port"] = static_cast<Json::UInt>(local_rpc_meta_.rpc_port);
         if (existing == desired) {
             return 0;
         }
-        storage_plugin_->remove(full_key);
     }
 
     LOG(INFO) << "Re-publishing RPC meta entry for " << server_name;
     Json::Value rpcMetaJSON;
     rpcMetaJSON["ip_or_host_name"] = local_rpc_meta_.ip_or_host_name;
-    rpcMetaJSON["rpc_port"] =
-        static_cast<Json::UInt64>(local_rpc_meta_.rpc_port);
+    rpcMetaJSON["rpc_port"] = static_cast<Json::UInt>(local_rpc_meta_.rpc_port);
     if (!storage_plugin_->set(full_key, rpcMetaJSON)) {
         LOG(ERROR) << "Failed to re-publish RPC meta entry for " << server_name;
         return ERR_METADATA;