diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 1f46c81f6..c995a2aa8 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -62,6 +62,18 @@ impl ErasureSetHealer { } } + fn effective_heal_page_object_concurrency_for_scan_mode(scan_mode: HealScanMode) -> usize { + if matches!(scan_mode, HealScanMode::Deep) { + 1 + } else { + Self::effective_heal_page_object_concurrency() + } + } + + fn is_object_not_found_message(message: &str) -> bool { + message.contains("File not found") || message.contains("not found") + } + pub fn new( storage: Arc, progress: Arc>, @@ -292,7 +304,7 @@ impl ErasureSetHealer { // 2. process objects with pagination to avoid loading all objects into memory let mut continuation_token: Option = None; let mut global_obj_idx = 0usize; - let page_concurrency_limit = Self::effective_heal_page_object_concurrency(); + let page_concurrency_limit = Self::effective_heal_page_object_concurrency_for_scan_mode(self.heal_opts.scan_mode); let in_flight = Arc::new(AtomicUsize::new(0)); loop { @@ -329,23 +341,50 @@ impl ErasureSetHealer { let in_flight = in_flight.clone(); let set_label = set_disk_id.to_string(); let heal_opts = self.heal_opts; - let permit = semaphore - .clone() - .acquire_owned() - .await - .map_err(|e| Error::other(format!("Failed to acquire page concurrency permit: {e}")))?; - - let current_in_flight = in_flight.fetch_add(1, Ordering::SeqCst) + 1; - gauge!( - "rustfs_heal_page_concurrency_current", - "set" => set_label.clone() - ) - .set(current_in_flight as f64); + let deep_scan = matches!(heal_opts.scan_mode, HealScanMode::Deep); + let semaphore = semaphore.clone(); page_tasks.push(async move { - let _permit = permit; + let permit = semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| Error::other(format!("Failed to acquire page concurrency permit: {e}"))); + + let _permit = match permit { + Ok(permit) => permit, + Err(err) => return (object_name, Err(err)), + }; + + let current_in_flight = in_flight.fetch_add(1, Ordering::SeqCst) + 1; + gauge!( + "rustfs_heal_page_concurrency_current", + "set" => set_label.clone() + ) + .set(current_in_flight as f64); + let result = if cancel_token.is_cancelled() { Err(Error::TaskCancelled) + } else if deep_scan { + match storage.heal_object(&bucket_name, &object_name, None, &heal_opts).await { + Ok((_result, None)) => Ok(true), + Ok((_, Some(err))) => { + let err_msg = err.to_string(); + if Self::is_object_not_found_message(&err_msg) { + Ok(false) + } else { + Err(Error::other(err)) + } + } + Err(err) => { + let err_msg = err.to_string(); + if Self::is_object_not_found_message(&err_msg) { + Ok(false) + } else { + Err(err) + } + } + } } else { let object_exists = match storage.object_exists(&bucket_name, &object_name).await { Ok(exists) => exists, @@ -676,6 +715,7 @@ impl ErasureSetHealer { #[cfg(test)] mod tests { use super::ErasureSetHealer; + use rustfs_common::heal_channel::HealScanMode; #[test] fn heal_page_object_concurrency_uses_default_when_env_is_unset() { @@ -702,4 +742,24 @@ mod tests { }); }); } + + #[test] + fn deep_scan_heal_page_object_concurrency_is_serial() { + temp_env::with_var(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("11"), || { + assert_eq!( + ErasureSetHealer::effective_heal_page_object_concurrency_for_scan_mode(HealScanMode::Deep), + 1 + ); + }); + } + + #[test] + fn normal_scan_heal_page_object_concurrency_uses_effective_limit() { + temp_env::with_var(rustfs_config::ENV_HEAL_PAGE_OBJECT_CONCURRENCY, Some("11"), || { + assert_eq!( + ErasureSetHealer::effective_heal_page_object_concurrency_for_scan_mode(HealScanMode::Normal), + 11 + ); + }); + } } diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index e4a503f0e..46ece118d 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -662,7 +662,11 @@ impl HealTask { let heal_opts = HealOpts { recursive: self.options.recursive, dry_run: self.options.dry_run, - remove: self.options.remove_corrupted, + remove: if self.options.recursive { + false + } else { + self.options.remove_corrupted + }, recreate: self.options.recreate_missing, scan_mode: self.options.scan_mode, update_parity: self.options.update_parity, @@ -1130,22 +1134,40 @@ impl HealTask { // Step 3: Heal bucket structure // Check control flags before each iteration to ensure timely cancellation. - // Each heal_bucket call may handle timeout/cancellation internally, see its implementation for details. + let bucket_heal_opts = HealOpts { + recursive: false, + dry_run: self.options.dry_run, + remove: false, + recreate: self.options.recreate_missing, + scan_mode: self.options.scan_mode, + update_parity: self.options.update_parity, + no_lock: false, + pool: self.options.pool_index, + set: self.options.set_index, + }; + for bucket in buckets.iter() { // Check control flags before starting each bucket heal self.check_control_flags().await?; - // heal_bucket internally uses await_with_control for timeout/cancellation handling - if let Err(err) = self.heal_bucket(bucket).await { - // Check if error is due to cancellation or timeout - if matches!(err, Error::TaskCancelled | Error::TaskTimeout) { - return Err(err); + let heal_result = self + .await_with_control(self.storage.heal_bucket(bucket, &bucket_heal_opts)) + .await; + match heal_result { + Ok(result) => { + self.record_result_item(result).await; + } + Err(err) => { + // Check if error is due to cancellation or timeout + if matches!(err, Error::TaskCancelled | Error::TaskTimeout) { + return Err(err); + } + info!("Bucket heal failed: {}", err.to_string()); } - info!("Bucket heal failed: {}", err.to_string()); } } - // Step 3: Create erasure set healer with resume support - info!("Step 3: Creating erasure set healer with resume support"); + // Create erasure set healer with resume support + info!("Creating erasure set healer with resume support"); let heal_opts = HealOpts { recursive: self.options.recursive, dry_run: self.options.dry_run, @@ -1217,6 +1239,8 @@ mod tests { struct MockStorage { listed: Mutex, healed_objects: Mutex>, + bucket_heal_opts: Mutex>, + object_heal_opts: Mutex>, } #[async_trait::async_trait] @@ -1285,9 +1309,10 @@ mod tests { _bucket: &str, object: &str, _version_id: Option<&str>, - _opts: &HealOpts, + opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { self.healed_objects.lock().unwrap().push(object.to_string()); + self.object_heal_opts.lock().unwrap().push(*opts); Ok(( HealResultItem { object_size: 1, @@ -1297,7 +1322,8 @@ mod tests { )) } - async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result { + async fn heal_bucket(&self, _bucket: &str, opts: &HealOpts) -> Result { + self.bucket_heal_opts.lock().unwrap().push(*opts); Ok(HealResultItem::default()) } @@ -1360,4 +1386,40 @@ mod tests { assert_eq!(result_items.len(), 3); assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2); } + + #[tokio::test] + async fn test_recursive_bucket_heal_does_not_remove_bucket_metadata() { + let storage = Arc::new(MockStorage::default()); + let request = HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + remove_corrupted: true, + recreate_missing: true, + scan_mode: HealScanMode::Deep, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ); + let task = HealTask::from_request(request, storage.clone()); + + task.heal_bucket("bucket-a") + .await + .expect("recursive bucket heal should succeed"); + + let bucket_opts = storage.bucket_heal_opts.lock().unwrap(); + assert_eq!(bucket_opts.len(), 1); + assert!(!bucket_opts[0].remove); + assert!(bucket_opts[0].recreate); + assert_eq!(bucket_opts[0].scan_mode, HealScanMode::Deep); + + let object_opts = storage.object_heal_opts.lock().unwrap(); + assert_eq!(object_opts.len(), 2); + assert!(object_opts.iter().all(|opts| opts.remove)); + assert!(object_opts.iter().all(|opts| opts.recreate)); + assert!(object_opts.iter().all(|opts| opts.scan_mode == HealScanMode::Deep)); + } }