fix(heal): aggregate status across cluster nodes (#4990)

* fix(rpc): bind internode auth to exact targets

* fix(heal): initialize the runtime atomically

* fix(heal): aggregate status across cluster nodes

---------

Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

crates/common/src/heal_channel.rs

@@ -243,6 +243,19 @@ pub enum HealAdmissionResult {
     Dropped(HealAdmissionDropReason),
 }
 
+/// Admission decision together with the canonical task identifier.
+///
+/// A merged request must return the identifier of the task that already owns
+/// the work instead of exposing the discarded request identifier as a new
+/// client token.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HealAdmissionReceipt {
+    /// Admission decision for the submitted request.
+    pub result: HealAdmissionResult,
+    /// Canonical identifier of the accepted or merged task.
+    pub task_id: String,
+}
+
 impl HealAdmissionResult {
     pub fn result_label(self) -> &'static str {
         match self {
@@ -382,8 +395,25 @@ pub type HealChannelSender = mpsc::UnboundedSender<HealChannelCommand>;
 /// Heal channel receiver
 pub type HealChannelReceiver = mpsc::UnboundedReceiver<HealChannelCommand>;
 
+/// Canonical-receipt start command kept separate from the legacy public enum.
+#[derive(Debug)]
+pub struct HealReceiptCommand {
+    /// Heal request to admit.
+    pub request: HealChannelRequest,
+    /// Completion channel for the admission receipt.
+    pub response_tx: oneshot::Sender<Result<HealAdmissionReceipt, String>>,
+}
+
+/// Canonical-receipt command receiver.
+pub type HealReceiptReceiver = mpsc::UnboundedReceiver<HealReceiptCommand>;
+
+struct HealChannelSenders {
+    command: HealChannelSender,
+    receipt: mpsc::UnboundedSender<HealReceiptCommand>,
+}
+
 /// Global heal channel sender
-static GLOBAL_HEAL_CHANNEL_SENDER: OnceLock<HealChannelSender> = OnceLock::new();
+static GLOBAL_HEAL_CHANNEL_SENDERS: OnceLock<HealChannelSenders> = OnceLock::new();
 
 type HealResponseSender = broadcast::Sender<HealChannelResponse>;
 
@@ -392,17 +422,24 @@ static GLOBAL_HEAL_RESPONSE_SENDER: OnceLock<HealResponseSender> = OnceLock::new
 
 /// Initialize global heal channel
 pub fn init_heal_channel() -> Result<HealChannelReceiver, &'static str> {
-    let (tx, rx) = mpsc::unbounded_channel();
-    if GLOBAL_HEAL_CHANNEL_SENDER.set(tx).is_ok() {
-        Ok(rx)
-    } else {
-        Err("Heal channel sender already initialized")
-    }
+    let (receiver, receipt_receiver) = init_heal_channels()?;
+    drop(receipt_receiver);
+    Ok(receiver)
+}
+
+/// Initialize the legacy command and canonical-receipt channels atomically.
+pub fn init_heal_channels() -> Result<(HealChannelReceiver, HealReceiptReceiver), &'static str> {
+    let (command, command_receiver) = mpsc::unbounded_channel();
+    let (receipt, receipt_receiver) = mpsc::unbounded_channel();
+    GLOBAL_HEAL_CHANNEL_SENDERS
+        .set(HealChannelSenders { command, receipt })
+        .map_err(|_| "Heal channel sender already initialized")?;
+    Ok((command_receiver, receipt_receiver))
 }
 
 /// Get global heal channel sender
 pub fn get_heal_channel_sender() -> Option<&'static HealChannelSender> {
-    GLOBAL_HEAL_CHANNEL_SENDER.get()
+    GLOBAL_HEAL_CHANNEL_SENDERS.get().map(|senders| &senders.command)
 }
 
 /// Send heal command through global channel
@@ -436,6 +473,21 @@ pub fn subscribe_heal_responses() -> broadcast::Receiver<HealChannelResponse> {
     heal_response_sender().subscribe()
 }
 
+/// Send heal start request and wait for structured admission feedback.
+pub async fn send_heal_request_with_receipt(request: HealChannelRequest) -> Result<HealAdmissionReceipt, String> {
+    let (response_tx, response_rx) = oneshot::channel();
+    let senders = GLOBAL_HEAL_CHANNEL_SENDERS
+        .get()
+        .ok_or_else(|| "Heal channel not initialized".to_string())?;
+    senders
+        .receipt
+        .send(HealReceiptCommand { request, response_tx })
+        .map_err(|err| format!("Failed to send heal receipt command: {err}"))?;
+    response_rx
+        .await
+        .map_err(|e| format!("Failed to receive heal admission response: {e}"))?
+}
+
 /// Send heal start request and wait for structured admission feedback.
 pub async fn send_heal_request_with_admission(request: HealChannelRequest) -> Result<HealAdmissionResult, String> {
     let (response_tx, response_rx) = oneshot::channel();

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -29,11 +29,11 @@ use rustfs_madmin::{
 };
 use rustfs_protos::evict_failed_connection;
 use rustfs_protos::proto_gen::node_service::{
-    CancelDecommissionRequest, ClearDecommissionRequest, DeleteBucketMetadataRequest, DeletePolicyRequest,
-    DeleteServiceAccountRequest, DeleteUserRequest, GetCpusRequest, GetLiveEventsRequest, GetMemInfoRequest, GetMetricsRequest,
-    GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, GetSysConfigRequest,
-    GetSysErrorsRequest, LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest,
-    LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest,
+    BackgroundHealStatusRequest, CancelDecommissionRequest, ClearDecommissionRequest, DeleteBucketMetadataRequest,
+    DeletePolicyRequest, DeleteServiceAccountRequest, DeleteUserRequest, GetCpusRequest, GetLiveEventsRequest, GetMemInfoRequest,
+    GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest,
+    GetSysConfigRequest, GetSysErrorsRequest, LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest,
+    LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest,
     LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ScannerActivityRequest,
     ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest, StartDecommissionRequest, StartProfilingRequest,
     StopRebalanceRequest, node_service_client::NodeServiceClient,
@@ -60,6 +60,7 @@ pub const PEER_RESTSUB_SYS: &str = "sub-sys";
 pub const PEER_RESTDRY_RUN: &str = "dry-run";
 pub const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = 1;
 pub const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = 2;
+const BACKGROUND_HEAL_STATUS_MAX_MESSAGE_SIZE: usize = 64 * 1024;
 const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
 const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
 const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024;
@@ -664,6 +665,38 @@ impl PeerRestClient {
         Err(Error::NotImplemented)
     }
 
+    pub async fn background_heal_status(&self) -> Result<Option<Vec<u8>>> {
+        self.finalize_result(
+            async {
+                let mut client = self
+                    .get_client()
+                    .await?
+                    .max_decoding_message_size(BACKGROUND_HEAL_STATUS_MAX_MESSAGE_SIZE);
+                let response = match client
+                    .background_heal_status(Request::new(BackgroundHealStatusRequest::default()))
+                    .await
+                {
+                    Ok(response) => response.into_inner(),
+                    Err(status) if status.code() == tonic::Code::Unimplemented => {
+                        // RUSTFS_COMPAT_TODO(heal-status-rpc-v1): accept old peers without node heal snapshots. Remove after the minimum supported RustFS peer version implements BackgroundHealStatus.
+                        return Ok(None);
+                    }
+                    Err(status) => return Err(status.into()),
+                };
+                if !response.success {
+                    return Err(Error::other(
+                        response
+                            .error_info
+                            .unwrap_or_else(|| "peer background heal status failed without an error".to_string()),
+                    ));
+                }
+                Ok(Some(response.bg_heal_state.to_vec()))
+            }
+            .await,
+        )
+        .await
+    }
+
     pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
         self.finalize_result(
             async {

crates/heal/src/heal/channel.rs

@@ -20,8 +20,8 @@ use crate::heal::{
 };
 use crate::{Error, Result};
 use rustfs_common::heal_channel::{
-    HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealChannelResponse,
-    HealRequestSource, HealScanMode, publish_heal_response,
+    HealAdmissionReceipt, HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest,
+    HealChannelResponse, HealReceiptCommand, HealReceiptReceiver, HealRequestSource, HealScanMode, publish_heal_response,
 };
 use rustfs_madmin::heal_commands::HealResultItem;
 use serde::Serialize;
@@ -79,8 +79,19 @@ impl HealChannelProcessor {
         }
     }
 
-    /// Start processing heal channel requests
-    pub async fn start(&mut self, mut receiver: HealChannelReceiver) -> Result<()> {
+    /// Start processing legacy heal channel requests.
+    pub async fn start(&mut self, receiver: HealChannelReceiver) -> Result<()> {
+        let (receipt_sender, receipt_receiver) = mpsc::unbounded_channel();
+        drop(receipt_sender);
+        self.start_with_receipts(receiver, receipt_receiver).await
+    }
+
+    /// Start processing legacy and canonical-receipt heal channel requests.
+    pub async fn start_with_receipts(
+        &mut self,
+        mut receiver: HealChannelReceiver,
+        mut receipt_receiver: HealReceiptReceiver,
+    ) -> Result<()> {
         info!(
             target: "rustfs::heal::channel",
             event = EVENT_HEAL_CHANNEL_STATE,
@@ -90,6 +101,7 @@ impl HealChannelProcessor {
             "Heal channel started"
         );
 
+        let mut receipt_channel_open = true;
         loop {
             tokio::select! {
                 command = receiver.recv() => {
@@ -120,6 +132,23 @@ impl HealChannelProcessor {
                         }
                     }
                 }
+                command = receipt_receiver.recv(), if receipt_channel_open => {
+                    let Some(HealReceiptCommand { request, response_tx }) = command else {
+                        receipt_channel_open = false;
+                        continue;
+                    };
+                    if let Err(e) = self.process_start_request(request, false, true, response_tx).await {
+                        error!(
+                            target: "rustfs::heal::channel",
+                            event = EVENT_HEAL_CHANNEL_REQUEST,
+                            component = LOG_COMPONENT_HEAL,
+                            subsystem = LOG_SUBSYSTEM_CHANNEL,
+                            state = "receipt_process_failed",
+                            error = %e,
+                            "Heal receipt request processing failed"
+                        );
+                    }
+                }
                 response = self.response_receiver.recv() => {
                     if let Some(response) = response {
                         // Handle response if needed
@@ -152,7 +181,7 @@ impl HealChannelProcessor {
     /// Process heal command
     async fn process_command(&self, command: HealChannelCommand) -> Result<()> {
         match command {
-            HealChannelCommand::Start { request, response_tx } => self.process_start_request(request, response_tx).await,
+            HealChannelCommand::Start { request, response_tx } => self.process_legacy_start_request(request, response_tx).await,
             HealChannelCommand::Query {
                 heal_path,
                 client_token,
@@ -166,11 +195,28 @@ impl HealChannelProcessor {
         }
     }
 
+    async fn process_legacy_start_request(
+        &self,
+        request: HealChannelRequest,
+        response_tx: oneshot::Sender<std::result::Result<HealAdmissionResult, String>>,
+    ) -> Result<()> {
+        let (receipt_tx, receipt_rx) = oneshot::channel();
+        self.process_start_request(request, true, false, receipt_tx).await?;
+        let result = receipt_rx
+            .await
+            .map_err(|err| Error::other(format!("heal receipt channel closed: {err}")))?
+            .map(|receipt| receipt.result);
+        let _ = response_tx.send(result);
+        Ok(())
+    }
+
     /// Process start request
     async fn process_start_request(
         &self,
         request: HealChannelRequest,
-        response_tx: oneshot::Sender<std::result::Result<HealAdmissionResult, String>>,
+        preserve_alias: bool,
+        publish_canonical_id: bool,
+        response_tx: oneshot::Sender<std::result::Result<HealAdmissionReceipt, String>>,
     ) -> Result<()> {
         debug!(
             target: "rustfs::heal::channel",
@@ -201,8 +247,13 @@ impl HealChannelProcessor {
         };
 
         // Submit to heal manager
-        match self.heal_manager.submit_heal_request(heal_request).await {
-            Ok(admission) => {
+        match self
+            .heal_manager
+            .submit_heal_request_with_receipt_and_alias(heal_request, preserve_alias)
+            .await
+        {
+            Ok(receipt) => {
+                let admission = receipt.result;
                 debug!(
                     target: "rustfs::heal::channel",
                     event = EVENT_HEAL_CHANNEL_REQUEST,
@@ -214,9 +265,13 @@ impl HealChannelProcessor {
                     "Heal admission decided"
                 );
 
-                let _ = response_tx.send(Ok(admission));
-
-                self.publish_response(admission_response(request.id, admission));
+                let response_id = if publish_canonical_id {
+                    receipt.task_id.clone()
+                } else {
+                    request.id.clone()
+                };
+                self.publish_response(admission_response(response_id, receipt.result));
+                let _ = response_tx.send(Ok(receipt));
             }
             Err(e) => {
                 let error_text = e.to_string();
@@ -537,6 +592,7 @@ mod tests {
         HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealRequestSource, HealScanMode,
     };
     use std::sync::Arc;
+    use std::time::Duration;
 
     // Mock storage for testing
     struct MockStorage;
@@ -1097,27 +1153,190 @@ mod tests {
 
         let (tx, rx) = oneshot::channel();
         processor
-            .process_start_request(request.clone(), tx)
+            .process_start_request(request.clone(), false, true, tx)
             .await
             .expect("first admission should succeed");
-        assert_eq!(
-            rx.await
-                .expect("oneshot should resolve")
-                .expect("admission should be returned"),
-            HealAdmissionResult::Accepted
-        );
+        let first = rx
+            .await
+            .expect("oneshot should resolve")
+            .expect("admission should be returned");
+        assert_eq!(first.result, HealAdmissionResult::Accepted);
+        assert_eq!(first.task_id, "admission-id");
+
+        let mut duplicate = request;
+        duplicate.id = "duplicate-id".to_string();
+        let (tx, rx) = oneshot::channel();
+        processor
+            .process_start_request(duplicate, false, true, tx)
+            .await
+            .expect("duplicate admission should succeed");
+        let merged = rx
+            .await
+            .expect("oneshot should resolve")
+            .expect("admission should be returned");
+        assert_eq!(merged.result, HealAdmissionResult::Merged);
+        assert_eq!(merged.task_id, "admission-id");
+    }
+
+    #[tokio::test]
+    async fn test_legacy_start_preserves_duplicate_alias_and_submitted_response_id() {
+        let manager = create_test_heal_manager();
+        let processor = HealChannelProcessor::new(manager.clone());
+        let request = HealChannelRequest {
+            id: "legacy-original-id".to_string(),
+            bucket: "bucket".to_string(),
+            object_prefix: Some("object".to_string()),
+            priority: HealChannelPriority::Low,
+            source: HealRequestSource::Admin,
+            ..Default::default()
+        };
+        let mut responses = rustfs_common::heal_channel::subscribe_heal_responses();
 
         let (tx, rx) = oneshot::channel();
         processor
-            .process_start_request(request, tx)
+            .process_command(HealChannelCommand::Start {
+                request: request.clone(),
+                response_tx: tx,
+            })
             .await
-            .expect("duplicate admission should succeed");
+            .expect("legacy start should be processed");
         assert_eq!(
             rx.await
-                .expect("oneshot should resolve")
-                .expect("admission should be returned"),
+                .expect("legacy response should arrive")
+                .expect("legacy start should succeed"),
+            HealAdmissionResult::Accepted
+        );
+
+        let mut duplicate = request;
+        duplicate.id = "legacy-duplicate-id".to_string();
+        let (tx, rx) = oneshot::channel();
+        processor
+            .process_command(HealChannelCommand::Start {
+                request: duplicate,
+                response_tx: tx,
+            })
+            .await
+            .expect("legacy duplicate should be processed");
+        assert_eq!(
+            rx.await
+                .expect("legacy duplicate response should arrive")
+                .expect("legacy duplicate should succeed"),
             HealAdmissionResult::Merged
         );
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let response = responses.recv().await.expect("legacy broadcast should stay open");
+                if response.request_id == "legacy-duplicate-id" {
+                    break;
+                }
+            }
+        })
+        .await
+        .expect("legacy broadcast must retain the submitted request id");
+        assert_eq!(
+            manager
+                .get_task_status_for_path("bucket/object", "legacy-duplicate-id")
+                .await
+                .expect("legacy alias should resolve"),
+            HealTaskStatus::Pending
+        );
+        manager
+            .cancel_task("legacy-duplicate-id")
+            .await
+            .expect("legacy alias should cancel the canonical task");
+    }
+
+    #[tokio::test]
+    async fn test_public_legacy_start_processes_commands_with_receipt_channel_closed() {
+        let manager = create_test_heal_manager();
+        let mut processor = HealChannelProcessor::new(manager);
+        let (command_tx, command_rx) = mpsc::unbounded_channel();
+        let processor_task = tokio::spawn(async move { processor.start(command_rx).await });
+        let (response_tx, response_rx) = oneshot::channel();
+        command_tx
+            .send(HealChannelCommand::Start {
+                request: HealChannelRequest {
+                    id: "legacy-start-loop-id".to_string(),
+                    bucket: "bucket".to_string(),
+                    object_prefix: Some("object".to_string()),
+                    source: HealRequestSource::Admin,
+                    ..Default::default()
+                },
+                response_tx,
+            })
+            .expect("legacy command should send");
+
+        assert_eq!(
+            tokio::time::timeout(Duration::from_secs(1), response_rx)
+                .await
+                .expect("legacy processor should not starve")
+                .expect("legacy response should arrive")
+                .expect("legacy start should succeed"),
+            HealAdmissionResult::Accepted
+        );
+        drop(command_tx);
+        tokio::time::timeout(Duration::from_secs(1), processor_task)
+            .await
+            .expect("legacy processor should stop when command channel closes")
+            .expect("legacy processor task should join")
+            .expect("legacy processor should stop cleanly");
+    }
+
+    #[tokio::test]
+    async fn test_receipt_channel_returns_canonical_id_end_to_end() {
+        let manager = create_test_heal_manager();
+        let mut processor = HealChannelProcessor::new(manager);
+        let (command_tx, command_rx) = mpsc::unbounded_channel();
+        let (receipt_tx, receipt_rx) = mpsc::unbounded_channel();
+        let processor_task = tokio::spawn(async move { processor.start_with_receipts(command_rx, receipt_rx).await });
+        let request = HealChannelRequest {
+            id: "receipt-original-id".to_string(),
+            bucket: "bucket".to_string(),
+            object_prefix: Some("object".to_string()),
+            source: HealRequestSource::Admin,
+            ..Default::default()
+        };
+
+        let (response_tx, response_rx) = oneshot::channel();
+        receipt_tx
+            .send(HealReceiptCommand {
+                request: request.clone(),
+                response_tx,
+            })
+            .expect("receipt command should send");
+        let accepted = tokio::time::timeout(Duration::from_secs(1), response_rx)
+            .await
+            .expect("receipt processor should not starve")
+            .expect("receipt response should arrive")
+            .expect("receipt start should succeed");
+        assert_eq!(accepted.result, HealAdmissionResult::Accepted);
+        assert_eq!(accepted.task_id, "receipt-original-id");
+
+        let mut duplicate = request;
+        duplicate.id = "receipt-duplicate-id".to_string();
+        let (response_tx, response_rx) = oneshot::channel();
+        receipt_tx
+            .send(HealReceiptCommand {
+                request: duplicate,
+                response_tx,
+            })
+            .expect("duplicate receipt command should send");
+        let merged = tokio::time::timeout(Duration::from_secs(1), response_rx)
+            .await
+            .expect("duplicate receipt should not starve")
+            .expect("duplicate receipt response should arrive")
+            .expect("duplicate receipt should succeed");
+        assert_eq!(merged.result, HealAdmissionResult::Merged);
+        assert_eq!(merged.task_id, "receipt-original-id");
+
+        drop(receipt_tx);
+        drop(command_tx);
+        tokio::time::timeout(Duration::from_secs(1), processor_task)
+            .await
+            .expect("receipt processor should stop when channels close")
+            .expect("receipt processor task should join")
+            .expect("receipt processor should stop cleanly");
     }
 
     #[tokio::test]
@@ -1147,7 +1366,7 @@ mod tests {
 
         let (tx, rx) = oneshot::channel();
         processor
-            .process_start_request(request, tx)
+            .process_start_request(request, false, true, tx)
             .await
             .expect("processor should surface invalid request through response channel");
         assert!(rx.await.expect("oneshot should resolve").is_err());

crates/heal/src/heal/manager.rs

@@ -19,9 +19,11 @@ use crate::heal::{
 };
 use crate::{Error, Result};
 use metrics::{counter, gauge};
-use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult, HealRequestSource};
+use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionReceipt, HealAdmissionResult, HealRequestSource};
 use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
 use rustfs_madmin::heal_commands::HealResultItem;
+#[cfg(test)]
+use std::sync::LazyLock;
 use std::{
     collections::{BinaryHeap, HashMap, HashSet},
     sync::Arc,
@@ -52,6 +54,54 @@ const EVENT_HEAL_UNCLEAN_SHUTDOWN: &str = "heal_unclean_shutdown";
 const MAX_RECOVERABLE_HEAL_RETRIES: u32 = 3;
 const MAX_RECOVERABLE_HEAL_RETRY_DELAY: Duration = Duration::from_secs(30);
 
+#[cfg(test)]
+struct RetryOwnershipTestHook {
+    task_id: String,
+    active_to_retrying_reached: Notify,
+    active_to_retrying_release: Notify,
+    retrying_to_queue_reached: Notify,
+    retrying_to_queue_release: Notify,
+}
+
+#[cfg(test)]
+static RETRY_OWNERSHIP_TEST_HOOK: LazyLock<Mutex<Option<Arc<RetryOwnershipTestHook>>>> = LazyLock::new(|| Mutex::new(None));
+
+#[cfg(test)]
+struct DuplicateAdmissionTestHook {
+    request_id: String,
+    active_lock_reached: Notify,
+    active_lock_release: Notify,
+}
+
+#[cfg(test)]
+static DUPLICATE_ADMISSION_TEST_HOOK: LazyLock<Mutex<Option<Arc<DuplicateAdmissionTestHook>>>> =
+    LazyLock::new(|| Mutex::new(None));
+
+#[cfg(test)]
+async fn pause_retry_ownership_transition(task_id: &str, to_queue: bool) {
+    let hook = RETRY_OWNERSHIP_TEST_HOOK.lock().await.clone();
+    let Some(hook) = hook.filter(|hook| hook.task_id == task_id) else {
+        return;
+    };
+    if to_queue {
+        hook.retrying_to_queue_reached.notify_one();
+        hook.retrying_to_queue_release.notified().await;
+    } else {
+        hook.active_to_retrying_reached.notify_one();
+        hook.active_to_retrying_release.notified().await;
+    }
+}
+
+#[cfg(test)]
+async fn pause_duplicate_admission_after_active_lock(request_id: &str) {
+    let hook = DUPLICATE_ADMISSION_TEST_HOOK.lock().await.clone();
+    let Some(hook) = hook.filter(|hook| hook.request_id == request_id) else {
+        return;
+    };
+    hook.active_lock_reached.notify_one();
+    hook.active_lock_release.notified().await;
+}
+
 type WorkloadSnapshotProviderRef = Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
 
 /// Priority queue wrapper for heal requests
@@ -151,8 +201,8 @@ pub struct HealTaskReport {
     pub progress: Option<HealProgress>,
 }
 
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
-#[serde(rename_all = "camelCase")]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct HealPriorityCounts {
     pub low: u64,
     pub normal: u64,
@@ -171,8 +221,8 @@ impl HealPriorityCounts {
     }
 }
 
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
-#[serde(rename_all = "camelCase")]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct HealSourceCounts {
     pub scanner: u64,
     pub admin: u64,
@@ -193,15 +243,18 @@ impl HealSourceCounts {
     }
 }
 
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
-#[serde(rename_all = "camelCase")]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct HealOperationsSnapshot {
     pub queue_length: u64,
     pub active_tasks: u64,
+    pub retrying_tasks: u64,
     pub queued_by_priority: HealPriorityCounts,
     pub active_by_priority: HealPriorityCounts,
+    pub retrying_by_priority: HealPriorityCounts,
     pub queued_by_source: HealSourceCounts,
     pub active_by_source: HealSourceCounts,
+    pub retrying_by_source: HealSourceCounts,
 }
 
 fn usize_to_u64_saturated(value: usize) -> u64 {
@@ -1391,110 +1444,47 @@ impl HealManager {
     }
 
     /// Submit heal request
-    pub async fn submit_heal_request(&self, request: HealRequest) -> Result<HealAdmissionResult> {
+    pub async fn submit_heal_request_with_receipt(&self, request: HealRequest) -> Result<HealAdmissionReceipt> {
+        self.submit_heal_request_with_receipt_and_alias(request, false).await
+    }
+
+    pub(crate) async fn submit_heal_request_with_receipt_and_alias(
+        &self,
+        request: HealRequest,
+        preserve_alias: bool,
+    ) -> Result<HealAdmissionReceipt> {
         let config = self.config.read().await;
         let dedup_key = PriorityHealQueue::make_dedup_key(&request);
 
-        // Keep this lock order aligned with the scheduler so a request cannot
-        // slip between queued and active states while duplicate admission runs.
-        let active_duplicate = {
-            let active_heals = self.active_heals.lock().await;
-            (!request.force_start)
-                .then(|| active_heal_for_dedup_key(&active_heals, &dedup_key))
-                .flatten()
-        };
-        if let Some((merged_task_id, _)) = active_duplicate {
-            let admission = Self::duplicate_admission_for_request(&request, &config);
-
-            match admission {
-                HealAdmissionResult::Merged => {
-                    self.insert_task_alias(&request.id, &merged_task_id).await;
-                    info!(
-                        target: "rustfs::heal::manager",
-                        event = EVENT_HEAL_QUEUE_ADMISSION,
-                        component = LOG_COMPONENT_HEAL,
-                        subsystem = LOG_SUBSYSTEM_MANAGER,
-                        request_id = %request.id,
-                        merged_task_id = %merged_task_id,
-                        priority = ?request.priority,
-                        result = "merged_active_duplicate",
-                        "Heal queue admission decided"
-                    );
-                }
-                HealAdmissionResult::Dropped(reason) => {
-                    warn!(
-                        target: "rustfs::heal::manager",
-                        event = EVENT_HEAL_QUEUE_ADMISSION,
-                        component = LOG_COMPONENT_HEAL,
-                        subsystem = LOG_SUBSYSTEM_MANAGER,
-                        request_id = %request.id,
-                        priority = ?request.priority,
-                        reason = reason.as_str(),
-                        result = "dropped_active_duplicate",
-                        "Heal queue admission decided"
-                    );
-                }
-                HealAdmissionResult::Accepted | HealAdmissionResult::Full => {}
-            }
-
-            return Ok(admission);
-        }
-
-        let retrying_duplicate = {
-            let retrying_heals = self.retrying_heals.lock().await;
-            (!request.force_start)
-                .then(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key))
-                .flatten()
-        };
-        if let Some((merged_task_id, _)) = retrying_duplicate {
-            let admission = Self::duplicate_admission_for_request(&request, &config);
-
-            match admission {
-                HealAdmissionResult::Merged => {
-                    self.insert_task_alias(&request.id, &merged_task_id).await;
-                    info!(
-                        target: "rustfs::heal::manager",
-                        event = EVENT_HEAL_QUEUE_ADMISSION,
-                        component = LOG_COMPONENT_HEAL,
-                        subsystem = LOG_SUBSYSTEM_MANAGER,
-                        request_id = %request.id,
-                        merged_task_id = %merged_task_id,
-                        priority = ?request.priority,
-                        result = "merged_retrying_duplicate",
-                        "Heal queue admission decided"
-                    );
-                }
-                HealAdmissionResult::Dropped(reason) => {
-                    warn!(
-                        target: "rustfs::heal::manager",
-                        event = EVENT_HEAL_QUEUE_ADMISSION,
-                        component = LOG_COMPONENT_HEAL,
-                        subsystem = LOG_SUBSYSTEM_MANAGER,
-                        request_id = %request.id,
-                        priority = ?request.priority,
-                        reason = reason.as_str(),
-                        result = "dropped_retrying_duplicate",
-                        "Heal queue admission decided"
-                    );
-                }
-                HealAdmissionResult::Accepted | HealAdmissionResult::Full => {}
-            }
-
-            return Ok(admission);
-        }
-
+        // Match the scheduler's active -> queue order and keep retry ownership
+        // in the same atomic view. Otherwise queue -> active and
+        // active -> retrying transitions can slip between duplicate checks.
+        let active_heals = self.active_heals.lock().await;
+        #[cfg(test)]
+        pause_duplicate_admission_after_active_lock(&request.id).await;
         let mut queue = self.heal_queue.lock().await;
-
-        let queued_duplicate = (!request.force_start)
-            .then(|| queue.request_for_dedup_key(&dedup_key).map(|queued| queued.id.clone()))
-            .flatten();
-        if let Some(merged_task_id) = queued_duplicate {
+        let retrying_heals = self.retrying_heals.lock().await;
+        let duplicate = (!request.force_start).then(|| {
+            active_heal_for_dedup_key(&active_heals, &dedup_key)
+                .map(|(task_id, _)| (task_id, "active"))
+                .or_else(|| {
+                    queue
+                        .request_for_dedup_key(&dedup_key)
+                        .map(|queued| (queued.id.clone(), "queued"))
+                })
+                .or_else(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| (task_id, "retrying")))
+        });
+        if let Some((merged_task_id, duplicate_state)) = duplicate.flatten() {
             let admission = Self::duplicate_admission_for_request(&request, &config);
+            drop(retrying_heals);
             drop(queue);
+            drop(active_heals);
 
             match admission {
                 HealAdmissionResult::Merged => {
-                    self.insert_task_alias(&request.id, &merged_task_id).await;
+                    if preserve_alias {
+                        self.insert_task_alias(&request.id, &merged_task_id).await;
+                    }
                     debug!(
                         target: "rustfs::heal::manager",
                         event = EVENT_HEAL_QUEUE_ADMISSION,
@@ -1503,8 +1493,9 @@ impl HealManager {
                         request_id = %request.id,
                         merged_task_id = %merged_task_id,
                         priority = ?request.priority,
+                        duplicate_state,
                         result = "merged_duplicate",
-                        "Heal queue request merged"
+                        "Heal queue admission decided"
                     );
                 }
                 HealAdmissionResult::Dropped(reason) => {
@@ -1516,25 +1507,45 @@ impl HealManager {
                         request_id = %request.id,
                         priority = ?request.priority,
                         reason = reason.as_str(),
+                        duplicate_state,
                         result = "dropped_duplicate",
-                        "Heal queue request dropped"
+                        "Heal queue admission decided"
                     );
                 }
                 HealAdmissionResult::Accepted | HealAdmissionResult::Full => {}
             }
 
-            return Ok(admission);
+            return Ok(HealAdmissionReceipt {
+                result: admission,
+                task_id: merged_task_id,
+            });
         }
 
+        let mut task_id = request.id.clone();
         let admission = Self::admit_request_to_queue(&mut queue, request, &config, "submit");
+        if admission == HealAdmissionResult::Merged
+            && let Some(queued) = queue.request_for_dedup_key(&dedup_key)
+        {
+            task_id.clone_from(&queued.id);
+        }
         let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
+        drop(retrying_heals);
         drop(queue);
+        drop(active_heals);
 
         if should_notify {
             self.notify.notify_one();
         }
 
-        Ok(admission)
+        Ok(HealAdmissionReceipt {
+            result: admission,
+            task_id,
+        })
+    }
+
+    /// Submit heal request.
+    pub async fn submit_heal_request(&self, request: HealRequest) -> Result<HealAdmissionResult> {
+        Ok(self.submit_heal_request_with_receipt_and_alias(request, true).await?.result)
     }
 
     /// Get task status
@@ -2012,43 +2023,51 @@ impl HealManager {
     }
 
     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)
-        };
+        // Match the scheduler's active -> queue order. Retry transitions also
+        // acquire active before retrying, so the three sets form one snapshot.
+        let active_heals = self.active_heals.lock().await;
+        let queue = self.heal_queue.lock().await;
+        let retrying_heals = self.retrying_heals.lock().await;
+        let (queued_by_priority, queued_by_source) = queue.operation_counts();
+        let mut active_by_priority = HealPriorityCounts::default();
+        let mut active_by_source = HealSourceCounts::default();
+        for task in active_heals.values() {
+            active_by_priority.increment(task.priority);
+            active_by_source.increment(task.source);
+        }
+        let mut retrying_by_priority = HealPriorityCounts::default();
+        let mut retrying_by_source = HealSourceCounts::default();
+        for retrying in retrying_heals.values() {
+            retrying_by_priority.increment(retrying.request.priority);
+            retrying_by_source.increment(retrying.request.source);
+        }
+        publish_active_heal_count(&active_heals);
+        publish_heal_queue_length(&queue);
 
         HealOperationsSnapshot {
-            queue_length,
-            active_tasks,
+            queue_length: usize_to_u64_saturated(queue.len()),
+            active_tasks: usize_to_u64_saturated(active_heals.len()),
+            retrying_tasks: usize_to_u64_saturated(retrying_heals.len()),
             queued_by_priority,
             active_by_priority,
+            retrying_by_priority,
             queued_by_source,
             active_by_source,
+            retrying_by_source,
         }
     }
 
     pub async fn active_progress_snapshot(&self) -> Option<HealProgress> {
-        let active_heals = self.active_heals.lock().await;
-        if active_heals.is_empty() {
+        let active_tasks = {
+            let active_heals = self.active_heals.lock().await;
+            active_heals.values().cloned().collect::<Vec<_>>()
+        };
+        if active_tasks.is_empty() {
             return None;
         }
 
         let mut snapshot = HealProgress::default();
-        for task in active_heals.values() {
+        for task in active_tasks {
             let progress = task.get_progress().await;
             snapshot.objects_scanned = snapshot.objects_scanned.saturating_add(progress.objects_scanned);
             snapshot.objects_healed = snapshot.objects_healed.saturating_add(progress.objects_healed);
@@ -2556,10 +2575,40 @@ impl HealManager {
                         retry_attempt: request.retry_attempts,
                     });
                     let retry_request_for_queue = retry_request;
+                    let retry_cancel_token = retry_request_for_queue.as_ref().map(|_| CancellationToken::new());
                     let mut active_heals_guard = active_heals_clone.lock().await;
-                    if let Some(completed_task) = active_heals_guard.remove(&task_id) {
+                    // Keep retry ownership continuous: status snapshots acquire
+                    // these locks in the same active -> retrying order.
+                    let mut retrying_heals_guard = if let (Some((request, _, error)), Some(cancel_token)) =
+                        (retry_request_for_queue.as_ref(), retry_cancel_token.as_ref())
+                    {
+                        let mut retrying = retrying_heals_clone.lock().await;
+                        if active_heals_guard.contains_key(&task_id) {
+                            retrying.insert(
+                                request.id.clone(),
+                                RetryingHeal {
+                                    request: request.clone(),
+                                    error: error.clone(),
+                                    cancel_token: cancel_token.clone(),
+                                },
+                            );
+                            #[cfg(test)]
+                            pause_retry_ownership_transition(&task_id, false).await;
+                        }
+                        Some(retrying)
+                    } else {
+                        None
+                    };
+                    let completed_task = active_heals_guard.remove(&task_id);
+                    if let Some(completed_task) = completed_task.as_ref() {
                         publish_active_heal_count(&active_heals_guard);
                         update_task_running_metric_for_task(&active_heals_guard, completed_task.as_ref());
+                    }
+                    let active_count = active_heals_guard.len();
+                    drop(retrying_heals_guard.take());
+                    drop(active_heals_guard);
+
+                    if let Some(completed_task) = completed_task {
                         let completed_status = if let Some(status) = retry_request_for_status {
                             status
                         } else {
@@ -2585,10 +2634,12 @@ impl HealManager {
                                 stats.update_task_completion(false);
                             }
                         }
-                        stats.update_running_tasks(active_heals_guard.len() as u64);
+                        stats.update_running_tasks(usize_to_u64_saturated(active_count));
                     }
 
-                    if let Some((retry_request, retry_delay, retry_error)) = retry_request_for_queue {
+                    if let (Some((retry_request, retry_delay, retry_error)), Some(retry_cancel_token)) =
+                        (retry_request_for_queue, retry_cancel_token)
+                    {
                         let retry_request_id = retry_request.id.clone();
                         let retry_attempt = retry_request.retry_attempts;
                         let retry_key = PriorityHealQueue::make_dedup_key(&retry_request);
@@ -2598,17 +2649,8 @@ impl HealManager {
                         let retrying_heals_for_spawn = retrying_heals_clone.clone();
                         let retry_completed_heals = completed_heals_clone.clone();
                         let retry_notify = notify_clone.clone();
-                        let retry_cancel_token = CancellationToken::new();
                         let retry_manager_cancel_token = manager_cancel_token.clone();
                         let retry_config = config_for_spawn.clone();
-                        retrying_heals_clone.lock().await.insert(
-                            retry_request_id.clone(),
-                            RetryingHeal {
-                                request: retry_request.clone(),
-                                error: retry_error.clone(),
-                                cancel_token: retry_cancel_token.clone(),
-                            },
-                        );
                         tokio::spawn(async move {
                             loop {
                                 tokio::select! {
@@ -2667,8 +2709,12 @@ impl HealManager {
                                     && retry_config.event_driven_scheduler_enable;
                                 match admission {
                                     HealAdmissionResult::Accepted => {
-                                        drop(queue);
+                                        // Transfer ownership while holding queue -> retrying,
+                                        // matching operations_snapshot's lock order.
+                                        #[cfg(test)]
+                                        pause_retry_ownership_transition(&retry_request_id, true).await;
                                         retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
+                                        drop(queue);
                                         retry_completed_heals.lock().await.remove(&retry_request_id);
                                         info!(
                                             target: "rustfs::heal::manager",
@@ -2689,8 +2735,8 @@ impl HealManager {
                                         return;
                                     }
                                     HealAdmissionResult::Merged => {
-                                        drop(queue);
                                         retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
+                                        drop(queue);
                                         info!(
                                             target: "rustfs::heal::manager",
                                             event = EVENT_HEAL_QUEUE_ADMISSION,
@@ -2963,8 +3009,8 @@ mod tests {
             Ok(Vec::new())
         }
 
-        async fn object_exists(&self, _bucket: &str, _object: &str) -> Result<bool> {
-            Ok(false)
+        async fn object_exists(&self, bucket: &str, _object: &str) -> Result<bool> {
+            Ok(bucket == "retry-transition")
         }
 
         async fn get_object_size(&self, _bucket: &str, _object: &str) -> Result<Option<u64>> {
@@ -2977,11 +3023,20 @@ mod tests {
 
         async fn heal_object(
             &self,
-            _bucket: &str,
+            bucket: &str,
             _object: &str,
             _version_id: Option<&str>,
             _opts: &HealOpts,
         ) -> Result<(HealResultItem, Option<Error>)> {
+            if bucket == "retry-transition" {
+                return Ok((
+                    HealResultItem::default(),
+                    Some(Error::Storage(EcstoreError::InsufficientReadQuorum(
+                        bucket.to_string(),
+                        "object".to_string(),
+                    ))),
+                ));
+            }
             Ok((HealResultItem::default(), None))
         }
 
@@ -3478,6 +3533,103 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_admin_duplicate_receipt_returns_canonical_task_without_alias() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+        let mut original = HealRequest::object("bucket".to_string(), "object".to_string(), None);
+        original.source = HealRequestSource::Admin;
+        let original_id = original.id.clone();
+        let mut duplicate = HealRequest::object("bucket".to_string(), "object".to_string(), None);
+        duplicate.source = HealRequestSource::Admin;
+        let duplicate_id = duplicate.id.clone();
+
+        let accepted = manager
+            .submit_heal_request_with_receipt(original)
+            .await
+            .expect("first request should be accepted");
+        let merged = manager
+            .submit_heal_request_with_receipt(duplicate)
+            .await
+            .expect("duplicate request should merge");
+
+        assert_eq!(accepted.result, HealAdmissionResult::Accepted);
+        assert_eq!(accepted.task_id, original_id);
+        assert_eq!(merged.result, HealAdmissionResult::Merged);
+        assert_eq!(merged.task_id, original_id);
+        assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
+    }
+
+    #[tokio::test]
+    async fn test_duplicate_admission_is_atomic_with_queue_to_active_transition() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = Arc::new(HealManager::new(storage.clone(), None));
+        let mut original = HealRequest::object("bucket".to_string(), "object".to_string(), None);
+        original.source = HealRequestSource::Admin;
+        let original_id = original.id.clone();
+        assert_eq!(
+            manager
+                .submit_heal_request(original)
+                .await
+                .expect("original request should be accepted"),
+            HealAdmissionResult::Accepted
+        );
+
+        let mut duplicate = HealRequest::object("bucket".to_string(), "object".to_string(), None);
+        duplicate.source = HealRequestSource::Admin;
+        let hook = Arc::new(DuplicateAdmissionTestHook {
+            request_id: duplicate.id.clone(),
+            active_lock_reached: Notify::new(),
+            active_lock_release: Notify::new(),
+        });
+        *DUPLICATE_ADMISSION_TEST_HOOK.lock().await = Some(hook.clone());
+
+        let duplicate_manager = manager.clone();
+        let duplicate_task = tokio::spawn(async move { duplicate_manager.submit_heal_request_with_receipt(duplicate).await });
+        tokio::time::timeout(Duration::from_secs(1), hook.active_lock_reached.notified())
+            .await
+            .expect("duplicate admission should reach the active lock hook");
+
+        let transition_manager = manager.clone();
+        let transition_storage = storage.clone();
+        let (attempting_active_tx, attempting_active_rx) = tokio::sync::oneshot::channel();
+        let (active_acquired_tx, mut active_acquired_rx) = tokio::sync::oneshot::channel();
+        let transition = tokio::spawn(async move {
+            let _ = attempting_active_tx.send(());
+            let mut active = transition_manager.active_heals.lock().await;
+            let _ = active_acquired_tx.send(());
+            let mut queue = transition_manager.heal_queue.lock().await;
+            let request = queue.pop_next().expect("original request should remain queued");
+            let task = Arc::new(HealTask::from_request(request, transition_storage));
+            active.insert(task.id.clone(), task);
+        });
+        tokio::time::timeout(Duration::from_secs(1), attempting_active_rx)
+            .await
+            .expect("transition should attempt the active lock")
+            .expect("transition attempt signal should be delivered");
+        assert!(matches!(
+            active_acquired_rx.try_recv(),
+            Err(tokio::sync::oneshot::error::TryRecvError::Empty)
+        ));
+
+        hook.active_lock_release.notify_one();
+        let receipt = tokio::time::timeout(Duration::from_secs(1), duplicate_task)
+            .await
+            .expect("duplicate admission should not hang")
+            .expect("duplicate task should join")
+            .expect("duplicate admission should succeed");
+        tokio::time::timeout(Duration::from_secs(1), transition)
+            .await
+            .expect("queue to active transition should not hang")
+            .expect("queue to active transition should join");
+        *DUPLICATE_ADMISSION_TEST_HOOK.lock().await = None;
+
+        assert_eq!(receipt.result, HealAdmissionResult::Merged);
+        assert_eq!(receipt.task_id, original_id);
+        assert_eq!(manager.get_queue_length().await, 0);
+        assert_eq!(manager.get_active_task_count().await, 1);
+    }
+
     #[tokio::test]
     async fn test_submit_heal_request_returns_merged_for_active_duplicate() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -3988,6 +4140,105 @@ mod tests {
         assert_eq!(snapshot.active_by_source.scanner, 0);
     }
 
+    #[tokio::test]
+    async fn test_operations_snapshot_counts_retry_backoff_as_owned_work() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new(storage, None);
+        let mut request = HealRequest::bucket("bucket-retry".to_string());
+        request.priority = HealPriority::Urgent;
+        request.source = HealRequestSource::Admin;
+        let task_id = request.id.clone();
+        manager.retrying_heals.lock().await.insert(
+            task_id,
+            RetryingHeal {
+                request,
+                error: "transient".to_string(),
+                cancel_token: CancellationToken::new(),
+            },
+        );
+
+        let snapshot = manager.operations_snapshot().await;
+
+        assert_eq!(snapshot.queue_length, 0);
+        assert_eq!(snapshot.active_tasks, 0);
+        assert_eq!(snapshot.retrying_tasks, 1);
+        assert_eq!(snapshot.retrying_by_priority.urgent, 1);
+        assert_eq!(snapshot.retrying_by_source.admin, 1);
+    }
+
+    #[tokio::test]
+    async fn test_scheduler_retry_transitions_keep_continuous_single_ownership() {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = Arc::new(HealManager::new(storage, None));
+        {
+            let mut config = manager.config.write().await;
+            config.enable_auto_heal = false;
+            config.heal_interval = Duration::from_millis(10);
+            config.event_driven_scheduler_enable = true;
+        }
+        manager.start().await.expect("manager should start");
+
+        let request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
+        let task_id = request.id.clone();
+        let hook = Arc::new(RetryOwnershipTestHook {
+            task_id: task_id.clone(),
+            active_to_retrying_reached: Notify::new(),
+            active_to_retrying_release: Notify::new(),
+            retrying_to_queue_reached: Notify::new(),
+            retrying_to_queue_release: Notify::new(),
+        });
+        *RETRY_OWNERSHIP_TEST_HOOK.lock().await = Some(hook.clone());
+
+        manager
+            .submit_heal_request(request)
+            .await
+            .expect("retry test request should be accepted");
+        tokio::time::timeout(Duration::from_secs(2), hook.active_to_retrying_reached.notified())
+            .await
+            .expect("active to retrying transition should be reached");
+
+        let snapshot_manager = manager.clone();
+        let mut snapshot = tokio::spawn(async move { snapshot_manager.operations_snapshot().await });
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), &mut snapshot).await.is_err(),
+            "snapshot must wait while active and retrying ownership locks are held"
+        );
+        hook.active_to_retrying_release.notify_one();
+        let snapshot = tokio::time::timeout(Duration::from_secs(1), snapshot)
+            .await
+            .expect("snapshot should resume after active to retrying handoff")
+            .expect("snapshot task should complete");
+        assert_eq!(snapshot.active_tasks + snapshot.queue_length + snapshot.retrying_tasks, 1);
+        assert_eq!(snapshot.retrying_tasks, 1);
+
+        tokio::time::timeout(Duration::from_secs(5), hook.retrying_to_queue_reached.notified())
+            .await
+            .expect("retrying to queue transition should be reached");
+        let snapshot_manager = manager.clone();
+        let mut snapshot = tokio::spawn(async move { snapshot_manager.operations_snapshot().await });
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), &mut snapshot).await.is_err(),
+            "snapshot must wait while queue ownership is transferred"
+        );
+        hook.retrying_to_queue_release.notify_one();
+        // The scheduler may immediately execute the retried request again;
+        // leave a permit so a second test-only active handoff cannot stall it.
+        hook.active_to_retrying_release.notify_one();
+        let snapshot = tokio::time::timeout(Duration::from_secs(1), snapshot)
+            .await
+            .expect("snapshot should resume after retrying to queue handoff")
+            .expect("snapshot task should complete");
+        assert_eq!(snapshot.active_tasks + snapshot.queue_length + snapshot.retrying_tasks, 1);
+
+        *RETRY_OWNERSHIP_TEST_HOOK.lock().await = None;
+        hook.active_to_retrying_release.notify_one();
+        hook.retrying_to_queue_release.notify_one();
+        tokio::time::timeout(Duration::from_secs(1), manager.stop())
+            .await
+            .expect("manager stop should not stall")
+            .expect("manager should stop");
+    }
+
     #[tokio::test]
     async fn test_active_progress_snapshot_sums_active_task_progress() {
         let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/lib.rs

@@ -156,10 +156,10 @@ pub async fn init_heal_manager_with_workload_provider(
         let channel_receiver = if force_channel_failure {
             Err("forced heal channel initialization failure")
         } else {
-            rustfs_common::heal_channel::init_heal_channel()
+            rustfs_common::heal_channel::init_heal_channels()
         };
-        let receiver = match channel_receiver {
-            Ok(receiver) => receiver,
+        let (receiver, receipt_receiver) = match channel_receiver {
+            Ok(receivers) => receivers,
             Err(err) => {
                 stop_initializing_manager(&heal_manager).await?;
                 return Err(Error::Config(err.to_string()));
@@ -175,7 +175,7 @@ pub async fn init_heal_manager_with_workload_provider(
 
         tokio::spawn(async move {
             let mut processor = channel_processor.lock().await;
-            if let Err(e) = processor.start(receiver).await {
+            if let Err(e) = processor.start_with_receipts(receiver, receipt_receiver).await {
                 error!(
                     target: "rustfs::heal",
                     event = EVENT_HEAL_RUNTIME_STATE,

docs/architecture/compat-cleanup-register.md

@@ -14,6 +14,7 @@ for later deletion.
 
 - `#4648` walk-dir stream completion capability: old clients can append fallback output to an already-used metacache writer after a terminal body error, so servers emit terminal walk errors only to clients that sign the `walk_dir_stream_completion=error-v1` query capability and its request-body digest. Remove the legacy clean-EOF path after the minimum supported RustFS peer version always advertises this capability.
 - `heal-rpc-auth-v2` internode gRPC authentication: servers temporarily accept legacy prefix signatures so old peers remain available during rolling upgrades. Remove the legacy fallback after the minimum supported RustFS peer version sends v2 authentication on every internode gRPC request.
+- `heal-status-rpc-v1` node heal status capability: new peers treat an unimplemented BackgroundHealStatus RPC as an explicitly incomplete rolling-upgrade response. Remove the fallback after the minimum supported RustFS peer version implements BackgroundHealStatus.
 
 ## Review Checklist
 

rustfs/src/admin/handlers/heal.rs

@@ -14,19 +14,22 @@
 
 use crate::admin::auth::{authenticate_request, validate_admin_request};
 use crate::admin::router::{AdminOperation, Operation, S3Router};
-use crate::admin::runtime_sources::{object_store_from_extensions, object_store_from_req};
+use crate::admin::runtime_sources::{app_context_from_req, object_store_from_extensions};
 use crate::admin::storage_api::access::spawn_traced;
 use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket;
 use crate::admin::storage_api::bucket::utils::is_valid_object_prefix;
 use crate::admin::storage_api::contract::heal::HealOperations as _;
 use crate::server::ADMIN_PREFIX;
 use crate::server::RemoteAddr;
-use crate::startup_background::{heal_enabled_from_env, scanner_enabled_from_env};
+use crate::storage::rpc::node_service::heal::{
+    NodeHealProgress, NodeHealStatusSnapshot, capture_node_heal_status, decode_node_heal_status,
+};
 use bytes::Bytes;
+use futures_util::future::join_all;
 use http::{HeaderMap, HeaderValue, Uri};
 use hyper::{Method, StatusCode};
 use matchit::Params;
-use rustfs_common::heal_channel::{HealChannelPriority, HealChannelRequest, HealOpts, HealRequestSource};
+use rustfs_common::heal_channel::{HealChannelPriority, HealChannelRequest, HealOpts, HealRequestSource, HealScanMode};
 use rustfs_config::MAX_HEAL_REQUEST_SIZE;
 use rustfs_heal::heal::utils::format_set_disk_id;
 use rustfs_policy::policy::action::{Action, AdminAction};
@@ -39,6 +42,7 @@ use std::collections::HashSet;
 use std::path::PathBuf;
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 use tokio::sync::mpsc;
+use tokio::time::{Duration, timeout};
 use tracing::{info, warn};
 
 const LOG_COMPONENT_ADMIN_API: &str = "admin_api";
@@ -46,6 +50,7 @@ const LOG_SUBSYSTEM_HEAL_ADMIN: &str = "heal_admin";
 const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected";
 const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed";
 const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted";
+const PEER_HEAL_STATUS_TIMEOUT: Duration = Duration::from_secs(5);
 
 #[derive(Debug, Default, Serialize, Deserialize)]
 struct HealInitParams {
@@ -208,6 +213,7 @@ struct BackgroundHealStatus<'a> {
     heal_queue_length: u64,
     heal_active_tasks: u64,
     heal_operations: rustfs_heal::HealOperationsSnapshot,
+    cluster_status_complete: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     progress: Option<BackgroundHealProgress>,
 }
@@ -221,6 +227,7 @@ enum HealRuntimeState {
     Active,
 }
 
+#[cfg(test)]
 fn background_heal_runtime_state(
     services_enabled: bool,
     initialized: bool,
@@ -232,7 +239,7 @@ fn background_heal_runtime_state(
     if !initialized {
         return HealRuntimeState::Uninitialized;
     }
-    if operations.queue_length > 0 || operations.active_tasks > 0 {
+    if operations.queue_length > 0 || operations.active_tasks > 0 || operations.retrying_tasks > 0 {
         HealRuntimeState::Active
     } else {
         HealRuntimeState::Idle
@@ -248,15 +255,228 @@ struct BackgroundHealProgress {
     bytes_processed: u64,
 }
 
-impl From<rustfs_heal::HealProgress> for BackgroundHealProgress {
-    fn from(progress: rustfs_heal::HealProgress) -> Self {
-        Self {
-            objects_scanned: progress.objects_scanned,
-            objects_healed: progress.objects_healed,
-            objects_failed: progress.objects_failed,
-            bytes_processed: progress.bytes_processed,
+#[derive(Debug)]
+struct ClusterHealStatusSnapshot {
+    info: BackgroundHealInfo,
+    state: HealRuntimeState,
+    operations: rustfs_heal::HealOperationsSnapshot,
+    progress: Option<BackgroundHealProgress>,
+    complete: bool,
+}
+
+fn add_priority_counts(total: &mut rustfs_heal::HealPriorityCounts, next: rustfs_heal::HealPriorityCounts) {
+    total.low = total.low.saturating_add(next.low);
+    total.normal = total.normal.saturating_add(next.normal);
+    total.high = total.high.saturating_add(next.high);
+    total.urgent = total.urgent.saturating_add(next.urgent);
+}
+
+fn add_source_counts(total: &mut rustfs_heal::HealSourceCounts, next: rustfs_heal::HealSourceCounts) {
+    total.scanner = total.scanner.saturating_add(next.scanner);
+    total.admin = total.admin.saturating_add(next.admin);
+    total.auto_heal = total.auto_heal.saturating_add(next.auto_heal);
+    total.internal = total.internal.saturating_add(next.internal);
+    total.read_repair = total.read_repair.saturating_add(next.read_repair);
+}
+
+fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_heal::HealOperationsSnapshot) {
+    total.queue_length = total.queue_length.saturating_add(next.queue_length);
+    total.active_tasks = total.active_tasks.saturating_add(next.active_tasks);
+    total.retrying_tasks = total.retrying_tasks.saturating_add(next.retrying_tasks);
+    add_priority_counts(&mut total.queued_by_priority, next.queued_by_priority);
+    add_priority_counts(&mut total.active_by_priority, next.active_by_priority);
+    add_priority_counts(&mut total.retrying_by_priority, next.retrying_by_priority);
+    add_source_counts(&mut total.queued_by_source, next.queued_by_source);
+    add_source_counts(&mut total.active_by_source, next.active_by_source);
+    add_source_counts(&mut total.retrying_by_source, next.retrying_by_source);
+}
+
+fn add_progress(total: &mut BackgroundHealProgress, next: NodeHealProgress) {
+    total.objects_scanned = total.objects_scanned.saturating_add(next.objects_scanned);
+    total.objects_healed = total.objects_healed.saturating_add(next.objects_healed);
+    total.objects_failed = total.objects_failed.saturating_add(next.objects_failed);
+    total.bytes_processed = total.bytes_processed.saturating_add(next.bytes_processed);
+}
+
+fn aggregate_cluster_heal_status(snapshots: Vec<NodeHealStatusSnapshot>) -> ClusterHealStatusSnapshot {
+    let mut info = BackgroundHealInfo::default();
+    let mut operations = rustfs_heal::HealOperationsSnapshot::default();
+    let mut progress = None;
+    let mut any_services_enabled = false;
+    let mut any_initialized = false;
+
+    for snapshot in snapshots {
+        let snapshot_info = snapshot.info();
+        any_services_enabled |= snapshot.services_enabled;
+        any_initialized |= snapshot.initialized;
+        info.bitrot_start_cycle = info.bitrot_start_cycle.max(snapshot_info.bitrot_start_cycle);
+        info.bitrot_start_time = info.bitrot_start_time.max(snapshot_info.bitrot_start_time);
+        if snapshot_info.current_scan_mode == HealScanMode::Deep
+            || (snapshot_info.current_scan_mode == HealScanMode::Normal && info.current_scan_mode == HealScanMode::Unknown)
+        {
+            info.current_scan_mode = snapshot_info.current_scan_mode;
+        }
+        add_operations(&mut operations, snapshot.operations);
+        if let Some(next) = snapshot.progress {
+            add_progress(
+                progress.get_or_insert(BackgroundHealProgress {
+                    objects_scanned: 0,
+                    objects_healed: 0,
+                    objects_failed: 0,
+                    bytes_processed: 0,
+                }),
+                next,
+            );
         }
     }
+
+    let state = if operations.queue_length > 0 || operations.active_tasks > 0 || operations.retrying_tasks > 0 {
+        HealRuntimeState::Active
+    } else if any_initialized {
+        HealRuntimeState::Idle
+    } else if any_services_enabled {
+        HealRuntimeState::Uninitialized
+    } else {
+        HealRuntimeState::Disabled
+    };
+
+    ClusterHealStatusSnapshot {
+        info,
+        state,
+        operations,
+        progress,
+        complete: true,
+    }
+}
+
+fn cluster_heal_status_unavailable(reason: &str) -> s3s::S3Error {
+    warn!(
+        event = EVENT_ADMIN_REQUEST_FAILED,
+        component = LOG_COMPONENT_ADMIN_API,
+        subsystem = LOG_SUBSYSTEM_HEAL_ADMIN,
+        operation = "background_heal_status",
+        result = "failed",
+        reason = reason,
+        "cluster heal status unavailable"
+    );
+    s3_error!(InternalError, "cluster heal status unavailable")
+}
+
+fn peer_topology_complete(
+    expected_nodes: usize,
+    remote_slots: usize,
+    unavailable_remote_slots: usize,
+    all_slots: usize,
+    available_all_slots: usize,
+) -> bool {
+    let expected_peers = expected_nodes.saturating_sub(1);
+    remote_slots == expected_peers
+        && unavailable_remote_slots == 0
+        && all_slots == expected_nodes
+        && available_all_slots == expected_peers
+}
+
+fn merge_peer_heal_statuses(
+    mut snapshots: Vec<NodeHealStatusSnapshot>,
+    peer_statuses: Vec<Result<Option<NodeHealStatusSnapshot>, String>>,
+    expected_nodes: usize,
+) -> S3Result<ClusterHealStatusSnapshot> {
+    for peer_status in peer_statuses {
+        match peer_status {
+            Ok(Some(snapshot)) => snapshots.push(snapshot),
+            Ok(None) => {
+                warn!(
+                    event = EVENT_ADMIN_REQUEST_FAILED,
+                    component = LOG_COMPONENT_ADMIN_API,
+                    subsystem = LOG_SUBSYSTEM_HEAL_ADMIN,
+                    operation = "background_heal_status",
+                    result = "degraded",
+                    reason = "peer_status_unsupported",
+                    "cluster heal status is partial during rolling upgrade"
+                );
+            }
+            Err(err) => {
+                warn!(
+                    event = EVENT_ADMIN_REQUEST_FAILED,
+                    component = LOG_COMPONENT_ADMIN_API,
+                    subsystem = LOG_SUBSYSTEM_HEAL_ADMIN,
+                    operation = "background_heal_status",
+                    result = "failed",
+                    reason = "peer_status_unavailable",
+                    error = %err,
+                    "cluster heal peer status unavailable"
+                );
+                return Err(cluster_heal_status_unavailable("peer_status_unavailable"));
+            }
+        }
+    }
+    let complete = snapshots.len() == expected_nodes;
+    let mut status = aggregate_cluster_heal_status(snapshots);
+    status.complete = complete;
+    if !complete && status.state != HealRuntimeState::Active {
+        return Err(cluster_heal_status_unavailable("peer_status_unsupported_without_known_active_work"));
+    }
+    Ok(status)
+}
+
+async fn query_peer_heal_status<E>(
+    host: &str,
+    request: impl std::future::Future<Output = Result<Option<Vec<u8>>, E>>,
+    request_timeout: Duration,
+) -> Result<Option<NodeHealStatusSnapshot>, String>
+where
+    E: std::fmt::Display,
+{
+    match timeout(request_timeout, request).await {
+        Ok(Ok(Some(status))) => decode_node_heal_status(&status)
+            .map(Some)
+            .map_err(|err| format!("peer {host}: {err}")),
+        Ok(Ok(None)) => Ok(None),
+        Ok(Err(err)) => Err(format!("peer {host}: {err}")),
+        Err(_) => Err(format!("peer {host}: heal status timed out")),
+    }
+}
+
+async fn read_cluster_heal_status(
+    local_info: BackgroundHealInfo,
+    notification_system: Option<&crate::admin::storage_api::runtime_sources::NotificationSys>,
+    expected_nodes: usize,
+) -> S3Result<ClusterHealStatusSnapshot> {
+    let snapshots = vec![capture_node_heal_status(local_info).await];
+    if expected_nodes == 1 {
+        return Ok(aggregate_cluster_heal_status(snapshots));
+    }
+    let Some(notification_system) = notification_system else {
+        return Err(cluster_heal_status_unavailable("notification_system_unavailable"));
+    };
+    if !peer_topology_complete(
+        expected_nodes,
+        notification_system.peer_clients.len(),
+        notification_system
+            .peer_clients
+            .iter()
+            .filter(|client| client.is_none())
+            .count(),
+        notification_system.all_peer_clients.len(),
+        notification_system
+            .all_peer_clients
+            .iter()
+            .filter(|client| client.is_some())
+            .count(),
+    ) {
+        return Err(cluster_heal_status_unavailable("peer_topology_incomplete"));
+    }
+
+    let peer_statuses = join_all(notification_system.peer_clients.iter().map(|client| async move {
+        let Some(client) = client else {
+            return Err("configured peer is unavailable".to_string());
+        };
+        let host = client.host.to_string();
+        query_peer_heal_status(&host, client.background_heal_status(), PEER_HEAL_STATUS_TIMEOUT).await
+    }))
+    .await;
+
+    merge_peer_heal_statuses(snapshots, peer_statuses, expected_nodes)
 }
 
 #[derive(Debug, Deserialize)]
@@ -398,6 +618,7 @@ fn encode_background_heal_status(
     state: HealRuntimeState,
     heal_operations: rustfs_heal::HealOperationsSnapshot,
     progress: Option<BackgroundHealProgress>,
+    cluster_status_complete: bool,
 ) -> S3Result<Vec<u8>> {
     let status = BackgroundHealStatus {
         info,
@@ -405,6 +626,7 @@ fn encode_background_heal_status(
         heal_queue_length: heal_operations.queue_length,
         heal_active_tasks: heal_operations.active_tasks,
         heal_operations,
+        cluster_status_complete,
         progress,
     };
     serde_json::to_vec(&status).map_err(|e| {
@@ -697,11 +919,9 @@ impl Operation for HealHandler {
             spawn_traced(async move {
                 // Create heal request through channel
                 let heal_request = build_heal_channel_request(&hip);
-                let client_token = heal_request.id.clone();
-
-                match rustfs_common::heal_channel::send_heal_request_with_admission(heal_request).await {
-                    Ok(admission) if admission.is_admitted() => {
-                        let resp_bytes = encode_heal_start_success(client_token, client_address);
+                match rustfs_common::heal_channel::send_heal_request_with_receipt(heal_request).await {
+                    Ok(receipt) if receipt.result.is_admitted() => {
+                        let resp_bytes = encode_heal_start_success(receipt.task_id, client_address);
                         match resp_bytes {
                             Ok(resp_bytes) => {
                                 let _ = tx_clone
@@ -721,13 +941,13 @@ impl Operation for HealHandler {
                             }
                         }
                     }
-                    Ok(admission) => {
+                    Ok(receipt) => {
                         let _ = tx_clone
                             .send(HealResp {
                                 api_err: Some(format!(
                                     "heal request not admitted: admission={}, reason={}",
-                                    admission.result_label(),
-                                    admission.reason_label()
+                                    receipt.result.result_label(),
+                                    receipt.result.reason_label()
                                 )),
                                 ..Default::default()
                             })
@@ -767,7 +987,7 @@ impl Operation for BackgroundHealStatusHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         validate_heal_admin_request(&req).await?;
 
-        let Some(store) = object_store_from_req(&req) else {
+        let Some(context) = app_context_from_req(&req) else {
             warn!(
                 event = EVENT_ADMIN_REQUEST_FAILED,
                 component = LOG_COMPONENT_ADMIN_API,
@@ -779,18 +999,25 @@ impl Operation for BackgroundHealStatusHandler {
             );
             return Err(s3_error!(InternalError, "server not initialized"));
         };
+        let store = context.object_store();
+        let Some(endpoints) = context.endpoints().handle() else {
+            return Err(cluster_heal_status_unavailable("endpoint_topology_unavailable"));
+        };
+        let expected_nodes = endpoints.get_nodes().len();
+        if expected_nodes == 0 {
+            return Err(cluster_heal_status_unavailable("endpoint_topology_empty"));
+        }
+        let notification_system = context.notification_system().handle();
 
-        let info = read_background_heal_info(store).await;
-        let heal_operations = rustfs_heal::current_heal_operations_snapshot().await;
-        let state = background_heal_runtime_state(
-            scanner_enabled_from_env() || heal_enabled_from_env(),
-            rustfs_heal::heal_runtime_initialized(),
-            &heal_operations,
-        );
-        let progress = rustfs_heal::current_heal_progress_snapshot()
-            .await
-            .map(BackgroundHealProgress::from);
-        let body = encode_background_heal_status(&info, state, heal_operations, progress)?;
+        let local_info = read_background_heal_info(store).await;
+        let cluster_status = read_cluster_heal_status(local_info, notification_system.as_deref(), expected_nodes).await?;
+        let body = encode_background_heal_status(
+            &cluster_status.info,
+            cluster_status.state,
+            cluster_status.operations,
+            cluster_status.progress,
+            cluster_status.complete,
+        )?;
         info!(
             event = EVENT_ADMIN_RESPONSE_EMITTED,
             component = LOG_COMPONENT_ADMIN_API,
@@ -808,13 +1035,14 @@ impl Operation for BackgroundHealStatusHandler {
 mod tests {
     use super::extract_heal_init_params;
     use super::{
-        BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState, background_heal_runtime_state,
-        build_heal_channel_request, encode_background_heal_status, encode_heal_start_success, encode_heal_task_status,
-        heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary, json_response,
-        map_heal_response, map_root_heal_status, should_handle_root_heal_directly, validate_heal_request_mode,
-        validate_heal_target,
+        BackgroundHealProgress, HealInitParams, HealResp, HealRuntimeState, aggregate_cluster_heal_status,
+        background_heal_runtime_state, build_heal_channel_request, encode_background_heal_status, encode_heal_start_success,
+        encode_heal_task_status, heal_channel_response_items, heal_channel_response_progress, heal_channel_response_summary,
+        json_response, map_heal_response, map_root_heal_status, merge_peer_heal_statuses, peer_topology_complete,
+        query_peer_heal_status, should_handle_root_heal_directly, validate_heal_request_mode, validate_heal_target,
     };
     use crate::admin::storage_api::error::StorageError;
+    use crate::storage::rpc::node_service::heal::{NodeHealProgress, NodeHealStatusSnapshot};
     use bytes::Bytes;
     use http::StatusCode;
     use http::Uri;
@@ -828,6 +1056,7 @@ mod tests {
     use serde_json::json;
     use time::{OffsetDateTime, format_description::well_known::Rfc3339};
     use tokio::sync::mpsc;
+    use tokio::time::Duration;
 
     #[test]
     fn test_heal_opts_serialization() {
@@ -1262,7 +1491,7 @@ mod tests {
             ..Default::default()
         };
 
-        let encoded = encode_background_heal_status(&info, HealRuntimeState::Active, operations, None)
+        let encoded = encode_background_heal_status(&info, HealRuntimeState::Active, operations, None, true)
             .expect("background heal info should serialize");
         let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize");
 
@@ -1278,6 +1507,7 @@ mod tests {
         assert!(json["healOperations"]["queuedByPriority"]["low"].is_u64());
         assert!(json["healOperations"]["queuedByPriority"]["high"].is_u64());
         assert_eq!(json["state"], "active");
+        assert_eq!(json["clusterStatusComplete"], true);
         assert!(json["progress"].is_null());
     }
 
@@ -1301,6 +1531,7 @@ mod tests {
             HealRuntimeState::Idle,
             rustfs_heal::HealOperationsSnapshot::default(),
             Some(progress),
+            true,
         )
         .expect("background heal info should serialize");
         let json: serde_json::Value = serde_json::from_slice(&encoded).expect("json should deserialize");
@@ -1311,6 +1542,199 @@ mod tests {
         assert_eq!(json["progress"]["bytesProcessed"], 4096);
     }
 
+    #[test]
+    fn test_cluster_heal_status_aggregates_active_peer_independent_of_request_node() {
+        let local = NodeHealStatusSnapshot::for_test(
+            true,
+            true,
+            BackgroundHealInfo {
+                bitrot_start_time: None,
+                bitrot_start_cycle: 7,
+                current_scan_mode: HealScanMode::Deep,
+            },
+            rustfs_heal::HealOperationsSnapshot {
+                queue_length: 2,
+                queued_by_source: rustfs_heal::HealSourceCounts {
+                    admin: 2,
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            Some(NodeHealProgress {
+                objects_scanned: 3,
+                objects_healed: 1,
+                objects_failed: 0,
+                bytes_processed: 100,
+            }),
+        );
+        let peer = NodeHealStatusSnapshot::for_test(
+            true,
+            true,
+            BackgroundHealInfo::default(),
+            rustfs_heal::HealOperationsSnapshot {
+                active_tasks: 1,
+                active_by_priority: rustfs_heal::HealPriorityCounts {
+                    high: 1,
+                    ..Default::default()
+                },
+                active_by_source: rustfs_heal::HealSourceCounts {
+                    admin: 1,
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            Some(NodeHealProgress {
+                objects_scanned: 5,
+                objects_healed: 4,
+                objects_failed: 1,
+                bytes_processed: 900,
+            }),
+        );
+
+        let local_first = aggregate_cluster_heal_status(vec![local.clone(), peer.clone()]);
+        let peer_first = aggregate_cluster_heal_status(vec![peer, local]);
+
+        assert_eq!(local_first.state, HealRuntimeState::Active);
+        assert_eq!(local_first.operations.queue_length, 2);
+        assert_eq!(local_first.operations.active_tasks, 1);
+        assert_eq!(local_first.operations.queued_by_source.admin, 2);
+        assert_eq!(local_first.operations.active_by_source.admin, 1);
+        assert_eq!(local_first.operations.active_by_priority.high, 1);
+        assert_eq!(local_first.info.bitrot_start_cycle, 7);
+        assert_eq!(local_first.info.current_scan_mode, HealScanMode::Deep);
+        let progress = local_first.progress.expect("cluster progress should be present");
+        assert_eq!(progress.objects_scanned, 8);
+        assert_eq!(progress.objects_healed, 5);
+        assert_eq!(progress.objects_failed, 1);
+        assert_eq!(progress.bytes_processed, 1000);
+
+        assert_eq!(peer_first.state, HealRuntimeState::Active);
+        assert_eq!(peer_first.operations, local_first.operations);
+        assert_eq!(peer_first.info.bitrot_start_cycle, local_first.info.bitrot_start_cycle);
+        assert_eq!(peer_first.info.current_scan_mode, local_first.info.current_scan_mode);
+    }
+
+    #[test]
+    fn test_cluster_heal_status_saturates_all_counter_groups() {
+        let priorities = |value| rustfs_heal::HealPriorityCounts {
+            low: value,
+            normal: value,
+            high: value,
+            urgent: value,
+        };
+        let sources = |value| rustfs_heal::HealSourceCounts {
+            scanner: value,
+            admin: value,
+            auto_heal: value,
+            internal: value,
+            read_repair: value,
+        };
+        let operations = |value| rustfs_heal::HealOperationsSnapshot {
+            queue_length: value,
+            active_tasks: value,
+            retrying_tasks: value,
+            queued_by_priority: priorities(value),
+            active_by_priority: priorities(value),
+            retrying_by_priority: priorities(value),
+            queued_by_source: sources(value),
+            active_by_source: sources(value),
+            retrying_by_source: sources(value),
+        };
+        let progress = |value| NodeHealProgress {
+            objects_scanned: value,
+            objects_healed: value,
+            objects_failed: value,
+            bytes_processed: value,
+        };
+        let saturated = NodeHealStatusSnapshot::for_test(
+            true,
+            true,
+            BackgroundHealInfo::default(),
+            operations(u64::MAX),
+            Some(progress(u64::MAX)),
+        );
+        let one = NodeHealStatusSnapshot::for_test(true, true, BackgroundHealInfo::default(), operations(1), Some(progress(1)));
+
+        let status = aggregate_cluster_heal_status(vec![saturated, one]);
+        assert_eq!(status.operations, operations(u64::MAX));
+        let progress = status.progress.expect("progress");
+        assert_eq!(progress.objects_scanned, u64::MAX);
+        assert_eq!(progress.objects_healed, u64::MAX);
+        assert_eq!(progress.objects_failed, u64::MAX);
+        assert_eq!(progress.bytes_processed, u64::MAX);
+    }
+
+    #[test]
+    fn test_peer_topology_validation_fails_closed_on_missing_slots() {
+        assert!(peer_topology_complete(4, 3, 0, 4, 3));
+        assert!(!peer_topology_complete(4, 2, 0, 4, 3));
+        assert!(!peer_topology_complete(4, 3, 1, 4, 2));
+        assert!(!peer_topology_complete(4, 3, 0, 3, 2));
+        assert!(peer_topology_complete(1, 0, 0, 1, 0));
+    }
+
+    #[test]
+    fn test_peer_status_merge_fails_closed_but_degrades_for_older_peers() {
+        let local = || {
+            NodeHealStatusSnapshot::for_test(
+                true,
+                true,
+                BackgroundHealInfo::default(),
+                rustfs_heal::HealOperationsSnapshot::default(),
+                None,
+            )
+        };
+        let error = merge_peer_heal_statuses(vec![local()], vec![Err("peer timeout".to_string())], 2)
+            .expect_err("peer failure must fail closed");
+        assert_eq!(error.message(), Some("cluster heal status unavailable"));
+
+        merge_peer_heal_statuses(vec![local()], vec![Ok(None)], 2).expect_err("unknown peer work must not be reported as idle");
+
+        let known_active = NodeHealStatusSnapshot::for_test(
+            true,
+            true,
+            BackgroundHealInfo::default(),
+            rustfs_heal::HealOperationsSnapshot {
+                active_tasks: 1,
+                ..Default::default()
+            },
+            None,
+        );
+        let partial = merge_peer_heal_statuses(vec![known_active], vec![Ok(None)], 2)
+            .expect("known active work may be reported as an explicit partial status");
+        assert!(!partial.complete);
+    }
+
+    #[tokio::test]
+    async fn test_peer_status_query_rejects_failure_decode_error_and_timeout() {
+        let failure = query_peer_heal_status(
+            "node-a",
+            std::future::ready(Err::<Option<Vec<u8>>, _>(std::io::Error::other("rpc failed"))),
+            Duration::from_secs(1),
+        )
+        .await
+        .expect_err("RPC failure must fail closed");
+        assert!(failure.contains("rpc failed"));
+
+        let decode = query_peer_heal_status(
+            "node-a",
+            std::future::ready(Ok::<_, std::io::Error>(Some(vec![0xc1]))),
+            Duration::from_secs(1),
+        )
+        .await
+        .expect_err("invalid payload must fail closed");
+        assert!(decode.contains("decode"));
+
+        let timed_out = query_peer_heal_status(
+            "node-a",
+            std::future::pending::<Result<Option<Vec<u8>>, std::io::Error>>(),
+            Duration::from_millis(1),
+        )
+        .await
+        .expect_err("timeout must fail closed");
+        assert!(timed_out.contains("timed out"));
+    }
+
     #[test]
     fn test_background_heal_runtime_state_distinguishes_disabled_and_uninitialized() {
         let operations = rustfs_heal::HealOperationsSnapshot::default();
@@ -1324,6 +1748,12 @@ mod tests {
             ..operations
         };
         assert_eq!(background_heal_runtime_state(true, true, &active), HealRuntimeState::Active);
+
+        let retrying = rustfs_heal::HealOperationsSnapshot {
+            retrying_tasks: 1,
+            ..Default::default()
+        };
+        assert_eq!(background_heal_runtime_state(true, true, &retrying), HealRuntimeState::Active);
     }
 
     #[test]

rustfs/src/admin/runtime_sources.rs

@@ -72,6 +72,13 @@ pub(crate) fn object_store_from_req<B>(req: &s3s::S3Request<B>) -> Option<Arc<EC
     object_store_from_extensions(&req.extensions)
 }
 
+pub(crate) fn app_context_from_req<B>(req: &s3s::S3Request<B>) -> Option<Arc<AppContext>> {
+    req.extensions
+        .get::<Arc<ServerContextSlot>>()
+        .and_then(|slot| slot.app_context())
+        .or_else(current_app_context)
+}
+
 /// Field-borrow form of [`object_store_from_req`] for handlers that have
 /// already moved other request fields (body, credentials) out of the request.
 pub(crate) fn object_store_from_extensions(extensions: &http::Extensions) -> Option<Arc<ECStore>> {

rustfs/src/storage/rpc/node_service.rs

@@ -46,6 +46,8 @@ use tokio_util::sync::CancellationToken;
 use tonic::{Request, Response, Status, Streaming};
 use tracing::{debug, error, info, warn};
 
+pub(crate) mod heal;
+
 const LOG_COMPONENT_STORAGE: &str = "storage";
 const LOG_SUBSYSTEM_RPC: &str = "rpc";
 const LOG_SUBSYSTEM_REBALANCE: &str = "rebalance";
@@ -1002,7 +1004,26 @@ impl Node for NodeService {
         &self,
         _request: Request<BackgroundHealStatusRequest>,
     ) -> Result<Response<BackgroundHealStatusResponse>, Status> {
-        Err(unimplemented_rpc("background_heal_status"))
+        if self.resolve_object_store().is_none() {
+            return Ok(Response::new(BackgroundHealStatusResponse {
+                success: false,
+                bg_heal_state: Bytes::new(),
+                error_info: Some("storage layer not initialized".to_string()),
+            }));
+        }
+        let snapshot = heal::capture_node_heal_status(rustfs_scanner::scanner::BackgroundHealInfo::default()).await;
+        match heal::encode_node_heal_status(&snapshot) {
+            Ok(bg_heal_state) => Ok(Response::new(BackgroundHealStatusResponse {
+                success: true,
+                bg_heal_state: bg_heal_state.into(),
+                error_info: None,
+            })),
+            Err(err) => Ok(Response::new(BackgroundHealStatusResponse {
+                success: false,
+                bg_heal_state: Bytes::new(),
+                error_info: Some(err),
+            })),
+        }
     }
 
     async fn get_metacache_listing(
@@ -2988,12 +3009,13 @@ mod tests {
                 .await,
             "get_all_bucket_stats",
         );
-        assert_unimplemented_status(
-            service
-                .background_heal_status(Request::new(BackgroundHealStatusRequest::default()))
-                .await,
-            "background_heal_status",
-        );
+        let heal_status = service
+            .background_heal_status(Request::new(BackgroundHealStatusRequest::default()))
+            .await
+            .expect("implemented heal status RPC should return a response")
+            .into_inner();
+        assert!(!heal_status.success);
+        assert_eq!(heal_status.error_info.as_deref(), Some("storage layer not initialized"));
         assert_unimplemented_status(
             service
                 .get_metacache_listing(Request::new(GetMetacacheListingRequest::default()))

rustfs/src/storage/rpc/node_service/heal.rs

@@ -0,0 +1,249 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::startup_background::{heal_enabled_from_env, scanner_enabled_from_env};
+use rmp_serde::Deserializer;
+use rustfs_common::heal_channel::HealScanMode;
+use rustfs_heal::HealOperationsSnapshot;
+use rustfs_scanner::scanner::BackgroundHealInfo;
+use serde::{Deserialize, Serialize};
+use std::io::Cursor;
+
+use super::super::encode_msgpack_map;
+
+const NODE_HEAL_STATUS_VERSION: u8 = 1;
+const NODE_HEAL_STATUS_MAX_SIZE: usize = 64 * 1024;
+
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub(crate) struct NodeHealProgress {
+    pub objects_scanned: u64,
+    pub objects_healed: u64,
+    pub objects_failed: u64,
+    pub bytes_processed: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct NodeHealInfo {
+    bitrot_start_time: Option<chrono::DateTime<chrono::Utc>>,
+    bitrot_start_cycle: u64,
+    current_scan_mode: HealScanMode,
+}
+
+impl From<BackgroundHealInfo> for NodeHealInfo {
+    fn from(info: BackgroundHealInfo) -> Self {
+        Self {
+            bitrot_start_time: info.bitrot_start_time,
+            bitrot_start_cycle: info.bitrot_start_cycle,
+            current_scan_mode: info.current_scan_mode,
+        }
+    }
+}
+
+impl From<NodeHealInfo> for BackgroundHealInfo {
+    fn from(info: NodeHealInfo) -> Self {
+        Self {
+            bitrot_start_time: info.bitrot_start_time,
+            bitrot_start_cycle: info.bitrot_start_cycle,
+            current_scan_mode: info.current_scan_mode,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub(crate) struct NodeHealStatusSnapshot {
+    version: u8,
+    pub services_enabled: bool,
+    pub initialized: bool,
+    info: NodeHealInfo,
+    pub operations: HealOperationsSnapshot,
+    pub progress: Option<NodeHealProgress>,
+}
+
+impl NodeHealStatusSnapshot {
+    #[cfg(test)]
+    pub(crate) fn for_test(
+        services_enabled: bool,
+        initialized: bool,
+        info: BackgroundHealInfo,
+        operations: HealOperationsSnapshot,
+        progress: Option<NodeHealProgress>,
+    ) -> Self {
+        Self {
+            version: NODE_HEAL_STATUS_VERSION,
+            services_enabled,
+            initialized,
+            info: info.into(),
+            operations,
+            progress,
+        }
+    }
+
+    pub(crate) fn info(&self) -> BackgroundHealInfo {
+        BackgroundHealInfo {
+            bitrot_start_time: self.info.bitrot_start_time,
+            bitrot_start_cycle: self.info.bitrot_start_cycle,
+            current_scan_mode: self.info.current_scan_mode,
+        }
+    }
+}
+
+pub(crate) async fn capture_node_heal_status(info: BackgroundHealInfo) -> NodeHealStatusSnapshot {
+    let progress = rustfs_heal::current_heal_progress_snapshot()
+        .await
+        .map(|progress| NodeHealProgress {
+            objects_scanned: progress.objects_scanned,
+            objects_healed: progress.objects_healed,
+            objects_failed: progress.objects_failed,
+            bytes_processed: progress.bytes_processed,
+        });
+
+    NodeHealStatusSnapshot {
+        version: NODE_HEAL_STATUS_VERSION,
+        services_enabled: scanner_enabled_from_env() || heal_enabled_from_env(),
+        initialized: rustfs_heal::heal_runtime_initialized(),
+        info: info.into(),
+        operations: rustfs_heal::current_heal_operations_snapshot().await,
+        progress,
+    }
+}
+
+pub(crate) fn encode_node_heal_status(snapshot: &NodeHealStatusSnapshot) -> Result<Vec<u8>, String> {
+    encode_msgpack_map(snapshot).map_err(|err| format!("failed to encode node heal status: {err}"))
+}
+
+pub(crate) fn decode_node_heal_status(data: &[u8]) -> Result<NodeHealStatusSnapshot, String> {
+    if data.len() > NODE_HEAL_STATUS_MAX_SIZE {
+        return Err("node heal status exceeds size limit".to_string());
+    }
+    let mut deserializer = Deserializer::new(Cursor::new(data));
+    let snapshot = NodeHealStatusSnapshot::deserialize(&mut deserializer)
+        .map_err(|err| format!("failed to decode node heal status: {err}"))?;
+    if usize::try_from(deserializer.get_ref().position()).ok() != Some(data.len()) {
+        return Err("node heal status contains trailing data".to_string());
+    }
+    if snapshot.version != NODE_HEAL_STATUS_VERSION {
+        return Err(format!("unsupported node heal status version: {}", snapshot.version));
+    }
+    Ok(snapshot)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        NODE_HEAL_STATUS_MAX_SIZE, NODE_HEAL_STATUS_VERSION, NodeHealProgress, NodeHealStatusSnapshot, decode_node_heal_status,
+        encode_node_heal_status,
+    };
+    use rustfs_heal::HealOperationsSnapshot;
+    use rustfs_scanner::scanner::BackgroundHealInfo;
+
+    #[test]
+    fn node_heal_status_round_trip_is_versioned() {
+        let snapshot = NodeHealStatusSnapshot::for_test(
+            true,
+            true,
+            BackgroundHealInfo::default(),
+            HealOperationsSnapshot {
+                queue_length: 2,
+                active_tasks: 1,
+                ..Default::default()
+            },
+            Some(NodeHealProgress {
+                objects_scanned: 7,
+                objects_healed: 5,
+                objects_failed: 1,
+                bytes_processed: 1024,
+            }),
+        );
+
+        let encoded = encode_node_heal_status(&snapshot).expect("snapshot should encode");
+        let decoded = decode_node_heal_status(&encoded).expect("snapshot should decode");
+        assert_eq!(decoded.version, NODE_HEAL_STATUS_VERSION);
+        assert_eq!(decoded.operations.queue_length, 2);
+        assert_eq!(decoded.progress, snapshot.progress);
+    }
+
+    #[test]
+    fn node_heal_status_rejects_unknown_version() {
+        let mut snapshot = NodeHealStatusSnapshot::for_test(
+            false,
+            false,
+            BackgroundHealInfo::default(),
+            HealOperationsSnapshot::default(),
+            None,
+        );
+        snapshot.version += 1;
+
+        let encoded = encode_node_heal_status(&snapshot).expect("snapshot should encode");
+        let err = decode_node_heal_status(&encoded).expect_err("unknown version should fail closed");
+        assert!(err.contains("unsupported node heal status version"));
+    }
+
+    #[test]
+    fn node_heal_status_accepts_fixed_v1_wire_keys() {
+        let fixture = serde_json::json!({
+            "version": 1,
+            "servicesEnabled": true,
+            "initialized": true,
+            "info": {"bitrotStartTime": null, "bitrotStartCycle": 9, "currentScanMode": 1},
+            "operations": {
+                "queueLength": 2, "activeTasks": 1, "retryingTasks": 0,
+                "queuedByPriority": {"low": 0, "normal": 2, "high": 0, "urgent": 0},
+                "activeByPriority": {"low": 0, "normal": 0, "high": 1, "urgent": 0},
+                "retryingByPriority": {"low": 0, "normal": 0, "high": 0, "urgent": 0},
+                "queuedBySource": {"scanner": 2, "admin": 0, "autoHeal": 0, "internal": 0, "readRepair": 0},
+                "activeBySource": {"scanner": 0, "admin": 1, "autoHeal": 0, "internal": 0, "readRepair": 0},
+                "retryingBySource": {"scanner": 0, "admin": 0, "autoHeal": 0, "internal": 0, "readRepair": 0}
+            },
+            "progress": null
+        });
+        let encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
+        let decoded = decode_node_heal_status(&encoded).expect("fixed v1 fixture should decode");
+        assert_eq!(decoded.info().bitrot_start_cycle, 9);
+        assert_eq!(decoded.operations.queue_length, 2);
+    }
+
+    #[test]
+    fn node_heal_status_rejects_nested_unknown_trailing_truncated_and_oversized_data() {
+        let mut fixture = serde_json::json!({
+            "version": 1, "servicesEnabled": true, "initialized": true,
+            "info": {"bitrotStartTime": null, "bitrotStartCycle": 0, "currentScanMode": 0, "unknown": true},
+            "operations": HealOperationsSnapshot::default(), "progress": null
+        });
+        let encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
+        assert!(
+            decode_node_heal_status(&encoded)
+                .expect_err("nested unknown field must fail")
+                .contains("unknown")
+        );
+
+        fixture["info"].as_object_mut().expect("info object").remove("unknown");
+        let mut encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
+        encoded.push(0);
+        assert!(
+            decode_node_heal_status(&encoded)
+                .expect_err("trailing data must fail")
+                .contains("trailing")
+        );
+        encoded.truncate(encoded.len() / 2);
+        assert!(decode_node_heal_status(&encoded).is_err());
+        assert!(
+            decode_node_heal_status(&vec![0; NODE_HEAL_STATUS_MAX_SIZE + 1])
+                .expect_err("oversized status must fail")
+                .contains("size limit")
+        );
+    }
+}