From 0b69f363d67b53fcc7bb4f6c0d45590e6b8deb26 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 3 Jun 2026 22:21:20 +0800 Subject: [PATCH] fix: prioritize manual heal queue admission (#3192) Co-authored-by: Henry Guo Co-authored-by: houseme --- crates/heal/src/heal/manager.rs | 128 ++++++++++++++++++++++++++++++ rustfs/src/admin/handlers/heal.rs | 24 +++++- 2 files changed, 149 insertions(+), 3 deletions(-) diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 439536ecf..8785184a6 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -149,6 +149,50 @@ impl PriorityHealQueue { QueuePushOutcome::Accepted } + fn can_displace_lower_priority(&self, priority: HealPriority) -> bool { + self.heap.iter().any(|item| item.priority < priority) + } + + fn push_displacing_lower_priority(&mut self, request: HealRequest) -> Option { + let mut retained = BinaryHeap::new(); + let mut displaced: Option = None; + + while let Some(item) = self.heap.pop() { + if item.priority < request.priority { + let should_displace = displaced + .as_ref() + .map(|current| { + item.priority < current.priority + || (item.priority == current.priority && item.sequence > current.sequence) + }) + .unwrap_or(true); + if should_displace { + if let Some(current) = displaced.replace(item) { + retained.push(current); + } + } else { + retained.push(item); + } + } else { + retained.push(item); + } + } + + 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); + item.request + }); + + if displaced.is_some() { + debug_assert_eq!(self.push(request), QueuePushOutcome::Accepted); + } + + displaced + } + /// Get statistics about queue contents by priority fn get_priority_stats(&self) -> HashMap { let mut stats = HashMap::new(); @@ -578,6 +622,37 @@ impl HealManager { } if queue_len >= queue_capacity && !request.force_start { + if queue.can_displace_lower_priority(request.priority) { + let request_id = request.id.clone(); + let priority = request.priority; + if let Some(displaced) = queue.push_displacing_lower_priority(request) { + publish_heal_queue_length(&queue); + warn!( + request_id = %request_id, + priority = ?priority, + displaced_request_id = %displaced.id, + displaced_priority = ?displaced.priority, + queue_len, + queue_capacity, + "Admitted heal request by displacing lower-priority queued work" + ); + drop(queue); + if config.event_driven_scheduler_enable { + self.notify.notify_one(); + } + return Ok(HealAdmissionResult::Accepted); + } + + warn!( + request_id = %request_id, + priority = ?priority, + queue_len, + queue_capacity, + "Heal queue was full and lower-priority work was unavailable for displacement" + ); + return Ok(HealAdmissionResult::Full); + } + let admission = Self::classify_full_admission(&request, &config); match admission { HealAdmissionResult::Dropped(reason) => { @@ -2135,6 +2210,59 @@ mod tests { ); } + #[tokio::test] + async fn test_high_priority_request_displaces_lower_priority_when_queue_full() { + let storage: Arc = Arc::new(MockStorage); + let manager = HealManager::new( + storage, + Some(HealConfig { + queue_size: 1, + ..HealConfig::default() + }), + ); + + let low = HealRequest::new( + HealType::Bucket { + bucket: "background-bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + let low_id = low.id.clone(); + let high = HealRequest::new( + HealType::Bucket { + bucket: "manual-bucket".to_string(), + }, + HealOptions::default(), + HealPriority::High, + ); + let high_id = high.id.clone(); + + assert_eq!( + manager + .submit_heal_request(low) + .await + .expect("low priority request should be accepted first"), + HealAdmissionResult::Accepted + ); + assert_eq!( + manager + .submit_heal_request(high) + .await + .expect("high priority request should be admitted by displacing lower priority work"), + HealAdmissionResult::Accepted + ); + assert_eq!(manager.get_queue_length().await, 1); + assert!(matches!(manager.get_task_status(&low_id).await, Err(Error::TaskNotFound { .. }))); + assert_eq!( + manager + .get_task_status(&high_id) + .await + .expect("high priority request should remain queued"), + HealTaskStatus::Pending + ); + } + #[tokio::test] async fn test_force_start_bypasses_duplicate_and_full_admission() { let storage: Arc = Arc::new(MockStorage); diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 8108fc365..ac1521202 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -164,6 +164,15 @@ struct HealTaskStatus { items: Vec, } +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct BackgroundHealStatus<'a> { + #[serde(flatten)] + info: &'a BackgroundHealInfo, + heal_queue_length: u64, + heal_active_tasks: u64, +} + #[derive(Debug, Deserialize)] struct HealTaskStatusPayload { summary: String, @@ -230,7 +239,7 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest { Some(hip.obj_prefix.clone()) }, hip.force_start, - Some(HealChannelPriority::Normal), + Some(HealChannelPriority::High), ); heal_request.pool_index = hip.hs.pool; @@ -278,7 +287,12 @@ fn heal_channel_response_items( } fn encode_background_heal_status(info: &BackgroundHealInfo) -> S3Result> { - serde_json::to_vec(info).map_err(|e| s3_error!(InternalError, "failed to serialize background heal status: {e}")) + let status = BackgroundHealStatus { + info, + heal_queue_length: rustfs_heal::current_heal_queue_length(), + heal_active_tasks: rustfs_heal::current_heal_active_tasks(), + }; + serde_json::to_vec(&status).map_err(|e| s3_error!(InternalError, "failed to serialize background heal status: {e}")) } fn validate_heal_request_mode(hip: &HealInitParams) -> S3Result<()> { @@ -569,7 +583,7 @@ mod tests { use http::StatusCode; use http::Uri; use matchit::Router; - use rustfs_common::heal_channel::{HealOpts, HealScanMode}; + use rustfs_common::heal_channel::{HealChannelPriority, HealOpts, HealScanMode}; use rustfs_ecstore::error::StorageError; use rustfs_scanner::scanner::BackgroundHealInfo; use s3s::{ @@ -686,6 +700,7 @@ mod tests { assert_eq!(request.bucket, "bucket"); assert_eq!(request.object_prefix.as_deref(), Some("prefix")); + assert_eq!(request.priority, HealChannelPriority::High); assert!(request.force_start); assert_eq!(request.scan_mode, Some(HealScanMode::Deep)); assert_eq!(request.recursive, Some(true)); @@ -858,6 +873,8 @@ mod tests { assert_eq!(json["bitrotStartCycle"], 42); assert_eq!(json["currentScanMode"], 2); assert!(json["bitrotStartTime"].is_null()); + assert!(json["healQueueLength"].is_u64()); + assert!(json["healActiveTasks"].is_u64()); } #[test] @@ -911,6 +928,7 @@ mod tests { assert_eq!(request.bucket, "bucket-a"); assert_eq!(request.object_prefix.as_deref(), Some("prefix-a")); + assert_eq!(request.priority, HealChannelPriority::High); assert!(request.force_start); assert_eq!(request.pool_index, Some(1)); assert_eq!(request.set_index, Some(2));