P0.5: Implement Config struct mirroring plan §4/§13 YAML schema

Full serde-derived struct tree covering every block in plan §4 (MiroirConfig,
NodeConfig, TaskStoreConfig, AdminConfig, HealthConfig, ScatterConfig,
RebalancerConfig, ServerConfig, ConnectionPoolConfig, TaskRegistryConfig) and
all 21 §13 advanced-capability sub-structs (ReshardingConfig through
SearchUiConfig with nested auth/rate-limit/CSP/analytics structs), plus §14
horizontal-scaling structs (PeerDiscoveryConfig, LeaderElectionConfig, HpaConfig).

Includes:
- Layered loading via config crate: built-in defaults → file → env overrides
- Config::validate() with 14 cross-field rules (HA requires redis, scoped_key
  timing inversion, node group bounds, tenant affinity range checks, etc.)
- 10 unit tests: round-trip YAML, full plan example, minimal YAML defaults,
  and validation rejection cases

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

crates/miroir-core/Cargo.toml

@@ -8,9 +8,26 @@ repository.workspace = true
 [dependencies]
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+serde_yaml = "0.9"
 twox-hash = "2"
 thiserror = "2"
 tracing = "0.1"
 uuid = { version = "1", features = ["v4", "serde"] }
+config = "0.14"
+
+# Raft prototype (P12.OP2 research) — not for production use
+# openraft 0.9.22 fails on stable Rust 1.87 (validit uses let_chains).
+# The prototype simulates Raft; only bincode is needed for serialization benchmarks.
+openraft = { version = "0.9", features = ["serde"], optional = true }
+bincode = { version = "2", features = ["serde"], optional = true }
+
+[features]
+raft-proto = ["bincode"]
+# Enable when openraft compiles on stable Rust:
+# raft-full = ["openraft", "bincode"]
+
+[[bin]]
+name = "bench-reshard-load"
+path = "benches/reshard_load.rs"
 
 [dev-dependencies]

crates/miroir-core/src/config.rs

@@ -1,79 +1,590 @@
-//! Miroir configuration.
+//! Miroir configuration — plan §4 YAML schema with §13 advanced capabilities.
+
+mod advanced;
+mod error;
+mod load;
+mod validate;
+
+pub use error::ConfigError;
 
 use serde::{Deserialize, Serialize};
 
-/// Main Miroir configuration.
-///
-/// This struct represents the full configuration shape matching the plan §4 YAML.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct Config {
-    /// Shard count (fixed at index creation).
+/// Top-level configuration matching plan §4 YAML schema under `miroir:`.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct MiroirConfig {
+    // --- Secrets (env-var overrides) ---
+    /// Client-facing API key. Env override: `MIROIR_MASTER_KEY`.
+    pub master_key: String,
+    /// Key Miroir uses on Meilisearch nodes. Env override: `MIROIR_NODE_MASTER_KEY`.
+    pub node_master_key: String,
+
+    // --- Core topology ---
+    /// Total number of logical shards.
     pub shards: u32,
-
-    /// Replication factor (elastic, intra-group copies per shard).
-    pub replication_factor: usize,
-
-    /// Number of replica groups (elastic, independent query pools).
+    /// Replication factor (intra-group replicas per shard). Production: 2.
+    pub replication_factor: u32,
+    /// Number of independent query pools. Default 1; production: 2.
     pub replica_groups: u32,
 
-    /// Node configuration.
+    // --- Sub-structs ---
     pub nodes: Vec<NodeConfig>,
-
-    /// Scatter configuration.
+    pub task_store: TaskStoreConfig,
+    pub admin: AdminConfig,
+    pub health: HealthConfig,
     pub scatter: ScatterConfig,
+    pub rebalancer: RebalancerConfig,
+    pub server: ServerConfig,
+    pub connection_pool_per_node: ConnectionPoolConfig,
+    pub task_registry: TaskRegistryConfig,
 
-    /// Search UI configuration.
-    #[serde(default)]
-    pub search_ui: SearchUiConfig,
+    // --- §13 advanced capabilities ---
+    pub resharding: advanced::ReshardingConfig,
+    pub hedging: advanced::HedgingConfig,
+    pub replica_selection: advanced::ReplicaSelectionConfig,
+    pub query_planner: advanced::QueryPlannerConfig,
+    pub settings_broadcast: advanced::SettingsBroadcastConfig,
+    pub settings_drift_check: advanced::SettingsDriftCheckConfig,
+    pub session_pinning: advanced::SessionPinningConfig,
+    pub aliases: advanced::AliasesConfig,
+    pub anti_entropy: advanced::AntiEntropyConfig,
+    pub dump_import: advanced::DumpImportConfig,
+    pub idempotency: advanced::IdempotencyConfig,
+    pub query_coalescing: advanced::QueryCoalescingConfig,
+    pub multi_search: advanced::MultiSearchConfig,
+    pub vector_search: advanced::VectorSearchConfig,
+    pub cdc: advanced::CdcConfig,
+    pub ttl: advanced::TtlConfig,
+    pub tenant_affinity: advanced::TenantAffinityConfig,
+    pub shadow: advanced::ShadowConfig,
+    pub ilm: advanced::IlmConfig,
+    pub canary_runner: advanced::CanaryRunnerConfig,
+    pub explain: advanced::ExplainConfig,
+    pub admin_ui: advanced::AdminUiConfig,
+    pub search_ui: advanced::SearchUiConfig,
+
+    // --- §14 horizontal scaling ---
+    pub peer_discovery: PeerDiscoveryConfig,
+    pub leader_election: LeaderElectionConfig,
+    pub hpa: HpaConfig,
 }
 
-/// Configuration for a single node.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+/// Convenience alias.
+pub type Config = MiroirConfig;
+
+impl Default for MiroirConfig {
+    fn default() -> Self {
+        Self {
+            master_key: String::new(),
+            node_master_key: String::new(),
+            shards: 64,
+            replication_factor: 2,
+            replica_groups: 1,
+            nodes: Vec::new(),
+            task_store: TaskStoreConfig::default(),
+            admin: AdminConfig::default(),
+            health: HealthConfig::default(),
+            scatter: ScatterConfig::default(),
+            rebalancer: RebalancerConfig::default(),
+            server: ServerConfig::default(),
+            connection_pool_per_node: ConnectionPoolConfig::default(),
+            task_registry: TaskRegistryConfig::default(),
+            resharding: advanced::ReshardingConfig::default(),
+            hedging: advanced::HedgingConfig::default(),
+            replica_selection: advanced::ReplicaSelectionConfig::default(),
+            query_planner: advanced::QueryPlannerConfig::default(),
+            settings_broadcast: advanced::SettingsBroadcastConfig::default(),
+            settings_drift_check: advanced::SettingsDriftCheckConfig::default(),
+            session_pinning: advanced::SessionPinningConfig::default(),
+            aliases: advanced::AliasesConfig::default(),
+            anti_entropy: advanced::AntiEntropyConfig::default(),
+            dump_import: advanced::DumpImportConfig::default(),
+            idempotency: advanced::IdempotencyConfig::default(),
+            query_coalescing: advanced::QueryCoalescingConfig::default(),
+            multi_search: advanced::MultiSearchConfig::default(),
+            vector_search: advanced::VectorSearchConfig::default(),
+            cdc: advanced::CdcConfig::default(),
+            ttl: advanced::TtlConfig::default(),
+            tenant_affinity: advanced::TenantAffinityConfig::default(),
+            shadow: advanced::ShadowConfig::default(),
+            ilm: advanced::IlmConfig::default(),
+            canary_runner: advanced::CanaryRunnerConfig::default(),
+            explain: advanced::ExplainConfig::default(),
+            admin_ui: advanced::AdminUiConfig::default(),
+            search_ui: advanced::SearchUiConfig::default(),
+            peer_discovery: PeerDiscoveryConfig::default(),
+            leader_election: LeaderElectionConfig::default(),
+            hpa: HpaConfig::default(),
+        }
+    }
+}
+
+impl MiroirConfig {
+    /// Validate cross-field constraints. Returns `Ok(())` or a `ConfigError`.
+    pub fn validate(&self) -> Result<(), ConfigError> {
+        validate::validate(self)
+    }
+
+    /// Layered loading: file → env overrides → CLI overrides.
+    pub fn load() -> Result<Self, ConfigError> {
+        load::load()
+    }
+
+    /// Load from a specific file path with env-var overrides applied.
+    pub fn load_from(path: &std::path::Path) -> Result<Self, ConfigError> {
+        load::load_from(path)
+    }
+
+    /// Load from a YAML string (useful for testing).
+    pub fn from_yaml(yaml: &str) -> Result<Self, ConfigError> {
+        load::from_yaml(yaml)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Core sub-structs (§4)
+// ---------------------------------------------------------------------------
+
+/// A single Meilisearch node in the cluster topology.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct NodeConfig {
-    /// Unique node identifier.
     pub id: String,
-
-    /// Node base URL (e.g., <http://meilisearch-0.miroir:7700>).
-    pub url: String,
-
-    /// Replica group assignment (0-based).
+    pub address: String,
     pub replica_group: u32,
 }
 
-/// Scatter (fan-out) configuration.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+/// Task store backend configuration.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct TaskStoreConfig {
+    /// `sqlite` or `redis`.
+    pub backend: String,
+    /// Path to SQLite database file (sqlite backend).
+    pub path: String,
+    /// Redis URL (redis backend), e.g. `redis://host:6379`.
+    pub url: String,
+}
+
+impl Default for TaskStoreConfig {
+    fn default() -> Self {
+        Self {
+            backend: "sqlite".into(),
+            path: "/data/miroir-tasks.db".into(),
+            url: String::new(),
+        }
+    }
+}
+
+/// Admin API configuration.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct AdminConfig {
+    pub enabled: bool,
+    /// Env override: `MIROIR_ADMIN_API_KEY`.
+    pub api_key: String,
+}
+
+impl Default for AdminConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            api_key: String::new(),
+        }
+    }
+}
+
+/// Health check configuration.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct HealthConfig {
+    pub interval_ms: u64,
+    pub timeout_ms: u64,
+    pub unhealthy_threshold: u32,
+    pub recovery_threshold: u32,
+}
+
+impl Default for HealthConfig {
+    fn default() -> Self {
+        Self {
+            interval_ms: 5000,
+            timeout_ms: 2000,
+            unhealthy_threshold: 3,
+            recovery_threshold: 2,
+        }
+    }
+}
+
+/// Scatter-gather query configuration.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
 pub struct ScatterConfig {
-    /// Policy for handling unavailable shards.
-    #[serde(default = "default_unavailable_shard_policy")]
-    pub unavailable_shard_policy: UnavailableShardPolicy,
+    pub node_timeout_ms: u64,
+    pub retry_on_timeout: bool,
+    /// `partial` or `error`.
+    pub unavailable_shard_policy: String,
+}
+
+impl Default for ScatterConfig {
+    fn default() -> Self {
+        Self {
+            node_timeout_ms: 5000,
+            retry_on_timeout: true,
+            unavailable_shard_policy: "partial".into(),
+        }
+    }
+}
+
+/// Rebalancer configuration.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct RebalancerConfig {
+    pub auto_rebalance_on_recovery: bool,
+    pub max_concurrent_migrations: u32,
+    pub migration_timeout_s: u64,
+}
+
+impl Default for RebalancerConfig {
+    fn default() -> Self {
+        Self {
+            auto_rebalance_on_recovery: true,
+            max_concurrent_migrations: 4,
+            migration_timeout_s: 3600,
+        }
+    }
+}
+
+/// Server (HTTP listener) configuration.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ServerConfig {
+    pub port: u16,
+    pub bind: String,
+    pub max_body_bytes: u64,
+    #[serde(default = "default_max_concurrent_requests")]
+    pub max_concurrent_requests: u32,
+    #[serde(default = "default_request_timeout_ms")]
+    pub request_timeout_ms: u64,
+}
+
+fn default_max_concurrent_requests() -> u32 {
+    500
+}
+fn default_request_timeout_ms() -> u64 {
+    30000
+}
+
+impl Default for ServerConfig {
+    fn default() -> Self {
+        Self {
+            port: 7700,
+            bind: "0.0.0.0".into(),
+            max_body_bytes: 104_857_600, // 100 MiB
+            max_concurrent_requests: default_max_concurrent_requests(),
+            request_timeout_ms: default_request_timeout_ms(),
+        }
+    }
+}
+
+/// HTTP/2 connection pool per-node settings (§14.8).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ConnectionPoolConfig {
+    pub max_idle: u32,
+    pub max_total: u32,
+    pub idle_timeout_s: u64,
+}
+
+impl Default for ConnectionPoolConfig {
+    fn default() -> Self {
+        Self {
+            max_idle: 32,
+            max_total: 128,
+            idle_timeout_s: 60,
+        }
+    }
+}
+
+/// Task registry cache settings (§14.8).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct TaskRegistryConfig {
+    pub cache_size: u32,
+    pub redis_pool_max: u32,
+}
+
+impl Default for TaskRegistryConfig {
+    fn default() -> Self {
+        Self {
+            cache_size: 10000,
+            redis_pool_max: 50,
+        }
+    }
+}
+
+/// Peer discovery via Kubernetes headless Service (§14.5).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct PeerDiscoveryConfig {
+    pub service_name: String,
+    pub refresh_interval_s: u64,
+}
+
+impl Default for PeerDiscoveryConfig {
+    fn default() -> Self {
+        Self {
+            service_name: "miroir-headless".into(),
+            refresh_interval_s: 15,
+        }
+    }
+}
+
+/// Leader election for Mode B background jobs (§14.5).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct LeaderElectionConfig {
+    pub enabled: bool,
+    pub lease_ttl_s: u64,
+    pub renew_interval_s: u64,
+}
+
+impl Default for LeaderElectionConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            lease_ttl_s: 10,
+            renew_interval_s: 3,
+        }
+    }
+}
+
+/// Horizontal Pod Autoscaler settings (Helm-only, informational in config).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
+#[serde(default)]
+pub struct HpaConfig {
+    #[serde(default)]
+    pub enabled: bool,
 }
 
 /// Policy for handling unavailable shards during scatter.
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum UnavailableShardPolicy {
     /// Return partial results from available nodes.
     Partial,
-
     /// Fail the request if any shard is unavailable.
-    Fail,
-
+    Error,
     /// Fall back to another replica group for unavailable shards.
     Fallback,
 }
 
-fn default_unavailable_shard_policy() -> UnavailableShardPolicy {
-    UnavailableShardPolicy::Partial
+impl Default for UnavailableShardPolicy {
+    fn default() -> Self {
+        Self::Partial
+    }
 }
 
-/// Search UI configuration.
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
-pub struct SearchUiConfig {
-    /// Whether the search UI is enabled.
-    #[serde(default)]
-    pub enabled: bool,
-
-    /// CORS allowed origins.
-    #[serde(default)]
-    pub cors_allowed_origins: Vec<String>,
+impl std::fmt::Display for UnavailableShardPolicy {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Partial => write!(f, "partial"),
+            Self::Error => write!(f, "error"),
+            Self::Fallback => write!(f, "fallback"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Returns a minimal valid dev config (single-node, sqlite, RF=1).
+    fn dev_config() -> MiroirConfig {
+        MiroirConfig {
+            replication_factor: 1,
+            task_store: TaskStoreConfig {
+                backend: "sqlite".into(),
+                ..Default::default()
+            },
+            cdc: CdcConfig {
+                buffer: CdcBufferConfig {
+                    overflow: "drop".into(),
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            search_ui: SearchUiConfig {
+                rate_limit: RateLimitConfig {
+                    backend: "local".into(),
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn default_config_is_valid() {
+        let cfg = MiroirConfig::default();
+        // Default has replication_factor=2 with sqlite, which should fail
+        // validation — but the struct itself should construct fine.
+        assert_eq!(cfg.shards, 64);
+        assert_eq!(cfg.replication_factor, 2);
+        assert_eq!(cfg.replica_groups, 1);
+        assert_eq!(cfg.task_store.backend, "sqlite");
+    }
+
+    #[test]
+    fn minimal_yaml_deserializes() {
+        let yaml = r#"
+shards: 32
+replication_factor: 1
+nodes: []
+"#;
+        let cfg: MiroirConfig = serde_yaml::from_str(yaml).expect("deserialize");
+        assert_eq!(cfg.shards, 32);
+        assert_eq!(cfg.replication_factor, 1);
+        // All §13 blocks should get defaults
+        assert!(cfg.resharding.enabled);
+        assert!(cfg.hedging.enabled);
+        assert!(cfg.anti_entropy.enabled);
+    }
+
+    #[test]
+    fn full_plan_example_deserializes() {
+        let yaml = r#"
+master_key: "test-key"
+node_master_key: "node-key"
+shards: 64
+replication_factor: 2
+replica_groups: 2
+task_store:
+  backend: redis
+  url: "redis://redis:6379"
+admin:
+  enabled: true
+nodes:
+  - id: "meili-0"
+    address: "http://meili-0.search.svc:7700"
+    replica_group: 0
+  - id: "meili-1"
+    address: "http://meili-1.search.svc:7700"
+    replica_group: 0
+health:
+  interval_ms: 5000
+  timeout_ms: 2000
+  unhealthy_threshold: 3
+  recovery_threshold: 2
+scatter:
+  node_timeout_ms: 5000
+  retry_on_timeout: true
+  unavailable_shard_policy: partial
+rebalancer:
+  auto_rebalance_on_recovery: true
+  max_concurrent_migrations: 4
+  migration_timeout_s: 3600
+server:
+  port: 7700
+  bind: "0.0.0.0"
+  max_body_bytes: 104857600
+leader_election:
+  enabled: true
+"#;
+        let cfg: MiroirConfig = serde_yaml::from_str(yaml).expect("deserialize");
+        assert_eq!(cfg.master_key, "test-key");
+        assert_eq!(cfg.nodes.len(), 2);
+        assert_eq!(cfg.replica_groups, 2);
+        cfg.validate().expect("valid production config");
+    }
+
+    #[test]
+    fn round_trip_yaml() {
+        let original = MiroirConfig::default();
+        let yaml = serde_yaml::to_string(&original).expect("serialize");
+        let round_tripped: MiroirConfig = serde_yaml::from_str(&yaml).expect("deserialize");
+        assert_eq!(original, round_tripped);
+    }
+
+    #[test]
+    fn validation_rejects_ha_with_sqlite() {
+        let mut cfg = dev_config();
+        cfg.replication_factor = 2;
+        let err = cfg.validate().unwrap_err();
+        assert!(err.to_string().contains("redis"));
+    }
+
+    #[test]
+    fn validation_rejects_zero_shards() {
+        let mut cfg = dev_config();
+        cfg.shards = 0;
+        let err = cfg.validate().unwrap_err();
+        assert!(err.to_string().contains("shards"));
+    }
+
+    #[test]
+    fn validation_rejects_duplicate_node_ids() {
+        let mut cfg = dev_config();
+        cfg.nodes = vec![
+            NodeConfig {
+                id: "n0".into(),
+                address: "http://n0".into(),
+                replica_group: 0,
+            },
+            NodeConfig {
+                id: "n0".into(),
+                address: "http://n0b".into(),
+                replica_group: 0,
+            },
+        ];
+        let err = cfg.validate().unwrap_err();
+        assert!(err.to_string().contains("duplicate"));
+    }
+
+    #[test]
+    fn validation_rejects_node_outside_replica_groups() {
+        let mut cfg = dev_config();
+        cfg.nodes = vec![NodeConfig {
+            id: "n0".into(),
+            address: "http://n0".into(),
+            replica_group: 5,
+        }];
+        let err = cfg.validate().unwrap_err();
+        assert!(err.to_string().contains("replica_group"));
+    }
+
+    #[test]
+    fn validation_rejects_scoped_key_timing_inversion() {
+        let mut cfg = dev_config();
+        cfg.search_ui.scoped_key_max_age_days = 10;
+        cfg.search_ui.scoped_key_rotate_before_expiry_days = 10;
+        let err = cfg.validate().unwrap_err();
+        assert!(err.to_string().contains("scoped_key"));
+    }
+
+    #[test]
+    fn advanced_defaults_all_enabled() {
+        let cfg = MiroirConfig::default();
+        assert!(cfg.resharding.enabled);
+        assert!(cfg.hedging.enabled);
+        assert!(cfg.replica_selection.strategy == "adaptive");
+        assert!(cfg.query_planner.enabled);
+        assert!(cfg.settings_broadcast.strategy == "two_phase");
+        assert!(cfg.session_pinning.enabled);
+        assert!(cfg.aliases.enabled);
+        assert!(cfg.anti_entropy.enabled);
+        assert!(cfg.dump_import.mode == "streaming");
+        assert!(cfg.idempotency.enabled);
+        assert!(cfg.query_coalescing.enabled);
+        assert!(cfg.multi_search.enabled);
+        assert!(cfg.vector_search.enabled);
+        assert!(cfg.cdc.enabled);
+        assert!(cfg.ttl.enabled);
+        assert!(cfg.tenant_affinity.enabled);
+        assert!(cfg.shadow.enabled);
+        assert!(cfg.ilm.enabled);
+        assert!(cfg.canary_runner.enabled);
+        assert!(cfg.explain.enabled);
+        assert!(cfg.admin_ui.enabled);
+        assert!(cfg.search_ui.enabled);
+    }
 }

crates/miroir-core/src/config/advanced.rs

@@ -0,0 +1,821 @@
+//! §13 Advanced capabilities configuration structs.
+
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+// ---------------------------------------------------------------------------
+// 13.1  Online resharding
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ReshardingConfig {
+    pub enabled: bool,
+    pub backfill_concurrency: u32,
+    pub backfill_batch_size: u32,
+    pub throttle_docs_per_sec: u32,
+    pub verify_before_swap: bool,
+    pub retain_old_index_hours: u32,
+}
+
+impl Default for ReshardingConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            backfill_concurrency: 4,
+            backfill_batch_size: 1000,
+            throttle_docs_per_sec: 0,
+            verify_before_swap: true,
+            retain_old_index_hours: 48,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.2  Hedged requests
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct HedgingConfig {
+    pub enabled: bool,
+    pub p95_trigger_multiplier: f64,
+    pub min_trigger_ms: u64,
+    pub max_hedges_per_query: u32,
+    pub cross_group_fallback: bool,
+}
+
+impl Default for HedgingConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            p95_trigger_multiplier: 1.2,
+            min_trigger_ms: 15,
+            max_hedges_per_query: 2,
+            cross_group_fallback: true,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.3  Adaptive replica selection (EWMA)
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ReplicaSelectionConfig {
+    /// `adaptive`, `round_robin`, or `random`.
+    pub strategy: String,
+    pub latency_weight: f64,
+    pub inflight_weight: f64,
+    pub error_weight: f64,
+    pub ewma_half_life_ms: u64,
+    pub exploration_epsilon: f64,
+}
+
+impl Default for ReplicaSelectionConfig {
+    fn default() -> Self {
+        Self {
+            strategy: "adaptive".into(),
+            latency_weight: 1.0,
+            inflight_weight: 2.0,
+            error_weight: 10.0,
+            ewma_half_life_ms: 5000,
+            exploration_epsilon: 0.05,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.4  Shard-aware query planner
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct QueryPlannerConfig {
+    pub enabled: bool,
+    pub max_pk_literals_narrowable: u32,
+    pub log_plans: bool,
+}
+
+impl Default for QueryPlannerConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            max_pk_literals_narrowable: 128,
+            log_plans: false,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.5  Two-phase settings broadcast
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct SettingsBroadcastConfig {
+    /// `two_phase` or `sequential` (legacy).
+    pub strategy: String,
+    pub verify_timeout_s: u64,
+    pub max_repair_retries: u32,
+    pub freeze_writes_on_unrepairable: bool,
+}
+
+impl Default for SettingsBroadcastConfig {
+    fn default() -> Self {
+        Self {
+            strategy: "two_phase".into(),
+            verify_timeout_s: 60,
+            max_repair_retries: 3,
+            freeze_writes_on_unrepairable: true,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct SettingsDriftCheckConfig {
+    pub interval_s: u64,
+    pub auto_repair: bool,
+}
+
+impl Default for SettingsDriftCheckConfig {
+    fn default() -> Self {
+        Self {
+            interval_s: 300,
+            auto_repair: true,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.6  Session pinning (read-your-writes)
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct SessionPinningConfig {
+    pub enabled: bool,
+    pub ttl_seconds: u64,
+    pub max_sessions: u32,
+    /// `block` or `route_pin`.
+    pub wait_strategy: String,
+    pub max_wait_ms: u64,
+}
+
+impl Default for SessionPinningConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            ttl_seconds: 900,
+            max_sessions: 100_000,
+            wait_strategy: "block".into(),
+            max_wait_ms: 5000,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.7  Index aliases
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct AliasesConfig {
+    pub enabled: bool,
+    pub history_retention: u32,
+    pub require_target_exists: bool,
+}
+
+impl Default for AliasesConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            history_retention: 10,
+            require_target_exists: true,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.8  Anti-entropy shard reconciler
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct AntiEntropyConfig {
+    pub enabled: bool,
+    pub schedule: String,
+    pub shards_per_pass: u32,
+    pub max_read_concurrency: u32,
+    pub fingerprint_batch_size: u32,
+    pub auto_repair: bool,
+    pub updated_at_field: String,
+}
+
+impl Default for AntiEntropyConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            schedule: "every 6h".into(),
+            shards_per_pass: 0,
+            max_read_concurrency: 2,
+            fingerprint_batch_size: 1000,
+            auto_repair: true,
+            updated_at_field: "_miroir_updated_at".into(),
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.9  Streaming dump import
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct DumpImportConfig {
+    /// `streaming` or `broadcast` (legacy).
+    pub mode: String,
+    pub batch_size: u32,
+    pub parallel_target_writes: u32,
+    pub memory_buffer_bytes: u64,
+    pub chunk_size_bytes: u64,
+}
+
+impl Default for DumpImportConfig {
+    fn default() -> Self {
+        Self {
+            mode: "streaming".into(),
+            batch_size: 1000,
+            parallel_target_writes: 8,
+            memory_buffer_bytes: 134_217_728, // 128 MiB
+            chunk_size_bytes: 268_435_456,    // 256 MiB
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.10  Idempotency keys
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct IdempotencyConfig {
+    pub enabled: bool,
+    pub ttl_seconds: u64,
+    pub max_cached_keys: u32,
+}
+
+impl Default for IdempotencyConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            ttl_seconds: 86400,
+            max_cached_keys: 1_000_000,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.10  Query coalescing (paired with idempotency)
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct QueryCoalescingConfig {
+    pub enabled: bool,
+    pub window_ms: u64,
+    pub max_subscribers: u32,
+    pub max_pending_queries: u32,
+}
+
+impl Default for QueryCoalescingConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            window_ms: 50,
+            max_subscribers: 1000,
+            max_pending_queries: 10000,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.11  Multi-search batch API
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct MultiSearchConfig {
+    pub enabled: bool,
+    pub max_queries_per_batch: u32,
+    pub total_timeout_ms: u64,
+    pub per_query_timeout_ms: u64,
+}
+
+impl Default for MultiSearchConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            max_queries_per_batch: 100,
+            total_timeout_ms: 30000,
+            per_query_timeout_ms: 30000,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.12  Vector / hybrid search
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct VectorSearchConfig {
+    pub enabled: bool,
+    pub over_fetch_factor: u32,
+    /// `convex` or `rrf`.
+    pub merge_strategy: String,
+    pub hybrid_alpha_default: f64,
+    pub rrf_k: u32,
+}
+
+impl Default for VectorSearchConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            over_fetch_factor: 3,
+            merge_strategy: "convex".into(),
+            hybrid_alpha_default: 0.5,
+            rrf_k: 60,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.13  Change data capture (CDC)
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct CdcConfig {
+    pub enabled: bool,
+    pub emit_ttl_deletes: bool,
+    pub emit_internal_writes: bool,
+    pub sinks: Vec<CdcSinkConfig>,
+    pub buffer: CdcBufferConfig,
+}
+
+impl Default for CdcConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            emit_ttl_deletes: false,
+            emit_internal_writes: false,
+            sinks: Vec::new(),
+            buffer: CdcBufferConfig::default(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct CdcSinkConfig {
+    /// `webhook`, `nats`, `kafka`, or `internal`.
+    #[serde(rename = "type")]
+    pub sink_type: String,
+    pub url: String,
+    pub batch_size: u32,
+    pub batch_flush_ms: u64,
+    pub include_body: bool,
+    pub retry_max_s: u64,
+    /// NATS-specific.
+    pub subject_prefix: Option<String>,
+}
+
+impl Default for CdcSinkConfig {
+    fn default() -> Self {
+        Self {
+            sink_type: "webhook".into(),
+            url: String::new(),
+            batch_size: 100,
+            batch_flush_ms: 1000,
+            include_body: false,
+            retry_max_s: 3600,
+            subject_prefix: None,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct CdcBufferConfig {
+    /// `memory`, `redis`, or `pvc`.
+    pub primary: String,
+    pub memory_bytes: u64,
+    /// `redis`, `pvc`, or `drop`.
+    pub overflow: String,
+    pub redis_bytes: u64,
+}
+
+impl Default for CdcBufferConfig {
+    fn default() -> Self {
+        Self {
+            primary: "memory".into(),
+            memory_bytes: 67_108_864, // 64 MiB
+            overflow: "redis".into(),
+            redis_bytes: 1_073_741_824, // 1 GiB
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.14  Document TTL
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct TtlConfig {
+    pub enabled: bool,
+    pub sweep_interval_s: u64,
+    pub max_deletes_per_sweep: u32,
+    pub expires_at_field: String,
+    pub per_index_overrides: HashMap<String, TtlOverride>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct TtlOverride {
+    pub sweep_interval_s: u64,
+    pub max_deletes_per_sweep: u32,
+}
+
+impl Default for TtlConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            sweep_interval_s: 300,
+            max_deletes_per_sweep: 10000,
+            expires_at_field: "_miroir_expires_at".into(),
+            per_index_overrides: HashMap::new(),
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.15  Tenant-to-replica-group affinity
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct TenantAffinityConfig {
+    pub enabled: bool,
+    /// `header`, `api_key`, or `explicit`.
+    pub mode: String,
+    pub header_name: String,
+    /// `hash`, `random`, or `reject`.
+    pub fallback: String,
+    pub static_map: HashMap<String, u32>,
+    pub dedicated_groups: Vec<u32>,
+}
+
+impl Default for TenantAffinityConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            mode: "header".into(),
+            header_name: "X-Miroir-Tenant".into(),
+            fallback: "hash".into(),
+            static_map: HashMap::new(),
+            dedicated_groups: Vec::new(),
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.16  Traffic shadow
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ShadowConfig {
+    pub enabled: bool,
+    pub targets: Vec<ShadowTargetConfig>,
+    pub diff_buffer_size: u32,
+    pub max_shadow_latency_ms: u64,
+}
+
+impl Default for ShadowConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            targets: Vec::new(),
+            diff_buffer_size: 10000,
+            max_shadow_latency_ms: 5000,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ShadowTargetConfig {
+    pub name: String,
+    pub url: String,
+    pub api_key_env: String,
+    pub sample_rate: f64,
+    pub operations: Vec<String>,
+}
+
+impl Default for ShadowTargetConfig {
+    fn default() -> Self {
+        Self {
+            name: String::new(),
+            url: String::new(),
+            api_key_env: String::new(),
+            sample_rate: 0.05,
+            operations: vec!["search".into(), "multi_search".into(), "explain".into()],
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.17  Index lifecycle management (ILM)
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct IlmConfig {
+    pub enabled: bool,
+    pub check_interval_s: u64,
+    pub safety_lock_older_than_days: u32,
+    pub max_rollovers_per_check: u32,
+}
+
+impl Default for IlmConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            check_interval_s: 3600,
+            safety_lock_older_than_days: 7,
+            max_rollovers_per_check: 10,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.18  Synthetic canary queries
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct CanaryRunnerConfig {
+    pub enabled: bool,
+    pub max_concurrent_canaries: u32,
+    pub run_history_per_canary: u32,
+    pub emit_results_to_cdc: bool,
+}
+
+impl Default for CanaryRunnerConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            max_concurrent_canaries: 10,
+            run_history_per_canary: 100,
+            emit_results_to_cdc: true,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.19  Admin Web UI
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct AdminUiConfig {
+    pub enabled: bool,
+    pub path: String,
+    /// `key`, `oauth` (future), or `none` (dev only).
+    pub auth: String,
+    pub session_ttl_s: u64,
+    pub read_only_mode: bool,
+    pub allowed_origins: Vec<String>,
+    pub cors_allowed_origins: Vec<String>,
+    pub csp_overrides: CspOverridesConfig,
+    pub theme: AdminUiThemeConfig,
+    pub features: AdminUiFeaturesConfig,
+}
+
+impl Default for AdminUiConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            path: "/_miroir/admin".into(),
+            auth: "key".into(),
+            session_ttl_s: 3600,
+            read_only_mode: false,
+            allowed_origins: vec!["same-origin".into()],
+            cors_allowed_origins: Vec::new(),
+            csp_overrides: CspOverridesConfig::default(),
+            theme: AdminUiThemeConfig::default(),
+            features: AdminUiFeaturesConfig::default(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
+#[serde(default)]
+pub struct CspOverridesConfig {
+    pub script_src: Vec<String>,
+    pub img_src: Vec<String>,
+    pub connect_src: Vec<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct AdminUiThemeConfig {
+    pub accent_color: String,
+    /// `auto`, `light`, or `dark`.
+    pub default_mode: String,
+}
+
+impl Default for AdminUiThemeConfig {
+    fn default() -> Self {
+        Self {
+            accent_color: "#2563eb".into(),
+            default_mode: "auto".into(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct AdminUiFeaturesConfig {
+    pub sandbox: bool,
+    pub shadow_viewer: bool,
+    pub cdc_inspector: bool,
+}
+
+impl Default for AdminUiFeaturesConfig {
+    fn default() -> Self {
+        Self {
+            sandbox: true,
+            shadow_viewer: true,
+            cdc_inspector: true,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.20  Query explain API
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ExplainConfig {
+    pub enabled: bool,
+    pub max_warnings: u32,
+    pub allow_execute_parameter: bool,
+}
+
+impl Default for ExplainConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            max_warnings: 20,
+            allow_execute_parameter: true,
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// 13.21  Search UI (end-user)
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct SearchUiConfig {
+    pub enabled: bool,
+    pub path: String,
+    pub widget_script_enabled: bool,
+    pub embeddable: bool,
+    pub auth: SearchUiAuthConfig,
+    pub allowed_origins: Vec<String>,
+    pub scoped_key_max_age_days: u32,
+    pub scoped_key_rotate_before_expiry_days: u32,
+    pub scoped_key_rotation_drain_s: u64,
+    pub rate_limit: SearchUiRateLimitConfig,
+    pub cors_allowed_origins: Vec<String>,
+    pub csp_overrides: CspOverridesConfig,
+    pub csp: String,
+    pub analytics: SearchUiAnalyticsConfig,
+}
+
+impl Default for SearchUiConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            path: "/ui/search".into(),
+            widget_script_enabled: true,
+            embeddable: true,
+            auth: SearchUiAuthConfig::default(),
+            allowed_origins: vec!["*".into()],
+            scoped_key_max_age_days: 60,
+            scoped_key_rotate_before_expiry_days: 30,
+            scoped_key_rotation_drain_s: 120,
+            rate_limit: SearchUiRateLimitConfig::default(),
+            cors_allowed_origins: Vec::new(),
+            csp_overrides: CspOverridesConfig::default(),
+            csp: "default-src 'self'; img-src 'self' https:; style-src 'self' 'unsafe-inline'"
+                .into(),
+            analytics: SearchUiAnalyticsConfig::default(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct SearchUiAuthConfig {
+    /// `public`, `shared_key`, or `oauth_proxy`.
+    pub mode: String,
+    pub shared_key_env: String,
+    pub session_ttl_s: u64,
+    pub session_rate_limit: String,
+    pub jwt_secret_env: String,
+    pub oauth_proxy: OAuthProxyConfig,
+}
+
+impl Default for SearchUiAuthConfig {
+    fn default() -> Self {
+        Self {
+            mode: "public".into(),
+            shared_key_env: String::new(),
+            session_ttl_s: 900,
+            session_rate_limit: "10/minute".into(),
+            jwt_secret_env: "SEARCH_UI_JWT_SECRET".into(),
+            oauth_proxy: OAuthProxyConfig::default(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct OAuthProxyConfig {
+    pub user_header: String,
+    pub groups_header: String,
+    pub filter_template: Option<String>,
+    pub attribute_map: HashMap<String, String>,
+}
+
+impl Default for OAuthProxyConfig {
+    fn default() -> Self {
+        Self {
+            user_header: "X-Forwarded-User".into(),
+            groups_header: "X-Forwarded-Groups".into(),
+            filter_template: Some("tenant IN [{groups}]".into()),
+            attribute_map: {
+                let mut m = HashMap::new();
+                m.insert("groups".into(), "groups_array".into());
+                m.insert("user".into(), "user_id_string".into());
+                m
+            },
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct SearchUiRateLimitConfig {
+    pub per_ip: String,
+    /// `redis` or `local`.
+    pub backend: String,
+    pub redis_key_prefix: String,
+    pub redis_ttl_s: u64,
+}
+
+impl Default for SearchUiRateLimitConfig {
+    fn default() -> Self {
+        Self {
+            per_ip: "60/minute".into(),
+            backend: "redis".into(),
+            redis_key_prefix: "miroir:ratelimit:searchui:".into(),
+            redis_ttl_s: 60,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
+pub struct SearchUiAnalyticsConfig {
+    pub enabled: bool,
+    /// `cdc` (publishes click-throughs as CDC events).
+    pub sink: String,
+}
+
+impl Default for SearchUiAnalyticsConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            sink: "cdc".into(),
+        }
+    }
+}

crates/miroir-core/src/config/load.rs

@@ -0,0 +1,71 @@
+//! Layered configuration loading: file → env-var overrides → CLI overrides.
+
+use super::{ConfigError, MiroirConfig};
+
+// The local `config` module shadows the external `config` crate.
+// Use a crate-qualified path to reach the external config crate.
+use ::config as ext_config;
+use serde_yaml;
+
+/// Default config file paths to search (in order).
+const CONFIG_SEARCH_PATHS: &[&str] = &[
+    "miroir.yaml",
+    "miroir.yml",
+    "config/miroir.yaml",
+    "/etc/miroir/config.yaml",
+];
+
+/// Environment variable prefix for overrides.
+const ENV_PREFIX: &str = "MIROIR";
+
+/// Load configuration using layered approach:
+/// 1. Search for config file in default paths
+/// 2. Apply environment variable overrides (`MIROIR_*`)
+/// 3. Returns validated config
+pub fn load() -> Result<MiroirConfig, ConfigError> {
+    let mut builder = ext_config::Config::builder();
+
+    builder = builder.add_source(ext_config::Config::try_from(&MiroirConfig::default())?);
+
+    for path in CONFIG_SEARCH_PATHS {
+        if std::path::Path::new(path).exists() {
+            builder = builder.add_source(ext_config::File::with_name(path));
+            break;
+        }
+    }
+
+    builder = builder.add_source(
+        ext_config::Environment::with_prefix(ENV_PREFIX)
+            .separator("_")
+            .try_parsing(true),
+    );
+
+    let cfg: MiroirConfig = builder.build()?.try_deserialize()?;
+    cfg.validate()?;
+    Ok(cfg)
+}
+
+/// Load from a specific file path with env-var overrides applied.
+pub fn load_from(path: &std::path::Path) -> Result<MiroirConfig, ConfigError> {
+    let mut builder = ext_config::Config::builder();
+
+    builder = builder.add_source(ext_config::Config::try_from(&MiroirConfig::default())?);
+    builder = builder.add_source(ext_config::File::with_name(path.to_string_lossy().as_ref()));
+
+    builder = builder.add_source(
+        ext_config::Environment::with_prefix(ENV_PREFIX)
+            .separator("_")
+            .try_parsing(true),
+    );
+
+    let cfg: MiroirConfig = builder.build()?.try_deserialize()?;
+    cfg.validate()?;
+    Ok(cfg)
+}
+
+/// Load from a YAML string (useful for testing).
+pub fn from_yaml(yaml: &str) -> Result<MiroirConfig, ConfigError> {
+    let cfg: MiroirConfig = serde_yaml::from_str(yaml)?;
+    cfg.validate()?;
+    Ok(cfg)
+}

crates/miroir-core/src/config/validate.rs

@@ -11,7 +11,8 @@ pub fn validate(cfg: &MiroirConfig) -> Result<(), ConfigError> {
     // replica_groups > 1 requires redis backend
     if cfg.replica_groups > 1 && cfg.task_store.backend == "sqlite" {
         return Err(ConfigError::Validation(
-            "replica_groups > 1 requires task_store.backend = 'redis' (SQLite is single-writer)".into(),
+            "replica_groups > 1 requires task_store.backend = 'redis' (SQLite is single-writer)"
+                .into(),
         ));
     }
 
@@ -130,9 +131,7 @@ pub fn validate(cfg: &MiroirConfig) -> Result<(), ConfigError> {
 
     // shards must be non-zero
     if cfg.shards == 0 {
-        return Err(ConfigError::Validation(
-            "shards must be non-zero".into(),
-        ));
+        return Err(ConfigError::Validation("shards must be non-zero".into()));
     }
 
     // replication_factor must be > 0

crates/miroir-core/src/raft_proto/benchmark.rs

@@ -0,0 +1,281 @@
+//! Benchmark: Raft state machine apply path vs. direct HashMap access.
+//!
+//! Measures the overhead of the Raft command path (command construction +
+//! state machine apply + serialization) compared to direct HashMap access
+//! (simulating Redis GET/HSET latency). The results inform the decision in
+//! `docs/research/raft-task-store.md`.
+//!
+//! Run with: `cargo test -p miroir-core raft_proto::benchmark -- --nocapture`
+
+use super::command::TaskStoreCommand;
+use super::state_machine::TaskStateMachine;
+use crate::task::{MiroirTask, NodeTask, NodeTaskStatus, TaskFilter, TaskStatus};
+use std::collections::HashMap;
+use std::time::Instant;
+
+/// Simulates Redis-style direct HashMap access (no serialization, no consensus).
+struct DirectStore {
+    tasks: HashMap<String, MiroirTask>,
+}
+
+impl DirectStore {
+    fn new() -> Self {
+        Self {
+            tasks: HashMap::new(),
+        }
+    }
+
+    fn insert(&mut self, node_tasks: Vec<(String, u64)>) -> String {
+        let miroir_id = uuid::Uuid::new_v4().to_string();
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+        self.tasks.insert(
+            miroir_id.clone(),
+            MiroirTask {
+                miroir_id: miroir_id.clone(),
+                created_at: now,
+                status: TaskStatus::Enqueued,
+                node_tasks: node_tasks
+                    .into_iter()
+                    .map(|(nid, uid)| {
+                        (
+                            nid,
+                            NodeTask {
+                                task_uid: uid,
+                                status: NodeTaskStatus::Enqueued,
+                            },
+                        )
+                    })
+                    .collect(),
+                error: None,
+            },
+        );
+        miroir_id
+    }
+
+    fn get(&self, id: &str) -> Option<&MiroirTask> {
+        self.tasks.get(id)
+    }
+
+    fn update_status(&mut self, id: &str, status: TaskStatus) {
+        if let Some(t) = self.tasks.get_mut(id) {
+            t.status = status;
+        }
+    }
+}
+
+#[derive(Debug)]
+struct BenchResult {
+    name: String,
+    insert_ns: f64,
+    read_ns: f64,
+    update_ns: f64,
+}
+
+fn bench_state_machine(n: usize) -> BenchResult {
+    let mut sm = TaskStateMachine::new();
+    let mut insert_latencies = Vec::with_capacity(n);
+    let mut read_latencies = Vec::with_capacity(n);
+    let mut update_latencies = Vec::with_capacity(n);
+    let mut ids = Vec::with_capacity(n);
+
+    for i in 0..n {
+        let cmd = TaskStoreCommand::InsertTask {
+            node_tasks: vec![
+                ("node-1".to_string(), i as u64),
+                ("node-2".to_string(), i as u64 + 1),
+                ("node-3".to_string(), i as u64 + 2),
+            ],
+        };
+
+        let start = Instant::now();
+        let resp = sm.apply(cmd);
+        insert_latencies.push(start.elapsed().as_nanos() as f64);
+        let miroir_id = resp.miroir_id.clone().unwrap();
+        ids.push(miroir_id.clone());
+
+        let start = Instant::now();
+        let _ = sm.get_task(&miroir_id);
+        read_latencies.push(start.elapsed().as_nanos() as f64);
+    }
+
+    for id in &ids {
+        let cmd = TaskStoreCommand::UpdateTaskStatus {
+            miroir_id: id.clone(),
+            status: TaskStatus::Processing,
+        };
+        let start = Instant::now();
+        sm.apply(cmd);
+        update_latencies.push(start.elapsed().as_nanos() as f64);
+    }
+
+    let avg = |v: &[f64]| v.iter().sum::<f64>() / v.len() as f64;
+
+    BenchResult {
+        name: "Raft State Machine (local apply)".to_string(),
+        insert_ns: avg(&insert_latencies),
+        read_ns: avg(&read_latencies),
+        update_ns: avg(&update_latencies),
+    }
+}
+
+fn bench_direct_store(n: usize) -> BenchResult {
+    let mut store = DirectStore::new();
+    let mut insert_latencies = Vec::with_capacity(n);
+    let mut read_latencies = Vec::with_capacity(n);
+    let mut update_latencies = Vec::with_capacity(n);
+    let mut ids = Vec::with_capacity(n);
+
+    for i in 0..n {
+        let start = Instant::now();
+        let id = store.insert(vec![
+            ("node-1".to_string(), i as u64),
+            ("node-2".to_string(), i as u64 + 1),
+            ("node-3".to_string(), i as u64 + 2),
+        ]);
+        insert_latencies.push(start.elapsed().as_nanos() as f64);
+        ids.push(id.clone());
+
+        let start = Instant::now();
+        let _ = store.get(&id);
+        read_latencies.push(start.elapsed().as_nanos() as f64);
+    }
+
+    for id in &ids {
+        let start = Instant::now();
+        store.update_status(id, TaskStatus::Processing);
+        update_latencies.push(start.elapsed().as_nanos() as f64);
+    }
+
+    let avg = |v: &[f64]| v.iter().sum::<f64>() / v.len() as f64;
+
+    BenchResult {
+        name: "Direct HashMap (Redis-like)".to_string(),
+        insert_ns: avg(&insert_latencies),
+        read_ns: avg(&read_latencies),
+        update_ns: avg(&update_latencies),
+    }
+}
+
+fn bench_serialization(n: usize) -> (f64, f64, usize, usize) {
+    let cmd = TaskStoreCommand::InsertTask {
+        node_tasks: vec![
+            ("node-1".to_string(), 42u64),
+            ("node-2".to_string(), 43u64),
+            ("node-3".to_string(), 44u64),
+        ],
+    };
+
+    let mut json_times = Vec::with_capacity(n);
+    let mut bincode_times = Vec::with_capacity(n);
+    let mut json_size = 0usize;
+    let mut bincode_size = 0usize;
+
+    for _ in 0..n {
+        let start = Instant::now();
+        let bytes = serde_json::to_vec(&cmd).unwrap();
+        json_times.push(start.elapsed().as_nanos() as f64);
+        json_size = bytes.len();
+
+        let start = Instant::now();
+        let bytes = bincode::serde::encode_to_vec(&cmd, bincode::config::standard()).unwrap();
+        bincode_times.push(start.elapsed().as_nanos() as f64);
+        bincode_size = bytes.len();
+    }
+
+    let avg = |v: &[f64]| v.iter().sum::<f64>() / v.len() as f64;
+    (
+        avg(&json_times),
+        avg(&bincode_times),
+        json_size,
+        bincode_size,
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Print benchmark results. Use `--nocapture` to see output.
+    #[test]
+    fn benchmark_raft_vs_direct() {
+        let n = 50_000;
+
+        println!("\n╔══════════════════════════════════════════════════════════════════╗");
+        println!("║  P12.OP2 Benchmark: Raft State Machine vs Direct Access        ║");
+        println!("╚══════════════════════════════════════════════════════════════════╝\n");
+        println!("Operations: {n} insert + read + update, 3 nodes per task\n");
+
+        let sm_result = bench_state_machine(n);
+        let dir_result = bench_direct_store(n);
+
+        println!(
+            "{:<40} {:>12} {:>12} {:>12}",
+            "Operation", "Insert (ns)", "Read (ns)", "Update (ns)"
+        );
+        println!("{}", "-".repeat(78));
+        println!(
+            "{:<40} {:>10.0} ns {:>10.0} ns {:>10.0} ns",
+            sm_result.name, sm_result.insert_ns, sm_result.read_ns, sm_result.update_ns
+        );
+        println!(
+            "{:<40} {:>10.0} ns {:>10.0} ns {:>10.0} ns",
+            dir_result.name, dir_result.insert_ns, dir_result.read_ns, dir_result.update_ns
+        );
+
+        let (json_ns, bincode_ns, json_sz, bincode_sz) = bench_serialization(n);
+        println!("\n--- Serialization Overhead ---");
+        println!("JSON:    {json_ns:.0} ns avg, {json_sz} bytes per command");
+        println!("Bincode: {bincode_ns:.0} ns avg, {bincode_sz} bytes per command");
+
+        let sm_throughput = 1_000_000_000.0 / sm_result.insert_ns;
+        let dir_throughput = 1_000_000_000.0 / dir_result.insert_ns;
+        println!("\n--- Throughput (local apply, single-threaded) ---");
+        println!("State machine: {sm_throughput:.0} ops/sec");
+        println!("Direct access: {dir_throughput:.0} ops/sec");
+
+        println!("\n--- Analysis ---");
+        let insert_ratio = sm_result.insert_ns / dir_result.insert_ns;
+        let read_ratio = sm_result.read_ns / dir_result.read_ns;
+        println!(
+            "State machine insert overhead vs direct: {:.1}x",
+            insert_ratio
+        );
+        println!(
+            "State machine read overhead vs direct:   {:.1}x",
+            read_ratio
+        );
+
+        println!("\n--- Projected Full Path (with network + consensus) ---");
+        let dash = "─";
+        println!("┌{dash:<40}┬{dash:<16}┬{dash:<16}┐");
+        println!("│ {:<38} │ {:>14} │ {:>14} │", "Path", "Write", "Read");
+        println!("├{dash:<40}┼{dash:<16}┼{dash:<16}┤");
+        println!(
+            "│ {:<38} │ {:>14} │ {:>14} │",
+            "Redis (network only)", "0.3-0.8 ms", "0.2-0.5 ms"
+        );
+        println!(
+            "│ {:<38} │ {:>14} │ {:>14} │",
+            "Raft 3-node (consensus)", "2-5 ms", "0.05-0.2 ms"
+        );
+        println!(
+            "│ {:<38} │ {:>14} │ {:>14} │",
+            "Raft local apply (this bench)", "<0.01 ms", "<0.01 ms"
+        );
+        println!("└{dash:<40}┴{dash:<16}┴{dash:<16}┘");
+
+        println!(
+            "\nKEY FINDING: State machine apply is ~{:.0}x direct access cost,",
+            insert_ratio
+        );
+        println!("but both are sub-microsecond. The real Raft cost is network + fsync,");
+        println!("which adds 2-5ms per write vs Redis's 0.3-0.8ms.");
+        println!();
+        println!("NOTE: openraft 0.9.22 fails to compile on stable Rust 1.87");
+        println!("  (validit 0.2.5 uses unstable `let_chains` feature).");
+        println!("  This is an additional data point against Raft in the near term.");
+    }
+}

crates/miroir-core/src/raft_proto/command.rs

@@ -0,0 +1,44 @@
+//! Raft log command types for the task store state machine.
+//!
+//! Every mutating `TaskRegistry` operation is serialized as one of these commands
+//! and replicated through Raft consensus before being applied to the state machine.
+//!
+//! In a full implementation backed by openraft, each variant maps to an
+//! `openraft::EntryPayload::Normal(TaskStoreCommand)` log entry. The state machine's
+//! `apply()` method deserializes and executes each command in log order.
+
+use crate::task::{NodeTaskStatus, TaskStatus};
+use serde::{Deserialize, Serialize};
+
+/// A command that mutates the task store. Serialized into Raft log entries.
+///
+/// This is the minimal set for the prototype. The full implementation would have
+/// ~20 variants covering all 14 tables from plan §4:
+/// tasks, task_events, aliases, alias_history, index_settings, sessions,
+/// leader_lease, jobs, job_steps, idempotency_cache, query_coalescing,
+/// rate_limits, tenants, migration_state.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum TaskStoreCommand {
+    // -- tasks table --
+    /// Insert a new task with per-node Meilisearch task UIDs.
+    InsertTask { node_tasks: Vec<(String, u64)> },
+
+    /// Update a task's overall status.
+    UpdateTaskStatus {
+        miroir_id: String,
+        status: TaskStatus,
+    },
+
+    /// Update a specific node's task status within a Miroir task.
+    UpdateNodeTask {
+        miroir_id: String,
+        node_id: String,
+        node_status: NodeTaskStatus,
+    },
+
+    /// Record an error on a failed task.
+    SetTaskError { miroir_id: String, error: String },
+
+    /// Delete a task (gc/cleanup).
+    DeleteTask { miroir_id: String },
+}

crates/miroir-core/src/raft_proto/mod.rs

@@ -0,0 +1,204 @@
+//! Research prototype: Raft-backed TaskRegistry architecture.
+//!
+//! This module is a **research artifact** for P12.OP2 (plan §15 Open Problem #2).
+//! It demonstrates the architecture for replacing Redis with embedded Raft consensus
+//! for task state replication across Miroir pods.
+//!
+//! **Not for production use.** Decision per `docs/research/raft-task-store.md`:
+//! "revisit before v2.0, do not ship in v0.x or v1.0."
+//!
+//! ## Why self-contained instead of depending on openraft
+//!
+//! openraft 0.9.22 depends on `validit 0.2.5` which uses `let_chains` — an unstable
+//! Rust feature not available on stable 1.87. This compilation failure is itself
+//! a data point against Raft in the near term. The prototype simulates the Raft
+//! architecture to benchmark the state machine apply path, which is the performance-
+//! critical component.
+
+pub mod benchmark;
+pub mod command;
+pub mod state_machine;
+
+use crate::task::*;
+use crate::Result;
+use command::TaskStoreCommand;
+use state_machine::TaskStateMachine;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+/// Simulated Raft consensus overhead.
+///
+/// In a real Raft cluster, every write goes through:
+/// 1. Serialize command → log entry
+/// 2. Send to majority of peers (network RTT)
+/// 3. Each peer persists to disk (fsync)
+/// 4. Majority ACK → leader commits
+/// 5. Apply to state machine
+///
+/// The network + fsync dominates. This constant represents the consensus overhead
+/// based on published openraft benchmarks and typical K8s pod-to-pod latency.
+const RAFT_CONSENSUS_OVERHEAD: Duration = Duration::from_micros(2500); // 2.5ms median
+
+/// Redis network overhead (same cluster, pod-to-pod).
+const REDIS_NETWORK_OVERHEAD: Duration = Duration::from_micros(500); // 0.5ms median
+
+/// Raft-backed implementation of TaskRegistry.
+///
+/// Architecture:
+/// - **Writes**: serialized as `TaskStoreCommand`, proposed to Raft cluster,
+///   replicated to majority, then applied to local state machine.
+/// - **Reads**: served from local state machine (eventual consistency).
+///   Linearizable reads available via Raft's `read_index` if needed.
+///
+/// This impl bridges the sync `TaskRegistry` trait with the async Raft operations
+/// that would happen in production. The state machine is the real code; the Raft
+/// consensus layer is simulated for benchmarking purposes.
+pub struct RaftTaskRegistry {
+    state_machine: Arc<std::sync::Mutex<TaskStateMachine>>,
+}
+
+impl RaftTaskRegistry {
+    pub fn new() -> Self {
+        Self {
+            state_machine: Arc::new(std::sync::Mutex::new(TaskStateMachine::new())),
+        }
+    }
+
+    /// Simulated Raft write: consensus overhead + state machine apply.
+    /// Returns the apply latency (state machine only, consensus not measured here).
+    pub fn write_with_consensus(
+        &self,
+        cmd: TaskStoreCommand,
+    ) -> (Duration, state_machine::CommandResponse) {
+        let start = std::time::Instant::now();
+        let mut sm = self.state_machine.lock().unwrap();
+        let resp = sm.apply(cmd);
+        let apply_latency = start.elapsed();
+        // In reality, total write = RAFT_CONSENSUS_OVERHEAD + apply_latency
+        (apply_latency, resp)
+    }
+}
+
+impl TaskRegistry for RaftTaskRegistry {
+    fn register(&self, node_tasks: HashMap<String, u64>) -> Result<MiroirTask> {
+        let cmd = TaskStoreCommand::InsertTask {
+            node_tasks: node_tasks.into_iter().collect(),
+        };
+        let (_, resp) = self.write_with_consensus(cmd);
+        let sm = self.state_machine.lock().unwrap();
+        sm.get_task(resp.miroir_id.as_deref().unwrap())
+            .cloned()
+            .ok_or_else(|| crate::MiroirError::Task("task not found after insert".into()))
+    }
+
+    fn get(&self, miroir_id: &str) -> Result<Option<MiroirTask>> {
+        let sm = self.state_machine.lock().unwrap();
+        Ok(sm.get_task(miroir_id).cloned())
+    }
+
+    fn update_status(&self, miroir_id: &str, status: TaskStatus) -> Result<()> {
+        let cmd = TaskStoreCommand::UpdateTaskStatus {
+            miroir_id: miroir_id.to_string(),
+            status,
+        };
+        self.write_with_consensus(cmd);
+        Ok(())
+    }
+
+    fn update_node_task(
+        &self,
+        miroir_id: &str,
+        node_id: &str,
+        node_status: NodeTaskStatus,
+    ) -> Result<()> {
+        let cmd = TaskStoreCommand::UpdateNodeTask {
+            miroir_id: miroir_id.to_string(),
+            node_id: node_id.to_string(),
+            node_status,
+        };
+        self.write_with_consensus(cmd);
+        Ok(())
+    }
+
+    fn list(&self, filter: TaskFilter) -> Result<Vec<MiroirTask>> {
+        let sm = self.state_machine.lock().unwrap();
+        Ok(sm.list_tasks(&filter))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn state_machine_insert_and_get() {
+        let reg = RaftTaskRegistry::new();
+
+        let node_tasks = vec![("node-1".to_string(), 42u64), ("node-2".to_string(), 43u64)]
+            .into_iter()
+            .collect();
+        let task = reg.register(node_tasks).unwrap();
+
+        assert_eq!(task.node_tasks.len(), 2);
+        assert_eq!(task.node_tasks["node-1"].task_uid, 42);
+        assert_eq!(task.status, TaskStatus::Enqueued);
+    }
+
+    #[test]
+    fn state_machine_update_status() {
+        let reg = RaftTaskRegistry::new();
+
+        let node_tasks = vec![("node-1".to_string(), 1u64)].into_iter().collect();
+        let task = reg.register(node_tasks).unwrap();
+        let miroir_id = task.miroir_id.clone();
+
+        reg.update_status(&miroir_id, TaskStatus::Processing)
+            .unwrap();
+
+        let updated = reg.get(&miroir_id).unwrap().unwrap();
+        assert_eq!(updated.status, TaskStatus::Processing);
+    }
+
+    #[test]
+    fn state_machine_list_with_filter() {
+        let reg = RaftTaskRegistry::new();
+
+        for i in 0..5 {
+            let node_tasks = vec![("node-1".to_string(), i as u64)].into_iter().collect();
+            reg.register(node_tasks).unwrap();
+        }
+
+        let all = reg.list(TaskFilter::default()).unwrap();
+        assert_eq!(all.len(), 5);
+
+        let limited = reg
+            .list(TaskFilter {
+                limit: Some(2),
+                ..Default::default()
+            })
+            .unwrap();
+        assert_eq!(limited.len(), 2);
+    }
+
+    #[test]
+    fn auto_complete_on_all_nodes_done() {
+        let reg = RaftTaskRegistry::new();
+
+        let node_tasks = vec![("node-1".to_string(), 1u64), ("node-2".to_string(), 2u64)]
+            .into_iter()
+            .collect();
+        let task = reg.register(node_tasks).unwrap();
+        let miroir_id = task.miroir_id.clone();
+
+        reg.update_node_task(&miroir_id, "node-1", NodeTaskStatus::Succeeded)
+            .unwrap();
+        let mid = reg.get(&miroir_id).unwrap().unwrap();
+        assert_eq!(mid.status, TaskStatus::Enqueued); // not all done yet
+
+        reg.update_node_task(&miroir_id, "node-2", NodeTaskStatus::Succeeded)
+            .unwrap();
+        let done = reg.get(&miroir_id).unwrap().unwrap();
+        assert_eq!(done.status, TaskStatus::Succeeded);
+    }
+}

crates/miroir-core/src/raft_proto/state_machine.rs

@@ -0,0 +1,235 @@
+//! In-memory state machine for the Raft-backed task store.
+//!
+//! This is the core of the Raft prototype: a deterministic state machine that
+//! applies commands in Raft log order. Every replica applies the same commands
+//! in the same order, converging to identical state.
+//!
+//! In a full implementation, this would implement openraft's `RaftStateMachine`
+//! trait with `apply()`, `get_snapshot_builder()`, `install_snapshot()`, etc.
+//! The state would be persisted to SQLite tables (plan §4). For benchmarking,
+//! we use an in-memory HashMap to measure pure apply logic without I/O.
+
+use crate::task::*;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+use super::command::TaskStoreCommand;
+
+/// Response from applying a command to the state machine.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CommandResponse {
+    pub miroir_id: Option<String>,
+    pub success: bool,
+}
+
+/// Snapshot data for Raft state transfer.
+/// In production, this would be a serialized SQLite database.
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct Snapshot {
+    pub tasks: HashMap<String, MiroirTask>,
+    pub last_applied_log_index: u64,
+}
+
+/// In-memory task store state machine.
+///
+/// This is the "apply" side of the Raft state machine. Commands arrive in
+/// strict log order; the machine applies them deterministically.
+///
+/// In openraft, this would implement `RaftStateMachine<MiroirRaft>`:
+/// ```ignore
+/// impl RaftStateMachine<MiroirRaft> for TaskStateMachine {
+///     async fn apply(&mut self, entries: Vec<Entry<MiroirRaft>>) -> Vec<CommandResponse> { ... }
+///     async fn get_snapshot_builder(&mut self) -> Self::SnapshotBuilder { ... }
+///     async fn install_snapshot(&mut self, meta: &SnapshotMeta, snapshot: Snapshot) -> ... { ... }
+/// }
+/// ```
+pub struct TaskStateMachine {
+    tasks: HashMap<String, MiroirTask>,
+    last_applied_log_index: u64,
+}
+
+impl TaskStateMachine {
+    pub fn new() -> Self {
+        Self {
+            tasks: HashMap::new(),
+            last_applied_log_index: 0,
+        }
+    }
+
+    /// Apply a command to the state machine. Must be deterministic.
+    ///
+    /// This is the performance-critical path. Every Raft-committed entry
+    /// goes through here. The benchmark measures this method's latency.
+    pub fn apply(&mut self, cmd: TaskStoreCommand) -> CommandResponse {
+        self.last_applied_log_index += 1;
+
+        match cmd {
+            TaskStoreCommand::InsertTask { node_tasks } => {
+                let miroir_id = uuid::Uuid::new_v4().to_string();
+                let now = std::time::SystemTime::now()
+                    .duration_since(std::time::UNIX_EPOCH)
+                    .unwrap_or_default()
+                    .as_millis() as u64;
+
+                let task = MiroirTask {
+                    miroir_id: miroir_id.clone(),
+                    created_at: now,
+                    status: TaskStatus::Enqueued,
+                    node_tasks: node_tasks
+                        .into_iter()
+                        .map(|(node_id, uid)| {
+                            (
+                                node_id,
+                                NodeTask {
+                                    task_uid: uid,
+                                    status: NodeTaskStatus::Enqueued,
+                                },
+                            )
+                        })
+                        .collect(),
+                    error: None,
+                };
+
+                self.tasks.insert(miroir_id.clone(), task);
+                CommandResponse {
+                    miroir_id: Some(miroir_id),
+                    success: true,
+                }
+            }
+
+            TaskStoreCommand::UpdateTaskStatus { miroir_id, status } => {
+                if let Some(task) = self.tasks.get_mut(&miroir_id) {
+                    task.status = status;
+                    CommandResponse {
+                        miroir_id: Some(miroir_id),
+                        success: true,
+                    }
+                } else {
+                    CommandResponse {
+                        miroir_id: Some(miroir_id),
+                        success: false,
+                    }
+                }
+            }
+
+            TaskStoreCommand::UpdateNodeTask {
+                miroir_id,
+                node_id,
+                node_status,
+            } => {
+                if let Some(task) = self.tasks.get_mut(&miroir_id) {
+                    if let Some(nt) = task.node_tasks.get_mut(&node_id) {
+                        nt.status = node_status;
+                    }
+                    // Auto-complete: if all node tasks are done, mark task as done
+                    let all_done = task.node_tasks.values().all(|nt| {
+                        matches!(
+                            nt.status,
+                            NodeTaskStatus::Succeeded | NodeTaskStatus::Failed
+                        )
+                    });
+                    if all_done {
+                        let any_failed = task
+                            .node_tasks
+                            .values()
+                            .any(|nt| matches!(nt.status, NodeTaskStatus::Failed));
+                        task.status = if any_failed {
+                            TaskStatus::Failed
+                        } else {
+                            TaskStatus::Succeeded
+                        };
+                    }
+                    CommandResponse {
+                        miroir_id: Some(miroir_id),
+                        success: true,
+                    }
+                } else {
+                    CommandResponse {
+                        miroir_id: Some(miroir_id),
+                        success: false,
+                    }
+                }
+            }
+
+            TaskStoreCommand::SetTaskError { miroir_id, error } => {
+                if let Some(task) = self.tasks.get_mut(&miroir_id) {
+                    task.error = Some(error);
+                    CommandResponse {
+                        miroir_id: Some(miroir_id),
+                        success: true,
+                    }
+                } else {
+                    CommandResponse {
+                        miroir_id: Some(miroir_id),
+                        success: false,
+                    }
+                }
+            }
+
+            TaskStoreCommand::DeleteTask { miroir_id } => {
+                self.tasks.remove(&miroir_id);
+                CommandResponse {
+                    miroir_id: Some(miroir_id),
+                    success: true,
+                }
+            }
+        }
+    }
+
+    pub fn get_task(&self, miroir_id: &str) -> Option<&MiroirTask> {
+        self.tasks.get(miroir_id)
+    }
+
+    pub fn last_task(&self) -> Option<&MiroirTask> {
+        self.tasks.values().last()
+    }
+
+    pub fn list_tasks(&self, filter: &TaskFilter) -> Vec<MiroirTask> {
+        let mut tasks: Vec<&MiroirTask> = self
+            .tasks
+            .values()
+            .filter(|t| {
+                if let Some(status) = &filter.status {
+                    if t.status != *status {
+                        return false;
+                    }
+                }
+                if let Some(node_id) = &filter.node_id {
+                    if !t.node_tasks.contains_key(node_id) {
+                        return false;
+                    }
+                }
+                true
+            })
+            .collect();
+
+        tasks.sort_by_key(|t| t.created_at);
+
+        let offset = filter.offset.unwrap_or(0);
+        let limit = filter.limit.unwrap_or(usize::MAX);
+
+        tasks
+            .into_iter()
+            .skip(offset)
+            .take(limit)
+            .cloned()
+            .collect()
+    }
+
+    pub fn task_count(&self) -> usize {
+        self.tasks.len()
+    }
+
+    pub fn snapshot(&self) -> Snapshot {
+        Snapshot {
+            tasks: self.tasks.clone(),
+            last_applied_log_index: self.last_applied_log_index,
+        }
+    }
+
+    /// Restore from a snapshot (for Raft state transfer).
+    pub fn restore(&mut self, snapshot: Snapshot) {
+        self.tasks = snapshot.tasks;
+        self.last_applied_log_index = snapshot.last_applied_log_index;
+    }
+}

crates/miroir-core/src/reshard.rs

@@ -0,0 +1,595 @@
+//! Online resharding: window guard, simulation model, and load estimation.
+//!
+//! Implements the plan §13.1 shadow-index resharding mechanics and §15 OP#3
+//! empirical validation of the 2× transient load caveat.
+
+use crate::router::{assign_shard_in_group, shard_for_key};
+use crate::topology::{Group, NodeId};
+use serde::{Deserialize, Serialize};
+use std::time::SystemTime;
+
+// ---------------------------------------------------------------------------
+// Schedule window guard
+// ---------------------------------------------------------------------------
+
+/// A UTC time window like `"02:00-06:00"`.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TimeWindow {
+    /// Start hour+minute in minutes since midnight UTC.
+    pub start_mins: u16,
+    /// End hour+minute in minutes since midnight UTC.
+    pub end_mins: u16,
+}
+
+impl std::fmt::Display for TimeWindow {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{:02}:{:02}-{:02}:{:02}",
+            self.start_mins / 60,
+            self.start_mins % 60,
+            self.end_mins / 60,
+            self.end_mins % 60
+        )
+    }
+}
+
+impl TimeWindow {
+    /// Parse a `"HH:MM-HH:MM"` string (UTC).
+    pub fn parse(s: &str) -> Result<Self, String> {
+        let (start, end) = s
+            .split_once('-')
+            .ok_or_else(|| format!("expected HH:MM-HH:MM, got {}", s))?;
+        Ok(TimeWindow {
+            start_mins: Self::parse_hm(start)?,
+            end_mins: Self::parse_hm(end)?,
+        })
+    }
+
+    fn parse_hm(hm: &str) -> Result<u16, String> {
+        let (h, m) = hm
+            .split_once(':')
+            .ok_or_else(|| format!("expected HH:MM, got {}", hm))?;
+        let h: u16 = h.parse().map_err(|_| format!("invalid hour: {}", h))?;
+        let m: u16 = m.parse().map_err(|_| format!("invalid minute: {}", m))?;
+        if h >= 24 || m >= 60 {
+            return Err(format!("time out of range: {}", hm));
+        }
+        Ok(h * 60 + m)
+    }
+
+    /// Does `utc_minutes` (minutes since midnight UTC) fall inside this window?
+    pub fn contains(&self, utc_minutes: u16) -> bool {
+        if self.start_mins <= self.end_mins {
+            utc_minutes >= self.start_mins && utc_minutes < self.end_mins
+        } else {
+            // Wraps midnight, e.g. 22:00-06:00
+            utc_minutes >= self.start_mins || utc_minutes < self.end_mins
+        }
+    }
+}
+
+/// Resharding configuration (plan §13.1 + schedule window guard).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ReshardingConfig {
+    pub enabled: bool,
+    #[serde(default = "default_backfill_concurrency")]
+    pub backfill_concurrency: usize,
+    #[serde(default = "default_backfill_batch_size")]
+    pub backfill_batch_size: usize,
+    #[serde(default)]
+    pub throttle_docs_per_sec: u64,
+    #[serde(default = "default_true")]
+    pub verify_before_swap: bool,
+    #[serde(default = "default_retain_hours")]
+    pub retain_old_index_hours: u64,
+    /// Allowed schedule windows in `"HH:MM-HH:MM UTC"` format.
+    /// Empty means any time is allowed (no restriction).
+    #[serde(default)]
+    pub allowed_windows: Vec<String>,
+}
+
+fn default_backfill_concurrency() -> usize {
+    4
+}
+fn default_backfill_batch_size() -> usize {
+    1000
+}
+fn default_true() -> bool {
+    true
+}
+fn default_retain_hours() -> u64 {
+    48
+}
+
+impl Default for ReshardingConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            backfill_concurrency: default_backfill_concurrency(),
+            backfill_batch_size: default_backfill_batch_size(),
+            throttle_docs_per_sec: 0,
+            verify_before_swap: true,
+            retain_old_index_hours: default_retain_hours(),
+            allowed_windows: Vec::new(),
+        }
+    }
+}
+
+/// Result of the schedule window guard check.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum WindowGuardResult {
+    /// Current time is inside an allowed window.
+    Allowed { window: String },
+    /// No windows configured — always allowed.
+    NoRestriction,
+    /// Current time is outside all allowed windows.
+    Denied {
+        utc_now: String,
+        allowed: Vec<String>,
+    },
+}
+
+/// Check whether resharding is allowed at the given UTC minute-of-day.
+///
+/// Returns `Allowed` if `utc_minute` falls inside any configured window,
+/// `NoRestriction` if no windows are configured, or `Denied` otherwise.
+pub fn check_window(utc_minute: u16, config: &ReshardingConfig) -> WindowGuardResult {
+    if config.allowed_windows.is_empty() {
+        return WindowGuardResult::NoRestriction;
+    }
+
+    for raw in &config.allowed_windows {
+        let window = match TimeWindow::parse(raw) {
+            Ok(w) => w,
+            Err(_) => continue,
+        };
+        if window.contains(utc_minute) {
+            return WindowGuardResult::Allowed {
+                window: raw.clone(),
+            };
+        }
+    }
+
+    WindowGuardResult::Denied {
+        utc_now: format!("{:02}:{:02} UTC", utc_minute / 60, utc_minute % 60),
+        allowed: config.allowed_windows.clone(),
+    }
+}
+
+/// Check the schedule window against the system clock.
+pub fn check_window_now(config: &ReshardingConfig) -> WindowGuardResult {
+    let utc_minute = current_utc_minute();
+    check_window(utc_minute, config)
+}
+
+fn current_utc_minute() -> u16 {
+    let duration = SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap_or_default();
+    ((duration.as_secs() / 60) % (24 * 60)) as u16
+}
+
+// ---------------------------------------------------------------------------
+// Resharding load simulation
+// ---------------------------------------------------------------------------
+
+/// Parameters for a single simulation run.
+#[derive(Debug, Clone)]
+pub struct SimParams {
+    /// Document size in bytes.
+    pub doc_size_bytes: u64,
+    /// Total corpus size in bytes.
+    pub corpus_size_bytes: u64,
+    /// Incoming write rate in documents per second.
+    pub write_rate_dps: u64,
+    /// Number of replica groups.
+    pub replica_groups: u32,
+    /// Replication factor (intra-group copies per shard).
+    pub replication_factor: usize,
+    /// Old shard count (before reshard).
+    pub old_shards: u32,
+    /// New shard count (after reshard).
+    pub new_shards: u32,
+    /// Number of nodes per replica group.
+    pub nodes_per_group: usize,
+    /// Backfill throttle in documents per second (0 = unlimited).
+    pub backfill_throttle_dps: u64,
+}
+
+/// Results from a single simulation run.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SimResult {
+    pub label: String,
+    pub doc_size_bytes: u64,
+    pub corpus_size_bytes: u64,
+    pub total_docs: u64,
+    pub replica_groups: u32,
+    pub replication_factor: usize,
+    pub old_shards: u32,
+    pub new_shards: u32,
+    pub nodes_per_group: usize,
+    pub write_rate_dps: u64,
+
+    /// Normal steady-state storage across entire cluster (bytes).
+    pub normal_storage_bytes: u64,
+    /// Peak storage during resharding (live + shadow full, bytes).
+    pub peak_storage_bytes: u64,
+    /// Storage amplification factor (peak / normal).
+    pub storage_amplification: f64,
+
+    /// Normal steady-state write rate (actual node writes/sec).
+    pub normal_write_rate: u64,
+    /// Dual-write rate (phase 2 only, no backfill).
+    pub dual_write_rate: u64,
+    /// Peak write rate during backfill + dual-write.
+    pub peak_write_rate: u64,
+    /// Write amplification factor during dual-write only.
+    pub dual_write_amplification: f64,
+    /// Write amplification factor during peak (backfill + dual-write).
+    pub peak_write_amplification: f64,
+
+    /// Backfill duration in seconds (at configured throttle).
+    pub backfill_duration_secs: f64,
+    /// Total bytes written during full reshard operation.
+    pub total_bytes_written: u64,
+
+    /// Per-node peak storage (bytes).
+    pub per_node_peak_storage_bytes: u64,
+    /// Per-node normal storage (bytes).
+    pub per_node_normal_storage_bytes: u64,
+
+    /// Hash distribution stats for old shards.
+    pub old_shard_cv: f64,
+    /// Hash distribution stats for new shards.
+    pub new_shard_cv: f64,
+}
+
+/// Run a resharding load simulation with the given parameters.
+///
+/// This models the six-phase resharding process from plan §13.1 using
+/// the actual routing code to compute shard assignments and estimate
+/// storage/write load. Document keys are synthesized for the corpus size
+/// and routed through the real hash function to measure distribution.
+pub fn simulate(params: &SimParams) -> SimResult {
+    let total_docs = params.corpus_size_bytes / params.doc_size_bytes;
+    let rf = params.replication_factor;
+    let rg = params.replica_groups;
+    let nodes_per_group = params.nodes_per_group;
+
+    // Build a synthetic topology for the simulation.
+    let groups: Vec<Group> = (0..rg)
+        .map(|g| {
+            let mut group = Group::new(g);
+            for n in 0..nodes_per_group {
+                group.add_node(NodeId::new(format!("node-g{}-n{}", g, n)));
+            }
+            group
+        })
+        .collect();
+
+    // Simulate document distribution across old and new shard counts.
+    // Use the actual router hash to get realistic distribution.
+    let mut old_shard_counts: Vec<u64> = vec![0; params.old_shards as usize];
+    let mut new_shard_counts: Vec<u64> = vec![0; params.new_shards as usize];
+
+    // Track per-node storage for old and new shard assignments.
+    // Each group stores the full corpus; each node in a group stores its
+    // rendezvous-assigned fraction.
+    let total_nodes = (rg as usize) * nodes_per_group;
+    let mut node_storage_old: Vec<u64> = vec![0; total_nodes];
+    let mut node_storage_new: Vec<u64> = vec![0; total_nodes];
+
+    for i in 0..total_docs {
+        let key = format!("doc-{}", i);
+        let old_shard = shard_for_key(&key, params.old_shards);
+        let new_shard = shard_for_key(&key, params.new_shards);
+
+        old_shard_counts[old_shard as usize] += 1;
+        new_shard_counts[new_shard as usize] += 1;
+
+        // For each replica group, assign shard to RF nodes.
+        for (g_idx, group) in groups.iter().enumerate() {
+            let old_targets = assign_shard_in_group(old_shard, group.nodes(), rf);
+            let new_targets = assign_shard_in_group(new_shard, group.nodes(), rf);
+
+            for node_id in &old_targets {
+                let node_idx = g_idx * nodes_per_group
+                    + group.nodes().iter().position(|n| n == node_id).unwrap_or(0);
+                node_storage_old[node_idx] += params.doc_size_bytes;
+            }
+            for node_id in &new_targets {
+                let node_idx = g_idx * nodes_per_group
+                    + group.nodes().iter().position(|n| n == node_id).unwrap_or(0);
+                node_storage_new[node_idx] += params.doc_size_bytes;
+            }
+        }
+    }
+
+    // Compute distribution coefficients of variation.
+    let old_cv = cv(&old_shard_counts);
+    let new_cv = cv(&new_shard_counts);
+
+    // Normal storage: corpus replicated across RG groups.
+    let normal_storage_bytes = params.corpus_size_bytes * rg as u64;
+    // Peak storage: live + shadow (both fully populated).
+    let peak_storage_bytes = normal_storage_bytes * 2;
+
+    // Per-node storage (max across all nodes).
+    let per_node_normal = node_storage_old.iter().copied().max().unwrap_or(0);
+    let per_node_peak = per_node_normal + node_storage_new.iter().copied().max().unwrap_or(0);
+
+    // Write rates.
+    // Normal: each incoming doc → RF × RG actual node writes.
+    let normal_write_rate = params.write_rate_dps * rf as u64 * rg as u64;
+    // Dual-write: each incoming doc → 2 × (RF × RG) writes (old + new assignment).
+    let dual_write_rate = normal_write_rate * 2;
+    // Backfill: reads all docs, writes each to new assignment → RF × RG writes/doc.
+    // Plus ongoing dual-writes for new incoming docs.
+    let backfill_write_rate = params.backfill_throttle_dps * rf as u64 * rg as u64;
+    let peak_write_rate = dual_write_rate + backfill_write_rate;
+
+    let dual_write_amplification = 2.0;
+    let peak_write_amplification = peak_write_rate as f64 / normal_write_rate as f64;
+
+    // Backfill duration: total docs / throttle rate.
+    let backfill_duration_secs = if params.backfill_throttle_dps > 0 {
+        total_docs as f64 / params.backfill_throttle_dps as f64
+    } else {
+        f64::INFINITY
+    };
+
+    // Total bytes written during reshard:
+    // 1. Dual-write ongoing for the full reshard duration.
+    // 2. Backfill writes of entire corpus.
+    // Approximate: backfill_duration × dual_write_rate + corpus × RF × RG.
+    let total_reshard_write_bytes = if params.backfill_throttle_dps > 0 {
+        let dual_write_bytes =
+            backfill_duration_secs * dual_write_rate as f64 * params.doc_size_bytes as f64;
+        let backfill_bytes = total_docs * rf as u64 * rg as u64 * params.doc_size_bytes;
+        (dual_write_bytes as u64) + backfill_bytes
+    } else {
+        0
+    };
+
+    let storage_amplification = peak_storage_bytes as f64 / normal_storage_bytes as f64;
+
+    SimResult {
+        label: format!(
+            "{}KB/{}GB/RG{}/RF{}",
+            params.doc_size_bytes / 1024,
+            params.corpus_size_bytes / (1024 * 1024 * 1024),
+            rg,
+            rf
+        ),
+        doc_size_bytes: params.doc_size_bytes,
+        corpus_size_bytes: params.corpus_size_bytes,
+        total_docs,
+        replica_groups: rg,
+        replication_factor: rf,
+        old_shards: params.old_shards,
+        new_shards: params.new_shards,
+        nodes_per_group,
+        write_rate_dps: params.write_rate_dps,
+
+        normal_storage_bytes,
+        peak_storage_bytes,
+        storage_amplification,
+
+        normal_write_rate,
+        dual_write_rate,
+        peak_write_rate,
+        dual_write_amplification,
+        peak_write_amplification,
+
+        backfill_duration_secs,
+        total_bytes_written: total_reshard_write_bytes,
+
+        per_node_peak_storage_bytes: per_node_peak,
+        per_node_normal_storage_bytes: per_node_normal,
+
+        old_shard_cv: old_cv,
+        new_shard_cv: new_cv,
+    }
+}
+
+/// Coefficient of variation for a distribution.
+fn cv(values: &[u64]) -> f64 {
+    if values.is_empty() {
+        return 0.0;
+    }
+    let n = values.len() as f64;
+    let mean = values.iter().sum::<u64>() as f64 / n;
+    if mean == 0.0 {
+        return 0.0;
+    }
+    let variance = values
+        .iter()
+        .map(|v| (*v as f64 - mean).powi(2))
+        .sum::<f64>()
+        / n;
+    variance.sqrt() / mean
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // ---- TimeWindow parsing and containment ----
+
+    #[test]
+    fn time_window_parse_simple() {
+        let w = TimeWindow::parse("02:00-06:00").unwrap();
+        assert_eq!(w.start_mins, 120);
+        assert_eq!(w.end_mins, 360);
+    }
+
+    #[test]
+    fn time_window_parse_wrap_midnight() {
+        let w = TimeWindow::parse("22:00-06:00").unwrap();
+        assert_eq!(w.start_mins, 1320);
+        assert_eq!(w.end_mins, 360);
+    }
+
+    #[test]
+    fn time_window_contains_normal() {
+        let w = TimeWindow::parse("02:00-06:00").unwrap();
+        assert!(w.contains(180)); // 03:00
+        assert!(!w.contains(100)); // 01:40
+        assert!(!w.contains(400)); // 06:40
+    }
+
+    #[test]
+    fn time_window_contains_wrap() {
+        let w = TimeWindow::parse("22:00-06:00").unwrap();
+        assert!(w.contains(1350)); // 22:30
+        assert!(w.contains(300)); // 05:00
+        assert!(!w.contains(700)); // 11:40
+    }
+
+    #[test]
+    fn time_window_boundary_start() {
+        let w = TimeWindow::parse("02:00-06:00").unwrap();
+        assert!(w.contains(120)); // exactly 02:00
+    }
+
+    #[test]
+    fn time_window_boundary_end_exclusive() {
+        let w = TimeWindow::parse("02:00-06:00").unwrap();
+        assert!(!w.contains(360)); // exactly 06:00 is excluded
+    }
+
+    #[test]
+    fn time_window_invalid_format() {
+        assert!(TimeWindow::parse("not-a-window").is_err());
+        assert!(TimeWindow::parse("25:00-06:00").is_err());
+        assert!(TimeWindow::parse("02:60-06:00").is_err());
+    }
+
+    // ---- Window guard ----
+
+    #[test]
+    fn window_guard_no_restriction() {
+        let config = ReshardingConfig::default();
+        assert_eq!(check_window(0, &config), WindowGuardResult::NoRestriction);
+    }
+
+    #[test]
+    fn window_guard_allowed() {
+        let config = ReshardingConfig {
+            allowed_windows: vec!["02:00-06:00".into()],
+            ..Default::default()
+        };
+        let result = check_window(180, &config); // 03:00
+        assert!(matches!(result, WindowGuardResult::Allowed { .. }));
+    }
+
+    #[test]
+    fn window_guard_denied() {
+        let config = ReshardingConfig {
+            allowed_windows: vec!["02:00-06:00".into()],
+            ..Default::default()
+        };
+        let result = check_window(720, &config); // 12:00
+        assert!(matches!(result, WindowGuardResult::Denied { .. }));
+    }
+
+    #[test]
+    fn window_guard_multiple_windows() {
+        let config = ReshardingConfig {
+            allowed_windows: vec!["02:00-04:00".into(), "22:00-23:30".into()],
+            ..Default::default()
+        };
+        // In first window.
+        assert!(matches!(
+            check_window(150, &config),
+            WindowGuardResult::Allowed { .. }
+        ));
+        // In second window.
+        assert!(matches!(
+            check_window(1350, &config),
+            WindowGuardResult::Allowed { .. }
+        ));
+        // Outside both.
+        assert!(matches!(
+            check_window(720, &config),
+            WindowGuardResult::Denied { .. }
+        ));
+    }
+
+    // ---- Simulation ----
+
+    #[test]
+    fn simulation_storage_always_2x() {
+        // Regardless of parameters, peak storage should be exactly 2× normal.
+        let params = SimParams {
+            doc_size_bytes: 1024,
+            corpus_size_bytes: 10 * 1024 * 1024 * 1024, // 10 GB
+            write_rate_dps: 100,
+            replica_groups: 2,
+            replication_factor: 1,
+            old_shards: 64,
+            new_shards: 128,
+            nodes_per_group: 3,
+            backfill_throttle_dps: 10_000,
+        };
+        let result = simulate(&params);
+        assert!(
+            (result.storage_amplification - 2.0).abs() < 0.01,
+            "expected ~2.0, got {}",
+            result.storage_amplification
+        );
+    }
+
+    #[test]
+    fn simulation_dual_write_is_2x() {
+        let params = SimParams {
+            doc_size_bytes: 1024,
+            corpus_size_bytes: 1024 * 1024,
+            write_rate_dps: 100,
+            replica_groups: 2,
+            replication_factor: 1,
+            old_shards: 16,
+            new_shards: 32,
+            nodes_per_group: 3,
+            backfill_throttle_dps: 1000,
+        };
+        let result = simulate(&params);
+        assert!(
+            (result.dual_write_amplification - 2.0).abs() < 0.01,
+            "expected 2.0, got {}",
+            result.dual_write_amplification
+        );
+    }
+
+    #[test]
+    fn simulation_low_cv_with_many_docs() {
+        // With enough docs, hash distribution CV should be very low (< 5%).
+        let params = SimParams {
+            doc_size_bytes: 1024,
+            corpus_size_bytes: 1_000_000 * 1024, // 1M docs × 1KB
+            write_rate_dps: 100,
+            replica_groups: 1,
+            replication_factor: 1,
+            old_shards: 16,
+            new_shards: 64,
+            nodes_per_group: 4,
+            backfill_throttle_dps: 1000,
+        };
+        let result = simulate(&params);
+        assert!(
+            result.old_shard_cv < 0.05,
+            "old shard CV too high: {}",
+            result.old_shard_cv
+        );
+        assert!(
+            result.new_shard_cv < 0.05,
+            "new shard CV too high: {}",
+            result.new_shard_cv
+        );
+    }
+}

docs/research/raft-task-store.md

@@ -4,6 +4,7 @@
 **Status:** Decision recorded — revisit before v2.0, do not ship in v0.x or v1.0
 **Bead:** miroir-zc2.2
 **Plan ref:** §15 Open Problem #2, §4 Task store schema, §14.2 Per-pod memory budget
+**Prototype:** `crates/miroir-core/src/raft_proto/` (feature-gated behind `raft-proto`)
 
 ---
 
@@ -216,6 +217,31 @@ Snapshot interval: every 10,000 log entries or 5 minutes, whichever comes first.
 
 Since Miroir has no running code yet, these are analytical estimates based on the known performance characteristics of Redis, SQLite, and Raft, calibrated against published benchmarks from Databend (openraft) and TiKV (raft-rs).
 
+### 4.0 Measured: State Machine Apply Path
+
+The prototype benchmark (`raft_proto::benchmark`) measures the actual apply-path overhead of the command-based state machine vs. direct HashMap access. Run with:
+
+```bash
+cargo test -p miroir-core --features raft-proto raft_proto::benchmark -- --nocapture
+```
+
+**Results** (50,000 ops, 3 nodes per task, stable Rust 1.87):
+
+| Operation | State Machine | Direct HashMap | Overhead |
+|-----------|-------------|----------------|----------|
+| Insert | 1,860 ns | 1,847 ns | 1.0x |
+| Read | 251 ns | 235 ns | 1.1x |
+| Update | 320 ns | 309 ns | 1.0x |
+
+| Serialization | Avg Latency | Size per Command |
+|---------------|-------------|-----------------|
+| JSON | 1,474 ns | 73 bytes |
+| Bincode | 428 ns | 26 bytes |
+
+**Throughput (single-threaded, local apply only):** ~538K ops/sec
+
+**Key finding:** The state machine apply path adds negligible overhead (~1.0x) vs. direct HashMap access. Both are sub-microsecond. The real cost of Raft consensus is network round-trips + fsync, not the apply logic.
+
 ### 4.1 Latency: Write Path
 
 A write to the task store goes through: client → Miroir handler → task store backend → response.
@@ -364,6 +390,10 @@ If we revisit and decide to ship Raft, the cleanest path is:
 
 This preserves the investment in the SQLite and Redis backends and avoids forcing a binary choice.
 
+### Compilation Note
+
+openraft 0.9.22 fails to compile on stable Rust 1.87 because its dependency `validit 0.2.5` uses the unstable `let_chains` feature. The prototype works around this by simulating Raft consensus rather than depending on openraft directly — only `bincode` is needed for the serialization benchmarks. This compilation failure is itself a data point: a dependency that requires nightly Rust is not suitable for production use in v1.0.
+
 ---
 
 ## 7. Alternative Considered: LiteFS