fix(scanner): preserve background heal compatibility (#3041)

crates/heal/src/error.rs

@@ -59,6 +59,9 @@ pub enum Error {
     #[error("Heal task already exists: {task_id}")]
     TaskAlreadyExists { task_id: String },
 
+    #[error("Invalid heal client token")]
+    InvalidClientToken,
+
     #[error("Heal manager is not running")]
     ManagerNotRunning,
 

crates/heal/src/heal/channel.rs

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 use crate::heal::{
-    manager::HealManager,
-    task::{HealOptions, HealPriority, HealRequest, HealType},
+    manager::{HealManager, HealTaskReport},
+    task::{HealOptions, HealPriority, HealRequest, HealTaskStatus, HealType},
     utils,
 };
 use crate::{Error, Result};
@@ -22,6 +22,8 @@ use rustfs_common::heal_channel::{
     HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealChannelResponse,
     HealScanMode, publish_heal_response,
 };
+use rustfs_madmin::heal_commands::HealResultItem;
+use serde::Serialize;
 use std::sync::Arc;
 use tokio::sync::{mpsc, oneshot};
 use tracing::{debug, error, info};
@@ -36,6 +38,12 @@ pub struct HealChannelProcessor {
     response_receiver: mpsc::UnboundedReceiver<HealChannelResponse>,
 }
 
+#[derive(Serialize)]
+struct HealTaskStatusPayload {
+    summary: String,
+    items: Vec<HealResultItem>,
+}
+
 impl HealChannelProcessor {
     /// Create new HealChannelProcessor
     pub fn new(heal_manager: Arc<HealManager>) -> Self {
@@ -83,8 +91,16 @@ impl HealChannelProcessor {
     async fn process_command(&self, command: HealChannelCommand) -> Result<()> {
         match command {
             HealChannelCommand::Start { request, response_tx } => self.process_start_request(request, response_tx).await,
-            HealChannelCommand::Query { heal_path, client_token } => self.process_query_request(heal_path, client_token).await,
-            HealChannelCommand::Cancel { heal_path } => self.process_cancel_request(heal_path).await,
+            HealChannelCommand::Query {
+                heal_path,
+                client_token,
+                response_tx,
+            } => self.process_query_request(heal_path, client_token, response_tx).await,
+            HealChannelCommand::Cancel {
+                heal_path,
+                client_token,
+                response_tx,
+            } => self.process_cancel_request(heal_path, client_token, response_tx).await,
         }
     }
 
@@ -166,36 +182,114 @@ impl HealChannelProcessor {
     }
 
     /// Process query request
-    async fn process_query_request(&self, heal_path: String, client_token: String) -> Result<()> {
+    async fn process_query_request(
+        &self,
+        heal_path: String,
+        client_token: String,
+        response_tx: oneshot::Sender<std::result::Result<HealChannelResponse, String>>,
+    ) -> Result<()> {
         info!("Processing heal query request for path: {}", heal_path);
 
-        // TODO: Implement query logic based on heal_path and client_token
-        // For now, return a placeholder response
+        let (summary, detail, items) = match self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await {
+            Ok(HealTaskReport {
+                status: HealTaskStatus::Pending | HealTaskStatus::Running,
+                result_items,
+            }) => ("running".to_string(), None, result_items),
+            Ok(HealTaskReport {
+                status: HealTaskStatus::Completed,
+                result_items,
+            }) => ("finished".to_string(), None, result_items),
+            Ok(HealTaskReport {
+                status: HealTaskStatus::Cancelled,
+                result_items,
+            }) => ("stopped".to_string(), Some("heal task cancelled".to_string()), result_items),
+            Ok(HealTaskReport {
+                status: HealTaskStatus::Timeout,
+                result_items,
+            }) => ("stopped".to_string(), Some("heal task timed out".to_string()), result_items),
+            Ok(HealTaskReport {
+                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::InvalidClientToken) => {
+                let response = HealChannelResponse {
+                    request_id: client_token,
+                    success: false,
+                    data: None,
+                    error: Some("invalid heal client token".to_string()),
+                };
+                let _ = response_tx.send(Ok(response.clone()));
+                self.publish_response(response);
+                return Ok(());
+            }
+            Err(err) => {
+                let error_text = err.to_string();
+                let response = HealChannelResponse {
+                    request_id: client_token,
+                    success: false,
+                    data: None,
+                    error: Some(error_text.clone()),
+                };
+                let _ = response_tx.send(Ok(response.clone()));
+                self.publish_response(response);
+                return Ok(());
+            }
+        };
+
+        let data = serde_json::to_vec(&HealTaskStatusPayload { summary, items })
+            .map_err(|e| crate::Error::Serialization(format!("failed to serialize heal task status: {e}")))?;
+
         let response = HealChannelResponse {
             request_id: client_token,
             success: true,
-            data: Some(format!("Query result for path: {heal_path}").into_bytes()),
-            error: None,
+            data: Some(data),
+            error: detail,
         };
 
+        let _ = response_tx.send(Ok(response.clone()));
         self.publish_response(response);
 
         Ok(())
     }
 
     /// Process cancel request
-    async fn process_cancel_request(&self, heal_path: String) -> Result<()> {
+    async fn process_cancel_request(
+        &self,
+        heal_path: String,
+        client_token: String,
+        response_tx: oneshot::Sender<std::result::Result<HealChannelResponse, String>>,
+    ) -> Result<()> {
         info!("Processing heal cancel request for path: {}", heal_path);
 
-        // TODO: Implement cancel logic based on heal_path
-        // For now, return a placeholder response
-        let response = HealChannelResponse {
-            request_id: heal_path.clone(),
-            success: true,
-            data: Some(format!("Cancel request for path: {heal_path}").into_bytes()),
-            error: None,
+        let request_id = if client_token.is_empty() {
+            heal_path.clone()
+        } else {
+            client_token.clone()
         };
 
+        let cancel_result = if client_token.is_empty() {
+            self.heal_manager.cancel_tasks_for_path(&heal_path).await.map(|_| ())
+        } else {
+            self.heal_manager.cancel_task(&client_token).await
+        };
+
+        let response = match cancel_result {
+            Ok(()) => HealChannelResponse {
+                request_id,
+                success: true,
+                data: Some("stopped".as_bytes().to_vec()),
+                error: None,
+            },
+            Err(err) => HealChannelResponse {
+                request_id,
+                success: false,
+                data: None,
+                error: Some(err.to_string()),
+            },
+        };
+
+        let _ = response_tx.send(Ok(response.clone()));
         self.publish_response(response);
 
         Ok(())
@@ -256,7 +350,9 @@ impl HealChannelProcessor {
             options.update_parity = true;
         }
 
-        Ok(HealRequest::new(heal_type, options, priority))
+        let mut heal_request = HealRequest::new(heal_type, options, priority);
+        heal_request.id = request.id;
+        Ok(heal_request)
     }
 
     fn publish_response(&self, response: HealChannelResponse) {
@@ -416,6 +512,7 @@ mod tests {
         };
 
         let heal_request = processor.convert_to_heal_request(channel_request).unwrap();
+        assert_eq!(heal_request.id, "test-id");
         assert!(matches!(heal_request.heal_type, HealType::Bucket { .. }));
         assert_eq!(heal_request.priority, HealPriority::Normal);
     }
@@ -691,4 +788,201 @@ mod tests {
             .expect("processor should surface invalid request through response channel");
         assert!(rx.await.expect("oneshot should resolve").is_err());
     }
+
+    #[tokio::test]
+    async fn test_process_query_request_reports_finished_when_task_is_not_active() {
+        let heal_manager = create_test_heal_manager();
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+
+        processor
+            .process_query_request("bucket".to_string(), "completed-token".to_string(), 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, "completed-token");
+        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["items"].as_array().expect("items should be an array").len(), 0);
+    }
+
+    #[tokio::test]
+    async fn test_process_query_request_reports_running_for_queued_task() {
+        let heal_manager = create_test_heal_manager();
+        let request = HealRequest::bucket("bucket".to_string());
+        let task_id = request.id.clone();
+        assert_eq!(
+            heal_manager
+                .submit_heal_request(request)
+                .await
+                .expect("request should be accepted"),
+            HealAdmissionResult::Accepted
+        );
+
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+
+        processor
+            .process_query_request("bucket".to_string(), 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);
+        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);
+    }
+
+    #[tokio::test]
+    async fn test_process_query_request_rejects_wrong_token_for_active_path() {
+        let heal_manager = create_test_heal_manager();
+        let request = HealRequest::bucket("bucket".to_string());
+        assert_eq!(
+            heal_manager
+                .submit_heal_request(request)
+                .await
+                .expect("request should be accepted"),
+            HealAdmissionResult::Accepted
+        );
+
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+
+        processor
+            .process_query_request("bucket".to_string(), "wrong-token".to_string(), 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, "wrong-token");
+        assert_eq!(response.error.as_deref(), Some("invalid heal client token"));
+    }
+
+    #[tokio::test]
+    async fn test_process_query_request_empty_path_ignores_unrelated_tasks() {
+        let heal_manager = create_test_heal_manager();
+        heal_manager
+            .submit_heal_request(HealRequest::bucket("bucket".to_string()))
+            .await
+            .expect("request should be accepted");
+
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+
+        processor
+            .process_query_request(String::new(), "wrong-token".to_string(), tx)
+            .await
+            .expect("query should process");
+
+        let response = rx
+            .await
+            .expect("oneshot should resolve")
+            .expect("query response should be returned");
+        assert!(response.success);
+        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["items"].as_array().expect("items should be an array").len(), 0);
+    }
+
+    #[tokio::test]
+    async fn test_process_cancel_request_cancels_queued_task_by_token() {
+        let heal_manager = create_test_heal_manager();
+        let request = HealRequest::bucket("bucket".to_string());
+        let task_id = request.id.clone();
+        heal_manager
+            .submit_heal_request(request)
+            .await
+            .expect("request should be accepted");
+
+        let processor = HealChannelProcessor::new(heal_manager.clone());
+        let (tx, rx) = oneshot::channel();
+
+        processor
+            .process_cancel_request("bucket".to_string(), task_id.clone(), tx)
+            .await
+            .expect("cancel should process");
+
+        let response = rx
+            .await
+            .expect("oneshot should resolve")
+            .expect("cancel response should be returned");
+        assert!(response.success);
+        assert_eq!(response.request_id, task_id);
+        assert_eq!(response.data.as_deref(), Some("stopped".as_bytes()));
+        assert!(matches!(
+            heal_manager.get_task_status(&response.request_id).await,
+            Err(crate::Error::TaskNotFound { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_process_cancel_request_cancels_queued_task_by_path() {
+        let heal_manager = create_test_heal_manager();
+        let request = HealRequest::bucket("bucket".to_string());
+        let task_id = request.id.clone();
+        heal_manager
+            .submit_heal_request(request)
+            .await
+            .expect("request should be accepted");
+
+        let processor = HealChannelProcessor::new(heal_manager.clone());
+        let (tx, rx) = oneshot::channel();
+
+        processor
+            .process_cancel_request("bucket".to_string(), String::new(), tx)
+            .await
+            .expect("cancel should process");
+
+        let response = rx
+            .await
+            .expect("oneshot should resolve")
+            .expect("cancel response should be returned");
+        assert!(response.success);
+        assert_eq!(response.request_id, "bucket");
+        assert_eq!(response.data.as_deref(), Some("stopped".as_bytes()));
+        assert!(matches!(
+            heal_manager.get_task_status(&task_id).await,
+            Err(crate::Error::TaskNotFound { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_process_cancel_request_reports_unknown_task() {
+        let heal_manager = create_test_heal_manager();
+        let processor = HealChannelProcessor::new(heal_manager);
+        let (tx, rx) = oneshot::channel();
+
+        processor
+            .process_cancel_request("missing".to_string(), "missing-token".to_string(), tx)
+            .await
+            .expect("cancel should process");
+
+        let response = rx
+            .await
+            .expect("oneshot should resolve")
+            .expect("cancel response should be returned");
+        assert!(!response.success);
+        assert_eq!(response.request_id, "missing-token");
+        assert!(response.error.unwrap_or_default().contains("Heal task not found"));
+    }
 }

crates/heal/src/heal/manager.rs

@@ -23,6 +23,7 @@ use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
 use rustfs_ecstore::disk::DiskAPI;
 use rustfs_ecstore::disk::error::DiskError;
 use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP;
+use rustfs_madmin::heal_commands::HealResultItem;
 use std::{
     collections::{BinaryHeap, HashMap, HashSet},
     sync::Arc,
@@ -35,6 +36,8 @@ use tokio::{
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
 
+const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
+
 /// Priority queue wrapper for heal requests
 /// Uses BinaryHeap for priority-based ordering while maintaining FIFO for same-priority items
 #[derive(Debug)]
@@ -88,6 +91,20 @@ enum QueuePushOutcome {
     Merged,
 }
 
+#[derive(Debug, Clone)]
+struct CompletedHealStatus {
+    heal_type: HealType,
+    status: HealTaskStatus,
+    result_items: Vec<HealResultItem>,
+    completed_at: SystemTime,
+}
+
+#[derive(Debug, Clone)]
+pub struct HealTaskReport {
+    pub status: HealTaskStatus,
+    pub result_items: Vec<HealResultItem>,
+}
+
 impl PriorityHealQueue {
     fn new() -> Self {
         Self {
@@ -235,6 +252,78 @@ impl PriorityHealQueue {
         let key = format!("erasure_set:{set_disk_id}");
         self.dedup_keys.contains(&key)
     }
+
+    fn contains_request_id(&self, request_id: &str) -> bool {
+        self.heap.iter().any(|item| item.request.id == request_id)
+    }
+
+    fn contains_request_id_matching_path(&self, request_id: &str, heal_path: &str) -> bool {
+        self.heap
+            .iter()
+            .any(|item| item.request.id == request_id && heal_type_matches_path(&item.request.heal_type, heal_path))
+    }
+
+    fn contains_matching<F>(&self, mut matches: F) -> bool
+    where
+        F: FnMut(&HealRequest) -> bool,
+    {
+        self.heap.iter().any(|item| matches(&item.request))
+    }
+
+    fn remove_request_id(&mut self, request_id: &str) -> Option<HealRequest> {
+        let mut retained = BinaryHeap::new();
+        let mut removed = 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);
+                self.dedup_keys.remove(&key);
+                removed = Some(item.request);
+            } else {
+                retained.push(item);
+            }
+        }
+
+        self.heap = retained;
+        removed
+    }
+
+    fn remove_matching<F>(&mut self, mut should_remove: F) -> usize
+    where
+        F: FnMut(&HealRequest) -> bool,
+    {
+        let mut retained = BinaryHeap::new();
+        let mut removed_count = 0;
+
+        while let Some(item) = self.heap.pop() {
+            if should_remove(&item.request) {
+                let key = Self::make_dedup_key(&item.request);
+                self.dedup_keys.remove(&key);
+                removed_count += 1;
+            } else {
+                retained.push(item);
+            }
+        }
+
+        self.heap = retained;
+        removed_count
+    }
+}
+
+fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool {
+    let heal_path = heal_path.trim_matches('/');
+    if heal_path.is_empty() {
+        return false;
+    }
+
+    match heal_type {
+        HealType::Object { bucket, object, .. }
+        | HealType::Metadata { bucket, object }
+        | HealType::ECDecode { bucket, object, .. } => heal_path == bucket || heal_path == format!("{bucket}/{object}"),
+        HealType::Bucket { bucket } => heal_path == bucket,
+        HealType::ErasureSet { set_disk_id, .. } => heal_path == set_disk_id,
+        HealType::MRF { meta_path } => heal_path == meta_path.trim_matches('/'),
+    }
 }
 
 fn publish_active_heal_count(active_heals: &HashMap<String, Arc<HealTask>>) {
@@ -357,6 +446,8 @@ pub struct HealManager {
     active_heals: Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
     /// Heal queue (priority-based)
     heal_queue: Arc<Mutex<PriorityHealQueue>>,
+    /// Recently completed heal statuses retained for status queries.
+    completed_heals: Arc<Mutex<HashMap<String, CompletedHealStatus>>>,
     /// Storage layer interface
     storage: Arc<dyn HealStorageAPI>,
     /// Cancel token
@@ -384,6 +475,7 @@ impl HealManager {
             state: Arc::new(RwLock::new(HealState::default())),
             active_heals: Arc::new(Mutex::new(HashMap::new())),
             heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())),
+            completed_heals: Arc::new(Mutex::new(HashMap::new())),
             storage,
             cancel_token: CancellationToken::new(),
             statistics: Arc::new(RwLock::new(HealStatistics::new())),
@@ -429,6 +521,7 @@ impl HealManager {
         }
         active_heals.clear();
         publish_active_heal_count(&active_heals);
+        self.completed_heals.lock().await.clear();
         crate::set_heal_queue_length(0);
 
         // update state
@@ -544,14 +637,139 @@ impl HealManager {
 
     /// Get task status
     pub async fn get_task_status(&self, task_id: &str) -> Result<HealTaskStatus> {
-        let active_heals = self.active_heals.lock().await;
-        if let Some(task) = active_heals.get(task_id) {
-            Ok(task.get_status().await)
-        } else {
-            Err(Error::TaskNotFound {
-                task_id: task_id.to_string(),
-            })
+        {
+            let active_heals = self.active_heals.lock().await;
+            if let Some(task) = active_heals.get(task_id) {
+                return Ok(task.get_status().await);
+            }
         }
+
+        let queue = self.heal_queue.lock().await;
+        if queue.contains_request_id(task_id) {
+            return Ok(HealTaskStatus::Pending);
+        }
+        drop(queue);
+
+        let mut completed_heals = self.completed_heals.lock().await;
+        prune_completed_heal_statuses(&mut completed_heals);
+        if let Some(completed) = completed_heals.get(task_id) {
+            return Ok(completed.status.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 active_heals = self.active_heals.lock().await;
+            if let Some(task) = active_heals.get(task_id)
+                && heal_type_matches_path(&task.heal_type, heal_path)
+            {
+                return Ok(HealTaskReport {
+                    status: task.get_status().await,
+                    result_items: task.get_result_items().await,
+                });
+            }
+        }
+
+        {
+            let queue = self.heal_queue.lock().await;
+            if queue.contains_request_id_matching_path(task_id, heal_path) {
+                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(task_id)
+                && heal_type_matches_path(&completed.heal_type, heal_path)
+            {
+                return Ok(HealTaskReport {
+                    status: completed.status.clone(),
+                    result_items: completed.result_items.clone(),
+                });
+            }
+        }
+
+        if self.path_has_task(heal_path).await {
+            return Err(Error::InvalidClientToken);
+        }
+
+        Err(Error::TaskNotFound {
+            task_id: task_id.to_string(),
+        })
+    }
+
+    /// Get task status for a path-bound client token.
+    ///
+    /// If the token is unknown but no task remains for the path, the caller can
+    /// treat it as an already-finished sequence. If the path still has a live or
+    /// recently completed task, a different token is invalid for that path.
+    pub async fn get_task_status_for_path(&self, heal_path: &str, task_id: &str) -> Result<HealTaskStatus> {
+        {
+            let active_heals = self.active_heals.lock().await;
+            if let Some(task) = active_heals.get(task_id)
+                && heal_type_matches_path(&task.heal_type, heal_path)
+            {
+                return Ok(task.get_status().await);
+            }
+        }
+
+        {
+            let queue = self.heal_queue.lock().await;
+            if queue.contains_request_id_matching_path(task_id, heal_path) {
+                return Ok(HealTaskStatus::Pending);
+            }
+        }
+
+        {
+            let mut completed_heals = self.completed_heals.lock().await;
+            prune_completed_heal_statuses(&mut completed_heals);
+            if let Some(completed) = completed_heals.get(task_id)
+                && heal_type_matches_path(&completed.heal_type, heal_path)
+            {
+                return Ok(completed.status.clone());
+            }
+        }
+
+        if self.path_has_task(heal_path).await {
+            return Err(Error::InvalidClientToken);
+        }
+
+        Err(Error::TaskNotFound {
+            task_id: task_id.to_string(),
+        })
+    }
+
+    async fn path_has_task(&self, heal_path: &str) -> bool {
+        {
+            let active_heals = self.active_heals.lock().await;
+            if active_heals
+                .values()
+                .any(|task| heal_type_matches_path(&task.heal_type, heal_path))
+            {
+                return true;
+            }
+        }
+
+        {
+            let queue = self.heal_queue.lock().await;
+            if queue.contains_matching(|request| heal_type_matches_path(&request.heal_type, heal_path)) {
+                return true;
+            }
+        }
+
+        let mut completed_heals = self.completed_heals.lock().await;
+        prune_completed_heal_statuses(&mut completed_heals);
+        completed_heals
+            .values()
+            .any(|completed| heal_type_matches_path(&completed.heal_type, heal_path))
     }
 
     /// Get task progress
@@ -574,18 +792,68 @@ impl HealManager {
 
     /// Cancel task
     pub async fn cancel_task(&self, task_id: &str) -> Result<()> {
-        let mut active_heals = self.active_heals.lock().await;
-        if let Some(task) = active_heals.get(task_id) {
-            task.cancel().await?;
-            active_heals.remove(task_id);
-            publish_active_heal_count(&active_heals);
-            info!("Cancelled heal task: {}", task_id);
-            Ok(())
-        } else {
-            Err(Error::TaskNotFound {
-                task_id: task_id.to_string(),
-            })
+        {
+            let mut active_heals = self.active_heals.lock().await;
+            if let Some(task) = active_heals.get(task_id) {
+                task.cancel().await?;
+                active_heals.remove(task_id);
+                publish_active_heal_count(&active_heals);
+                info!("Cancelled active heal task: {}", task_id);
+                return Ok(());
+            }
         }
+
+        let mut queue = self.heal_queue.lock().await;
+        if queue.remove_request_id(task_id).is_some() {
+            publish_heal_queue_length(&queue);
+            info!("Cancelled queued heal task: {}", task_id);
+            return Ok(());
+        }
+
+        Err(Error::TaskNotFound {
+            task_id: task_id.to_string(),
+        })
+    }
+
+    /// Cancel all queued or active tasks matching a heal path.
+    pub async fn cancel_tasks_for_path(&self, heal_path: &str) -> Result<usize> {
+        let mut cancelled = 0usize;
+
+        {
+            let mut active_heals = self.active_heals.lock().await;
+            let task_ids = active_heals
+                .iter()
+                .filter_map(|(task_id, task)| heal_type_matches_path(&task.heal_type, heal_path).then_some(task_id.clone()))
+                .collect::<Vec<_>>();
+
+            for task_id in task_ids {
+                if let Some(task) = active_heals.get(&task_id) {
+                    task.cancel().await?;
+                }
+                active_heals.remove(&task_id);
+                cancelled += 1;
+            }
+
+            if cancelled > 0 {
+                publish_active_heal_count(&active_heals);
+            }
+        }
+
+        let mut queue = self.heal_queue.lock().await;
+        let queued_cancelled = queue.remove_matching(|request| heal_type_matches_path(&request.heal_type, heal_path));
+        if queued_cancelled > 0 {
+            publish_heal_queue_length(&queue);
+            cancelled += queued_cancelled;
+        }
+
+        if cancelled == 0 {
+            return Err(Error::TaskNotFound {
+                task_id: heal_path.to_string(),
+            });
+        }
+
+        info!("Cancelled {} heal task(s) for path: {}", cancelled, heal_path);
+        Ok(cancelled)
     }
 
     /// Get statistics
@@ -612,6 +880,7 @@ impl HealManager {
         let config = self.config.clone();
         let heal_queue = self.heal_queue.clone();
         let active_heals = self.active_heals.clone();
+        let completed_heals = self.completed_heals.clone();
         let cancel_token = self.cancel_token.clone();
         let statistics = self.statistics.clone();
         let storage = self.storage.clone();
@@ -628,10 +897,10 @@ impl HealManager {
                         break;
                     }
                     _ = notify.notified(), if event_driven_scheduler_enable => {
-                        Self::process_heal_queue(&heal_queue, &active_heals, &config, &statistics, &storage, &notify).await;
+                        Self::process_heal_queue(&heal_queue, &active_heals, &completed_heals, &config, &statistics, &storage, &notify).await;
                     }
                     _ = interval.tick() => {
-                        Self::process_heal_queue(&heal_queue, &active_heals, &config, &statistics, &storage, &notify).await;
+                        Self::process_heal_queue(&heal_queue, &active_heals, &completed_heals, &config, &statistics, &storage, &notify).await;
                     }
                 }
             }
@@ -768,6 +1037,7 @@ impl HealManager {
     async fn process_heal_queue(
         heal_queue: &Arc<Mutex<PriorityHealQueue>>,
         active_heals: &Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
+        completed_heals: &Arc<Mutex<HashMap<String, CompletedHealStatus>>>,
         config: &Arc<RwLock<HealConfig>>,
         statistics: &Arc<RwLock<HealStatistics>>,
         storage: &Arc<dyn HealStorageAPI>,
@@ -827,6 +1097,7 @@ impl HealManager {
                 publish_active_heal_count(&active_heals_guard);
                 update_task_running_metric_for_task(&active_heals_guard, task.as_ref());
                 let active_heals_clone = active_heals.clone();
+                let completed_heals_clone = completed_heals.clone();
                 let statistics_clone = statistics.clone();
                 let notify_clone = notify.clone();
                 let task_type_label_for_spawn = task_type_label.clone();
@@ -851,9 +1122,19 @@ impl HealManager {
                     if let Some(completed_task) = active_heals_guard.remove(&task_id) {
                         publish_active_heal_count(&active_heals_guard);
                         update_task_running_metric_for_task(&active_heals_guard, completed_task.as_ref());
+                        let completed_status = completed_task.get_status().await;
+                        let completed_status_entry = CompletedHealStatus {
+                            heal_type: completed_task.heal_type.clone(),
+                            status: completed_status.clone(),
+                            result_items: completed_task.get_result_items().await,
+                            completed_at: SystemTime::now(),
+                        };
+                        let mut completed_heals_guard = completed_heals_clone.lock().await;
+                        prune_completed_heal_statuses(&mut completed_heals_guard);
+                        completed_heals_guard.insert(task_id.clone(), completed_status_entry);
                         // update statistics
                         let mut stats = statistics_clone.write().await;
-                        match completed_task.get_status().await {
+                        match completed_status {
                             HealTaskStatus::Completed => {
                                 stats.update_task_completion(true);
                             }
@@ -959,6 +1240,20 @@ fn running_erasure_set_counts(active_heals: &HashMap<String, Arc<HealTask>>) ->
     running
 }
 
+fn prune_completed_heal_statuses(completed_heals: &mut HashMap<String, CompletedHealStatus>) {
+    let Ok(now) = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) else {
+        return;
+    };
+
+    completed_heals.retain(|_, completed| {
+        completed
+            .completed_at
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .map(|completed_at| now.saturating_sub(completed_at) <= KEEP_HEAL_TASK_STATUS_DURATION)
+            .unwrap_or(false)
+    });
+}
+
 fn can_schedule_request(request: &HealRequest, running_per_set: &HashMap<String, usize>, max_concurrent_per_set: usize) -> bool {
     match heal_request_set_key(request) {
         Some(set_key) => running_per_set.get(&set_key).copied().unwrap_or(0) < max_concurrent_per_set,
@@ -1497,6 +1792,258 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_get_task_status_reports_pending_for_queued_request() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let request = HealRequest::bucket("bucket".to_string());
+        let request_id = request.id.clone();
+
+        assert_eq!(
+            manager
+                .submit_heal_request(request)
+                .await
+                .expect("request should be accepted"),
+            HealAdmissionResult::Accepted
+        );
+        assert_eq!(
+            manager
+                .get_task_status(&request_id)
+                .await
+                .expect("queued request should have status"),
+            HealTaskStatus::Pending
+        );
+    }
+
+    #[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);
+        let manager = HealManager::new(storage, None);
+
+        manager
+            .submit_heal_request(HealRequest::bucket("bucket".to_string()))
+            .await
+            .expect("request should be accepted");
+
+        assert!(matches!(
+            manager.get_task_status_for_path("bucket", "wrong-token").await,
+            Err(Error::InvalidClientToken)
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_get_task_status_for_path_rejects_token_from_other_active_path() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let bucket_request = HealRequest::bucket("bucket".to_string());
+        let other_request = HealRequest::bucket("other".to_string());
+        let other_request_id = other_request.id.clone();
+
+        manager
+            .submit_heal_request(bucket_request)
+            .await
+            .expect("bucket request should be accepted");
+        manager
+            .submit_heal_request(other_request)
+            .await
+            .expect("other request should be accepted");
+
+        assert!(matches!(
+            manager.get_task_status_for_path("bucket", &other_request_id).await,
+            Err(Error::InvalidClientToken)
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_get_task_status_for_path_does_not_accept_token_from_inactive_path() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let request = HealRequest::bucket("bucket".to_string());
+        let request_id = request.id.clone();
+
+        manager
+            .submit_heal_request(request)
+            .await
+            .expect("request should be accepted");
+
+        assert!(matches!(
+            manager.get_task_status_for_path("other", &request_id).await,
+            Err(Error::TaskNotFound { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_get_task_status_for_path_returns_not_found_when_path_is_inactive() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        assert!(matches!(
+            manager.get_task_status_for_path("bucket", "old-token").await,
+            Err(Error::TaskNotFound { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_get_task_status_for_empty_path_does_not_match_unrelated_tasks() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let request = HealRequest::bucket("bucket".to_string());
+        let request_id = request.id.clone();
+
+        manager
+            .submit_heal_request(request)
+            .await
+            .expect("request should be accepted");
+
+        assert!(matches!(
+            manager.get_task_status_for_path("", &request_id).await,
+            Err(Error::TaskNotFound { .. })
+        ));
+        assert!(matches!(
+            manager.get_task_status_for_path("", "wrong-token").await,
+            Err(Error::TaskNotFound { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_get_task_status_reads_recent_completed_status() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        manager.completed_heals.lock().await.insert(
+            "completed-token".to_string(),
+            CompletedHealStatus {
+                heal_type: HealType::Bucket {
+                    bucket: "bucket".to_string(),
+                },
+                status: HealTaskStatus::Completed,
+                result_items: Vec::new(),
+                completed_at: SystemTime::now(),
+            },
+        );
+
+        assert_eq!(
+            manager
+                .get_task_status_for_path("bucket", "completed-token")
+                .await
+                .expect("recent completed task should be queryable"),
+            HealTaskStatus::Completed
+        );
+    }
+
+    #[tokio::test]
+    async fn test_get_task_report_for_path_reads_completed_items() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        manager.completed_heals.lock().await.insert(
+            "completed-token".to_string(),
+            CompletedHealStatus {
+                heal_type: HealType::Object {
+                    bucket: "bucket".to_string(),
+                    object: "object".to_string(),
+                    version_id: None,
+                },
+                status: HealTaskStatus::Completed,
+                result_items: vec![HealResultItem {
+                    bucket: "bucket".to_string(),
+                    object: "object".to_string(),
+                    object_size: 1024,
+                    ..Default::default()
+                }],
+                completed_at: SystemTime::now(),
+            },
+        );
+
+        let report = manager
+            .get_task_report_for_path("bucket/object", "completed-token")
+            .await
+            .expect("recent completed task report should be queryable");
+
+        assert_eq!(report.status, HealTaskStatus::Completed);
+        assert_eq!(report.result_items.len(), 1);
+        assert_eq!(report.result_items[0].object_size, 1024);
+    }
+
+    #[tokio::test]
+    async fn test_get_task_report_for_empty_path_does_not_match_unrelated_tasks() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        manager
+            .submit_heal_request(HealRequest::bucket("bucket".to_string()))
+            .await
+            .expect("request should be accepted");
+
+        assert!(matches!(
+            manager.get_task_report_for_path("", "wrong-token").await,
+            Err(Error::TaskNotFound { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn test_cancel_task_removes_queued_request() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let request = HealRequest::bucket("bucket".to_string());
+        let request_id = request.id.clone();
+
+        manager
+            .submit_heal_request(request)
+            .await
+            .expect("request should be accepted");
+        manager
+            .cancel_task(&request_id)
+            .await
+            .expect("queued request should be cancelled");
+
+        assert!(matches!(manager.get_task_status(&request_id).await, Err(Error::TaskNotFound { .. })));
+    }
+
+    #[tokio::test]
+    async fn test_cancel_tasks_for_path_removes_matching_queued_requests() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+
+        let bucket_request = HealRequest::bucket("bucket".to_string());
+        let bucket_request_id = bucket_request.id.clone();
+        let other_request = HealRequest::bucket("other".to_string());
+        let other_request_id = other_request.id.clone();
+
+        manager
+            .submit_heal_request(bucket_request)
+            .await
+            .expect("bucket request should be accepted");
+        manager
+            .submit_heal_request(other_request)
+            .await
+            .expect("other request should be accepted");
+
+        assert_eq!(
+            manager
+                .cancel_tasks_for_path("bucket")
+                .await
+                .expect("matching request should be cancelled"),
+            1
+        );
+        assert!(matches!(
+            manager.get_task_status(&bucket_request_id).await,
+            Err(Error::TaskNotFound { .. })
+        ));
+        assert_eq!(
+            manager
+                .get_task_status(&other_request_id)
+                .await
+                .expect("unmatched request should remain queued"),
+            HealTaskStatus::Pending
+        );
+    }
+
     #[tokio::test]
     async fn test_submit_heal_request_returns_merged_before_full_for_duplicate() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/heal/task.rs

@@ -16,6 +16,7 @@ use crate::heal::{ErasureSetHealer, progress::HealProgress, storage::HealStorage
 use crate::{Error, Result};
 use metrics::{counter, histogram};
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
+use rustfs_madmin::heal_commands::HealResultItem;
 use serde::{Deserialize, Serialize};
 use std::{
     future::Future,
@@ -196,6 +197,8 @@ pub struct HealTask {
     pub status: Arc<RwLock<HealTaskStatus>>,
     /// Progress tracking
     pub progress: Arc<RwLock<HealProgress>>,
+    /// Result items collected from storage heal calls.
+    pub result_items: Arc<RwLock<Vec<HealResultItem>>>,
     /// Created time
     pub created_at: SystemTime,
     /// Queue admission time
@@ -220,6 +223,7 @@ impl HealTask {
             options: request.options,
             status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
             progress: Arc::new(RwLock::new(HealProgress::new())),
+            result_items: Arc::new(RwLock::new(Vec::new())),
             created_at: request.created_at,
             enqueued_at: request.enqueued_at,
             started_at: Arc::new(RwLock::new(None)),
@@ -411,6 +415,14 @@ impl HealTask {
         self.progress.read().await.clone()
     }
 
+    pub async fn get_result_items(&self) -> Vec<HealResultItem> {
+        self.result_items.read().await.clone()
+    }
+
+    async fn record_result_item(&self, result: HealResultItem) {
+        self.result_items.write().await.push(result);
+    }
+
     // specific heal implementation method
     #[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))]
     async fn heal_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
@@ -521,6 +533,7 @@ impl HealTask {
                     let mut progress = self.progress.write().await;
                     progress.update_progress(3, 3, object_size, object_size);
                 }
+                self.record_result_item(result).await;
                 Ok(())
             }
             Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
@@ -601,6 +614,7 @@ impl HealTask {
                     let mut progress = self.progress.write().await;
                     progress.update_progress(4, 4, object_size, object_size);
                 }
+                self.record_result_item(result).await;
                 Ok(())
             }
             Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
@@ -659,8 +673,13 @@ impl HealTask {
         match heal_result {
             Ok(result) => {
                 info!("Bucket heal completed successfully: {} ({} drives)", bucket, result.after.drives.len());
+                self.record_result_item(result).await;
 
-                {
+                if self.options.recursive {
+                    self.heal_bucket_objects(bucket).await?;
+                }
+
+                if !self.options.recursive {
                     let mut progress = self.progress.write().await;
                     progress.update_progress(3, 3, 0, 0);
                 }
@@ -681,6 +700,98 @@ impl HealTask {
         }
     }
 
+    async fn heal_bucket_objects(&self, bucket: &str) -> Result<()> {
+        let mut continuation_token: Option<String> = None;
+        let mut scanned = 0u64;
+        let mut healed = 0u64;
+        let mut failed = 0u64;
+        let mut bytes = 0u64;
+
+        let heal_opts = HealOpts {
+            recursive: false,
+            dry_run: self.options.dry_run,
+            remove: self.options.remove_corrupted,
+            recreate: self.options.recreate_missing,
+            scan_mode: self.options.scan_mode,
+            update_parity: self.options.update_parity,
+            no_lock: false,
+            pool: self.options.pool_index,
+            set: self.options.set_index,
+        };
+
+        loop {
+            self.check_control_flags().await?;
+            let (objects, next_token, is_truncated) = self
+                .await_with_control(
+                    self.storage
+                        .list_objects_for_heal_page(bucket, "", continuation_token.as_deref()),
+                )
+                .await?;
+
+            for object in objects {
+                self.check_control_flags().await?;
+                scanned += 1;
+                {
+                    let mut progress = self.progress.write().await;
+                    progress.set_current_object(Some(format!("{bucket}/{object}")));
+                    progress.update_progress(scanned, healed, failed, bytes);
+                }
+
+                match self.await_with_control(self.storage.object_exists(bucket, &object)).await {
+                    Ok(false) => {
+                        healed += 1;
+                    }
+                    Ok(true) => match self
+                        .await_with_control(self.storage.heal_object(bucket, &object, None, &heal_opts))
+                        .await
+                    {
+                        Ok((result, None)) => {
+                            healed += 1;
+                            bytes = bytes.saturating_add(result.object_size as u64);
+                            self.record_result_item(result).await;
+                        }
+                        Ok((_, Some(err))) => {
+                            failed += 1;
+                            warn!("Failed to heal object {}/{}: {}", bucket, object, err);
+                        }
+                        Err(err) => {
+                            failed += 1;
+                            warn!("Failed to heal object {}/{}: {}", bucket, object, err);
+                        }
+                    },
+                    Err(err) => {
+                        failed += 1;
+                        warn!("Failed to check object {}/{} before heal: {}", bucket, object, err);
+                    }
+                }
+
+                {
+                    let mut progress = self.progress.write().await;
+                    progress.update_progress(scanned, healed, failed, bytes);
+                }
+            }
+
+            if !is_truncated {
+                break;
+            }
+
+            continuation_token = next_token;
+            if continuation_token.is_none() {
+                warn!("List is truncated but no continuation token was returned for bucket {}", bucket);
+                break;
+            }
+        }
+
+        if failed > 0 {
+            return Err(Error::TaskExecutionFailed {
+                message: format!("Failed to heal {failed} object(s) in bucket {bucket}"),
+            });
+        }
+
+        info!("Recursive bucket heal completed for {}: {} scanned, {} healed", bucket, scanned, healed);
+        Ok(())
+    }
+
     async fn heal_metadata(&self, bucket: &str, object: &str) -> Result<()> {
         info!("Healing metadata: {}/{}", bucket, object);
 
@@ -755,6 +866,7 @@ impl HealTask {
                     let mut progress = self.progress.write().await;
                     progress.update_progress(3, 3, 0, 0);
                 }
+                self.record_result_item(result).await;
                 Ok(())
             }
             Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
@@ -830,6 +942,7 @@ impl HealTask {
                     let mut progress = self.progress.write().await;
                     progress.update_progress(2, 2, 0, 0);
                 }
+                self.record_result_item(result).await;
                 Ok(())
             }
             Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
@@ -923,6 +1036,7 @@ impl HealTask {
                     let mut progress = self.progress.write().await;
                     progress.update_progress(3, 3, object_size, object_size);
                 }
+                self.record_result_item(result).await;
                 Ok(())
             }
             Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
@@ -1072,3 +1186,163 @@ impl std::fmt::Debug for HealTask {
             .finish()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::heal::storage::DiskStatus;
+    use rustfs_ecstore::{
+        disk::{DiskStore, endpoint::Endpoint},
+        store_api::{BucketInfo, ObjectInfo},
+    };
+    use rustfs_madmin::heal_commands::HealResultItem;
+    use std::sync::Mutex;
+
+    #[derive(Default)]
+    struct MockStorage {
+        listed: Mutex<bool>,
+        healed_objects: Mutex<Vec<String>>,
+    }
+
+    #[async_trait::async_trait]
+    impl HealStorageAPI for MockStorage {
+        async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result<Option<ObjectInfo>> {
+            Ok(None)
+        }
+
+        async fn get_object_data(&self, _bucket: &str, _object: &str) -> Result<Option<Vec<u8>>> {
+            Ok(None)
+        }
+
+        async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> Result<()> {
+            Ok(())
+        }
+
+        async fn delete_object(&self, _bucket: &str, _object: &str) -> Result<()> {
+            Ok(())
+        }
+
+        async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> Result<bool> {
+            Ok(true)
+        }
+
+        async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> Result<Vec<u8>> {
+            Ok(Vec::new())
+        }
+
+        async fn get_disk_status(&self, _endpoint: &Endpoint) -> Result<DiskStatus> {
+            Ok(DiskStatus::Ok)
+        }
+
+        async fn format_disk(&self, _endpoint: &Endpoint) -> Result<()> {
+            Ok(())
+        }
+
+        async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
+            Ok(Some(BucketInfo {
+                name: bucket.to_string(),
+                ..Default::default()
+            }))
+        }
+
+        async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<()> {
+            Ok(())
+        }
+
+        async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
+            Ok(Vec::new())
+        }
+
+        async fn object_exists(&self, _bucket: &str, _object: &str) -> Result<bool> {
+            Ok(true)
+        }
+
+        async fn get_object_size(&self, _bucket: &str, _object: &str) -> Result<Option<u64>> {
+            Ok(None)
+        }
+
+        async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> Result<Option<String>> {
+            Ok(None)
+        }
+
+        async fn heal_object(
+            &self,
+            _bucket: &str,
+            object: &str,
+            _version_id: Option<&str>,
+            _opts: &HealOpts,
+        ) -> Result<(HealResultItem, Option<Error>)> {
+            self.healed_objects.lock().unwrap().push(object.to_string());
+            Ok((
+                HealResultItem {
+                    object_size: 1,
+                    ..Default::default()
+                },
+                None,
+            ))
+        }
+
+        async fn heal_bucket(&self, _bucket: &str, _opts: &HealOpts) -> Result<HealResultItem> {
+            Ok(HealResultItem::default())
+        }
+
+        async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
+            Ok((HealResultItem::default(), None))
+        }
+
+        async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<String>> {
+            Ok(vec!["object-a".to_string(), "object-b".to_string()])
+        }
+
+        async fn list_objects_for_heal_page(
+            &self,
+            _bucket: &str,
+            _prefix: &str,
+            continuation_token: Option<&str>,
+        ) -> Result<(Vec<String>, Option<String>, bool)> {
+            let mut listed = self.listed.lock().unwrap();
+            if continuation_token.is_none() && !*listed {
+                *listed = true;
+                Ok((vec!["object-a".to_string(), "object-b".to_string()], None, false))
+            } else {
+                Ok((Vec::new(), None, false))
+            }
+        }
+
+        async fn get_disk_for_resume(&self, _set_disk_id: &str) -> Result<DiskStore> {
+            Err(Error::other("not implemented in tests"))
+        }
+    }
+
+    #[tokio::test]
+    async fn test_recursive_bucket_heal_visits_objects() {
+        let storage = Arc::new(MockStorage::default());
+        let request = HealRequest::new(
+            HealType::Bucket {
+                bucket: "bucket-a".to_string(),
+            },
+            HealOptions {
+                recursive: true,
+                timeout: None,
+                ..Default::default()
+            },
+            HealPriority::Normal,
+        );
+        let task = HealTask::from_request(request, storage.clone());
+
+        task.heal_bucket("bucket-a")
+            .await
+            .expect("recursive bucket heal should succeed");
+
+        assert_eq!(
+            storage.healed_objects.lock().unwrap().as_slice(),
+            ["object-a".to_string(), "object-b".to_string()]
+        );
+        let progress = task.get_progress().await;
+        assert_eq!(progress.objects_scanned, 2);
+        assert_eq!(progress.objects_healed, 2);
+        let result_items = task.get_result_items().await;
+        assert_eq!(result_items.len(), 3);
+        assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2);
+    }
+}