feat(heal,scanner): best-effort repaired notices from the MRF consumer (#6283)

The scanner's pending-heal ledger and the MRF journal tracked the same
damaged objects with no cross-talk: once the consumer landed an intent
with the heal manager, the ledger's retry entry for that target kept
re-submitting a heal the manager already owned (backlog#1894 axis B).

Fan the acceptance out: both dispatch sites in the MRF queue (the live
consumer and the startup replay) record a compact MrfRepairedEvent
(bucket, object, version bytes) in a bounded process-wide ring owned by
rustfs-common. The scanner drains its own bucket's notices at the top
of retry_pending_scanner_heals and clears the matching Object-kind
ledger entries in one batched retain + sync (a mass-recovery first
sweep must not turn into thousands of full-table ledger clones on the
scan task), with nil notice UUIDs mapping to None per the repo-wide
defensive-UUID invariant so unversioned entries match unversioned
notices only. Notices are best-effort by design — a lost or capped-out
notice leaves the entry to expire through its own attempts/age limits,
because the ledger is a retry oracle, not a source of truth; other
buckets' notices stay queued for their own scanners. Neither persistent
format changes; old nodes that keep double-booking remain harmless.

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

crates/common/src/mrf_channel.rs

@@ -148,6 +148,62 @@ fn unix_now_ms() -> u64 {
         .unwrap_or(0)
 }
 
+/// A repair the MRF consumer landed, fanned out so retry ledgers can drop
+/// entries the journal no longer tracks (backlog#1894 axis B). The payload
+/// mirrors the intent identity so consumers match without re-parsing.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct MrfRepairedEvent {
+    pub bucket: Arc<str>,
+    pub object: Arc<str>,
+    pub version_id: Option<[u8; 16]>,
+}
+
+/// Bound on the repaired-event backlog. Notices are best-effort hints; when
+/// the ring is full the oldest are dropped and the affected ledger entries
+/// simply expire through their own attempts/age limits.
+const MRF_REPAIRED_EVENT_CAP: usize = 4096;
+
+static MRF_REPAIRED_EVENTS: OnceLock<std::sync::Mutex<std::collections::VecDeque<MrfRepairedEvent>>> = OnceLock::new();
+
+/// Record that the MRF consumer landed a repair. Never blocks: the critical
+/// section is a deque push under a std mutex.
+pub fn note_mrf_repaired(bucket: &str, object: &str, version_id: Option<[u8; 16]>) {
+    let registry = MRF_REPAIRED_EVENTS.get_or_init(|| std::sync::Mutex::new(std::collections::VecDeque::new()));
+    let Ok(mut events) = registry.lock() else {
+        return;
+    };
+    if events.len() >= MRF_REPAIRED_EVENT_CAP {
+        events.pop_front();
+    }
+    events.push_back(MrfRepairedEvent {
+        bucket: Arc::from(bucket),
+        object: Arc::from(object),
+        version_id,
+    });
+}
+
+/// Take the repair notices recorded for `bucket`, leaving other buckets'
+/// notices in place for their own scanners.
+pub fn take_mrf_repaired_events_for(bucket: &str) -> Vec<MrfRepairedEvent> {
+    let Some(registry) = MRF_REPAIRED_EVENTS.get() else {
+        return Vec::new();
+    };
+    let Ok(mut events) = registry.lock() else {
+        return Vec::new();
+    };
+    let mut taken = Vec::new();
+    let mut retained = std::collections::VecDeque::with_capacity(events.len());
+    while let Some(event) = events.pop_front() {
+        if event.bucket.as_ref() == bucket {
+            taken.push(event);
+        } else {
+            retained.push_back(event);
+        }
+    }
+    *events = retained;
+    taken
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -200,4 +256,32 @@ mod tests {
         assert!(!try_send_mrf_intent(MrfKind::MetadataCorruption, "b", "o", None));
         set_mrf_delivery_enabled(true);
     }
+
+    #[test]
+    fn repaired_events_take_is_bucket_scoped_and_cap_bounded() {
+        // Distinct buckets keep their notices until their own scanner takes
+        // them; a take for one bucket leaves the others' notices in place.
+        note_mrf_repaired("bucket-a", "object-1", None);
+        note_mrf_repaired("bucket-b", "object-2", None);
+        note_mrf_repaired("bucket-a", "object-3", None);
+
+        let taken_a = take_mrf_repaired_events_for("bucket-a");
+        assert_eq!(taken_a.len(), 2);
+        assert_eq!(taken_a[0].object.as_ref(), "object-1");
+        assert_eq!(taken_a[1].object.as_ref(), "object-3");
+        assert!(take_mrf_repaired_events_for("bucket-a").is_empty(), "take is destructive per bucket");
+
+        let taken_b = take_mrf_repaired_events_for("bucket-b");
+        assert_eq!(taken_b.len(), 1);
+        assert_eq!(taken_b[0].object.as_ref(), "object-2");
+
+        // Cap bound: flooding the ring drops the oldest notices rather than
+        // growing unbounded.
+        for i in 0..=(MRF_REPAIRED_EVENT_CAP + 8) {
+            note_mrf_repaired("flood-bucket", &format!("object-{i}"), None);
+        }
+        let flooded = take_mrf_repaired_events_for("flood-bucket");
+        assert_eq!(flooded.len(), MRF_REPAIRED_EVENT_CAP);
+        assert_eq!(flooded[0].object.as_ref(), "object-9", "the oldest notices past the cap are dropped");
+    }
 }

crates/heal/src/heal/mrf_queue.rs

@@ -409,8 +409,13 @@ impl MrfRuntime {
             let request = build_heal_request(&intent);
             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.
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
+                // 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);
+                }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts >= MRF_MAX_ATTEMPTS {
@@ -514,7 +519,9 @@ async fn replay_into(
         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) => {}
+                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

@@ -127,6 +127,20 @@ async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() {
         "MRF intent must reach the manager queue as an Urgent request (snapshot: {:?})",
         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"
+    );
 }
 
 /// A journal left behind by a previous process must be replayed into the

crates/scanner/src/scanner_folder.rs

@@ -700,6 +700,16 @@ fn pending_scanner_heal_identity(entry: &PendingScannerHeal) -> (u8, &str, Optio
     (kind, entry.bucket.as_str(), entry.object.as_deref(), entry.version_id.as_deref())
 }
 
+/// Decode an MRF repaired-notice version id for ledger matching. A nil UUID
+/// means "no value" per the repo-wide defensive-UUID invariant, so it maps
+/// to `None` and matches unversioned ledger entries only.
+fn mrf_repaired_version_id(version_id: Option<[u8; 16]>) -> Option<String> {
+    version_id
+        .map(uuid::Uuid::from_bytes)
+        .filter(|uuid| !uuid.is_nil())
+        .map(|uuid| uuid.to_string())
+}
+
 fn sort_pending_scanner_heals_for_retry(entries: &mut [PendingScannerHeal]) {
     entries.sort_by(|a, b| {
         a.last_attempt
@@ -1632,6 +1642,32 @@ impl FolderScanner {
         }
     }
 
+    /// Batched variant of [`Self::clear_pending_scanner_heal`] for repaired
+    /// notices (backlog#1894 axis B): one retain pass and one ledger sync
+    /// for the whole notice set, so a mass-recovery first sweep cannot turn
+    /// into thousands of full-table clones on the scan task. Only Object
+    /// entries match — bucket-level heals are never the MRF consumer's work.
+    fn clear_pending_scanner_heals_for_repaired(&mut self, events: &[rustfs_common::mrf_channel::MrfRepairedEvent]) {
+        // Pre-resolve the notice version strings once; each ledger entry then
+        // compares against plain Option<&str>.
+        let targets: Vec<(&str, &str, Option<String>)> = events
+            .iter()
+            .map(|event| (event.bucket.as_ref(), event.object.as_ref(), mrf_repaired_version_id(event.version_id)))
+            .collect();
+        let before = self.new_cache.info.pending_heals.len();
+        self.new_cache.info.pending_heals.retain(|entry| {
+            entry.kind != PendingScannerHealKind::Object
+                || !targets.iter().any(|(bucket, object, version)| {
+                    entry.bucket.as_str() == *bucket
+                        && entry.object.as_deref() == Some(*object)
+                        && entry.version_id.as_deref() == version.as_deref()
+                })
+        });
+        if self.new_cache.info.pending_heals.len() != before {
+            self.sync_pending_heals();
+        }
+    }
+
     fn record_pending_scanner_heal(
         &mut self,
         kind: PendingScannerHealKind,
@@ -1970,6 +2006,14 @@ impl FolderScanner {
         }
 
         let bucket = self.new_cache.info.name.clone();
+        // Backlog#1894 axis B: repairs the MRF consumer landed hand the
+        // manager the heal task, so the matching pending-ledger entries are
+        // retried nowhere — drop them here. Best-effort: a lost notice just
+        // leaves the entry to expire through its own attempts/age limits.
+        let repaired = rustfs_common::mrf_channel::take_mrf_repaired_events_for(&bucket);
+        if !repaired.is_empty() {
+            self.clear_pending_scanner_heals_for_repaired(&repaired);
+        }
         for pending in pending_scanner_heal_retry_candidates(&self.new_cache.info.pending_heals, &bucket) {
             if !self.should_heal().await {
                 break;
@@ -4324,6 +4368,86 @@ mod tests {
         }
     }
 
+    /// The nil-UUID branch of the defensive-UUID invariant: a nil version in
+    /// a repaired notice means "no value" and must match unversioned ledger
+    /// entries only.
+    #[test]
+    fn test_mrf_repaired_version_id_maps_nil_to_none() {
+        assert_eq!(mrf_repaired_version_id(None), None);
+        assert_eq!(mrf_repaired_version_id(Some([0u8; 16])), None);
+        let uuid = Uuid::new_v4();
+        assert_eq!(mrf_repaired_version_id(Some(*uuid.as_bytes())), Some(uuid.to_string()));
+    }
+
+    /// Full wiring of backlog#1894 axis B: notes taken for the scanned bucket
+    /// clear exactly the matching Object ledger entries — bucket-level
+    /// entries, other buckets' entries, and version-mismatched entries
+    /// survive; a real (non-nil) version matches only the same version.
+    #[tokio::test]
+    async fn test_mrf_repaired_notices_clear_matching_ledger_entries() {
+        use rustfs_common::mrf_channel::note_mrf_repaired;
+
+        let (mut scanner, temp_dir) = build_test_scanner().await;
+        let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir);
+        scanner.new_cache.info.name = "bucket".to_string();
+        scanner.update_cache.info.name = "bucket".to_string();
+        scanner.heal_object_select = 1;
+
+        let version = Uuid::new_v4().to_string();
+        scanner.new_cache.info.pending_heals = vec![
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-a"), None, 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-b"), Some(&version), 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-c"), None, 1, 1),
+            pending_heal(
+                PendingScannerHealKind::Object,
+                "bucket",
+                Some("object-c"),
+                Some("00000000-0000-0000-0000-000000000001"),
+                1,
+                1,
+            ),
+            pending_heal(PendingScannerHealKind::Bucket, "bucket", None, None, 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "other-bucket", Some("object-a"), None, 1, 1),
+        ];
+
+        note_mrf_repaired("bucket", "object-a", None);
+        note_mrf_repaired("bucket", "object-b", Some(*Uuid::parse_str(&version).unwrap().as_bytes()));
+        // A nil-UUID notice for object-c means "no value": it clears the
+        // unversioned entry but must not touch the versioned one.
+        note_mrf_repaired("bucket", "object-c", Some([0u8; 16]));
+        // A notice for a target the ledger does not track must be a no-op.
+        note_mrf_repaired("bucket", "object-untracked", None);
+
+        scanner
+            .retry_pending_scanner_heals()
+            .await
+            .expect("retry pass should succeed");
+
+        let survivors: Vec<(PendingScannerHealKind, &str, Option<&str>, Option<&str>)> = scanner
+            .new_cache
+            .info
+            .pending_heals
+            .iter()
+            .map(|entry| (entry.kind, entry.bucket.as_str(), entry.object.as_deref(), entry.version_id.as_deref()))
+            .collect();
+        // Cleared: object-a (no version), object-b (exact version match), and
+        // object-c's unversioned entry (the nil branch matched no-version
+        // only — the versioned object-c entry survives).
+        assert_eq!(
+            survivors,
+            vec![
+                (
+                    PendingScannerHealKind::Object,
+                    "bucket",
+                    Some("object-c"),
+                    Some("00000000-0000-0000-0000-000000000001")
+                ),
+                (PendingScannerHealKind::Bucket, "bucket", None, None),
+                (PendingScannerHealKind::Object, "other-bucket", Some("object-a"), None),
+            ]
+        );
+    }
+
     #[test]
     fn test_pending_heal_reconstructs_bucket_request() {
         let pending = pending_heal(PendingScannerHealKind::Bucket, "bucket", None, None, 1, 1);