From a247c793596c4db6b0e0a46e37fc2e2896777341 Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 20 Aug 2026 20:28:22 +0800 Subject: [PATCH] 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 --- crates/heal/src/heal/manager.rs | 38 +++++++++++++--- crates/heal/src/heal/manager/queue.rs | 33 +++++++------- crates/heal/src/heal/manager/scheduler.rs | 13 ++++++ crates/heal/src/heal/manager/tests.rs | 41 ++++++++++++++++++ crates/scanner/src/scanner_folder.rs | 1 + crates/scanner/src/scanner_folder/ledger.rs | 41 ++++++++++++++++-- crates/scanner/src/scanner_folder/tests.rs | 48 ++++++++++++++++++++- 7 files changed, 187 insertions(+), 28 deletions(-) diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 7c5c535b3..a45407f05 100644 --- a/crates/heal/src/heal/manager.rs +++ b/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>, active_heals: &'a Arc>>>, completed_heals: &'a Arc>>>, + task_aliases: &'a Arc>>, retrying_heals: &'a Arc>>, replacement_recovery_anchors: &'a Arc>>, config: &'a Arc>, @@ -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", diff --git a/crates/heal/src/heal/manager/queue.rs b/crates/heal/src/heal/manager/queue.rs index 9d33c660d..ed7887af8 100644 --- a/crates/heal/src/heal/manager/queue.rs +++ b/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 { 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 { 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 { 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); diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index a22bee4ad..00012e2ad 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/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)) = diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 0587371fc..8b955b172 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/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 = 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 = Arc::new(MockStorage); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index f2be32329..693dd6ebc 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/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(); diff --git a/crates/scanner/src/scanner_folder/ledger.rs b/crates/scanner/src/scanner_folder/ledger.rs index 40a7feacb..7ac03983c 100644 --- a/crates/scanner/src/scanner_folder/ledger.rs +++ b/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; diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index 01c9f03a6..dc4460b15 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/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;