From 540f5ac00c495e2608a6cc1b907e4ed73c3ae059 Mon Sep 17 00:00:00 2001 From: jedarden Date: Sun, 24 May 2026 16:48:57 -0400 Subject: [PATCH] fix(config): implement P6.1 pod resource envelope + fix compilation errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit implements P6.1 (Pod resource envelope + limits/requests) per plan §14.8 and fixes several pre-existing compilation errors. ## P6.1 Implementation (plan §14.1-14.3, §14.8) - Config defaults already match plan §14.8 envelope: - Server: max_body_bytes=104857600 (100MiB), max_concurrent_requests=500 - Connection pool: max_idle=32, max_total=128, idle_timeout_s=60 - Task registry: cache_size=10000, redis_pool_max=50 - Idempotency: max_cached_keys=1000000, ttl_seconds=86400 - Session pinning: max_sessions=100000 - Query coalescing: max_subscribers=1000, max_pending_queries=10000 - Anti-entropy: max_read_concurrency=2, fingerprint_batch_size=1000 - Resharding: backfill_concurrency=4, backfill_batch_size=1000 - Peer discovery: service_name="miroir-headless", refresh_interval_s=15 - Leader election: lease_ttl_s=10, renew_interval_s=3 (fixed from 30/5) - Helm values.yaml already has correct resource limits: - limits: cpu=2000m, memory=3584Mi (3.5GiB under 3.75GB node limit) - requests: cpu=500m, memory=1Gi ## Compilation Fixes - Made RebalanceJob, ShardState fields public (for admin API access) - Added jobs() accessor method to RebalancerWorker - Added MiroirCode variants: InvalidRequest, NotFound, InternalError - Fixed AdminUiAssets to be public (for rust-embed) - Added include-exclude feature to rust-embed dependency - Fixed DumpImportManager to accept Arc> (matching proxy state) - Re-exported DumpImportConfig from dump_import to avoid duplication - Fixed topology API usage (use .shards instead of .shard_count(), .nodes() instead of .all_nodes()) - Fixed HeaderMap iteration in search.rs (use .as_ref() instead of .as_str()) - Fixed AntiEntropyWorkerConfig defaults to match plan §14.8 (lease_ttl_secs=10, renew_interval_ms=3000) - Added from_code_str entries for new MiroirCode variants Closes: miroir-m9q.1 Co-Authored-By: Claude Opus 4.7 --- Cargo.lock | 24 ++++++++++++ crates/miroir-core/src/api_error.rs | 25 +++++++++++-- crates/miroir-core/src/config/advanced.rs | 26 ++----------- crates/miroir-core/src/dump_import.rs | 19 ++++++---- .../rebalancer_worker/anti_entropy_worker.rs | 6 +-- .../miroir-core/src/rebalancer_worker/mod.rs | 37 +++++++++++-------- crates/miroir-proxy/Cargo.toml | 2 +- crates/miroir-proxy/src/admin_ui.rs | 2 +- .../src/routes/admin_endpoints.rs | 9 ++--- crates/miroir-proxy/src/routes/search.rs | 3 +- 10 files changed, 91 insertions(+), 62 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9f3cd36..ae7aa9d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -467,6 +467,16 @@ dependencies = [ "tinyvec", ] +[[package]] +name = "bstr" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63044e1ae8e69f3b5a92c736ca6269b8d12fa7efe39bf34ddb06d102cf0e2cab" +dependencies = [ + "memchr", + "serde", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -1480,6 +1490,19 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280" +[[package]] +name = "globset" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52dfc19153a48bde0cbd630453615c8151bce3a5adfac7a0aebfbf0a1e1f57e3" +dependencies = [ + "aho-corasick", + "bstr", + "log", + "regex-automata", + "regex-syntax", +] + [[package]] name = "h2" version = "0.4.14" @@ -3771,6 +3794,7 @@ version = "8.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5bcdef0be6fe7f6fa333b1073c949729274b05f123a0ad7efcb8efd878e5c3b1" dependencies = [ + "globset", "sha2", "walkdir", ] diff --git a/crates/miroir-core/src/api_error.rs b/crates/miroir-core/src/api_error.rs index 3f13d55..678118f 100644 --- a/crates/miroir-core/src/api_error.rs +++ b/crates/miroir-core/src/api_error.rs @@ -40,7 +40,10 @@ impl std::fmt::Display for ErrorType { /// Miroir-specific error codes with associated metadata. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum MiroirCode { + InvalidRequest, PrimaryKeyRequired, + NotFound, + InternalError, NoQuorum, ShardUnavailable, ReservedField, @@ -58,8 +61,11 @@ pub enum MiroirCode { impl MiroirCode { /// All variants, used for iteration in tests. - pub const ALL: [MiroirCode; 14] = [ + pub const ALL: [MiroirCode; 17] = [ + MiroirCode::InvalidRequest, MiroirCode::PrimaryKeyRequired, + MiroirCode::NotFound, + MiroirCode::InternalError, MiroirCode::NoQuorum, MiroirCode::ShardUnavailable, MiroirCode::ReservedField, @@ -78,7 +84,10 @@ impl MiroirCode { /// Returns the error code string (e.g., `"miroir_no_quorum"`). pub fn as_str(&self) -> &'static str { match self { + Self::InvalidRequest => "miroir_invalid_request", Self::PrimaryKeyRequired => "miroir_primary_key_required", + Self::NotFound => "miroir_not_found", + Self::InternalError => "miroir_internal_error", Self::NoQuorum => "miroir_no_quorum", Self::ShardUnavailable => "miroir_shard_unavailable", Self::ReservedField => "miroir_reserved_field", @@ -98,7 +107,9 @@ impl MiroirCode { /// Returns the Meilisearch-compatible error type category. pub fn error_type(&self) -> ErrorType { match self { - Self::PrimaryKeyRequired + Self::InvalidRequest + | Self::PrimaryKeyRequired + | Self::NotFound | Self::ReservedField | Self::IdempotencyKeyReused | Self::MultiAliasNotWritable @@ -110,7 +121,8 @@ impl MiroirCode { | Self::MissingCsrf | Self::CsrfMismatch => ErrorType::Auth, - Self::NoQuorum + Self::InternalError + | Self::NoQuorum | Self::ShardUnavailable | Self::SettingsVersionStale | Self::Timeout => ErrorType::System, @@ -120,11 +132,13 @@ impl MiroirCode { /// Returns the HTTP status code for this error. pub fn http_status(&self) -> u16 { match self { - Self::PrimaryKeyRequired | Self::ReservedField => 400, + Self::InvalidRequest | Self::PrimaryKeyRequired | Self::ReservedField => 400, + Self::NotFound => 404, Self::JwtInvalid | Self::InvalidAuth | Self::MissingCsrf => 401, Self::JwtScopeDenied | Self::CsrfMismatch => 403, Self::IdempotencyKeyReused | Self::MultiAliasNotWritable => 409, Self::IndexAlreadyExists => 409, + Self::InternalError => 500, Self::Timeout => 504, Self::NoQuorum | Self::ShardUnavailable | Self::SettingsVersionStale => 503, } @@ -141,7 +155,10 @@ impl MiroirCode { /// Parse a code string back to a [`MiroirCode`]. pub fn from_code_str(s: &str) -> Option { match s { + "miroir_invalid_request" => Some(Self::InvalidRequest), "miroir_primary_key_required" => Some(Self::PrimaryKeyRequired), + "miroir_not_found" => Some(Self::NotFound), + "miroir_internal_error" => Some(Self::InternalError), "miroir_no_quorum" => Some(Self::NoQuorum), "miroir_shard_unavailable" => Some(Self::ShardUnavailable), "miroir_reserved_field" => Some(Self::ReservedField), diff --git a/crates/miroir-core/src/config/advanced.rs b/crates/miroir-core/src/config/advanced.rs index 3fdd17f..b38a43e 100644 --- a/crates/miroir-core/src/config/advanced.rs +++ b/crates/miroir-core/src/config/advanced.rs @@ -3,6 +3,9 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; +// Re-export DumpImportConfig from dump_import module to avoid duplication +pub use crate::dump_import::DumpImportConfig; + // --------------------------------------------------------------------------- // 13.1 Online resharding // --------------------------------------------------------------------------- @@ -252,29 +255,6 @@ impl Default for AntiEntropyConfig { // 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 // --------------------------------------------------------------------------- diff --git a/crates/miroir-core/src/dump_import.rs b/crates/miroir-core/src/dump_import.rs index 9d784fc..14a5509 100644 --- a/crates/miroir-core/src/dump_import.rs +++ b/crates/miroir-core/src/dump_import.rs @@ -16,7 +16,7 @@ use std::sync::Arc; use tokio::sync::RwLock; /// Dump import configuration. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct DumpImportConfig { /// Import mode: "streaming" or "broadcast". #[serde(default = "default_mode")] @@ -134,14 +134,14 @@ pub struct DumpImportManager { /// Active imports (ID -> status). active_imports: Arc>>, /// Topology for routing. - topology: Arc, + topology: Arc>, /// HTTP client for posting documents. client: Arc, } impl DumpImportManager { /// Create a new dump import manager. - pub fn new(config: DumpImportConfig, topology: Arc, client: C) -> Self { + pub fn new(config: DumpImportConfig, topology: Arc>, client: C) -> Self { Self { config, active_imports: Arc::new(RwLock::new(HashMap::new())), @@ -226,7 +226,7 @@ impl DumpImportManager { dump_data: Vec, primary_key: String, shard_count: u32, - topology: Arc, + topology: Arc>, config: DumpImportConfig, imports: Arc>>, client: Arc, @@ -244,6 +244,9 @@ impl DumpImportManager { let mut processed = 0u64; let bytes_read = dump_data.len() as u64; + // Acquire topology read lock for routing + let topology = topology.read().await; + for line in data_str.lines() { if line.is_empty() { continue; @@ -476,7 +479,7 @@ mod tests { #[tokio::test] async fn test_get_status_nonexistent() { - let topology = Arc::new(Topology::new(64, 2, 1)); + let topology = Arc::new(RwLock::new(Topology::new(64, 2, 1))); let client = MockNodeClient::default(); let manager = DumpImportManager::new(DumpImportConfig::default(), topology, client); @@ -490,7 +493,7 @@ mod tests { mode: "broadcast".into(), ..Default::default() }; - let topology = Arc::new(Topology::new(64, 2, 1)); + let topology = Arc::new(RwLock::new(Topology::new(64, 2, 1))); let client = MockNodeClient::default(); let manager = DumpImportManager::new(config, topology, client); @@ -516,7 +519,7 @@ mod tests { 0, )); - let topology = Arc::new(topology); + let topology = Arc::new(RwLock::new(topology)); // Create mock client let mut client = MockNodeClient::default(); @@ -569,7 +572,7 @@ mod tests { #[tokio::test] async fn test_import_invalid_json() { - let topology = Arc::new(Topology::new(64, 2, 1)); + let topology = Arc::new(RwLock::new(Topology::new(64, 2, 1))); let client = MockNodeClient::default(); let manager = DumpImportManager::new(DumpImportConfig::default(), topology, client); diff --git a/crates/miroir-core/src/rebalancer_worker/anti_entropy_worker.rs b/crates/miroir-core/src/rebalancer_worker/anti_entropy_worker.rs index 6a56308..b9ce52e 100644 --- a/crates/miroir-core/src/rebalancer_worker/anti_entropy_worker.rs +++ b/crates/miroir-core/src/rebalancer_worker/anti_entropy_worker.rs @@ -54,8 +54,8 @@ impl Default for AntiEntropyWorkerConfig { fn default() -> Self { Self { interval_s: 6 * 3600, // 6 hours - lease_renewal_interval_ms: 5000, // 5 seconds - lease_ttl_secs: 30, // 30 seconds + lease_renewal_interval_ms: 3000, // 3 seconds (plan §14.8) + lease_ttl_secs: 10, // 10 seconds (plan §14.8) } } } @@ -774,6 +774,6 @@ mod tests { let config = AntiEntropyWorkerConfig::default(); assert_eq!(config.interval_s, 6 * 3600); assert_eq!(config.lease_ttl_secs, 10); - assert_eq!(config.lease_renewal_interval_ms, 2000); + assert_eq!(config.lease_renewal_interval_ms, 3000); // 3 seconds per plan §14.8 } } diff --git a/crates/miroir-core/src/rebalancer_worker/mod.rs b/crates/miroir-core/src/rebalancer_worker/mod.rs index 3ca5e37..d3f9840 100644 --- a/crates/miroir-core/src/rebalancer_worker/mod.rs +++ b/crates/miroir-core/src/rebalancer_worker/mod.rs @@ -113,20 +113,20 @@ pub struct ShardMigrationProgress { /// Per-shard migration state for the worker. #[derive(Debug, Clone, Serialize, Deserialize)] -struct ShardState { +pub struct ShardState { /// Current phase. - phase: ShardMigrationPhase, + pub phase: ShardMigrationPhase, /// Documents migrated so far. - docs_migrated: u64, + pub docs_migrated: u64, /// Last offset for pagination resume. - last_offset: u32, + pub last_offset: u32, /// Source node for migration. - source_node: Option, + pub source_node: Option, /// Target node for migration. - target_node: String, + pub target_node: String, /// When this shard migration started. #[serde(skip, default = "Instant::now")] - started_at: Instant, + pub started_at: Instant, } /// Migration phases for a single shard. @@ -150,25 +150,25 @@ pub enum ShardMigrationPhase { /// State machine for a rebalance job (per index). #[derive(Debug, Clone, Serialize, Deserialize)] -struct RebalanceJob { +pub struct RebalanceJob { /// Job ID. - id: RebalanceJobId, + pub id: RebalanceJobId, /// Index UID being rebalanced. - index_uid: String, + pub index_uid: String, /// Replica group being rebalanced. - replica_group: u32, + pub replica_group: u32, /// Per-shard migration state. - shards: HashMap, + pub shards: HashMap, /// Job started at. #[serde(skip, default = "Instant::now")] - started_at: Instant, + pub started_at: Instant, /// Job completed at (if finished). #[serde(skip, default)] - completed_at: Option, + pub completed_at: Option, /// Total documents migrated. - total_docs_migrated: u64, + pub total_docs_migrated: u64, /// Whether the job is paused. - paused: bool, + pub paused: bool, } /// Configuration for the rebalancer worker. @@ -287,6 +287,11 @@ impl RebalancerWorker { self.event_tx.clone() } + /// Get all active rebalance jobs. + pub async fn jobs(&self) -> HashMap { + self.jobs.read().await.clone() + } + /// Start the background worker. /// /// This runs in a loop: diff --git a/crates/miroir-proxy/Cargo.toml b/crates/miroir-proxy/Cargo.toml index 95b6616..cc25c56 100644 --- a/crates/miroir-proxy/Cargo.toml +++ b/crates/miroir-proxy/Cargo.toml @@ -48,7 +48,7 @@ hex = "0.4" # HTTP / Tower tower = "0.5" -rust-embed = { version = "8", features = ["debug-embed"] } +rust-embed = { version = "8", features = ["debug-embed", "include-exclude"] } mime_guess = "2" # OpenTelemetry (optional - use feature flag to enable) diff --git a/crates/miroir-proxy/src/admin_ui.rs b/crates/miroir-proxy/src/admin_ui.rs index cffa446..fff6c16 100644 --- a/crates/miroir-proxy/src/admin_ui.rs +++ b/crates/miroir-proxy/src/admin_ui.rs @@ -28,7 +28,7 @@ pub use crate::routes::admin_endpoints; #[exclude = "*.swp"] #[exclude = "*.DS_Store"] #[exclude = ".git"] -struct AdminUiAssets; +pub struct AdminUiAssets; /// Serve the Admin Web UI. /// diff --git a/crates/miroir-proxy/src/routes/admin_endpoints.rs b/crates/miroir-proxy/src/routes/admin_endpoints.rs index eb131b2..9470b81 100644 --- a/crates/miroir-proxy/src/routes/admin_endpoints.rs +++ b/crates/miroir-proxy/src/routes/admin_endpoints.rs @@ -1803,7 +1803,7 @@ where // Check if there's already a rebalance job for this index let job_id = miroir_core::rebalancer_worker::RebalanceJobId::new(&index_uid); let has_existing_job = { - let jobs = worker.jobs.read().await; + let jobs = worker.jobs().await; jobs.contains_key(&job_id) }; @@ -2758,7 +2758,7 @@ where // Get current shard count from topology let topology = app_state.topology.read().await; - let old_shards = topology.shard_count(); + let old_shards = topology.shards; drop(topology); // Validate new_shards > old_shards (only scaling up is supported) @@ -2769,9 +2769,8 @@ where // Get node addresses for shadow creation let topology = app_state.topology.read().await; let node_addresses: Vec = topology - .all_nodes() - .iter() - .map(|n| n.address().to_string()) + .nodes() + .map(|n| n.address.clone()) .collect(); drop(topology); diff --git a/crates/miroir-proxy/src/routes/search.rs b/crates/miroir-proxy/src/routes/search.rs index 4b5dcfe..3c27828 100644 --- a/crates/miroir-proxy/src/routes/search.rs +++ b/crates/miroir-proxy/src/routes/search.rs @@ -323,7 +323,8 @@ async fn search_handler( // Convert HeaderMap to HashMap for tenant resolution let mut headers_map = std::collections::HashMap::new(); for (name, value) in headers.iter() { - if let (Some(name_str), Ok(value_str)) = (name.as_str(), value.to_str()) { + let name_str: &str = name.as_ref(); + if let Ok(value_str) = value.to_str() { headers_map.insert(name_str.to_string(), value_str.to_string()); } }