perf(heal): trim scanner and heal queue hot paths (#6307)

Cache heal queue dedup keys, avoid retry request double construction, clear task aliases after terminal completion, and age out stale scanner pending-heal ledger entries during retry sweeps.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/heal/src/heal/manager.rs

@@ -294,7 +294,7 @@ fn completed_status_is_retrying(status: &HealTaskStatus) -> bool {
     matches!(status, HealTaskStatus::Retrying { .. })
 }
 
-fn retry_request_for_result(task: &HealTask, result: &Result<()>) -> Option<(HealRequest, Duration, String)> {
+fn retry_budget_for_result(task: &HealTask, result: &Result<()>) -> Option<(Duration, String)> {
     let Err(err) = result else {
         return None;
     };
@@ -310,14 +310,35 @@ fn retry_request_for_result(task: &HealTask, result: &Result<()>) -> Option<(Hea
         return None;
     }
 
-    let request = task.retry_request();
-    let delay = recoverable_heal_retry_delay(request.retry_attempts);
-    Some((request, delay, error))
+    let retry_attempt = task.retry_attempts.saturating_add(1);
+    let delay = recoverable_heal_retry_delay(retry_attempt);
+    Some((delay, error))
+}
+
+#[cfg(test)]
+fn retry_request_for_result(task: &HealTask, result: &Result<()>) -> Option<(HealRequest, Duration, String)> {
+    let (delay, error) = retry_budget_for_result(task, result)?;
+    Some((task.retry_request(), delay, error))
 }
 
 async fn retry_request_for_result_with_budget(task: &HealTask, result: &Result<()>) -> Option<(HealRequest, Duration, String)> {
-    let (_, delay, error) = retry_request_for_result(task, result)?;
-    let request = task.retry_request_with_remaining_timeout().await.ok()?;
+    let (delay, error) = retry_budget_for_result(task, result)?;
+    let request = match task.retry_request_with_remaining_timeout().await {
+        Ok(request) => request,
+        Err(err) => {
+            debug!(
+                target: "rustfs::heal::manager",
+                event = EVENT_HEAL_QUEUE_ADMISSION,
+                component = LOG_COMPONENT_HEAL,
+                subsystem = LOG_SUBSYSTEM_MANAGER,
+                task_id = %task.id,
+                error = %err,
+                result = "retry_budget_exhausted",
+                "Heal retry admission decided"
+            );
+            return None;
+        }
+    };
     Some((request, delay, error))
 }
 
@@ -593,6 +614,7 @@ struct HealQueueContext<'a> {
     heal_queue: &'a Arc<Mutex<PriorityHealQueue>>,
     active_heals: &'a Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
     completed_heals: &'a Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
+    task_aliases: &'a Arc<Mutex<HashMap<String, HealTaskAlias>>>,
     retrying_heals: &'a Arc<Mutex<HashMap<String, RetryingHeal>>>,
     replacement_recovery_anchors: &'a Arc<std::sync::Mutex<HashMap<String, String>>>,
     config: &'a Arc<RwLock<HealConfig>>,
@@ -899,7 +921,9 @@ impl HealManager {
             QueuePushOutcome::Accepted => {
                 publish_heal_queue_length(queue);
                 Self::record_admission_metric(source, HealAdmissionResult::Accepted, context);
-                if matches!(priority, HealPriority::High | HealPriority::Urgent) {
+                if matches!(priority, HealPriority::High | HealPriority::Urgent)
+                    && tracing::enabled!(target: "rustfs::heal::manager", tracing::Level::DEBUG)
+                {
                     let stats = queue.get_priority_stats();
                     debug!(
                         target: "rustfs::heal::manager",

crates/heal/src/heal/manager/queue.rs

@@ -41,6 +41,7 @@ pub(super) struct PriorityHealQueue {
 pub(super) struct PriorityQueueItem {
     pub(super) priority: HealPriority,
     pub(super) sequence: u64,
+    pub(super) dedup_key: String,
     pub(super) request: HealRequest,
 }
 
@@ -135,8 +136,7 @@ impl PriorityHealQueue {
 
     pub(super) fn pop_next(&mut self) -> Option<HealRequest> {
         self.heap.pop().map(|item| {
-            let key = Self::make_dedup_key(&item.request);
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
             item.request
         })
     }
@@ -157,7 +157,7 @@ impl PriorityHealQueue {
         // request that opens the key becomes the named representative for merge
         // receipts (taken before `request` moves into the heap).
         self.dedup_keys
-            .entry(key)
+            .entry(key.clone())
             .or_insert_with(|| DedupKeyEntry {
                 refcount: 0,
                 representative_request_id: request.id.clone(),
@@ -167,6 +167,7 @@ impl PriorityHealQueue {
         self.heap.push(PriorityQueueItem {
             priority: request.priority,
             sequence: self.sequence,
+            dedup_key: key,
             request,
         });
         QueuePushOutcome::Accepted
@@ -204,9 +205,8 @@ impl PriorityHealQueue {
         self.heap = retained;
 
         let displaced = displaced.map(|item| {
-            let key = Self::make_dedup_key(&item.request);
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
-            self.refresh_dedup_representative(&key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+            self.refresh_dedup_representative(&item.dedup_key);
             item.request
         });
 
@@ -244,8 +244,7 @@ impl PriorityHealQueue {
     #[cfg(test)]
     pub(super) fn pop(&mut self) -> Option<HealRequest> {
         self.heap.pop().map(|item| {
-            let key = Self::make_dedup_key(&item.request);
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
             item.request
         })
     }
@@ -284,8 +283,7 @@ impl PriorityHealQueue {
 
         (
             selected.map(|item| {
-                let key = Self::make_dedup_key(&item.request);
-                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
                 item.request
             }),
             skipped,
@@ -379,7 +377,7 @@ impl PriorityHealQueue {
         if let Some(id) = self
             .heap
             .iter()
-            .find(|item| Self::make_dedup_key(&item.request) == key)
+            .find(|item| item.dedup_key == key)
             .map(|item| item.request.id.clone())
             && let Some(entry) = self.dedup_keys.get_mut(key)
         {
@@ -397,11 +395,13 @@ impl PriorityHealQueue {
     pub(super) fn remove_request_id(&mut self, request_id: &str) -> Option<HealRequest> {
         let mut retained = BinaryHeap::new();
         let mut removed = None;
+        let mut affected_key = None;
 
         while let Some(item) = self.heap.pop() {
             if removed.is_none() && item.request.id == request_id {
-                let key = Self::make_dedup_key(&item.request);
+                let key = item.dedup_key.clone();
                 Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
+                affected_key = Some(key);
                 removed = Some(item.request);
             } else {
                 retained.push(item);
@@ -409,8 +409,8 @@ impl PriorityHealQueue {
         }
 
         self.heap = retained;
-        if let Some(removed) = removed.as_ref() {
-            self.refresh_dedup_representative(&Self::make_dedup_key(removed));
+        if let Some(key) = affected_key.as_deref() {
+            self.refresh_dedup_representative(key);
         }
         removed
     }
@@ -425,9 +425,8 @@ impl PriorityHealQueue {
 
         while let Some(item) = self.heap.pop() {
             if should_remove(&item.request) {
-                let key = Self::make_dedup_key(&item.request);
-                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
-                affected_keys.push(key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+                affected_keys.push(item.dedup_key);
                 removed.push(item.request);
             } else {
                 retained.push(item);

crates/heal/src/heal/manager/scheduler.rs

@@ -21,6 +21,7 @@ impl HealManager {
         let heal_queue = self.heal_queue.clone();
         let active_heals = self.active_heals.clone();
         let completed_heals = self.completed_heals.clone();
+        let task_aliases = self.task_aliases.clone();
         let retrying_heals = self.retrying_heals.clone();
         let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
         let cancel_token = self.cancel_token.clone();
@@ -51,6 +52,7 @@ impl HealManager {
                             heal_queue: &heal_queue,
                             active_heals: &active_heals,
                             completed_heals: &completed_heals,
+                            task_aliases: &task_aliases,
                             retrying_heals: &retrying_heals,
                             replacement_recovery_anchors: &replacement_recovery_anchors,
                             config: &config,
@@ -67,6 +69,7 @@ impl HealManager {
                             heal_queue: &heal_queue,
                             active_heals: &active_heals,
                             completed_heals: &completed_heals,
+                            task_aliases: &task_aliases,
                             retrying_heals: &retrying_heals,
                             replacement_recovery_anchors: &replacement_recovery_anchors,
                             config: &config,
@@ -92,6 +95,7 @@ impl HealManager {
             heal_queue,
             active_heals,
             completed_heals,
+            task_aliases,
             retrying_heals,
             replacement_recovery_anchors,
             config,
@@ -175,6 +179,7 @@ impl HealManager {
                 let active_heals_clone = active_heals.clone();
                 let heal_queue_clone = heal_queue.clone();
                 let completed_heals_clone = completed_heals.clone();
+                let task_aliases_clone = task_aliases.clone();
                 let retrying_heals_clone = retrying_heals.clone();
                 let replacement_recovery_anchors_clone = replacement_recovery_anchors.clone();
                 let statistics_clone = statistics.clone();
@@ -295,6 +300,7 @@ impl HealManager {
                         } else {
                             completed_task.get_status().await
                         };
+                        let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. });
                         let completed_progress = completed_task.get_progress().await;
                         // Single snapshot of the retained window: the task is
                         // finished and already off the active map, so there is
@@ -326,6 +332,13 @@ impl HealManager {
                             }
                         }
                         stats.update_running_tasks(usize_to_u64_saturated(active_count));
+                        drop(stats);
+                        if terminal_completion {
+                            task_aliases_clone
+                                .lock()
+                                .await
+                                .retain(|alias_id, alias| alias_id != &task_id && alias.task_id != task_id);
+                        }
                     }
 
                     if let (Some((retry_request, retry_delay, retry_error)), Some(retry_cancel_token)) =

crates/heal/src/heal/manager/tests.rs

@@ -84,6 +84,7 @@ async fn process_manager_queue_once(manager: &HealManager) {
         heal_queue: &manager.heal_queue,
         active_heals: &manager.active_heals,
         completed_heals: &manager.completed_heals,
+        task_aliases: &manager.task_aliases,
         retrying_heals: &manager.retrying_heals,
         replacement_recovery_anchors: &manager.replacement_recovery_anchors,
         config: &manager.config,
@@ -840,6 +841,46 @@ async fn test_admin_duplicate_receipt_returns_canonical_task_without_alias() {
     assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
 }
 
+#[tokio::test]
+async fn test_task_alias_is_removed_after_terminal_completion() {
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = HealManager::new(storage, None);
+    let original = HealRequest::object("bucket".to_string(), "object".to_string(), None);
+    let original_id = original.id.clone();
+    let duplicate = HealRequest::object("bucket".to_string(), "object".to_string(), None);
+    let duplicate_id = duplicate.id.clone();
+
+    assert_eq!(
+        manager
+            .submit_heal_request(original)
+            .await
+            .expect("first request should be accepted"),
+        HealAdmissionResult::Accepted
+    );
+    assert_eq!(
+        manager
+            .submit_heal_request(duplicate)
+            .await
+            .expect("duplicate request should merge"),
+        HealAdmissionResult::Merged
+    );
+    assert_eq!(manager.canonical_task_id(&duplicate_id).await, original_id);
+
+    process_manager_queue_once(&manager).await;
+    tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            if matches!(manager.get_task_status(&original_id).await, Ok(HealTaskStatus::Completed)) {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("task should complete promptly");
+
+    assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
+}
+
 #[tokio::test]
 async fn test_duplicate_admission_is_atomic_with_queue_to_active_transition() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/scanner/src/scanner_folder.rs

@@ -193,6 +193,7 @@ fn emit_scanner_alert_event(event_name: &str, bucket: &str, object: &str, size:
     });
 }
 const MAX_PENDING_SCANNER_HEALS_PER_BUCKET: usize = 10_000;
+const MAX_PENDING_SCANNER_HEAL_AGE_SECS: u64 = 24 * 60 * 60;
 
 static SCANNER_ALERT_METRICS_ONCE: Once = Once::new();
 

crates/scanner/src/scanner_folder/ledger.rs

@@ -100,19 +100,52 @@ impl FolderScanner {
             last_admission_result: result.result_label().to_string(),
             last_admission_reason: result.reason_label().to_string(),
         });
-        self.prune_pending_scanner_heals();
-        self.sync_pending_heals();
+        if self.prune_pending_scanner_heal_capacity() == 0 {
+            self.sync_pending_heals();
+        }
     }
 
     pub(super) fn prune_pending_scanner_heals(&mut self) {
+        let now = Self::now_secs();
+        let before_expiry = self.new_cache.info.pending_heals.len();
+        self.new_cache
+            .info
+            .pending_heals
+            .retain(|entry| now.saturating_sub(entry.first_seen) <= MAX_PENDING_SCANNER_HEAL_AGE_SECS);
+        let expired = before_expiry.saturating_sub(self.new_cache.info.pending_heals.len());
+        if expired > 0 {
+            counter!(
+                METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL,
+                "bucket" => self.new_cache.info.name.clone()
+            )
+            .increment(u64::try_from(expired).unwrap_or(u64::MAX));
+            warn!(
+                target: "rustfs::scanner::folder",
+                event = EVENT_SCANNER_HEAL_ADMISSION,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_HEAL,
+                bucket = %self.new_cache.info.name,
+                pruned = expired,
+                remaining = self.new_cache.info.pending_heals.len(),
+                state = "pending_heal_expired",
+                "Scanner pending heal ledger expired old entries"
+            );
+            self.sync_pending_heals();
+        }
+
+        self.prune_pending_scanner_heal_capacity();
+    }
+
+    fn prune_pending_scanner_heal_capacity(&mut self) -> usize {
         let len = self.new_cache.info.pending_heals.len();
         if len <= MAX_PENDING_SCANNER_HEALS_PER_BUCKET {
-            return;
+            return 0;
         }
 
         sort_pending_scanner_heals_for_retry(&mut self.new_cache.info.pending_heals);
         let remove_count = len.saturating_sub(MAX_PENDING_SCANNER_HEALS_PER_BUCKET);
         self.new_cache.info.pending_heals.drain(..remove_count);
+        self.sync_pending_heals();
         counter!(
             METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL,
             "bucket" => self.new_cache.info.name.clone()
@@ -129,6 +162,7 @@ impl FolderScanner {
             state = "pending_heal_pruned",
             "Scanner pending heal ledger pruned oldest entries"
         );
+        remove_count
     }
 
     pub(super) fn update_pending_scanner_heal_after_admission(
@@ -174,6 +208,7 @@ impl FolderScanner {
         if !repaired.is_empty() {
             self.clear_pending_scanner_heals_for_repaired(&repaired);
         }
+        self.prune_pending_scanner_heals();
         for pending in pending_scanner_heal_retry_candidates(&self.new_cache.info.pending_heals, &bucket) {
             if !self.should_heal().await {
                 break;

crates/scanner/src/scanner_folder/tests.rs

@@ -1018,7 +1018,7 @@ fn pending_heal(
         object: object.map(ToOwned::to_owned),
         version_id: version_id.map(ToOwned::to_owned),
         scan_mode: HealScanMode::Deep,
-        first_seen: 1,
+        first_seen: FolderScanner::now_secs(),
         last_attempt,
         attempts,
         last_admission_result: "full".to_string(),
@@ -1155,6 +1155,52 @@ fn test_pending_heal_retry_candidates_respect_cap_and_order() {
     assert_eq!(candidates.last().and_then(|entry| entry.object.as_deref()), Some("object-127"));
 }
 
+#[tokio::test]
+async fn test_pending_heal_prune_expires_stale_entries() {
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir);
+    scanner.new_cache.info.name = "bucket".to_string();
+    scanner.update_cache.info.name = "bucket".to_string();
+
+    let mut stale = pending_heal(PendingScannerHealKind::Object, "bucket", Some("stale"), None, 1, 1);
+    stale.first_seen = FolderScanner::now_secs().saturating_sub(MAX_PENDING_SCANNER_HEAL_AGE_SECS + 1);
+    let fresh = pending_heal(PendingScannerHealKind::Object, "bucket", Some("fresh"), None, 1, 1);
+    scanner.new_cache.info.pending_heals = vec![stale, fresh];
+
+    scanner.prune_pending_scanner_heals();
+
+    assert_eq!(scanner.new_cache.info.pending_heals.len(), 1);
+    assert_eq!(scanner.new_cache.info.pending_heals[0].object.as_deref(), Some("fresh"));
+    assert_eq!(scanner.update_cache.info.pending_heals, scanner.new_cache.info.pending_heals);
+    assert!(scanner.pending_heals_changed);
+}
+
+#[tokio::test]
+async fn test_pending_heal_update_keeps_stale_entry_until_retry_prune() {
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir);
+    scanner.new_cache.info.name = "bucket".to_string();
+    scanner.update_cache.info.name = "bucket".to_string();
+
+    let mut stale = pending_heal(PendingScannerHealKind::Object, "bucket", Some("object"), None, 1, 1);
+    stale.first_seen = FolderScanner::now_secs().saturating_sub(MAX_PENDING_SCANNER_HEAL_AGE_SECS + 1);
+    scanner.new_cache.info.pending_heals = vec![stale];
+
+    scanner.update_pending_scanner_heal_after_admission(
+        PendingScannerHealKind::Object,
+        "bucket",
+        Some("object"),
+        None,
+        HealScanMode::Deep,
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull),
+    );
+
+    assert_eq!(scanner.new_cache.info.pending_heals.len(), 1);
+    assert_eq!(scanner.new_cache.info.pending_heals[0].attempts, 2);
+    assert_eq!(scanner.new_cache.info.pending_heals[0].object.as_deref(), Some("object"));
+    assert_eq!(scanner.update_cache.info.pending_heals, scanner.new_cache.info.pending_heals);
+}
+
 #[tokio::test]
 async fn test_pending_heal_queue_full_deduplicates_object_entry() {
     let (mut scanner, temp_dir) = build_test_scanner().await;