diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index 1a74bd7f9..896033c36 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -286,6 +286,23 @@ impl Sets { self.get_disks(self.get_hashed_set_index(key)) } + fn get_disks_for_heal_object(&self, key: &str, opts: &HealOpts) -> Result> { + match opts.set { + Some(set_idx) => self.disk_set.get(set_idx).cloned().ok_or_else(|| { + StorageError::InvalidArgument( + "heal".to_string(), + "set".to_string(), + format!( + "invalid heal set index {set_idx} for pool {} with {} sets", + self.pool_idx, + self.disk_set.len() + ), + ) + }), + None => Ok(self.get_disks_by_key(key)), + } + } + pub(crate) async fn storage_info_snapshot(&self) -> rustfs_madmin::StorageInfo { let mut futures = Vec::with_capacity(self.disk_set.len()); @@ -1101,7 +1118,7 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets { version_id: &str, opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { - self.get_disks_by_key(object) + self.get_disks_for_heal_object(object, opts)? .heal_object(bucket, object, version_id, opts) .await } @@ -1431,6 +1448,53 @@ mod tests { (temp_dirs, sets) } + #[tokio::test] + async fn heal_object_uses_explicit_set_scope() { + let (_temp_dirs, sets) = two_set_test_sets().await; + let selected = sets + .get_disks_for_heal_object( + "object", + &HealOpts { + set: Some(1), + ..Default::default() + }, + ) + .expect("requested set should be selected"); + + assert!(Arc::ptr_eq(&selected, &sets.disk_set[1])); + } + + #[tokio::test] + async fn heal_object_without_set_scope_keeps_hash_routing() { + let (_temp_dirs, sets) = two_set_test_sets().await; + let object = "object"; + let selected = sets + .get_disks_for_heal_object(object, &HealOpts::default()) + .expect("hash-routed set should be selected"); + + assert!(Arc::ptr_eq(&selected, &sets.get_disks_by_key(object))); + } + + #[tokio::test] + async fn heal_object_rejects_invalid_set_scope() { + let (_temp_dirs, sets) = two_set_test_sets().await; + let err = sets + .get_disks_for_heal_object( + "object", + &HealOpts { + set: Some(2), + ..Default::default() + }, + ) + .expect_err("out-of-range set scope must fail closed"); + + assert!( + matches!(err, StorageError::InvalidArgument(_, ref field, ref reason) + if field == "set" && reason.contains("invalid heal set index 2 for pool 0 with 2 sets")), + "unexpected invalid set error: {err:?}" + ); + } + #[tokio::test] async fn delete_prefix_surfaces_a_hard_error_from_any_set() { let (_temp_dirs, sets) = two_set_test_sets().await; diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index a77381014..fb1b19d6d 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -21,7 +21,27 @@ const LOG_SUBSYSTEM_HEAL: &str = "heal"; const EVENT_HEAL_FORMAT_COMPLETED: &str = "heal_format_completed"; const EVENT_HEAL_OBJECT_STARTED: &str = "heal_object_started"; +fn invalid_heal_pool_index(pool_idx: usize, pool_count: usize) -> Error { + StorageError::InvalidArgument( + "heal".to_string(), + "pool".to_string(), + format!("invalid heal pool index {pool_idx} for {pool_count} pools"), + ) +} + impl ECStore { + fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result>> { + match opts.pool { + Some(pool_idx) => Ok(vec![ + self.pools + .get(pool_idx) + .cloned() + .ok_or_else(|| invalid_heal_pool_index(pool_idx, self.pools.len()))?, + ]), + None => Ok(self.pools.clone()), + } + } + #[instrument(skip(self))] pub(super) async fn handle_heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { let mut r = HealResultItem { @@ -105,8 +125,10 @@ impl ECStore { ); let object = encode_dir_object(object); - let mut futures = Vec::with_capacity(self.pools.len()); - for pool in self.pools.iter() { + let pools = self.get_pools_for_heal_object(opts)?; + + let mut futures = Vec::with_capacity(pools.len()); + for pool in pools.iter() { if self.is_suspended(pool.pool_idx).await { continue; } @@ -178,6 +200,82 @@ mod tests { use crate::layout::endpoints::{Endpoints, PoolEndpoints}; use crate::store::init_format::{load_format_erasure, save_format_file}; + async fn minimal_heal_pool(pool_idx: usize) -> Arc { + let format = FormatV3::new(1, 1); + let endpoint_url = format!("http://127.0.0.1:{}/data", 19000 + pool_idx); + let mut endpoint = Endpoint::try_from(endpoint_url.as_str()).expect("endpoint should parse"); + endpoint.set_pool_index(pool_idx); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + + Sets::new( + vec![None], + &PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 1, + endpoints: Endpoints::from(vec![endpoint]), + cmd_line: String::new(), + platform: String::new(), + }, + &format, + pool_idx, + 0, + ) + .await + .expect("minimal pool should build") + } + + async fn minimal_heal_store() -> ECStore { + ECStore { + id: Uuid::new_v4(), + disk_map: HashMap::new(), + pools: vec![minimal_heal_pool(0).await, minimal_heal_pool(1).await], + peer_sys: S3PeerSys { + clients: Vec::new(), + pools_count: 2, + }, + pool_meta: RwLock::new(PoolMeta::default()), + rebalance_meta: RwLock::new(None), + decommission_cancelers: RwLock::new(Vec::new()), + start_gate: Mutex::new(()), + pool_meta_save_gate: Mutex::new(()), + ctx: crate::runtime::instance::bootstrap_ctx(), + bucket_fence_registry: std::sync::Arc::default(), + } + } + + #[tokio::test] + async fn heal_object_pool_scope_selects_only_requested_pool() { + let store = minimal_heal_store().await; + let pools = store + .get_pools_for_heal_object(&HealOpts { + pool: Some(1), + ..Default::default() + }) + .expect("requested pool should be selected"); + + assert_eq!(pools.len(), 1); + assert!(Arc::ptr_eq(&pools[0], &store.pools[1])); + } + + #[tokio::test] + async fn heal_object_pool_scope_rejects_invalid_pool() { + let store = minimal_heal_store().await; + let err = store + .get_pools_for_heal_object(&HealOpts { + pool: Some(2), + ..Default::default() + }) + .expect_err("out-of-range pool scope must fail closed"); + + assert!( + matches!(err, StorageError::InvalidArgument(_, ref field, ref reason) + if field == "pool" && reason.contains("invalid heal pool index 2 for 2 pools")), + "unexpected invalid pool error: {err:?}" + ); + } + #[tokio::test] async fn handle_heal_format_continues_after_a_pool_error() { let canonical_format = FormatV3::new(1, 3); diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 024dd1d52..595943575 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -2475,7 +2475,7 @@ impl HealManager { return; } - let mut running_per_set = running_erasure_set_counts(&active_heals_guard); + let mut running_per_set = running_heal_set_counts(&active_heals_guard); let mut tasks_started = 0usize; let mut delayed_by_mainline_throttle = false; @@ -2856,6 +2856,14 @@ impl std::fmt::Debug for HealManager { fn heal_request_set_key(request: &HealRequest) -> Option { match &request.heal_type { HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()), + HealType::Object { .. } => heal_options_set_key(&request.options), + _ => None, + } +} + +fn heal_options_set_key(options: &HealOptions) -> Option { + match (options.pool_index, options.set_index) { + (Some(pool), Some(set)) => Some(format!("pool_{pool}_set_{set}")), _ => None, } } @@ -2905,16 +2913,24 @@ fn update_task_running_metric_for_task(active_heals: &HashMap>) -> HashMap { +fn running_heal_set_counts(active_heals: &HashMap>) -> HashMap { let mut running = HashMap::new(); for task in active_heals.values() { - if let HealType::ErasureSet { set_disk_id, .. } = &task.heal_type { - *running.entry(set_disk_id.clone()).or_insert(0) += 1; + if let Some(set_key) = heal_request_set_key_for_task(task) { + *running.entry(set_key).or_insert(0) += 1; } } running } +fn heal_request_set_key_for_task(task: &HealTask) -> Option { + match &task.heal_type { + HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()), + HealType::Object { .. } => heal_options_set_key(&task.options), + _ => None, + } +} + fn prune_completed_heal_statuses(completed_heals: &mut HashMap) { let Ok(now) = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) else { return; @@ -3523,6 +3539,30 @@ mod tests { assert!(can_schedule_request(&request, &running, 2)); } + #[test] + fn test_can_schedule_scoped_object_request_respects_per_set_limit() { + let options = HealOptions { + pool_index: Some(0), + set_index: Some(1), + ..Default::default() + }; + let request = HealRequest::new( + HealType::Object { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + }, + options, + HealPriority::Normal, + ); + + let mut running = HashMap::new(); + running.insert("pool_0_set_1".to_string(), 1); + + assert!(!can_schedule_request(&request, &running, 1)); + assert!(can_schedule_request(&request, &running, 2)); + } + #[tokio::test] async fn test_submit_heal_request_returns_merged_for_duplicate() { let storage: Arc = Arc::new(MockStorage); @@ -5136,7 +5176,7 @@ mod tests { } #[test] - fn test_running_erasure_set_counts_groups_only_erasure_tasks() { + fn test_running_heal_set_counts_groups_set_scoped_tasks() { let storage: Arc = Arc::new(MockStorage); let erasure_task = Arc::new(HealTask::from_request( HealRequest::new( @@ -5149,6 +5189,23 @@ mod tests { ), storage.clone(), )); + let scoped_options = HealOptions { + pool_index: Some(0), + set_index: Some(1), + ..Default::default() + }; + let scoped_object_task = Arc::new(HealTask::from_request( + HealRequest::new( + HealType::Object { + bucket: "bucket".to_string(), + object: "scoped-object".to_string(), + version_id: None, + }, + scoped_options, + HealPriority::Normal, + ), + storage.clone(), + )); let object_task = Arc::new(HealTask::from_request( HealRequest::new( HealType::Object { @@ -5164,10 +5221,11 @@ mod tests { let mut active = HashMap::new(); active.insert(erasure_task.id.clone(), erasure_task); + active.insert(scoped_object_task.id.clone(), scoped_object_task); active.insert(object_task.id.clone(), object_task); - let counts = running_erasure_set_counts(&active); - assert_eq!(counts.get("pool_0_set_1"), Some(&1)); + let counts = running_heal_set_counts(&active); + assert_eq!(counts.get("pool_0_set_1"), Some(&2)); assert_eq!(counts.len(), 1); }