From d6efb655880a8ae3ac68a7950bf3a97f5d4d609c Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 20 Aug 2026 01:31:26 +0800 Subject: [PATCH] feat(heal,scanner): best-effort repaired notices from the MRF consumer (#6283) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/common/src/mrf_channel.rs | 84 +++++++++++++++++ crates/heal/src/heal/mrf_queue.rs | 13 ++- crates/heal/tests/mrf_pipeline_test.rs | 14 +++ crates/scanner/src/scanner_folder.rs | 124 +++++++++++++++++++++++++ 4 files changed, 232 insertions(+), 3 deletions(-) diff --git a/crates/common/src/mrf_channel.rs b/crates/common/src/mrf_channel.rs index f0a91a238..0217f7047 100644 --- a/crates/common/src/mrf_channel.rs +++ b/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, + pub object: Arc, + 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>> = 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 { + 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"); + } } diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 5101a5e12..8a7e1e3b5 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/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 { diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index 36f07a5a7..10bb03ce8 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/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 diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 4b279eea2..70db02ba4 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/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 { + 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)> = 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);