fix(heal): emit MRF repair notices on completion (#6309)

Move MRF repaired-event fan-out from admission to successful terminal completion so scanner pending-heal ledgers only clear after the canonical heal task actually finishes. Track notice ownership across duplicate admission, retry merge, cancellation, and queue displacement.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/heal/src/heal/manager.rs

@@ -27,7 +27,7 @@ use rustfs_madmin::heal_commands::HealResultItem;
 use std::sync::LazyLock;
 use std::{
     collections::{BinaryHeap, HashMap, HashSet},
-    sync::Arc,
+    sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard},
     time::{Duration, SystemTime},
 };
 use tokio::{
@@ -113,6 +113,51 @@ async fn pause_duplicate_admission_after_active_lock(request_id: &str) {
 
 type WorkloadSnapshotProviderRef = Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct MrfRepairNoticeTarget {
+    bucket: Arc<str>,
+    object: Arc<str>,
+    version_id: Option<[u8; 16]>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct HealAdmissionDecision {
+    result: HealAdmissionResult,
+    displaced_task_id: Option<String>,
+}
+
+impl HealAdmissionDecision {
+    const fn new(result: HealAdmissionResult) -> Self {
+        Self {
+            result,
+            displaced_task_id: None,
+        }
+    }
+
+    fn accepted_with_displacement(displaced_task_id: String) -> Self {
+        Self {
+            result: HealAdmissionResult::Accepted,
+            displaced_task_id: Some(displaced_task_id),
+        }
+    }
+}
+
+fn lock_mrf_repair_notice_targets(
+    registry: &StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>,
+) -> StdMutexGuard<'_, HashMap<String, Vec<MrfRepairNoticeTarget>>> {
+    match registry.lock() {
+        Ok(guard) => guard,
+        Err(poisoned) => poisoned.into_inner(),
+    }
+}
+
+async fn remove_task_aliases_for_task(registry: &Arc<Mutex<HashMap<String, HealTaskAlias>>>, task_id: &str) {
+    registry
+        .lock()
+        .await
+        .retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
+}
+
 #[derive(Debug, Clone)]
 pub struct HealTaskReport {
     pub status: HealTaskStatus,
@@ -272,10 +317,6 @@ fn publish_heal_queue_length(queue: &PriorityHealQueue) {
     crate::set_heal_queue_length(queue.len());
 }
 
-fn active_heals_contains_dedup_key(active_heals: &HashMap<String, Arc<HealTask>>, key: &str) -> bool {
-    active_heal_for_dedup_key(active_heals, key).is_some()
-}
-
 fn active_heal_for_dedup_key(active_heals: &HashMap<String, Arc<HealTask>>, key: &str) -> Option<(String, HealType)> {
     active_heals
         .iter()
@@ -581,6 +622,10 @@ pub struct HealManager {
     task_aliases: Arc<Mutex<HashMap<String, HealTaskAlias>>>,
     /// Heal tasks waiting for a retry backoff to expire.
     retrying_heals: Arc<Mutex<HashMap<String, RetryingHeal>>>,
+    /// MRF repaired-event targets keyed by canonical heal task id. Admission
+    /// only registers ownership; the scheduler emits the event after a real
+    /// successful completion.
+    mrf_repair_notice_targets: Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>,
     /// Surviving disks that hold durable replacement intents, keyed by task ID.
     /// This is rebuilt from durable state after restart and never crosses the
     /// public request boundary.
@@ -616,6 +661,7 @@ struct HealQueueContext<'a> {
     completed_heals: &'a Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
     task_aliases: &'a Arc<Mutex<HashMap<String, HealTaskAlias>>>,
     retrying_heals: &'a Arc<Mutex<HashMap<String, RetryingHeal>>>,
+    mrf_repair_notice_targets: &'a Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>,
     replacement_recovery_anchors: &'a Arc<std::sync::Mutex<HashMap<String, String>>>,
     config: &'a Arc<RwLock<HealConfig>>,
     statistics: &'a Arc<RwLock<HealStatistics>>,
@@ -779,12 +825,27 @@ impl HealManager {
         .increment(1);
     }
 
+    fn remove_mrf_repair_notice_targets_for_task(&self, task_id: &str) {
+        lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).remove(task_id);
+    }
+
+    fn insert_mrf_repair_notice_target(
+        registry: &mut HashMap<String, Vec<MrfRepairNoticeTarget>>,
+        task_id: &str,
+        target: MrfRepairNoticeTarget,
+    ) {
+        let task_targets = registry.entry(task_id.to_string()).or_default();
+        if !task_targets.contains(&target) {
+            task_targets.push(target);
+        }
+    }
+
     fn admit_request_to_queue(
         queue: &mut PriorityHealQueue,
         request: HealRequest,
         config: &HealConfig,
         context: &'static str,
-    ) -> HealAdmissionResult {
+    ) -> HealAdmissionDecision {
         let queue_len = queue.len();
         publish_heal_queue_length(queue);
         let queue_capacity = config.queue_size;
@@ -813,7 +874,7 @@ impl HealManager {
                         result = "accepted_by_displacement",
                         "Heal queue request accepted by displacement"
                     });
-                    return HealAdmissionResult::Accepted;
+                    return HealAdmissionDecision::accepted_with_displacement(displaced.id);
                 }
 
                 demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", {
@@ -830,7 +891,7 @@ impl HealManager {
                     "Heal queue request rejected without displacement"
                 });
                 Self::record_admission_metric(source, HealAdmissionResult::Full, context);
-                return HealAdmissionResult::Full;
+                return HealAdmissionDecision::new(HealAdmissionResult::Full);
             }
 
             let admission = Self::classify_full_admission(&request, config);
@@ -869,7 +930,7 @@ impl HealManager {
                 }
                 HealAdmissionResult::Accepted | HealAdmissionResult::Merged => {}
             }
-            return admission;
+            return HealAdmissionDecision::new(admission);
         }
 
         if let Some(admission) = Self::classify_pressure_admission(&request, queue_len, queue_capacity) {
@@ -892,7 +953,7 @@ impl HealManager {
                     "Heal queue request dropped under pressure"
                 );
             }
-            return admission;
+            return HealAdmissionDecision::new(admission);
         }
 
         if queue_capacity > 0 {
@@ -956,7 +1017,7 @@ impl HealManager {
                     result = "accepted",
                     "Heal queue request accepted"
                 );
-                HealAdmissionResult::Accepted
+                HealAdmissionDecision::new(HealAdmissionResult::Accepted)
             }
             QueuePushOutcome::Merged => {
                 Self::record_admission_metric(source, HealAdmissionResult::Merged, context);
@@ -973,7 +1034,7 @@ impl HealManager {
                     result = "merged_duplicate",
                     "Heal queue request merged"
                 );
-                HealAdmissionResult::Merged
+                HealAdmissionDecision::new(HealAdmissionResult::Merged)
             }
         }
     }
@@ -1001,10 +1062,7 @@ impl HealManager {
     }
 
     async fn remove_aliases_for_task(&self, task_id: &str) {
-        self.task_aliases
+        remove_task_aliases_for_task(&self.task_aliases, task_id).await;
-            .lock()
-            .await
-            .retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
     }
 
     fn block_replacement_recovery_set(&self, set_disk_id: &str) {
@@ -1049,6 +1107,7 @@ impl HealManager {
             completed_heals: Arc::new(Mutex::new(HashMap::new())),
             task_aliases: Arc::new(Mutex::new(HashMap::new())),
             retrying_heals: Arc::new(Mutex::new(HashMap::new())),
+            mrf_repair_notice_targets: Arc::new(StdMutex::new(HashMap::new())),
             replacement_recovery_anchors: Arc::new(std::sync::Mutex::new(HashMap::new())),
             replacement_recovery_blocked_sets: Arc::new(std::sync::Mutex::new(HashSet::new())),
             storage,
@@ -1152,6 +1211,7 @@ impl HealManager {
         self.completed_heals.lock().await.clear();
         self.task_aliases.lock().await.clear();
         self.retrying_heals.lock().await.clear();
+        lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).clear();
         crate::set_heal_queue_length(0);
 
         // update state
@@ -1178,6 +1238,35 @@ impl HealManager {
         &self,
         request: HealRequest,
         preserve_alias: bool,
+    ) -> Result<HealAdmissionReceipt> {
+        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, None)
+            .await
+    }
+
+    pub(crate) async fn submit_mrf_heal_request_with_receipt(
+        &self,
+        request: HealRequest,
+        bucket: Arc<str>,
+        object: Arc<str>,
+        version_id: Option<[u8; 16]>,
+    ) -> Result<HealAdmissionReceipt> {
+        self.submit_heal_request_with_receipt_alias_and_mrf_notice(
+            request,
+            true,
+            Some(MrfRepairNoticeTarget {
+                bucket,
+                object,
+                version_id,
+            }),
+        )
+        .await
+    }
+
+    async fn submit_heal_request_with_receipt_alias_and_mrf_notice(
+        &self,
+        request: HealRequest,
+        preserve_alias: bool,
+        mrf_notice_target: Option<MrfRepairNoticeTarget>,
     ) -> Result<HealAdmissionReceipt> {
         // HS-06 forceStart semantics (admin only): MinIO stops the old task
         // first and then starts the new one. Cancel any active admin task
@@ -1254,6 +1343,12 @@ impl HealManager {
                 } else {
                     Self::duplicate_admission_for_request(&request, &config)
                 };
+            if matches!(admission, HealAdmissionResult::Merged)
+                && let Some(target) = mrf_notice_target
+            {
+                let mut targets = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets);
+                Self::insert_mrf_repair_notice_target(&mut targets, &merged_task_id, target);
+            }
             drop(retrying_heals);
             drop(queue);
             drop(active_heals);
@@ -1356,17 +1451,32 @@ impl HealManager {
         }
 
         let mut task_id = request.id.clone();
-        let admission = Self::admit_request_to_queue(&mut queue, request, &config, "submit");
+        let admission_decision = Self::admit_request_to_queue(&mut queue, request, &config, "submit");
+        let admission = admission_decision.result;
         if admission == HealAdmissionResult::Merged
             && let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key)
         {
             task_id = queued_id.to_owned();
         }
         let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
+        let displaced_task_id = admission_decision.displaced_task_id;
+        if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
+            && let Some(target) = mrf_notice_target
+        {
+            let mut targets = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets);
+            Self::insert_mrf_repair_notice_target(&mut targets, &task_id, target);
+        }
+        if let Some(displaced_task_id) = &displaced_task_id {
+            lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).remove(displaced_task_id);
+        }
         drop(retrying_heals);
         drop(queue);
         drop(active_heals);
 
+        if let Some(displaced_task_id) = displaced_task_id {
+            self.remove_aliases_for_task(&displaced_task_id).await;
+        }
+
         if should_notify {
             self.notify.notify_one();
         }
@@ -1603,6 +1713,7 @@ impl HealManager {
                 );
                 drop(active_heals);
                 self.remove_aliases_for_task(&canonical_task_id).await;
+                self.remove_mrf_repair_notice_targets_for_task(&canonical_task_id);
                 return Ok(());
             }
         }
@@ -1614,6 +1725,7 @@ impl HealManager {
                 drop(retrying_heals);
                 self.completed_heals.lock().await.remove(&canonical_task_id);
                 self.remove_aliases_for_task(&canonical_task_id).await;
+                self.remove_mrf_repair_notice_targets_for_task(&canonical_task_id);
                 info!(
                     target: "rustfs::heal::manager",
                     event = EVENT_HEAL_MANAGER_STATE,
@@ -1641,6 +1753,7 @@ impl HealManager {
             );
             drop(queue);
             self.remove_aliases_for_task(&canonical_task_id).await;
+            self.remove_mrf_repair_notice_targets_for_task(&canonical_task_id);
             return Ok(());
         }
 
@@ -1674,6 +1787,7 @@ impl HealManager {
             drop(active_heals);
             for task_id in &task_ids {
                 self.remove_aliases_for_task(task_id).await;
+                self.remove_mrf_repair_notice_targets_for_task(task_id);
             }
         }
 
@@ -1703,6 +1817,7 @@ impl HealManager {
 
                 for task_id in &task_ids {
                     self.remove_aliases_for_task(task_id).await;
+                    self.remove_mrf_repair_notice_targets_for_task(task_id);
                 }
             }
         }
@@ -1716,6 +1831,7 @@ impl HealManager {
         drop(queue);
         for request in &queued_cancelled {
             self.remove_aliases_for_task(&request.id).await;
+            self.remove_mrf_repair_notice_targets_for_task(&request.id);
         }
 
         if cancelled == 0 {

crates/heal/src/heal/manager/auto_scan.rs

@@ -20,6 +20,8 @@ impl HealManager {
         let config = self.config.clone();
         let heal_queue = self.heal_queue.clone();
         let active_heals = self.active_heals.clone();
+        let task_aliases = self.task_aliases.clone();
+        let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
         let storage = self.storage.clone();
         let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
         let replacement_recovery_blocked_sets = self.replacement_recovery_blocked_sets.clone();
@@ -475,7 +477,8 @@ impl HealManager {
                             let endpoint_count = req.heal_endpoints.len();
                             let config = config.read().await;
                             let mut queue = heal_queue.lock().await;
-                            let admission = Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan");
+                            let admission_decision = Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan");
+                            let admission = admission_decision.result;
                             let should_notify =
                                 matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
                             if matches!(admission, HealAdmissionResult::Accepted)
@@ -488,6 +491,10 @@ impl HealManager {
                             }
                             drop(queue);
                             drop(config);
+                            if let Some(displaced_task_id) = admission_decision.displaced_task_id {
+                                remove_task_aliases_for_task(&task_aliases, &displaced_task_id).await;
+                                lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id);
+                            }
                             if matches!(admission, HealAdmissionResult::Accepted) {
                                 if should_notify {
                                     notify.notify_one();

crates/heal/src/heal/manager/scheduler.rs

@@ -23,6 +23,7 @@ impl HealManager {
         let completed_heals = self.completed_heals.clone();
         let task_aliases = self.task_aliases.clone();
         let retrying_heals = self.retrying_heals.clone();
+        let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
         let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
         let cancel_token = self.cancel_token.clone();
         let statistics = self.statistics.clone();
@@ -54,6 +55,7 @@ impl HealManager {
                             completed_heals: &completed_heals,
                             task_aliases: &task_aliases,
                             retrying_heals: &retrying_heals,
+                            mrf_repair_notice_targets: &mrf_repair_notice_targets,
                             replacement_recovery_anchors: &replacement_recovery_anchors,
                             config: &config,
                             statistics: &statistics,
@@ -71,6 +73,7 @@ impl HealManager {
                             completed_heals: &completed_heals,
                             task_aliases: &task_aliases,
                             retrying_heals: &retrying_heals,
+                            mrf_repair_notice_targets: &mrf_repair_notice_targets,
                             replacement_recovery_anchors: &replacement_recovery_anchors,
                             config: &config,
                             statistics: &statistics,
@@ -97,6 +100,7 @@ impl HealManager {
             completed_heals,
             task_aliases,
             retrying_heals,
+            mrf_repair_notice_targets,
             replacement_recovery_anchors,
             config,
             statistics,
@@ -181,6 +185,7 @@ impl HealManager {
                 let completed_heals_clone = completed_heals.clone();
                 let task_aliases_clone = task_aliases.clone();
                 let retrying_heals_clone = retrying_heals.clone();
+                let mrf_repair_notice_targets_clone = mrf_repair_notice_targets.clone();
                 let replacement_recovery_anchors_clone = replacement_recovery_anchors.clone();
                 let statistics_clone = statistics.clone();
                 let notify_clone = notify.clone();
@@ -301,6 +306,7 @@ impl HealManager {
                             completed_task.get_status().await
                         };
                         let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. });
+                        let successful_completion = matches!(completed_status, HealTaskStatus::Completed);
                         let completed_progress = completed_task.get_progress().await;
                         // Single snapshot of the retained window: the task is
                         // finished and already off the active map, so there is
@@ -319,6 +325,7 @@ impl HealManager {
                         let mut completed_heals_guard = completed_heals_clone.lock().await;
                         prune_completed_heal_statuses(&mut completed_heals_guard);
                         completed_heals_guard.insert(task_id.clone(), Arc::new(completed_status_entry));
+                        drop(completed_heals_guard);
                         // update statistics
                         let mut stats = statistics_clone.write().await;
                         match completed_status {
@@ -334,6 +341,10 @@ impl HealManager {
                         stats.update_running_tasks(usize_to_u64_saturated(active_count));
                         drop(stats);
                         if terminal_completion {
+                            let notice_targets = take_mrf_repair_notice_targets(&mrf_repair_notice_targets_clone, &task_id);
+                            if successful_completion {
+                                emit_mrf_repaired_events(notice_targets);
+                            }
                             task_aliases_clone
                                 .lock()
                                 .await
@@ -351,6 +362,8 @@ impl HealManager {
                         let retry_active_heals = active_heals_clone.clone();
                         let retry_heal_queue = heal_queue_clone.clone();
                         let retrying_heals_for_spawn = retrying_heals_clone.clone();
+                        let retry_task_aliases = task_aliases_clone.clone();
+                        let retry_mrf_repair_notice_targets = mrf_repair_notice_targets_clone.clone();
                         let retry_completed_heals = completed_heals_clone.clone();
                         let retry_notify = notify_clone.clone();
                         let retry_manager_cancel_token = manager_cancel_token.clone();
@@ -386,12 +399,17 @@ impl HealManager {
                                     }
                                 }
 
-                                let active_duplicate = {
+                                let active_duplicate_task_id = {
                                     let active_heals_guard = retry_active_heals.lock().await;
-                                    active_heals_contains_dedup_key(&active_heals_guard, &retry_key)
+                                    active_heal_for_dedup_key(&active_heals_guard, &retry_key).map(|(task_id, _)| task_id)
                                 };
-                                if active_duplicate {
+                                if let Some(active_duplicate_task_id) = active_duplicate_task_id {
                                     retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
+                                    move_mrf_repair_notice_targets(
+                                        &retry_mrf_repair_notice_targets,
+                                        &retry_request_id,
+                                        &active_duplicate_task_id,
+                                    );
                                     debug!(
                                         target: "rustfs::heal::manager",
                                         event = EVENT_HEAL_QUEUE_ADMISSION,
@@ -407,8 +425,9 @@ impl HealManager {
                                 }
 
                                 let mut queue = retry_heal_queue.lock().await;
-                                let admission =
+                                let admission_decision =
                                     Self::admit_request_to_queue(&mut queue, retry_request.clone(), &retry_config, "retry");
+                                let admission = admission_decision.result;
                                 let should_notify = matches!(admission, HealAdmissionResult::Accepted)
                                     && retry_config.event_driven_scheduler_enable;
                                 match admission {
@@ -418,7 +437,15 @@ impl HealManager {
                                         #[cfg(test)]
                                         pause_retry_ownership_transition(&retry_request_id, true).await;
                                         retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
+                                        let displaced_task_id = admission_decision.displaced_task_id;
                                         drop(queue);
+                                        if let Some(displaced_task_id) = displaced_task_id {
+                                            remove_task_aliases_for_task(&retry_task_aliases, &displaced_task_id).await;
+                                            remove_mrf_repair_notice_targets(
+                                                &retry_mrf_repair_notice_targets,
+                                                &displaced_task_id,
+                                            );
+                                        }
                                         retry_completed_heals.lock().await.remove(&retry_request_id);
                                         debug!(
                                             target: "rustfs::heal::manager",
@@ -439,8 +466,17 @@ impl HealManager {
                                         return;
                                     }
                                     HealAdmissionResult::Merged => {
+                                        let merged_task_id =
+                                            queue.queued_request_id_for_dedup_key(&retry_key).map(ToOwned::to_owned);
                                         retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
                                         drop(queue);
+                                        if let Some(merged_task_id) = merged_task_id {
+                                            move_mrf_repair_notice_targets(
+                                                &retry_mrf_repair_notice_targets,
+                                                &retry_request_id,
+                                                &merged_task_id,
+                                            );
+                                        }
                                         debug!(
                                             target: "rustfs::heal::manager",
                                             event = EVENT_HEAL_QUEUE_ADMISSION,
@@ -597,6 +633,43 @@ pub(super) fn running_heal_set_counts(active_heals: &HashMap<String, Arc<HealTas
     running
 }
 
+fn remove_mrf_repair_notice_targets(registry: &Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>, task_id: &str) {
+    lock_mrf_repair_notice_targets(registry).remove(task_id);
+}
+
+fn take_mrf_repair_notice_targets(
+    registry: &Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>,
+    task_id: &str,
+) -> Vec<MrfRepairNoticeTarget> {
+    lock_mrf_repair_notice_targets(registry).remove(task_id).unwrap_or_default()
+}
+
+fn move_mrf_repair_notice_targets(
+    registry: &Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>,
+    from_task_id: &str,
+    to_task_id: &str,
+) {
+    if from_task_id == to_task_id {
+        return;
+    }
+    let mut registry = lock_mrf_repair_notice_targets(registry);
+    let Some(moving) = registry.remove(from_task_id) else {
+        return;
+    };
+    let targets = registry.entry(to_task_id.to_string()).or_default();
+    for target in moving {
+        if !targets.contains(&target) {
+            targets.push(target);
+        }
+    }
+}
+
+fn emit_mrf_repaired_events(targets: Vec<MrfRepairNoticeTarget>) {
+    for target in targets {
+        rustfs_common::mrf_channel::note_mrf_repaired(&target.bucket, &target.object, target.version_id);
+    }
+}
+
 pub(super) fn heal_request_set_key_for_task(task: &HealTask) -> Option<String> {
     match &task.heal_type {
         HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()),

crates/heal/src/heal/manager/tests.rs

@@ -86,6 +86,7 @@ async fn process_manager_queue_once(manager: &HealManager) {
         completed_heals: &manager.completed_heals,
         task_aliases: &manager.task_aliases,
         retrying_heals: &manager.retrying_heals,
+        mrf_repair_notice_targets: &manager.mrf_repair_notice_targets,
         replacement_recovery_anchors: &manager.replacement_recovery_anchors,
         config: &manager.config,
         statistics: &manager.statistics,
@@ -2356,6 +2357,84 @@ async fn test_cancel_task_removes_queued_request() {
     assert!(matches!(manager.get_task_status(&request_id).await, Err(Error::TaskNotFound { .. })));
 }
 
+#[tokio::test]
+async fn test_mrf_repaired_notice_waits_for_successful_completion() {
+    let bucket = "mrf-completion-success";
+    let object = "object";
+    let version_id = Some([9u8; 16]);
+    let _ = rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket);
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = HealManager::new(storage, None);
+
+    let mut request = HealRequest::object(bucket.to_string(), object.to_string(), None);
+    request.source = HealRequestSource::Mrf;
+    let receipt = manager
+        .submit_mrf_heal_request_with_receipt(request, Arc::from(bucket), Arc::from(object), version_id)
+        .await
+        .expect("MRF request should be admitted");
+    assert_eq!(receipt.result, HealAdmissionResult::Accepted);
+    assert!(
+        manager
+            .mrf_repair_notice_targets
+            .lock()
+            .expect("mrf repair notice registry poisoned")
+            .contains_key(&receipt.task_id),
+        "MRF notice ownership must be registered before the scheduler can observe the queued task"
+    );
+
+    assert!(
+        rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket).is_empty(),
+        "admission alone must not clear the scanner pending-heal ledger"
+    );
+
+    process_manager_queue_once(&manager).await;
+    for _ in 0..100 {
+        let events = rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket);
+        if !events.is_empty() {
+            assert_eq!(events.len(), 1);
+            assert_eq!(events[0].object.as_ref(), object);
+            assert_eq!(events[0].version_id, version_id);
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("successful MRF-owned heal should emit one repaired event");
+}
+
+#[tokio::test]
+async fn test_mrf_repaired_notice_removed_on_queued_cancel_without_event() {
+    let bucket = "mrf-completion-cancel";
+    let object = "object";
+    let _ = rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket);
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = HealManager::new(storage, None);
+
+    let mut request = HealRequest::object(bucket.to_string(), object.to_string(), None);
+    request.source = HealRequestSource::Mrf;
+    let receipt = manager
+        .submit_mrf_heal_request_with_receipt(request, Arc::from(bucket), Arc::from(object), None)
+        .await
+        .expect("MRF request should be admitted");
+
+    manager
+        .cancel_task(&receipt.task_id)
+        .await
+        .expect("queued MRF request should cancel");
+
+    assert!(
+        rustfs_common::mrf_channel::take_mrf_repaired_events_for(bucket).is_empty(),
+        "cancelled MRF-owned heal must not emit a repaired event"
+    );
+    assert!(
+        manager
+            .mrf_repair_notice_targets
+            .lock()
+            .expect("mrf repair notice registry poisoned")
+            .is_empty(),
+        "cancel must discard completion notice ownership"
+    );
+}
+
 #[tokio::test]
 async fn test_cancel_tasks_for_path_removes_matching_queued_requests() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -2570,6 +2649,54 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
     );
 }
 
+#[tokio::test]
+async fn test_displacing_registered_mrf_task_drops_notice_ownership() {
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = HealManager::new(
+        storage,
+        Some(HealConfig {
+            queue_size: 1,
+            ..HealConfig::default()
+        }),
+    );
+
+    let mut low = HealRequest::object("bucket".to_string(), "object".to_string(), None);
+    low.source = HealRequestSource::Mrf;
+    low.priority = HealPriority::Low;
+    let high = HealRequest::new(
+        HealType::Bucket {
+            bucket: "manual-bucket".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::High,
+    );
+
+    assert_eq!(
+        manager
+            .submit_mrf_heal_request_with_receipt(low, Arc::from("bucket"), Arc::from("object"), None)
+            .await
+            .expect("low priority MRF request should be accepted first")
+            .result,
+        HealAdmissionResult::Accepted
+    );
+
+    assert_eq!(
+        manager
+            .submit_heal_request(high)
+            .await
+            .expect("high priority request should displace low priority work"),
+        HealAdmissionResult::Accepted
+    );
+    assert!(
+        manager
+            .mrf_repair_notice_targets
+            .lock()
+            .expect("mrf repair notice registry poisoned")
+            .is_empty(),
+        "displaced MRF-owned task cannot reach completion, so its notice ownership must be dropped"
+    );
+}
+
 #[tokio::test]
 async fn test_submit_heal_request_drops_read_repair_under_pressure() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/heal/mrf_queue.rs

@@ -356,6 +356,18 @@ pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest {
     request
 }
 
+async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> crate::Result<HealAdmissionResult> {
+    let receipt = manager
+        .submit_mrf_heal_request_with_receipt(
+            build_heal_request(intent),
+            intent.bucket.clone(),
+            intent.object.clone(),
+            intent.version_id,
+        )
+        .await?;
+    Ok(receipt.result)
+}
+
 struct MrfRuntime {
     queue: MrfQueue,
     config: MrfConsumerConfig,
@@ -409,16 +421,11 @@ impl MrfRuntime {
             // attempts counter) changes the encoded snapshot; mark it dirty
             // either way.
             self.dirty = true;
-            let request = build_heal_request(&intent);
+            match submit_mrf_heal_request(manager, &intent).await {
-            match manager.submit_heal_request(request).await {
                 // Accepted intents leave the pending set; the next flush persists the
-                // smaller snapshot, which is the journal's compaction. Fan out a
+                // smaller snapshot. The scanner ledger is cleared later, when the
-                // best-effort repaired notice so retry ledgers (the scanner's
+                // canonical heal task reaches a successful terminal completion.
-                // pending-heal oracle) can drop entries whose repair the manager
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
-                // now owns (backlog#1894 axis B).
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
-                    rustfs_common::mrf_channel::note_mrf_repaired(&intent.bucket, &intent.object, intent.version_id);
-                }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts >= MRF_MAX_ATTEMPTS {
@@ -520,11 +527,8 @@ async fn replay_into(
     // stays armed in `queue` for the consumer's retry loop.
     if backoff_until.is_none() {
         while let Some(mut intent) = queue.pop_front() {
-            let request = build_heal_request(&intent);
+            match submit_mrf_heal_request(manager, &intent).await {
-            match manager.submit_heal_request(request).await {
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
-                    rustfs_common::mrf_channel::note_mrf_repaired(&intent.bucket, &intent.object, intent.version_id);
-                }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts < MRF_MAX_ATTEMPTS {

crates/heal/tests/mrf_pipeline_test.rs

@@ -128,18 +128,9 @@ async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() {
         manager.operations_snapshot().await
     );
 
-    // Axis B (backlog#1894): the accepted dispatch must also fan out a
-    // repaired notice for the intent's bucket so the scanner ledger can drop
-    // its retry entry for the same target. Polled: the queue observation
-    // above can land between the manager push and the consumer's notice.
-    let noticed = wait_until(Duration::from_secs(10), || async {
-        !mrf_channel::take_mrf_repaired_events_for("mrf-bucket").is_empty()
-    })
-    .await;
-    assert!(noticed, "accepted intent must fan out a repaired notice");
     assert!(
         mrf_channel::take_mrf_repaired_events_for("mrf-bucket").is_empty(),
-        "notice take is destructive"
+        "accepted intent must wait for successful heal completion before repaired notice fan-out"
     );
 }