fix: rebuild wiped disks during admin heal (#3084)

* fix: rebuild wiped disks during admin heal

* Preserve forceStart heal admission semantics

* Address heal regression test review comments

* fix(heal): address admin heal review comments

* fix(heal): fix force-start dedup and test polling

* fix(heal): address unresolved review comments

* fix(heal): simplify dedup key for clippy

---------

Co-authored-by: houseme <housemecn@gmail.com>

crates/heal/src/heal/channel.rs

@@ -331,7 +331,7 @@ impl HealChannelProcessor {
         };
 
         // Build HealOptions with all available fields
-        let mut options = HealOptions {
+        let options = HealOptions {
             scan_mode: request.scan_mode.unwrap_or(HealScanMode::Normal),
             remove_corrupted: request.remove_corrupted.unwrap_or(false),
             recreate_missing: request.recreate_missing.unwrap_or(true),
@@ -343,15 +343,13 @@ impl HealChannelProcessor {
             set_index: request.set_index,
         };
 
-        // Apply force_start overrides
-        if request.force_start {
-            options.remove_corrupted = true;
-            options.recreate_missing = true;
-            options.update_parity = true;
-        }
-
         let mut heal_request = HealRequest::new(heal_type, options, priority);
         heal_request.id = request.id;
+        // force_start controls admission/queue semantics only. Do not reinterpret it as
+        // destructive heal options: admin clients commonly pass forceStart=true together
+        // with remove=false, and turning that into remove_corrupted=true can delete the
+        // remaining healthy bucket volumes before object shards are rebuilt.
+        heal_request.force_start = request.force_start;
         Ok(heal_request)
     }
 
@@ -664,13 +662,14 @@ mod tests {
             timeout_seconds: None,
             pool_index: None,
             set_index: None,
-            force_start: true, // Should override the above false values
+            force_start: true, // Admission force only; must not override explicit heal options.
         };
 
         let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
-        assert!(heal_request.options.remove_corrupted);
-        assert!(heal_request.options.recreate_missing);
-        assert!(heal_request.options.update_parity);
+        assert!(heal_request.force_start);
+        assert!(!heal_request.options.remove_corrupted);
+        assert!(!heal_request.options.recreate_missing);
+        assert!(!heal_request.options.update_parity);
     }
 
     #[tokio::test]

crates/heal/src/heal/erasure_healer.rs

@@ -35,6 +35,7 @@ pub struct ErasureSetHealer {
     progress: Arc<RwLock<HealProgress>>,
     cancel_token: tokio_util::sync::CancellationToken,
     disk: DiskStore,
+    heal_opts: HealOpts,
 }
 
 impl ErasureSetHealer {
@@ -66,12 +67,14 @@ impl ErasureSetHealer {
         progress: Arc<RwLock<HealProgress>>,
         cancel_token: tokio_util::sync::CancellationToken,
         disk: DiskStore,
+        heal_opts: HealOpts,
     ) -> Self {
         Self {
             storage,
             progress,
             cancel_token,
             disk,
+            heal_opts,
         }
     }
 
@@ -325,6 +328,7 @@ impl ErasureSetHealer {
                 let cancel_token = self.cancel_token.clone();
                 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()
@@ -375,12 +379,6 @@ impl ErasureSetHealer {
                         if !object_exists {
                             Ok(false)
                         } else {
-                            let heal_opts = HealOpts {
-                                scan_mode: HealScanMode::Normal,
-                                remove: true,
-                                recreate: true,
-                                ..Default::default()
-                            };
                             match storage.heal_object(&bucket_name, &object_name, None, &heal_opts).await {
                                 Ok((_result, None)) => Ok(true),
                                 Ok((_, Some(err))) => Err(Error::other(err)),

crates/heal/src/heal/manager.rs

@@ -25,7 +25,7 @@ use rustfs_ecstore::disk::error::DiskError;
 use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP;
 use rustfs_madmin::heal_commands::HealResultItem;
 use std::{
-    collections::{BinaryHeap, HashMap, HashSet},
+    collections::{BinaryHeap, HashMap},
     sync::Arc,
     time::{Duration, SystemTime},
 };
@@ -46,8 +46,8 @@ struct PriorityHealQueue {
     heap: BinaryHeap<PriorityQueueItem>,
     /// Sequence counter for FIFO ordering within same priority
     sequence: u64,
-    /// Set of request keys to prevent duplicates
-    dedup_keys: HashSet<String>,
+    /// Deduplication key reference counts for queued requests
+    dedup_keys: HashMap<String, usize>,
 }
 
 /// Wrapper for heap items to implement proper ordering
@@ -110,7 +110,7 @@ impl PriorityHealQueue {
         Self {
             heap: BinaryHeap::new(),
             sequence: 0,
-            dedup_keys: HashSet::new(),
+            dedup_keys: HashMap::new(),
         }
     }
 
@@ -121,7 +121,7 @@ impl PriorityHealQueue {
     fn pop_next(&mut self) -> Option<HealRequest> {
         self.heap.pop().map(|item| {
             let key = Self::make_dedup_key(&item.request);
-            self.dedup_keys.remove(&key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
             item.request
         })
     }
@@ -133,12 +133,13 @@ impl PriorityHealQueue {
     fn push(&mut self, request: HealRequest) -> QueuePushOutcome {
         let key = Self::make_dedup_key(&request);
 
-        // Check for duplicates
-        if self.dedup_keys.contains(&key) {
+        // Check for duplicates unless the caller explicitly forces admission.
+        if self.dedup_keys.contains_key(&key) && !request.force_start {
             return QueuePushOutcome::Merged;
         }
-
-        self.dedup_keys.insert(key);
+        // Track dedup keys for both normal and forced requests so queued forced work
+        // also reserves the dedup key for later non-forced duplicates.
+        *self.dedup_keys.entry(key).or_insert(0) += 1;
         self.sequence += 1;
         self.heap.push(PriorityQueueItem {
             priority: request.priority,
@@ -161,7 +162,7 @@ impl PriorityHealQueue {
     fn pop(&mut self) -> Option<HealRequest> {
         self.heap.pop().map(|item| {
             let key = Self::make_dedup_key(&item.request);
-            self.dedup_keys.remove(&key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
             item.request
         })
     }
@@ -201,7 +202,7 @@ impl PriorityHealQueue {
         (
             selected.map(|item| {
                 let key = Self::make_dedup_key(&item.request);
-                self.dedup_keys.remove(&key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
                 item.request
             }),
             skipped,
@@ -240,17 +241,27 @@ impl PriorityHealQueue {
         }
     }
 
+    fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, usize>, key: &str) {
+        if let Some(count) = dedup_keys.get_mut(key) {
+            if *count <= 1 {
+                dedup_keys.remove(key);
+            } else {
+                *count -= 1;
+            }
+        }
+    }
+
     /// Check if a request with the same key already exists in the queue
     #[allow(dead_code)]
     fn contains_key(&self, request: &HealRequest) -> bool {
         let key = Self::make_dedup_key(request);
-        self.dedup_keys.contains(&key)
+        self.dedup_keys.contains_key(&key)
     }
 
     /// Check if an erasure set heal request for a specific set_disk_id exists
     fn contains_erasure_set(&self, set_disk_id: &str) -> bool {
         let key = format!("erasure_set:{set_disk_id}");
-        self.dedup_keys.contains(&key)
+        self.dedup_keys.contains_key(&key)
     }
 
     fn contains_request_id(&self, request_id: &str) -> bool {
@@ -277,7 +288,7 @@ impl PriorityHealQueue {
         while let Some(item) = self.heap.pop() {
             if removed.is_none() && item.request.id == request_id {
                 let key = Self::make_dedup_key(&item.request);
-                self.dedup_keys.remove(&key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
                 removed = Some(item.request);
             } else {
                 retained.push(item);
@@ -298,7 +309,7 @@ impl PriorityHealQueue {
         while let Some(item) = self.heap.pop() {
             if should_remove(&item.request) {
                 let key = Self::make_dedup_key(&item.request);
-                self.dedup_keys.remove(&key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
                 removed_count += 1;
             } else {
                 retained.push(item);
@@ -541,7 +552,7 @@ impl HealManager {
         publish_heal_queue_length(&queue);
         let queue_capacity = config.queue_size;
 
-        if queue.contains_key(&request) {
+        if !request.force_start && queue.contains_key(&request) {
             let admission = if request.priority == HealPriority::Low && !config.low_priority_merge_enable {
                 HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped)
             } else {
@@ -566,7 +577,7 @@ impl HealManager {
             return Ok(admission);
         }
 
-        if queue_len >= queue_capacity {
+        if queue_len >= queue_capacity && !request.force_start {
             let admission = Self::classify_full_admission(&request, &config);
             match admission {
                 HealAdmissionResult::Dropped(reason) => {
@@ -2124,6 +2135,122 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_force_start_bypasses_duplicate_and_full_admission() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(
+            storage,
+            Some(HealConfig {
+                queue_size: 1,
+                low_priority_drop_when_full: true,
+                ..HealConfig::default()
+            }),
+        );
+
+        let normal = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+        let mut forced_duplicate = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+        forced_duplicate.force_start = true;
+
+        let subsequent_duplicate = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+
+        assert_eq!(
+            manager
+                .submit_heal_request(normal)
+                .await
+                .expect("first request should be accepted"),
+            HealAdmissionResult::Accepted
+        );
+        assert_eq!(
+            manager
+                .submit_heal_request(forced_duplicate)
+                .await
+                .expect("force start should bypass duplicate/full policy"),
+            HealAdmissionResult::Accepted
+        );
+        assert_eq!(
+            manager
+                .submit_heal_request(subsequent_duplicate)
+                .await
+                .expect("subsequent non-force duplicate should be merged"),
+            HealAdmissionResult::Merged
+        );
+    }
+
+    #[tokio::test]
+    async fn test_force_start_marks_dedup_key_for_future_duplicates() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(
+            storage,
+            Some(HealConfig {
+                queue_size: 1,
+                ..HealConfig::default()
+            }),
+        );
+
+        let normal = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+        let mut forced = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+        forced.force_start = true;
+        let duplicate = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+
+        assert_eq!(
+            manager
+                .submit_heal_request(normal)
+                .await
+                .expect("first request should be accepted"),
+            HealAdmissionResult::Accepted
+        );
+        assert_eq!(
+            manager
+                .submit_heal_request(forced)
+                .await
+                .expect("forced request should bypass duplicate/full admission"),
+            HealAdmissionResult::Accepted
+        );
+        assert_eq!(
+            manager
+                .submit_heal_request(duplicate)
+                .await
+                .expect("non-forced duplicate should merge while forced request is queued"),
+            HealAdmissionResult::Merged
+        );
+    }
+
     #[test]
     fn test_running_erasure_set_counts_groups_only_erasure_tasks() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/heal/task.rs

@@ -133,6 +133,8 @@ pub struct HealRequest {
     pub options: HealOptions,
     /// Priority
     pub priority: HealPriority,
+    /// Whether this request should bypass queue admission dedup/full policies.
+    pub force_start: bool,
     /// Created time
     pub created_at: SystemTime,
     /// Queue admission time used for scheduler delay metrics
@@ -147,6 +149,7 @@ impl HealRequest {
             heal_type,
             options,
             priority,
+            force_start: false,
             created_at: now,
             enqueued_at: now,
         }
@@ -1143,7 +1146,19 @@ impl HealTask {
 
         // Step 3: Create erasure set healer with resume support
         info!("Step 3: Creating erasure set healer with resume support");
-        let erasure_healer = ErasureSetHealer::new(self.storage.clone(), self.progress.clone(), self.cancel_token.clone(), disk);
+        let heal_opts = HealOpts {
+            recursive: self.options.recursive,
+            dry_run: self.options.dry_run,
+            remove: self.options.remove_corrupted,
+            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,
+        };
+        let erasure_healer =
+            ErasureSetHealer::new(self.storage.clone(), self.progress.clone(), self.cancel_token.clone(), disk, heal_opts);
 
         {
             let mut progress = self.progress.write().await;