diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index a45407f05..bad6d4b5b 100644 --- a/crates/heal/src/heal/manager.rs +++ b/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; +#[derive(Debug, Clone, PartialEq, Eq)] +struct MrfRepairNoticeTarget { + bucket: Arc, + object: Arc, + version_id: Option<[u8; 16]>, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct HealAdmissionDecision { + result: HealAdmissionResult, + displaced_task_id: Option, +} + +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>>, +) -> StdMutexGuard<'_, HashMap>> { + match registry.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } +} + +async fn remove_task_aliases_for_task(registry: &Arc>>, 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>, key: &str) -> bool { - active_heal_for_dedup_key(active_heals, key).is_some() -} - fn active_heal_for_dedup_key(active_heals: &HashMap>, key: &str) -> Option<(String, HealType)> { active_heals .iter() @@ -581,6 +622,10 @@ pub struct HealManager { task_aliases: Arc>>, /// Heal tasks waiting for a retry backoff to expire. retrying_heals: Arc>>, + /// 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>>>, /// 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>>>, task_aliases: &'a Arc>>, retrying_heals: &'a Arc>>, + mrf_repair_notice_targets: &'a Arc>>>, replacement_recovery_anchors: &'a Arc>>, config: &'a Arc>, statistics: &'a Arc>, @@ -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>, + 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 - .lock() - .await - .retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id); + remove_task_aliases_for_task(&self.task_aliases, task_id).await; } 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 { + 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, + object: Arc, + version_id: Option<[u8; 16]>, + ) -> Result { + 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, ) -> Result { // 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 { diff --git a/crates/heal/src/heal/manager/auto_scan.rs b/crates/heal/src/heal/manager/auto_scan.rs index d828b446c..10ac136a4 100644 --- a/crates/heal/src/heal/manager/auto_scan.rs +++ b/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(); diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index 00012e2ad..0841d0e1c 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/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>>>, task_id: &str) { + lock_mrf_repair_notice_targets(registry).remove(task_id); +} + +fn take_mrf_repair_notice_targets( + registry: &Arc>>>, + task_id: &str, +) -> Vec { + lock_mrf_repair_notice_targets(registry).remove(task_id).unwrap_or_default() +} + +fn move_mrf_repair_notice_targets( + registry: &Arc>>>, + 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) { + 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 { match &task.heal_type { HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()), diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 8b955b172..b4e8737d1 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/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 = 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 = 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 = 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 = 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 = Arc::new(MockStorage); diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 362a32966..c1c4ea027 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/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 { + 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 manager.submit_heal_request(request).await { + match submit_mrf_heal_request(manager, &intent).await { // Accepted intents leave the pending set; the next flush persists the - // smaller snapshot, which is the journal's compaction. Fan out a - // best-effort repaired notice so retry ledgers (the scanner's - // pending-heal oracle) can drop entries whose repair the manager - // 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); - } + // smaller snapshot. The scanner ledger is cleared later, when the + // canonical heal task reaches a successful terminal completion. + Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {} 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 manager.submit_heal_request(request).await { - Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => { - rustfs_common::mrf_channel::note_mrf_repaired(&intent.bucket, &intent.object, intent.version_id); - } + match submit_mrf_heal_request(manager, &intent).await { + Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {} Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { intent.attempts = intent.attempts.saturating_add(1); if intent.attempts < MRF_MAX_ATTEMPTS { diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index 10bb03ce8..137cdd75d 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/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" ); }