[Store] Add opt-in replica placement shadow evaluator

mooncake-store/include/master_config.h

@@ -1,16 +1,24 @@
 #pragma once
 
+#include <chrono>
 #include <optional>
 #include <stdexcept>
 #include <string_view>
 
 #include <glog/logging.h>
 
 #include "config_helper.h"
+#include "replica_placement_shadow.h"
 #include "types.h"
 
 namespace mooncake {
 
+struct MasterReplicaPlacementShadowConfig {
+    ReplicaPlacementShadowConfig evaluator;
+    bool auto_collect_master_signals{true};
+    std::chrono::nanoseconds signal_refresh_interval{std::chrono::seconds(1)};
+};
+
 // Forwarded to the HA serve phase via MasterServiceSupervisorConfig.
 class HttpMetadataServer;
 
@@ -133,6 +141,12 @@ struct MasterConfig {
     // rich clusters may safely raise it.
     uint32_t promotion_max_per_heartbeat = 1;
 
+    // Dynamic replica placement observation. Empty keeps the feature fully
+    // disabled; targets have no built-in production defaults and must be
+    // supplied by the operator before the evaluator can be constructed.
+    std::optional<MasterReplicaPlacementShadowConfig>
+        replica_placement_shadow_config;
+
     // KV Events publisher (RFC #1527) for cache-aware indexers.
     bool enable_kv_events = false;
     std::string kv_events_bind_endpoint;
@@ -225,6 +239,8 @@ class MasterServiceSupervisorConfig {
     uint32_t promotion_admission_threshold = 2;
     uint32_t promotion_queue_limit = 50000;
     uint32_t promotion_max_per_heartbeat = 1;
+    std::optional<MasterReplicaPlacementShadowConfig>
+        replica_placement_shadow_config;
     bool enable_kv_events = false;
     std::string kv_events_bind_endpoint;
     std::string kv_events_model_name;
@@ -279,6 +295,8 @@ class MasterServiceSupervisorConfig {
         promotion_admission_threshold = config.promotion_admission_threshold;
         promotion_queue_limit = config.promotion_queue_limit;
         promotion_max_per_heartbeat = config.promotion_max_per_heartbeat;
+        replica_placement_shadow_config =
+            config.replica_placement_shadow_config;
         enable_kv_events = config.enable_kv_events;
         kv_events_bind_endpoint = config.kv_events_bind_endpoint;
         kv_events_model_name = config.kv_events_model_name;
@@ -441,6 +459,8 @@ class WrappedMasterServiceConfig {
     uint32_t promotion_admission_threshold = 2;
     uint32_t promotion_queue_limit = 50000;
     uint32_t promotion_max_per_heartbeat = 1;
+    std::optional<MasterReplicaPlacementShadowConfig>
+        replica_placement_shadow_config;
     bool enable_kv_events = false;
     std::string kv_events_bind_endpoint;
     std::string kv_events_model_name;
@@ -526,6 +546,8 @@ class WrappedMasterServiceConfig {
         promotion_admission_threshold = config.promotion_admission_threshold;
         promotion_queue_limit = config.promotion_queue_limit;
         promotion_max_per_heartbeat = config.promotion_max_per_heartbeat;
+        replica_placement_shadow_config =
+            config.replica_placement_shadow_config;
         enable_kv_events = config.enable_kv_events;
         kv_events_bind_endpoint = config.kv_events_bind_endpoint;
         kv_events_model_name = config.kv_events_model_name;
@@ -635,6 +657,8 @@ class WrappedMasterServiceConfig {
         promotion_admission_threshold = config.promotion_admission_threshold;
         promotion_queue_limit = config.promotion_queue_limit;
         promotion_max_per_heartbeat = config.promotion_max_per_heartbeat;
+        replica_placement_shadow_config =
+            config.replica_placement_shadow_config;
         enable_kv_events = config.enable_kv_events;
         kv_events_bind_endpoint = config.kv_events_bind_endpoint;
         kv_events_model_name = config.kv_events_model_name;
@@ -1057,6 +1081,8 @@ class MasterServiceConfig {
     uint32_t promotion_admission_threshold = 2;
     uint32_t promotion_queue_limit = 50000;
     uint32_t promotion_max_per_heartbeat = 1;
+    std::optional<MasterReplicaPlacementShadowConfig>
+        replica_placement_shadow_config;
     bool enable_kv_events = false;
     std::string kv_events_bind_endpoint;
     std::string kv_events_model_name;
@@ -1138,6 +1164,8 @@ class MasterServiceConfig {
         promotion_admission_threshold = config.promotion_admission_threshold;
         promotion_queue_limit = config.promotion_queue_limit;
         promotion_max_per_heartbeat = config.promotion_max_per_heartbeat;
+        replica_placement_shadow_config =
+            config.replica_placement_shadow_config;
         enable_kv_events = config.enable_kv_events;
         kv_events_bind_endpoint = config.kv_events_bind_endpoint;
         kv_events_model_name = config.kv_events_model_name;

mooncake-store/include/master_metric_manager.h

@@ -12,6 +12,8 @@
 
 namespace mooncake {
 
+struct ReplicaPlacementShadowResult;
+
 class MasterMetricManager {
    public:
     // --- Singleton Access ---
@@ -389,6 +391,12 @@ class MasterMetricManager {
     int64_t get_update_task_requests();
     int64_t get_update_task_failures();
 
+    // Replica placement SHADOW metrics have fixed-cardinality labels derived
+    // only from enums. Object keys, tenants and endpoints are deliberately not
+    // accepted by this API.
+    void observe_replica_placement_shadow(
+        const ReplicaPlacementShadowResult& result);
+
     // --- Serialization ---
     /**
      * @brief Serializes all managed metrics into Prometheus text format.
@@ -686,6 +694,11 @@ class MasterMetricManager {
     ylt::metric::dynamic_counter_2t tenant_quota_reject_total_;
     ylt::metric::dynamic_counter_1t tenant_evict_bytes_total_;
 
+    ylt::metric::dynamic_counter_2t replica_placement_shadow_observations_;
+    ylt::metric::dynamic_counter_2t replica_placement_shadow_add_intents_;
+    ylt::metric::dynamic_counter_2t replica_placement_shadow_remove_intents_;
+    ylt::metric::dynamic_counter_1t replica_placement_shadow_degraded_;
+
     // Snapshot Metrics
     ylt::metric::histogram_t snapshot_duration_ms_;
     ylt::metric::counter_t snapshot_success_;

mooncake-store/include/master_service.h

@@ -13,6 +13,7 @@
 #include <mutex>
 #include <optional>
 #include <shared_mutex>
+#include <span>
 #include <string>
 #include <thread>
 #include <unordered_map>
@@ -115,6 +116,14 @@ class MasterService {
         const UUID& segment_id);
     std::optional<TenantQuotaSnapshot> GetTenantQuotaSnapshotForTesting(
         const std::string& tenant_id) const;
+    [[nodiscard]] bool ReplicaPlacementShadowEnabled() const noexcept;
+    [[nodiscard]] ReplicaPlacementSignalPublishStatus
+    PublishReplicaPlacementSignalSnapshot(
+        ReplicaPlacementSignalSnapshot snapshot);
+    [[nodiscard]] uint64_t ReplicaPlacementShadowSignalGeneration()
+        const noexcept;
+    [[nodiscard]] std::optional<ReplicaPlacementShadowCountersSnapshot>
+    GetReplicaPlacementShadowCounters() const noexcept;
     bool IsTenantQuotaEnabled() const;
     std::vector<TenantQuotaSnapshot> ListTenantQuotaSnapshots() const;
     std::optional<TenantQuotaSnapshot> GetTenantQuotaSnapshot(
@@ -1938,6 +1947,26 @@ class MasterService {
     // from any GetReplicaList caller without additional locking.
     std::unique_ptr<CountMinSketch> promotion_sketch_;
 
+    // Optional read-path observer. It owns no task queue and cannot mutate
+    // metadata; only explicit configuration constructs it.
+    std::unique_ptr<ReplicaPlacementShadowEvaluator>
+        replica_placement_shadow_evaluator_;
+    bool replica_placement_shadow_auto_collect_signals_{false};
+    std::chrono::nanoseconds replica_placement_shadow_refresh_interval_{
+        std::chrono::seconds(1)};
+    std::atomic<bool> replica_placement_shadow_external_signal_source_{false};
+    std::mutex replica_placement_shadow_refresh_mutex_;
+    std::optional<ReplicaPlacementSignalClock::TimePoint>
+        replica_placement_shadow_last_refresh_;
+
+    void ObserveReplicaPlacementShadow(
+        const ObjectIdentity& object_id,
+        std::span<const Replica::Descriptor> replicas);
+    void RefreshReplicaPlacementShadowSignals();
+    [[nodiscard]] ReplicaPlacementSignalSnapshot
+    CollectReplicaPlacementShadowSignals(
+        ReplicaPlacementSignalClock::TimePoint observed_at);
+
     const std::string ha_backend_type_;
 
     const std::string ha_backend_connstring_;

mooncake-store/include/replica_placement_policy.h

@@ -0,0 +1,99 @@
+// Copyright 2026 Mooncake Authors
+
+#pragma once
+
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <span>
+
+namespace mooncake {
+
+enum class ReplicaPlacementTier : uint8_t {
+    MEMORY = 0,
+    LOCAL_DISK,
+    NOF_SSD,
+    REMOTE_STORE,
+    COUNT,
+};
+
+enum class ReplicaTemperature : uint8_t {
+    COLD = 0,
+    WARM,
+    HOT,
+    COUNT,
+};
+
+inline constexpr size_t kReplicaPlacementTierCount =
+    static_cast<size_t>(ReplicaPlacementTier::COUNT);
+inline constexpr size_t kReplicaTemperatureCount =
+    static_cast<size_t>(ReplicaTemperature::COUNT);
+
+struct ReplicaTierTarget {
+    uint32_t desired{0};
+    bool required{false};
+};
+
+struct ReplicaPlacementPolicyConfig {
+    std::array<std::array<ReplicaTierTarget, kReplicaPlacementTierCount>,
+               kReplicaTemperatureCount>
+        targets{};
+    uint32_t min_complete_replicas{1};
+    uint32_t max_total_replicas{16};
+};
+
+enum class ReplicaTierSignalState : uint8_t {
+    UNKNOWN = 0,
+    AVAILABLE,
+    UNAVAILABLE,
+};
+
+struct ReplicaTierObservation {
+    uint32_t complete{0};
+    uint32_t pending{0};
+    ReplicaTierSignalState allocation_state{ReplicaTierSignalState::UNKNOWN};
+    ReplicaTierSignalState health_state{ReplicaTierSignalState::UNKNOWN};
+};
+
+enum class ReplicaPlacementDegradedReason : uint8_t {
+    REQUIRED_TIER_UNAVAILABLE = 0,
+    REQUIRED_TIER_UNHEALTHY,
+    REQUIRED_TIER_SIGNAL_UNKNOWN,
+    MIN_COMPLETE_UNSATISFIED,
+};
+
+struct ReplicaTierAdjustment {
+    ReplicaPlacementTier tier{ReplicaPlacementTier::MEMORY};
+    uint32_t add{0};
+    uint32_t remove{0};
+    uint32_t effective_target{0};
+};
+
+struct ReplicaPlacementPlan {
+    std::array<ReplicaTierAdjustment, kReplicaPlacementTierCount> adjustments{};
+    std::array<ReplicaPlacementDegradedReason, kReplicaPlacementTierCount + 1>
+        degraded_reasons{};
+    size_t degraded_reason_count{0};
+
+    [[nodiscard]] bool degraded() const noexcept {
+        return degraded_reason_count != 0;
+    }
+};
+
+// Pure, immutable policy. It produces intent only; callers remain responsible
+// for admission, source/target selection, task submission, and transactional
+// metadata changes.
+class ReplicaPlacementPolicy final {
+   public:
+    explicit ReplicaPlacementPolicy(ReplicaPlacementPolicyConfig config);
+
+    [[nodiscard]] ReplicaPlacementPlan Plan(
+        ReplicaTemperature temperature,
+        std::span<const ReplicaTierObservation, kReplicaPlacementTierCount>
+            observations) const;
+
+   private:
+    ReplicaPlacementPolicyConfig config_;
+};
+
+}  // namespace mooncake