feat(heal): expose scanner-aware operations status (#3483)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

crates/heal/src/heal/channel.rs

@@ -432,6 +432,7 @@ impl HealChannelProcessor {
 
         let mut heal_request = HealRequest::new(heal_type, options, priority);
         heal_request.id = request.id;
+        heal_request.source = request.source;
         // 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
@@ -480,7 +481,9 @@ mod tests {
     use super::*;
     use crate::heal::manager::HealConfig;
     use crate::heal::storage::HealStorageAPI;
-    use rustfs_common::heal_channel::{HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealScanMode};
+    use rustfs_common::heal_channel::{
+        HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealRequestSource, HealScanMode,
+    };
     use std::sync::Arc;
 
     // Mock storage for testing
@@ -611,6 +614,7 @@ mod tests {
             pool_index: None,
             set_index: None,
             force_start: false,
+            source: HealRequestSource::Internal,
         };
 
         let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
@@ -641,11 +645,13 @@ mod tests {
             pool_index: Some(0),
             set_index: Some(1),
             force_start: false,
+            source: HealRequestSource::Scanner,
         };
 
         let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
         assert!(matches!(heal_request.heal_type, HealType::Object { .. }));
         assert_eq!(heal_request.priority, HealPriority::High);
+        assert_eq!(heal_request.source, HealRequestSource::Scanner);
         assert_eq!(heal_request.options.scan_mode, HealScanMode::Deep);
         assert!(heal_request.options.remove_corrupted);
         assert!(heal_request.options.recreate_missing);
@@ -673,6 +679,7 @@ mod tests {
             pool_index: None,
             set_index: None,
             force_start: false,
+            source: HealRequestSource::Internal,
         };
 
         let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
@@ -702,6 +709,7 @@ mod tests {
             pool_index: None,
             set_index: None,
             force_start: false,
+            source: HealRequestSource::Internal,
         };
 
         let result = processor.convert_to_heal_request(channel_request);
@@ -738,6 +746,7 @@ mod tests {
                 pool_index: None,
                 set_index: None,
                 force_start: false,
+                source: HealRequestSource::Internal,
             };
 
             let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
@@ -767,6 +776,7 @@ mod tests {
             pool_index: None,
             set_index: None,
             force_start: true, // Admission force only; must not override explicit heal options.
+            source: HealRequestSource::Internal,
         };
 
         let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
@@ -798,6 +808,7 @@ mod tests {
             pool_index: None,
             set_index: None,
             force_start: false,
+            source: HealRequestSource::Internal,
         };
 
         let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
@@ -833,6 +844,7 @@ mod tests {
             pool_index: None,
             set_index: None,
             force_start: false,
+            source: HealRequestSource::Internal,
         };
 
         let (tx, rx) = oneshot::channel();
@@ -882,6 +894,7 @@ mod tests {
             pool_index: None,
             set_index: None,
             force_start: false,
+            source: HealRequestSource::Internal,
         };
 
         let (tx, rx) = oneshot::channel();

crates/heal/src/heal/manager.rs

@@ -19,7 +19,7 @@ use crate::heal::{
 };
 use crate::{Error, Result};
 use metrics::{counter, gauge};
-use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
+use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult, HealRequestSource};
 use rustfs_ecstore::disk::DiskAPI;
 use rustfs_ecstore::disk::error::DiskError;
 use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP;
@@ -115,6 +115,61 @@ pub struct HealTaskReport {
     pub result_items: Vec<HealResultItem>,
 }
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct HealPriorityCounts {
+    pub low: u64,
+    pub normal: u64,
+    pub high: u64,
+    pub urgent: u64,
+}
+
+impl HealPriorityCounts {
+    fn increment(&mut self, priority: HealPriority) {
+        match priority {
+            HealPriority::Low => self.low += 1,
+            HealPriority::Normal => self.normal += 1,
+            HealPriority::High => self.high += 1,
+            HealPriority::Urgent => self.urgent += 1,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct HealSourceCounts {
+    pub scanner: u64,
+    pub admin: u64,
+    pub auto_heal: u64,
+    pub internal: u64,
+}
+
+impl HealSourceCounts {
+    fn increment(&mut self, source: HealRequestSource) {
+        match source {
+            HealRequestSource::Scanner => self.scanner += 1,
+            HealRequestSource::Admin => self.admin += 1,
+            HealRequestSource::AutoHeal => self.auto_heal += 1,
+            HealRequestSource::Internal => self.internal += 1,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct HealOperationsSnapshot {
+    pub queue_length: u64,
+    pub active_tasks: u64,
+    pub queued_by_priority: HealPriorityCounts,
+    pub active_by_priority: HealPriorityCounts,
+    pub queued_by_source: HealSourceCounts,
+    pub active_by_source: HealSourceCounts,
+}
+
+fn usize_to_u64_saturated(value: usize) -> u64 {
+    u64::try_from(value).unwrap_or(u64::MAX)
+}
+
 impl PriorityHealQueue {
     fn new() -> Self {
         Self {
@@ -212,6 +267,16 @@ impl PriorityHealQueue {
         stats
     }
 
+    fn operation_counts(&self) -> (HealPriorityCounts, HealSourceCounts) {
+        let mut priority = HealPriorityCounts::default();
+        let mut source = HealSourceCounts::default();
+        for item in &self.heap {
+            priority.increment(item.request.priority);
+            source.increment(item.request.source);
+        }
+        (priority, source)
+    }
+
     #[cfg(test)]
     fn pop(&mut self) -> Option<HealRequest> {
         self.heap.pop().map(|item| {
@@ -1095,6 +1160,36 @@ impl HealManager {
         queue.len()
     }
 
+    pub async fn operations_snapshot(&self) -> HealOperationsSnapshot {
+        let (queue_length, queued_by_priority, queued_by_source) = {
+            let queue = self.heal_queue.lock().await;
+            let (priority, source) = queue.operation_counts();
+            publish_heal_queue_length(&queue);
+            (usize_to_u64_saturated(queue.len()), priority, source)
+        };
+
+        let (active_tasks, active_by_priority, active_by_source) = {
+            let active_heals = self.active_heals.lock().await;
+            let mut priority = HealPriorityCounts::default();
+            let mut source = HealSourceCounts::default();
+            for task in active_heals.values() {
+                priority.increment(task.priority);
+                source.increment(task.source);
+            }
+            publish_active_heal_count(&active_heals);
+            (usize_to_u64_saturated(active_heals.len()), priority, source)
+        };
+
+        HealOperationsSnapshot {
+            queue_length,
+            active_tasks,
+            queued_by_priority,
+            active_by_priority,
+            queued_by_source,
+            active_by_source,
+        }
+    }
+
     /// Start scheduler
     async fn start_scheduler(&self) -> Result<()> {
         let config = self.config.clone();
@@ -1304,7 +1399,7 @@ impl HealManager {
                             }
 
                             // enqueue erasure set heal request for this disk
-                            let req = HealRequest::new(
+                            let mut req = HealRequest::new(
                                 HealType::ErasureSet {
                                     buckets: buckets.clone(),
                                     set_disk_id: set_disk_id.clone(),
@@ -1312,6 +1407,7 @@ impl HealManager {
                                 HealOptions::default(),
                                 HealPriority::Normal,
                             );
+                            req.source = HealRequestSource::AutoHeal;
                             let mut queue = heal_queue.lock().await;
                             if matches!(queue.push(req), QueuePushOutcome::Accepted) {
                                 publish_heal_queue_length(&queue);
@@ -1623,8 +1719,8 @@ fn can_schedule_request(request: &HealRequest, running_per_set: &HashMap<String,
 mod tests {
     use super::*;
     use crate::heal::storage::HealStorageAPI;
-    use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealType};
-    use rustfs_common::heal_channel::HealOpts;
+    use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};
+    use rustfs_common::heal_channel::{HealOpts, HealRequestSource};
     use rustfs_ecstore::disk::{DiskStore, endpoint::Endpoint};
     use rustfs_madmin::heal_commands::HealResultItem;
     use rustfs_storage_api::BucketInfo;
@@ -2172,6 +2268,85 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_operations_snapshot_counts_queue_by_source_and_priority() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let mut scanner_request = HealRequest::new(
+            HealType::Object {
+                bucket: "bucket-a".to_string(),
+                object: "object-a".to_string(),
+                version_id: None,
+            },
+            HealOptions::default(),
+            HealPriority::Low,
+        );
+        scanner_request.source = HealRequestSource::Scanner;
+
+        let mut admin_request = HealRequest::bucket("bucket-b".to_string());
+        admin_request.priority = HealPriority::High;
+        admin_request.source = HealRequestSource::Admin;
+
+        let mut auto_request = HealRequest::new(
+            HealType::ErasureSet {
+                buckets: vec!["bucket-c".to_string()],
+                set_disk_id: "0-0".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::Normal,
+        );
+        auto_request.source = HealRequestSource::AutoHeal;
+
+        manager
+            .submit_heal_request(scanner_request)
+            .await
+            .expect("scanner request should be accepted");
+        manager
+            .submit_heal_request(admin_request)
+            .await
+            .expect("admin request should be accepted");
+        manager
+            .submit_heal_request(auto_request)
+            .await
+            .expect("auto request should be accepted");
+
+        let snapshot = manager.operations_snapshot().await;
+
+        assert_eq!(snapshot.queue_length, 3);
+        assert_eq!(snapshot.active_tasks, 0);
+        assert_eq!(snapshot.queued_by_priority.low, 1);
+        assert_eq!(snapshot.queued_by_priority.normal, 1);
+        assert_eq!(snapshot.queued_by_priority.high, 1);
+        assert_eq!(snapshot.queued_by_priority.urgent, 0);
+        assert_eq!(snapshot.queued_by_source.scanner, 1);
+        assert_eq!(snapshot.queued_by_source.admin, 1);
+        assert_eq!(snapshot.queued_by_source.auto_heal, 1);
+        assert_eq!(snapshot.queued_by_source.internal, 0);
+    }
+
+    #[tokio::test]
+    async fn test_operations_snapshot_counts_active_by_source_and_priority() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let mut request = HealRequest::bucket("bucket-a".to_string());
+        request.priority = HealPriority::High;
+        request.source = HealRequestSource::Admin;
+        let task = Arc::new(HealTask::from_request(request, manager.storage.clone()));
+        let task_id = task.id.clone();
+
+        manager.active_heals.lock().await.insert(task_id, task);
+
+        let snapshot = manager.operations_snapshot().await;
+
+        assert_eq!(snapshot.queue_length, 0);
+        assert_eq!(snapshot.active_tasks, 1);
+        assert_eq!(snapshot.active_by_priority.high, 1);
+        assert_eq!(snapshot.active_by_source.admin, 1);
+        assert_eq!(snapshot.active_by_source.scanner, 0);
+    }
+
     #[tokio::test]
     async fn test_get_task_status_for_path_rejects_wrong_token_when_path_is_active() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/heal/mod.rs

@@ -23,6 +23,6 @@ pub mod task;
 pub mod utils;
 
 pub use erasure_healer::ErasureSetHealer;
-pub use manager::HealManager;
+pub use manager::{HealManager, HealOperationsSnapshot, HealPriorityCounts, HealSourceCounts};
 pub use resume::{CheckpointManager, ResumeCheckpoint, ResumeManager, ResumeState, ResumeUtils};
 pub use task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};

crates/heal/src/heal/task.rs

@@ -15,7 +15,7 @@
 use crate::heal::{ErasureSetHealer, progress::HealProgress, storage::HealStorageAPI};
 use crate::{Error, Result};
 use metrics::{counter, histogram};
-use rustfs_common::heal_channel::{HealOpts, HealScanMode};
+use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode};
 use rustfs_ecstore::{
     data_usage::DATA_USAGE_CACHE_NAME,
     disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
@@ -169,6 +169,8 @@ pub struct HealRequest {
     pub options: HealOptions,
     /// Priority
     pub priority: HealPriority,
+    /// Origin of the request for operational status.
+    pub source: HealRequestSource,
     /// Whether this request should bypass queue admission dedup/full policies.
     pub force_start: bool,
     /// Created time
@@ -185,6 +187,7 @@ impl HealRequest {
             heal_type,
             options,
             priority,
+            source: HealRequestSource::Internal,
             force_start: false,
             created_at: now,
             enqueued_at: now,
@@ -232,6 +235,10 @@ pub struct HealTask {
     pub heal_type: HealType,
     /// Heal options
     pub options: HealOptions,
+    /// Priority inherited from the request
+    pub priority: HealPriority,
+    /// Origin inherited from the request
+    pub source: HealRequestSource,
     /// Task status
     pub status: Arc<RwLock<HealTaskStatus>>,
     /// Progress tracking
@@ -260,6 +267,8 @@ impl HealTask {
             id: request.id,
             heal_type: request.heal_type,
             options: request.options,
+            priority: request.priority,
+            source: request.source,
             status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
             progress: Arc::new(RwLock::new(HealProgress::new())),
             result_items: Arc::new(RwLock::new(Vec::new())),

crates/heal/src/lib.rs

@@ -16,7 +16,10 @@ mod error;
 pub mod heal;
 
 pub use error::{Error, Result};
-pub use heal::{HealManager, HealOptions, HealPriority, HealRequest, HealType, channel::HealChannelProcessor};
+pub use heal::{
+    HealManager, HealOperationsSnapshot, HealOptions, HealPriority, HealPriorityCounts, HealRequest, HealSourceCounts, HealType,
+    channel::HealChannelProcessor,
+};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, OnceLock};
 use tokio_util::sync::CancellationToken;
@@ -138,10 +141,26 @@ pub fn current_heal_queue_length() -> u64 {
     GLOBAL_HEAL_QUEUE_LENGTH.load(Ordering::Relaxed)
 }
 
+pub async fn current_heal_operations_snapshot() -> HealOperationsSnapshot {
+    if let Some(manager) = get_heal_manager() {
+        manager.operations_snapshot().await
+    } else {
+        HealOperationsSnapshot {
+            queue_length: current_heal_queue_length(),
+            active_tasks: current_heal_active_tasks(),
+            ..Default::default()
+        }
+    }
+}
+
+fn usize_to_u64_saturated(value: usize) -> u64 {
+    u64::try_from(value).unwrap_or(u64::MAX)
+}
+
 pub(crate) fn set_heal_active_tasks(count: usize) {
-    GLOBAL_HEAL_ACTIVE_TASKS.store(count as u64, Ordering::Relaxed);
+    GLOBAL_HEAL_ACTIVE_TASKS.store(usize_to_u64_saturated(count), Ordering::Relaxed);
 }
 
 pub(crate) fn set_heal_queue_length(count: usize) {
-    GLOBAL_HEAL_QUEUE_LENGTH.store(count as u64, Ordering::Relaxed);
+    GLOBAL_HEAL_QUEUE_LENGTH.store(usize_to_u64_saturated(count), Ordering::Relaxed);
 }