[Store] Extract per-object metadata and runtime state headers from MasterService

mooncake-store/include/object_runtime_state.h

@@ -0,0 +1,127 @@
+#pragma once
+
+// Per-key runtime state records: the in-flight replication, offloading and
+// promotion bookkeeping that belongs to one object key. Moved out of
+// MasterService unchanged; each record is small state, not a container.
+
+#include <chrono>
+#include <cstdint>
+#include <string>
+#include <vector>
+
+#include "replica.h"
+#include "types.h"
+
+namespace mooncake {
+
+struct DynamicReplicaPending {
+    UUID proposal_id{};
+    UUID lease_id{};
+    std::string source_segment;
+    std::string target_segment;
+    std::string target_domain;
+    uint64_t version_epoch{0};
+    int64_t expire_at_ms_epoch{0};
+    UUID task_id{};
+};
+
+struct ReplicationTask {
+    UUID client_id;
+    std::chrono::system_clock::time_point start_time;
+    enum class Type {
+        COPY,
+        MOVE,
+    } type;
+    ReplicaID source_id;
+    std::vector<ReplicaID> replica_ids;
+    uint64_t pending_quota_charge_bytes{0};
+    UUID dynamic_replication_lease_id{};
+    uint64_t dynamic_replication_version_epoch{0};
+    bool durable_cleanup_pending{false};
+};
+
+struct OffloadingTask {
+    ReplicaID source_id;
+    std::chrono::system_clock::time_point start_time;
+    // Clients whose LocalDiskSegment::offloading_objects hold a mirror
+    // for this key. One marker can cover several mirrors, since the
+    // offload is pushed once per completed MEMORY replica and those
+    // replicas may live on different clients.
+    std::vector<UUID> mirror_clients;
+};
+
+// Tracks an in-flight LOCAL_DISK -> MEMORY copy. The source
+// LOCAL_DISK replica is refcnt-pinned for the duration of the task
+// so it cannot be evicted.
+//
+// alloc_id pins down which staged PROCESSING MEMORY replica
+enum class PromotionQueueResult {
+    kQueued,
+    kDisabled,
+    kFrequencyRejected,
+    kWatermarkRejected,
+    kQueueCapRejected,
+    kAlreadyInFlight,
+    kMemoryReplicaPresent,
+    kNoLocalDiskSource,
+    kNotFound,
+    kPushFailed,
+};
+
+enum class PromotionCandidateReason {
+    kWatermark,
+    kQueueCap,
+    kPushFailed,
+    kExecutionFailed,
+};
+
+struct PromotionCandidate {
+    uint8_t sketch_score{0};
+    std::chrono::steady_clock::time_point first_seen;
+    std::chrono::steady_clock::time_point last_seen;
+    std::chrono::steady_clock::time_point retry_after;
+    PromotionCandidateReason last_reason{PromotionCandidateReason::kQueueCap};
+    ErrorCode last_error{ErrorCode::OK};
+    uint32_t retry_count{0};
+    // Execution failures in this admission chain (AllocStart / TE-write /
+    // SSD failures reported via NotifyPromotionFailure). Propagated into
+    // PromotionTask at admission so the bound survives the candidate's
+    // consumption; reset only when a genuinely new chain starts (fresh
+    // insert with 0, e.g. after a give-up or a success).
+    uint32_t execution_failures{0};
+};
+
+// NotifyPromotionSuccess should commit, so a concurrent Put on the
+// same key cannot be confused with ours. 0 until
+// PromotionAllocStart records the new replica.
+//
+// start_time is the reaper deadline anchor. Set at task admission
+// and reset at PromotionAllocStart so each phase (queue-wait and
+// active-transfer) gets its own full put_start_release_timeout_sec_
+// window. Without the reset a backlogged task could enter active
+// transfer with little TTL left, and the reaper could free the
+// staged replica via EraseReplicaByID mid-RDMA-write.
+//
+// holder_id is the client owning the source LOCAL_DISK segment and
+// the only one authorized to commit (NotifyPromotionSuccess) or
+// abort (NotifyPromotionFailure) the task. Without it, any client
+// knowing the key could flip the staged PROCESSING replica to
+// COMPLETE before the holder's RDMA write landed, exposing torn
+// data to readers.
+struct PromotionTask {
+    ReplicaID source_id;    // the LOCAL_DISK replica being promoted
+    ReplicaID alloc_id{0};  // the new MEMORY replica staged by AllocStart
+    uint64_t object_size;
+    uint64_t pending_quota_charge_bytes{0};
+    std::chrono::system_clock::time_point start_time;
+    UUID holder_id;  // owner of source LOCAL_DISK; only Notifier allowed
+    // Execution failures so far in this admission chain. Read by
+    // NotifyPromotionFailure before the task is erased and re-recorded as
+    // execution_failures+1 until kMaxPromotionExecutionFailures. Note the
+    // asymmetry with PromotionCandidate::execution_failures: admission
+    // copies candidate -> task verbatim, failure re-record writes
+    // task+1 -> candidate.
+    uint32_t execution_failures{0};
+};
+
+}  // namespace mooncake

mooncake-store/src/master_service.cpp

@@ -2314,15 +2314,15 @@ tl::expected<void, ErrorCode> MasterService::PersistStaleHandleCleanupForHA(
     return {};
 }
 
-std::unordered_map<std::string, MasterService::ObjectMetadata>::iterator
+std::unordered_map<std::string, ObjectMetadata>::iterator
 MasterService::EraseMetadata(
     TenantState& tenant_state,
     std::unordered_map<std::string, ObjectMetadata>::iterator it,
     const TenantId& tenant_id) {
     return EraseMetadata(tenant_state, it, tenant_id, QuotaEraseMode::kFull);
 }
 
-std::unordered_map<std::string, MasterService::ObjectMetadata>::iterator
+std::unordered_map<std::string, ObjectMetadata>::iterator
 MasterService::EraseMetadata(
     TenantState& tenant_state,
     std::unordered_map<std::string, ObjectMetadata>::iterator it,
@@ -2334,7 +2334,7 @@ MasterService::EraseMetadata(
 // associated per-key state: offloading_tasks (with dec_refcnt),
 // processing_keys, replication_tasks, and promotion tasks.
 // Callers no longer need to clean these up manually before calling.
-std::unordered_map<std::string, MasterService::ObjectMetadata>::iterator
+std::unordered_map<std::string, ObjectMetadata>::iterator
 MasterService::EraseMetadata(
     TenantState& tenant_state,
     std::unordered_map<std::string, ObjectMetadata>::iterator it,
@@ -2683,7 +2683,8 @@ auto MasterService::ResolveSoftPinRequest(const ReplicateConfig& config) const
                            << ", error=ttl_requires_enable";
                 return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
             }
-            return ResolvedSoftPinRequest{config.soft_pin_action, 0};
+            return ResolvedSoftPinRequest{config.soft_pin_action,
+                                          std::chrono::milliseconds::zero()};
         case SoftPinAction::ENABLE: {
             const uint64_t ttl_ms =
                 config.soft_pin_ttl_ms.value_or(default_kv_soft_pin_ttl_);
@@ -2693,7 +2694,8 @@ auto MasterService::ResolveSoftPinRequest(const ReplicateConfig& config) const
                            << ", error=soft_pin_ttl_exceeds_limit";
                 return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
             }
-            return ResolvedSoftPinRequest{config.soft_pin_action, ttl_ms};
+            return ResolvedSoftPinRequest{config.soft_pin_action,
+                                          std::chrono::milliseconds(ttl_ms)};
         }
     }
     LOG(ERROR) << "soft_pin_action="
@@ -9593,7 +9595,7 @@ tl::expected<UUID, ErrorCode> MasterService::SubmitDynamicReplicaCopyTask(
              .dynamic_replication_version_epoch = version_epoch});
 }
 
-MasterService::PromotionQueueResult MasterService::TryPushPromotionQueue(
+PromotionQueueResult MasterService::TryPushPromotionQueue(
     const ObjectIdentity& object_id, bool record_candidate) {
     if (!promotion_on_hit_ || !promotion_sketch_) {
         return PromotionQueueResult::kDisabled;
@@ -13076,8 +13078,7 @@ MasterService::MetadataSerializer::DeserializeShard(const msgpack::object& obj,
 
 tl::expected<void, SerializationError>
 MasterService::MetadataSerializer::SerializeMetadata(
-    const MasterService::ObjectMetadata& metadata,
-    MsgpackPacker& packer) const {
+    const ObjectMetadata& metadata, MsgpackPacker& packer) const {
     // Pack ObjectMetadata using array structure for efficiency
     // Format: [client_id, put_start_time, size, lease_timeout,
     // has_soft_pin_timeout, soft_pin_timeout, replicas_count, data_type,
@@ -13141,7 +13142,7 @@ MasterService::MetadataSerializer::SerializeMetadata(
     return {};
 }
 
-tl::expected<std::unique_ptr<MasterService::ObjectMetadata>, SerializationError>
+tl::expected<std::unique_ptr<ObjectMetadata>, SerializationError>
 MasterService::MetadataSerializer::DeserializeMetadata(
     const msgpack::object& obj) const {
     // Check if input is a valid array

mooncake-store/tests/ha/master_service_ha_test.cpp

@@ -645,12 +645,11 @@ class MasterServiceHATest : public ::testing::Test {
         auto& tenant_state =
             service->GetOrCreateTenantState(shard_access.get(), tenant);
         tenant_state.promotion_tasks.emplace(
-            key, MasterService::PromotionTask{
-                     .source_id = 0,
-                     .alloc_id = alloc_id,
-                     .object_size = object_size,
-                     .start_time = std::chrono::system_clock::now(),
-                     .holder_id = holder_id});
+            key, PromotionTask{.source_id = 0,
+                               .alloc_id = alloc_id,
+                               .object_size = object_size,
+                               .start_time = std::chrono::system_clock::now(),
+                               .holder_id = holder_id});
     }
 
     static bool SnapshotManagerCreatedForTesting(const MasterService& service) {

mooncake-store/tests/master_service_test.cpp

@@ -303,16 +303,18 @@ TEST_F(MasterServiceTest, SoftPinDeadlineCalculationSaturatesAtMaximum) {
     using Clock = std::chrono::system_clock;
 
     const auto normal_now = Clock::time_point(std::chrono::seconds(10));
-    EXPECT_EQ(ComputeSoftPinDeadlineForTest(normal_now, 25),
+    EXPECT_EQ(ComputeSoftPinDeadlineForTest(normal_now,
+                                            std::chrono::milliseconds(25)),
               normal_now + std::chrono::milliseconds(25));
-    EXPECT_EQ(ComputeSoftPinDeadlineForTest(
-                  normal_now, std::numeric_limits<uint64_t>::max()),
+    EXPECT_EQ(ComputeSoftPinDeadlineForTest(normal_now,
+                                            std::chrono::milliseconds::max()),
               Clock::time_point::max());
 
     const auto near_max =
         Clock::time_point::max() - std::chrono::milliseconds(5);
-    EXPECT_EQ(ComputeSoftPinDeadlineForTest(near_max, 10),
-              Clock::time_point::max());
+    EXPECT_EQ(
+        ComputeSoftPinDeadlineForTest(near_max, std::chrono::milliseconds(10)),
+        Clock::time_point::max());
 }
 
 #ifdef USE_NOF

mooncake-store/tests/master_service_test_fixture.h

@@ -252,9 +252,9 @@ class MasterServiceTest : public ::testing::Test {
     }
 
     std::chrono::system_clock::time_point ComputeSoftPinDeadlineForTest(
-        const std::chrono::system_clock::time_point& now, uint64_t ttl_ms) {
-        return MasterService::ObjectMetadata::ComputeSoftPinDeadline(now,
-                                                                     ttl_ms);
+        const std::chrono::system_clock::time_point& now,
+        std::chrono::milliseconds ttl) {
+        return ObjectMetadata::ComputeSoftPinDeadline(now, ttl);
     }
 
     std::string WriteTenantPolicyFile(

mooncake-store/tests/promotion_on_hit_test.cpp

@@ -143,7 +143,7 @@ class PromotionOnHitTest : public ::testing::Test {
         const auto result =
             service->TryPushPromotionQueue(MasterService::ObjectIdentity{
                 .tenant_id = tenant_id, .user_key = key});
-        return result == MasterService::PromotionQueueResult::kAlreadyInFlight;
+        return result == PromotionQueueResult::kAlreadyInFlight;
     }
 
     static void MarkReplicaCompleteForTesting(MasterService* service,