fix(heal): track root erasure-set rebuild status (#3625)

crates/heal/src/heal/channel.rs

@@ -262,7 +262,13 @@ impl HealChannelProcessor {
             "Heal query received"
         );
 
-        let (summary, detail, items) = match self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await {
+        let report = if heal_path.trim_matches('/').is_empty() {
+            self.heal_manager.get_task_report(&client_token).await
+        } else {
+            self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await
+        };
+
+        let (summary, detail, items) = match report {
             Ok(HealTaskReport {
                 status: HealTaskStatus::Pending | HealTaskStatus::Running,
                 result_items,
@@ -291,7 +297,9 @@ impl HealChannelProcessor {
                 status: HealTaskStatus::Failed { error },
                 result_items,
             }) => ("stopped".to_string(), Some(error), result_items),
-            Err(crate::Error::TaskNotFound { .. }) => ("finished".to_string(), None, Vec::new()),
+            Err(crate::Error::TaskNotFound { .. }) => {
+                ("notFound".to_string(), Some("heal task not found or expired".to_string()), Vec::new())
+            }
             Err(crate::Error::InvalidClientToken) => {
                 let response = HealChannelResponse {
                     request_id: client_token,
@@ -908,7 +916,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_process_query_request_reports_finished_when_task_is_not_active() {
+    async fn test_process_query_request_reports_not_found_when_task_is_unknown() {
         let heal_manager = create_test_heal_manager();
         let processor = HealChannelProcessor::new(heal_manager);
         let (tx, rx) = oneshot::channel();
@@ -924,10 +932,11 @@ mod tests {
             .expect("query response should be returned");
         assert!(response.success);
         assert_eq!(response.request_id, "completed-token");
+        assert_eq!(response.error.as_deref(), Some("heal task not found or expired"));
         let payload: serde_json::Value =
             serde_json::from_slice(response.data.as_deref().expect("status payload should be present"))
                 .expect("status payload should be json");
-        assert_eq!(payload["summary"], "finished");
+        assert_eq!(payload["summary"], "notFound");
         assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
     }
 
@@ -1018,7 +1027,46 @@ mod tests {
         let payload: serde_json::Value =
             serde_json::from_slice(response.data.as_deref().expect("status payload should be present"))
                 .expect("status payload should be json");
-        assert_eq!(payload["summary"], "finished");
+        assert_eq!(payload["summary"], "notFound");
+        assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
+    }
+
+    #[tokio::test]
+    async fn test_process_query_request_empty_path_uses_client_token_directly() {
+        let heal_manager = create_test_heal_manager();
+        let request = HealRequest::new(
+            HealType::ErasureSet {
+                buckets: vec![],
+                set_disk_id: "pool_0_set_1".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::High,
+        );
+        let task_id = request.id.clone();
+        heal_manager
+            .submit_heal_request(request)
+            .await
+            .expect("request should be accepted");
+
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+
+        processor
+            .process_query_request(String::new(), task_id.clone(), tx)
+            .await
+            .expect("query should process");
+
+        let response = rx
+            .await
+            .expect("oneshot should resolve")
+            .expect("query response should be returned");
+        assert!(response.success);
+        assert_eq!(response.request_id, task_id);
+        assert!(response.error.is_none());
+        let payload: serde_json::Value =
+            serde_json::from_slice(response.data.as_deref().expect("status payload should be present"))
+                .expect("status payload should be json");
+        assert_eq!(payload["summary"], "running");
         assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
     }
 

crates/heal/src/heal/manager.rs

@@ -1200,6 +1200,65 @@ impl HealManager {
         })
     }
 
+    pub async fn get_task_report(&self, task_id: &str) -> Result<HealTaskReport> {
+        let canonical_task_id = self.canonical_task_id(task_id).await;
+        {
+            let active_heals = self.active_heals.lock().await;
+            if let Some(task) = active_heals.get(&canonical_task_id) {
+                return Ok(HealTaskReport {
+                    status: task.get_status().await,
+                    result_items: task.get_result_items().await,
+                });
+            }
+        }
+
+        {
+            let retrying_heals = self.retrying_heals.lock().await;
+            if let Some(retrying) = retrying_heals.get(&canonical_task_id) {
+                return Ok(HealTaskReport {
+                    status: retrying.status(),
+                    result_items: Vec::new(),
+                });
+            }
+        }
+
+        {
+            let mut completed_heals = self.completed_heals.lock().await;
+            prune_completed_heal_statuses(&mut completed_heals);
+            if let Some(completed) = completed_heals.get(&canonical_task_id)
+                && completed_status_is_retrying(&completed.status)
+            {
+                return Ok(HealTaskReport {
+                    status: completed.status.clone(),
+                    result_items: completed.result_items.clone(),
+                });
+            }
+        }
+
+        {
+            let queue = self.heal_queue.lock().await;
+            if queue.contains_request_id(&canonical_task_id) {
+                return Ok(HealTaskReport {
+                    status: HealTaskStatus::Pending,
+                    result_items: Vec::new(),
+                });
+            }
+        }
+
+        let mut completed_heals = self.completed_heals.lock().await;
+        prune_completed_heal_statuses(&mut completed_heals);
+        if let Some(completed) = completed_heals.get(&canonical_task_id) {
+            return Ok(HealTaskReport {
+                status: completed.status.clone(),
+                result_items: completed.result_items.clone(),
+            });
+        }
+
+        Err(Error::TaskNotFound {
+            task_id: task_id.to_string(),
+        })
+    }
+
     pub async fn get_task_report_for_path(&self, heal_path: &str, task_id: &str) -> Result<HealTaskReport> {
         let canonical_task_id = self.canonical_task_id(task_id).await;
         {
@@ -3245,6 +3304,35 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn test_get_task_report_queries_queued_task_by_token_without_path() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let request = HealRequest::new(
+            HealType::ErasureSet {
+                buckets: vec![],
+                set_disk_id: "pool_0_set_1".to_string(),
+            },
+            HealOptions::default(),
+            HealPriority::High,
+        );
+        let request_id = request.id.clone();
+
+        manager
+            .submit_heal_request(request)
+            .await
+            .expect("request should be accepted");
+
+        let report = manager
+            .get_task_report(&request_id)
+            .await
+            .expect("queued task should be queryable by token");
+
+        assert_eq!(report.status, HealTaskStatus::Pending);
+        assert!(report.result_items.is_empty());
+    }
+
     #[tokio::test]
     async fn test_get_task_status_reads_recent_completed_status() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

rustfs/src/admin/handlers/heal.rs

@@ -26,6 +26,7 @@ use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_common::heal_channel::{HealChannelPriority, HealChannelRequest, HealOpts, HealRequestSource};
 use rustfs_config::MAX_HEAL_REQUEST_SIZE;
+use rustfs_heal::heal::utils::format_set_disk_id;
 use rustfs_policy::policy::action::{Action, AdminAction};
 use rustfs_scanner::scanner::{BackgroundHealInfo, read_background_heal_info};
 use rustfs_storage_api::HealOperations as _;
@@ -247,7 +248,9 @@ fn encode_heal_task_status(
 }
 
 fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest {
-    let recursive = if !hip.bucket.is_empty() && hip.obj_prefix.is_empty() {
+    let root_erasure_set_target =
+        hip.bucket.is_empty() && hip.obj_prefix.is_empty() && matches!((hip.hs.pool, hip.hs.set), (Some(_), Some(_)));
+    let recursive = if (!hip.bucket.is_empty() && hip.obj_prefix.is_empty()) || root_erasure_set_target {
         true
     } else {
         hip.hs.recursive
@@ -265,6 +268,9 @@ fn build_heal_channel_request(hip: &HealInitParams) -> HealChannelRequest {
 
     heal_request.pool_index = hip.hs.pool;
     heal_request.set_index = hip.hs.set;
+    if root_erasure_set_target && let (Some(pool), Some(set)) = (hip.hs.pool, hip.hs.set) {
+        heal_request.disk = Some(format_set_disk_id(pool, set));
+    }
     heal_request.scan_mode = Some(hip.hs.scan_mode);
     heal_request.remove_corrupted = Some(hip.hs.remove);
     heal_request.recreate_missing = Some(hip.hs.recreate);
@@ -335,14 +341,25 @@ fn encode_background_heal_status(
 
 fn validate_heal_request_mode(hip: &HealInitParams) -> S3Result<()> {
     if hip.bucket.is_empty() && hip.client_token.is_empty() && !hip.force_stop {
-        return Err(s3_error!(InvalidRequest, "starting heal without a bucket target is not supported"));
+        return match (hip.hs.pool, hip.hs.set) {
+            (Some(_), Some(_)) => Ok(()),
+            (Some(_), None) | (None, Some(_)) => {
+                Err(s3_error!(InvalidRequest, "root heal erasure-set target requires both pool and set"))
+            }
+            (None, None) => Err(s3_error!(InvalidRequest, "starting heal without a bucket target is not supported")),
+        };
     }
 
     Ok(())
 }
 
 fn should_handle_root_heal_directly(hip: &HealInitParams) -> bool {
-    hip.bucket.is_empty() && hip.obj_prefix.is_empty() && hip.client_token.is_empty() && !hip.force_stop
+    hip.bucket.is_empty()
+        && hip.obj_prefix.is_empty()
+        && hip.client_token.is_empty()
+        && !hip.force_stop
+        && hip.hs.pool.is_none()
+        && hip.hs.set.is_none()
 }
 
 fn map_root_heal_status(heal_err: Option<crate::admin::storage_compat::Error>) -> S3Result<()> {
@@ -836,6 +853,33 @@ mod tests {
         assert_eq!(request.set_index, Some(2));
     }
 
+    #[test]
+    fn test_root_heal_channel_request_with_pool_set_targets_erasure_set() {
+        let hip = HealInitParams {
+            hs: HealOpts {
+                scan_mode: HealScanMode::Deep,
+                recreate: true,
+                pool: Some(1),
+                set: Some(2),
+                ..Default::default()
+            },
+            force_start: true,
+            ..Default::default()
+        };
+
+        let request = build_heal_channel_request(&hip);
+
+        assert_eq!(request.bucket, "");
+        assert_eq!(request.disk.as_deref(), Some("pool_1_set_2"));
+        assert_eq!(request.priority, HealChannelPriority::High);
+        assert_eq!(request.source, HealRequestSource::Admin);
+        assert_eq!(request.pool_index, Some(1));
+        assert_eq!(request.set_index, Some(2));
+        assert_eq!(request.scan_mode, Some(HealScanMode::Deep));
+        assert_eq!(request.recursive, Some(true));
+        assert_eq!(request.recreate_missing, Some(true));
+    }
+
     #[test]
     fn test_bucket_heal_channel_request_defaults_to_recursive() {
         let hip = HealInitParams {
@@ -895,6 +939,37 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_validate_heal_request_mode_allows_root_erasure_set_target() {
+        validate_heal_request_mode(&HealInitParams {
+            hs: HealOpts {
+                pool: Some(1),
+                set: Some(2),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .expect("root heal with pool and set should start tracked erasure-set rebuild");
+    }
+
+    #[test]
+    fn test_validate_heal_request_mode_rejects_partial_root_erasure_set_target() {
+        let err = validate_heal_request_mode(&HealInitParams {
+            hs: HealOpts {
+                pool: Some(1),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .expect_err("root heal must provide both pool and set");
+
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+        assert!(
+            err.to_string()
+                .contains("root heal erasure-set target requires both pool and set")
+        );
+    }
+
     #[test]
     fn test_should_handle_root_heal_directly_for_root_start_modes() {
         assert!(should_handle_root_heal_directly(&HealInitParams::default()));
@@ -918,6 +993,14 @@ mod tests {
             bucket: "bucket".to_string(),
             ..Default::default()
         }));
+        assert!(!should_handle_root_heal_directly(&HealInitParams {
+            hs: HealOpts {
+                pool: Some(1),
+                set: Some(2),
+                ..Default::default()
+            },
+            ..Default::default()
+        }));
     }
 
     #[test]