From 32cc7c8fcf963c41c8adb0783b795be0bd003b02 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 23 Aug 2026 16:45:22 +0800 Subject: [PATCH] fix(heal): coalesce duplicate MRF intents (#6425) Co-authored-by: houseme --- crates/common/src/mrf_channel.rs | 343 ++++++++++++++- .../src/set_disk/core/io_primitives.rs | 7 +- crates/ecstore/src/set_disk/ops/object.rs | 23 +- crates/heal/src/heal/manager.rs | 57 ++- crates/heal/src/heal/manager/auto_scan.rs | 13 +- crates/heal/src/heal/manager/queue.rs | 6 +- crates/heal/src/heal/manager/scheduler.rs | 37 +- crates/heal/src/heal/mrf_queue.rs | 409 +++++++++++++++--- crates/heal/tests/mrf_pipeline_test.rs | 74 +++- crates/scanner/src/scanner_folder.rs | 5 +- .../src/scanner_folder/item_actions.rs | 11 +- crates/scanner/src/scanner_folder/tests.rs | 16 +- 12 files changed, 903 insertions(+), 98 deletions(-) diff --git a/crates/common/src/mrf_channel.rs b/crates/common/src/mrf_channel.rs index 0217f7047..f83b78f96 100644 --- a/crates/common/src/mrf_channel.rs +++ b/crates/common/src/mrf_channel.rs @@ -23,21 +23,32 @@ //! unconsumed intents is the consumer's job (see `rustfs-heal` //! `heal::mrf_queue`), mirroring MinIO's `.heal/mrf/list.bin`. +use std::collections::HashMap; +use std::collections::hash_map::RandomState; +use std::hash::{BuildHasher, Hash}; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::AtomicUsize; use std::sync::{ - Arc, OnceLock, + Arc, Mutex, OnceLock, atomic::{AtomicBool, Ordering}, }; +use std::time::{Duration, Instant}; use tokio::sync::mpsc; use uuid::Uuid; /// Bounded capacity of the global MRF channel. Backpressure is resolved by /// dropping (and counting) intents, never by blocking the producer. const MRF_CHANNEL_CAPACITY: usize = 8192; +const MRF_COALESCER_SHARDS: usize = 16; +const MRF_COALESCER_MAX_KEYS: usize = 8192; +const MRF_COALESCER_MAX_BYTES: usize = 16 * 1024 * 1024; +const MRF_COALESCER_TTL: Duration = Duration::from_secs(60); +const MRF_MAX_IDENTITY_COMPONENT: usize = 1024; /// Why an intent was produced. Drives the heal priority mapping on the /// consumer side (DecodeFailure -> Urgent, MetadataCorruption -> High, /// PartialWrite -> Normal). -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub enum MrfKind { /// Erasure decode failed while serving a read (read path). DecodeFailure, @@ -67,12 +78,52 @@ pub struct MrfIntent { /// Version the intent targets, as raw UUID bytes. pub version_id: Option<[u8; 16]>, pub kind: MrfKind, + /// Stable erasure-set scope when the producer has it. Kept optional so + /// metadata corruption and legacy producers do not invent a scope. + pub scope: Option, + /// Generation of the node-local ingress lease. It is not persisted in + /// the journal; replayed records acquire a fresh lease when re-enqueued. + pub lease: Option, pub enqueued_at_ms: u64, /// Times this intent has already been offered to the heal manager. /// Dropped by the consumer once it reaches `MRF_MAX_ATTEMPTS`. pub attempts: u8, } +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct MrfScope { + pub pool_index: u32, + pub set_index: u32, +} + +/// Opaque generation used to release exactly the admission that created an +/// ingress entry. A generation prevents a late terminal callback from +/// deleting a newer retry for the same identity (ABA). +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct MrfIngressLease(u64); + +impl MrfIngressLease { + const fn new(value: u64) -> Self { + Self(value) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum MrfDropReason { + Disabled, + Uninitialized, + Full, + OversizedIdentity, + CoalescerFull, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum MrfIngressResult { + Enqueued, + Coalesced, + Dropped(MrfDropReason), +} + /// Consumer-side retry ceiling before an intent is given up on. pub const MRF_MAX_ATTEMPTS: u8 = 3; @@ -87,6 +138,159 @@ impl MrfIntent { static GLOBAL_MRF_SENDER: OnceLock> = OnceLock::new(); +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +struct MrfIdentityKey { + kind: MrfKind, + bucket: Arc, + object: Arc, + version_id: Option<[u8; 16]>, + scope: Option, +} + +#[derive(Debug)] +struct IngressEntry { + lease: MrfIngressLease, + expires_at: Instant, + bytes: usize, +} + +type MrfCoalescerShard = Mutex>; +type MrfCoalescer = Box<[MrfCoalescerShard]>; + +static MRF_COALESCER: OnceLock = OnceLock::new(); +static NEXT_MRF_LEASE: AtomicU64 = AtomicU64::new(1); +static MRF_COALESCER_COUNT: AtomicUsize = AtomicUsize::new(0); +static MRF_COALESCER_BYTES: AtomicUsize = AtomicUsize::new(0); +static MRF_HASH_STATE: OnceLock = OnceLock::new(); + +fn coalescer() -> &'static [MrfCoalescerShard] { + MRF_COALESCER.get_or_init(|| { + (0..MRF_COALESCER_SHARDS) + .map(|_| Mutex::new(HashMap::new())) + .collect::>() + .into_boxed_slice() + }) +} + +fn key_shard(key: &MrfIdentityKey) -> usize { + let hash = MRF_HASH_STATE.get_or_init(RandomState::new).hash_one(key); + usize::try_from(hash).unwrap_or(0) % MRF_COALESCER_SHARDS +} + +fn canonical_version(version_id: Option) -> Option<[u8; 16]> { + version_id + .filter(|version| !version.is_nil()) + .map(|version| *version.as_bytes()) +} + +fn canonical_identity( + kind: MrfKind, + version_id: Option<[u8; 16]>, + scope: Option, +) -> (Option<[u8; 16]>, Option) { + let version_id = version_id.filter(|bytes| *bytes != [0; 16]); + match kind { + MrfKind::MetadataCorruption => (None, None), + MrfKind::DecodeFailure | MrfKind::PartialWrite => (version_id, scope), + } +} + +fn identity_estimated_bytes(key: &MrfIdentityKey) -> usize { + 64usize + .saturating_add(key.bucket.len()) + .saturating_add(key.object.len()) + .saturating_add(key.version_id.map_or(0, |_| 16)) + .saturating_add(key.scope.map_or(0, |_| 8)) +} + +fn reserve(counter: &AtomicUsize, limit: usize, amount: usize) -> bool { + let mut current = counter.load(Ordering::Relaxed); + loop { + let Some(next) = current.checked_add(amount) else { + return false; + }; + if next > limit { + return false; + } + match counter.compare_exchange_weak(current, next, Ordering::Relaxed, Ordering::Relaxed) { + Ok(_) => return true, + Err(observed) => current = observed, + } + } +} + +fn coalescer_admit(key: MrfIdentityKey) -> Result { + let shard = key_shard(&key); + let mut entries = coalescer()[shard] + .lock() + .map_err(|_| MrfIngressResult::Dropped(MrfDropReason::CoalescerFull))?; + let now = Instant::now(); + let before = entries.len(); + let mut expired_bytes = 0usize; + entries.retain(|_, entry| { + if entry.expires_at > now { + true + } else { + expired_bytes = expired_bytes.saturating_add(entry.bytes); + false + } + }); + let evicted = before.saturating_sub(entries.len()); + if evicted > 0 { + MRF_COALESCER_COUNT.fetch_sub(evicted, Ordering::Relaxed); + MRF_COALESCER_BYTES.fetch_sub(expired_bytes, Ordering::Relaxed); + let evicted = u64::try_from(evicted).unwrap_or(u64::MAX); + metrics::counter!("rustfs_heal_mrf_coalescer_expired_total").increment(evicted); + metrics::counter!("rustfs_heal_mrf_coalescer_evictions_total").increment(evicted); + } + if entries.contains_key(&key) { + metrics::counter!("rustfs_heal_mrf_coalesced_total").increment(1); + return Err(MrfIngressResult::Coalesced); + } + let bytes = identity_estimated_bytes(&key); + let count_reserved = reserve(&MRF_COALESCER_COUNT, MRF_COALESCER_MAX_KEYS, 1); + let bytes_reserved = count_reserved && reserve(&MRF_COALESCER_BYTES, MRF_COALESCER_MAX_BYTES, bytes); + if !count_reserved || !bytes_reserved { + if count_reserved { + MRF_COALESCER_COUNT.fetch_sub(1, Ordering::Relaxed); + } + metrics::counter!("rustfs_heal_mrf_dropped_total", "reason" => "coalescer_full").increment(1); + return Err(MrfIngressResult::Dropped(MrfDropReason::CoalescerFull)); + } + let lease = MrfIngressLease::new(NEXT_MRF_LEASE.fetch_add(1, Ordering::Relaxed)); + if entries + .insert( + key, + IngressEntry { + lease, + expires_at: now + MRF_COALESCER_TTL, + bytes, + }, + ) + .is_some() + { + MRF_COALESCER_COUNT.fetch_sub(1, Ordering::Relaxed); + MRF_COALESCER_BYTES.fetch_sub(bytes, Ordering::Relaxed); + metrics::counter!("rustfs_heal_mrf_coalesced_total").increment(1); + return Err(MrfIngressResult::Coalesced); + } + Ok(lease) +} + +fn coalescer_release(key: &MrfIdentityKey, lease: Option) { + let Some(lease) = lease else { + return; + }; + if let Ok(mut entries) = coalescer()[key_shard(key)].lock() { + let should_remove = entries.get(key).is_some_and(|entry| entry.lease == lease); + if should_remove { + let bytes = entries.remove(key).map(|entry| entry.bytes).unwrap_or(0); + MRF_COALESCER_COUNT.fetch_sub(1, Ordering::Relaxed); + MRF_COALESCER_BYTES.fetch_sub(bytes, Ordering::Relaxed); + } + } +} + /// Delivery kill-switch, set from `RUSTFS_HEAL_MRF_ENABLE`. Producers check /// this before touching the channel so the disabled path stays allocation- and /// sync-free. @@ -122,21 +326,90 @@ pub fn init_mrf_channel() -> Result, &'static str> { /// This runs on IO error paths, so it stays synchronous and cheap: one /// bounded allocation for the two `Arc` handles plus the channel slot. pub fn try_send_mrf_intent(kind: MrfKind, bucket: &str, object: &str, version_id: Option) -> bool { + matches!( + try_send_mrf_intent_typed(kind, bucket, object, version_id, None), + MrfIngressResult::Enqueued + ) +} + +/// Typed ingress result. `Coalesced` means an equivalent in-flight channel +/// intent already exists; it is not a second executable or durable admission. +pub fn try_send_mrf_intent_typed( + kind: MrfKind, + bucket: &str, + object: &str, + version_id: Option, + scope: Option, +) -> MrfIngressResult { if !mrf_delivery_enabled() { - return false; + return MrfIngressResult::Dropped(MrfDropReason::Disabled); } let Some(sender) = GLOBAL_MRF_SENDER.get() else { - return false; + return MrfIngressResult::Dropped(MrfDropReason::Uninitialized); }; - let intent = MrfIntent { + if bucket.len() > MRF_MAX_IDENTITY_COMPONENT || object.len() > MRF_MAX_IDENTITY_COMPONENT { + return MrfIngressResult::Dropped(MrfDropReason::OversizedIdentity); + } + let (version_id, scope) = canonical_identity(kind, canonical_version(version_id), scope); + let key = MrfIdentityKey { + kind, bucket: Arc::from(bucket), object: Arc::from(object), - version_id: version_id.map(|vid| *vid.as_bytes()), + version_id, + scope, + }; + let lease = match coalescer_admit(key.clone()) { + Ok(lease) => lease, + Err(result) => return result, + }; + let intent = MrfIntent { + bucket: key.bucket.clone(), + object: key.object.clone(), + version_id: key.version_id, kind, + scope, + lease: Some(lease), enqueued_at_ms: unix_now_ms(), attempts: 0, }; - sender.try_send(intent).is_ok() + match sender.try_send(intent) { + Ok(()) => MrfIngressResult::Enqueued, + Err(mpsc::error::TrySendError::Full(_)) => { + coalescer_release(&key, Some(lease)); + metrics::counter!("rustfs_heal_mrf_dropped_total", "reason" => "channel_full").increment(1); + MrfIngressResult::Dropped(MrfDropReason::Full) + } + Err(mpsc::error::TrySendError::Closed(_)) => { + coalescer_release(&key, Some(lease)); + MrfIngressResult::Dropped(MrfDropReason::Uninitialized) + } + } +} + +/// Release the ingress key once the consumer owns the intent. +pub fn release_mrf_intent(intent: &MrfIntent) { + release_mrf_identity(intent.kind, &intent.bucket, &intent.object, intent.version_id, intent.scope, intent.lease); +} + +pub fn release_mrf_identity( + kind: MrfKind, + bucket: &str, + object: &str, + version_id: Option<[u8; 16]>, + scope: Option, + lease: Option, +) { + let (version_id, scope) = canonical_identity(kind, version_id, scope); + coalescer_release( + &MrfIdentityKey { + kind, + bucket: Arc::from(bucket), + object: Arc::from(object), + version_id, + scope, + }, + lease, + ); } fn unix_now_ms() -> u64 { @@ -144,7 +417,8 @@ fn unix_now_ms() -> u64 { // failure would be a bug rather than something to handle here. std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) + .ok() + .and_then(|d| u64::try_from(d.as_millis()).ok()) .unwrap_or(0) } @@ -215,12 +489,60 @@ mod tests { object: Arc::from("object"), version_id: Some([0u8; 16]), kind: MrfKind::DecodeFailure, + scope: None, + lease: None, enqueued_at_ms: 0, attempts: 0, }; assert!(intent.estimated_bytes() >= intent.bucket.len() + intent.object.len()); } + #[test] + fn ingress_duplicate_identity_coalesces_and_releases_for_retry() { + let key = MrfIdentityKey { + kind: MrfKind::DecodeFailure, + bucket: Arc::from("ingress-test-bucket"), + object: Arc::from("ingress-test-object"), + version_id: Some([9; 16]), + scope: Some(MrfScope { + pool_index: 3, + set_index: 4, + }), + }; + let lease = coalescer_admit(key.clone()).expect("first identity should be admitted"); + for _ in 0..999 { + assert_eq!(coalescer_admit(key.clone()), Err(MrfIngressResult::Coalesced)); + } + coalescer_release(&key, Some(lease)); + let retry_lease = coalescer_admit(key.clone()).expect("released identity must admit a retry"); + coalescer_release(&key, Some(retry_lease)); + } + + #[test] + fn ingress_identity_preserves_kind_scope_and_version_boundaries() { + let (nil_version, nil_scope) = canonical_identity( + MrfKind::DecodeFailure, + Some([0; 16]), + Some(MrfScope { + pool_index: 1, + set_index: 2, + }), + ); + assert_eq!(nil_version, None, "nil UUID is the unversioned identity"); + assert!(nil_scope.is_some()); + + let (metadata_version, metadata_scope) = canonical_identity( + MrfKind::MetadataCorruption, + Some([7; 16]), + Some(MrfScope { + pool_index: 1, + set_index: 2, + }), + ); + assert_eq!(metadata_version, None); + assert_eq!(metadata_scope, None); + } + #[tokio::test] async fn try_send_delivers_and_respects_capacity() { let mut receiver = init_mrf_channel().expect("first initialization should succeed"); @@ -230,6 +552,7 @@ mod tests { let intent = receiver.recv().await.expect("intent should arrive"); assert_eq!(intent.kind, MrfKind::DecodeFailure); assert_eq!(intent.bucket.as_ref(), "b"); + release_mrf_intent(&intent); // Disable delivery: producers become no-ops. set_mrf_delivery_enabled(false); @@ -239,8 +562,8 @@ mod tests { // Fill the bounded channel past capacity: excess intents are dropped, // never blocking. let mut accepted = 0; - for _ in 0..(MRF_CHANNEL_CAPACITY + 64) { - if try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None) { + for index in 0..(MRF_CHANNEL_CAPACITY + 64) { + if try_send_mrf_intent(MrfKind::PartialWrite, "b", &format!("o-{index}"), None) { accepted += 1; } } diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 853eff9fd..a7310b44f 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -1427,8 +1427,11 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter( // Reservation won: this sighting owns the repair records for the object, // including the durable journal intent when the caller asked for one. - if let Some((kind, version_uuid)) = mrf_intent { - rustfs_common::mrf_channel::try_send_mrf_intent(kind, bucket, object, version_uuid); + if let Some((kind, version_uuid)) = mrf_intent + && let (Ok(pool_index), Ok(set_index)) = (u32::try_from(pool_index), u32::try_from(set_index)) + { + let scope = rustfs_common::mrf_channel::MrfScope { pool_index, set_index }; + let _ = rustfs_common::mrf_channel::try_send_mrf_intent_typed(kind, bucket, object, version_uuid, Some(scope)); } let mut request = rustfs_common::heal_channel::create_heal_request_with_options( diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 1ceb8acdf..86f75eb8d 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -6642,12 +6642,23 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> { // MRF journal intent: partial-write recovery must survive a restart // (HS-01); the heal request below remains the in-memory fast path. - rustfs_common::mrf_channel::try_send_mrf_intent( - rustfs_common::mrf_channel::MrfKind::PartialWrite, - bucket, - object, - uuid::Uuid::try_parse(version_id).ok(), - ); + let version_uuid = if version_id.is_empty() { + Some(None) + } else { + uuid::Uuid::try_parse(version_id).ok().map(Some) + }; + if let Some(version_uuid) = version_uuid + && let (Ok(pool_index), Ok(set_index)) = (u32::try_from(self.pool_index), u32::try_from(self.set_index)) + { + let scope = rustfs_common::mrf_channel::MrfScope { pool_index, set_index }; + let _ = rustfs_common::mrf_channel::try_send_mrf_intent_typed( + rustfs_common::mrf_channel::MrfKind::PartialWrite, + bucket, + object, + version_uuid, + Some(scope), + ); + } let mut request = rustfs_common::heal_channel::create_heal_request_with_options( bucket.to_string(), Some(object.to_string()), diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index fc8255e0b..30ef94446 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -119,6 +119,9 @@ struct MrfRepairNoticeTarget { bucket: Arc, object: Arc, version_id: Option<[u8; 16]>, + kind: rustfs_common::mrf_channel::MrfKind, + scope: Option, + lease: Option, } #[derive(Debug, Clone)] @@ -889,7 +892,19 @@ impl HealManager { } 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); + let targets = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).remove(task_id); + if let Some(targets) = targets { + for target in targets { + rustfs_common::mrf_channel::release_mrf_identity( + target.kind, + &target.bucket, + &target.object, + target.version_id, + target.scope, + target.lease, + ); + } + } } fn insert_mrf_repair_notice_target( @@ -1279,7 +1294,20 @@ impl HealManager { } self.task_aliases.lock().await.clear(); self.retrying_heals.lock().await.clear(); - lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).clear(); + let mrf_targets = { + let mut registry = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets); + registry.drain().flat_map(|(_, targets)| targets).collect::>() + }; + for target in mrf_targets { + rustfs_common::mrf_channel::release_mrf_identity( + target.kind, + &target.bucket, + &target.object, + target.version_id, + target.scope, + target.lease, + ); + } crate::set_heal_queue_length(0); // update state @@ -1311,12 +1339,32 @@ impl HealManager { .await } + #[cfg(test)] pub(crate) async fn submit_mrf_heal_request_with_receipt( &self, request: HealRequest, bucket: Arc, object: Arc, version_id: Option<[u8; 16]>, + ) -> Result { + let kind = match &request.heal_type { + HealType::Metadata { .. } => rustfs_common::mrf_channel::MrfKind::MetadataCorruption, + HealType::ECDecode { .. } => rustfs_common::mrf_channel::MrfKind::DecodeFailure, + _ => rustfs_common::mrf_channel::MrfKind::PartialWrite, + }; + self.submit_mrf_heal_request_with_receipt_and_identity(request, bucket, object, version_id, kind, None, None) + .await + } + + pub(crate) async fn submit_mrf_heal_request_with_receipt_and_identity( + &self, + request: HealRequest, + bucket: Arc, + object: Arc, + version_id: Option<[u8; 16]>, + kind: rustfs_common::mrf_channel::MrfKind, + scope: Option, + lease: Option, ) -> Result { self.submit_heal_request_with_receipt_alias_and_mrf_notice( request, @@ -1325,6 +1373,9 @@ impl HealManager { bucket, object, version_id, + kind, + scope, + lease, }), ) .await @@ -1539,7 +1590,7 @@ impl HealManager { 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); + self.remove_mrf_repair_notice_targets_for_task(displaced_task_id); } drop(retrying_heals); drop(queue); diff --git a/crates/heal/src/heal/manager/auto_scan.rs b/crates/heal/src/heal/manager/auto_scan.rs index f7eb4482d..d457d3ee9 100644 --- a/crates/heal/src/heal/manager/auto_scan.rs +++ b/crates/heal/src/heal/manager/auto_scan.rs @@ -506,7 +506,18 @@ impl HealManager { &displaced_terminal, ) .await; - lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id); + if let Some(targets) = lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id) { + for target in targets { + rustfs_common::mrf_channel::release_mrf_identity( + target.kind, + &target.bucket, + &target.object, + target.version_id, + target.scope, + target.lease, + ); + } + } } if matches!(admission, HealAdmissionResult::Accepted) { if should_notify { diff --git a/crates/heal/src/heal/manager/queue.rs b/crates/heal/src/heal/manager/queue.rs index 5bf0ae1e6..28bedaba4 100644 --- a/crates/heal/src/heal/manager/queue.rs +++ b/crates/heal/src/heal/manager/queue.rs @@ -299,7 +299,11 @@ impl PriorityHealQueue { /// Create a deduplication key from a heal request pub(super) fn make_dedup_key(request: &HealRequest) -> String { - Self::make_dedup_key_for_type(&request.heal_type) + let base = Self::make_dedup_key_for_type(&request.heal_type); + match (&request.heal_type, request.options.set_key()) { + (HealType::Object { .. } | HealType::ECDecode { .. }, Some(scope)) => format!("{base}:scope:{scope}"), + _ => base, + } } pub(super) fn make_dedup_key_for_type(heal_type: &HealType) -> String { diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index c76a037ae..c0deae524 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -349,6 +349,8 @@ impl HealManager { let notice_targets = take_mrf_repair_notice_targets(&mrf_repair_notice_targets_clone, &task_id); if successful_completion { emit_mrf_repaired_events(notice_targets); + } else { + release_mrf_repair_notice_targets(notice_targets); } task_aliases_clone .lock() @@ -638,7 +640,19 @@ pub(super) fn running_heal_set_counts(active_heals: &HashMap>>>, task_id: &str) { - lock_mrf_repair_notice_targets(registry).remove(task_id); + let targets = lock_mrf_repair_notice_targets(registry).remove(task_id); + if let Some(targets) = targets { + for target in targets { + rustfs_common::mrf_channel::release_mrf_identity( + target.kind, + &target.bucket, + &target.object, + target.version_id, + target.scope, + target.lease, + ); + } + } } fn take_mrf_repair_notice_targets( @@ -671,6 +685,27 @@ fn move_mrf_repair_notice_targets( 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); + rustfs_common::mrf_channel::release_mrf_identity( + target.kind, + &target.bucket, + &target.object, + target.version_id, + target.scope, + target.lease, + ); + } +} + +fn release_mrf_repair_notice_targets(targets: Vec) { + for target in targets { + rustfs_common::mrf_channel::release_mrf_identity( + target.kind, + &target.bucket, + &target.object, + target.version_id, + target.scope, + target.lease, + ); } } diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index c1c4ea027..d3ad2abdd 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -37,7 +37,7 @@ use crate::heal::manager::HealManager; use metrics::{counter, gauge}; use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIntent}; -use std::collections::VecDeque; +use std::collections::{HashSet, VecDeque}; use std::sync::Arc; use std::time::Duration; use tokio::sync::mpsc; @@ -48,15 +48,27 @@ use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealType}; /// Journal location inside the metadata bucket, following the resume-state /// layout. pub(crate) const MRF_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal.bin"; +/// The scoped path is the authoritative snapshot for new readers and carries +/// both v1 and v2 records. The legacy path is only a v1 compatibility mirror; +/// older readers ignore the authoritative path, while new readers never merge +/// the two files. This prevents a partial two-file flush from fabricating a +/// mixed epoch. +pub(crate) const MRF_SCOPED_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal-scoped.bin"; /// Record format tag. const MRF_JOURNAL_FORMAT: u8 = 1; /// Record layout version. const MRF_JOURNAL_VERSION: u8 = 1; +const MRF_JOURNAL_VERSION_SCOPED: u8 = 2; /// Fixed header size: format, version, kind, attempts, enqueued_at_ms, /// has_version flag. const MRF_RECORD_FIXED_HEAD: usize = 1 + 1 + 1 + 1 + 8 + 1; +const MRF_MAX_IDENTITY_COMPONENT: usize = 1024; + +fn metric_f64(value: usize) -> f64 { + f64::from(u32::try_from(value).unwrap_or(u32::MAX)) +} #[derive(Debug, Clone)] pub(crate) struct MrfConsumerConfig { @@ -101,40 +113,90 @@ impl Default for MrfConsumerConfig { /// incoming intent (never a resident one) and counts the loss. pub(crate) struct MrfQueue { pending: VecDeque, + pending_keys: HashSet, bytes: usize, capacity: usize, byte_budget: usize, } +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct MrfQueueKey { + kind: rustfs_common::mrf_channel::MrfKind, + bucket: Arc, + object: Arc, + version_id: Option<[u8; 16]>, + scope: Option, +} + +fn queue_key(intent: &MrfIntent) -> MrfQueueKey { + let version_id = intent.version_id.filter(|bytes| *bytes != [0; 16]); + let scope = (!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption)) + .then_some(intent.scope) + .flatten(); + MrfQueueKey { + kind: intent.kind, + bucket: intent.bucket.clone(), + object: intent.object.clone(), + version_id, + scope, + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum MrfQueuePushResult { + Enqueued, + Coalesced, + Rejected, +} + impl MrfQueue { pub(crate) fn new(capacity: usize, byte_budget: usize) -> Self { Self { pending: VecDeque::new(), + pending_keys: HashSet::new(), bytes: 0, capacity, byte_budget, } } - /// Returns `false` (after counting) when either ceiling would be crossed. - pub(crate) fn try_push(&mut self, intent: MrfIntent) -> bool { + pub(crate) fn try_push_typed(&mut self, intent: MrfIntent) -> MrfQueuePushResult { + if intent.bucket.len() > MRF_MAX_IDENTITY_COMPONENT || intent.object.len() > MRF_MAX_IDENTITY_COMPONENT { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "identity_oversized").increment(1); + return MrfQueuePushResult::Rejected; + } + let key = queue_key(&intent); + if self.pending_keys.contains(&key) { + counter!("rustfs_heal_mrf_coalesced_total", "layer" => "queue").increment(1); + return MrfQueuePushResult::Coalesced; + } let cost = intent.estimated_bytes(); if self.pending.len() >= self.capacity || self.bytes + cost > self.byte_budget { counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1); - return false; + return MrfQueuePushResult::Rejected; } self.bytes += cost; + self.pending_keys.insert(key); self.pending.push_back(intent); - true + MrfQueuePushResult::Enqueued + } + + /// Bool compatibility adapter: only a newly executable queue item is + /// reported as accepted; a coalesced duplicate is not durable admission. + #[cfg(test)] + pub(crate) fn try_push(&mut self, intent: MrfIntent) -> bool { + matches!(self.try_push_typed(intent), MrfQueuePushResult::Enqueued) } pub(crate) fn pop_front(&mut self) -> Option { let intent = self.pending.pop_front()?; + self.pending_keys.remove(&queue_key(&intent)); self.bytes = self.bytes.saturating_sub(intent.estimated_bytes()); Some(intent) } pub(crate) fn push_back(&mut self, intent: MrfIntent) { + self.pending_keys.insert(queue_key(&intent)); self.bytes += intent.estimated_bytes(); self.pending.push_back(intent); } @@ -157,10 +219,24 @@ impl MrfQueue { // --------------------------------------------------------------------------- /// Append one encoded record to `out`. -pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec) { +pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec) -> bool { + let Ok(bucket_len) = u32::try_from(intent.bucket.len()) else { + return false; + }; + let Ok(object_len) = u32::try_from(intent.object.len()) else { + return false; + }; + let scope = (!matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption)) + .then_some(intent.scope) + .flatten(); + let version_id = intent.version_id.filter(|bytes| *bytes != [0; 16]); let start = out.len(); out.push(MRF_JOURNAL_FORMAT); - out.push(MRF_JOURNAL_VERSION); + out.push(if scope.is_some() { + MRF_JOURNAL_VERSION_SCOPED + } else { + MRF_JOURNAL_VERSION + }); out.push(match intent.kind { rustfs_common::mrf_channel::MrfKind::DecodeFailure => 1, rustfs_common::mrf_channel::MrfKind::MetadataCorruption => 2, @@ -168,27 +244,36 @@ pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec) { }); out.push(intent.attempts); out.extend_from_slice(&intent.enqueued_at_ms.to_le_bytes()); - match intent.version_id { + match version_id { Some(bytes) => { out.push(1); out.extend_from_slice(&bytes); } None => out.push(0), } - out.extend_from_slice(&(intent.bucket.len() as u32).to_le_bytes()); - out.extend_from_slice(&(intent.object.len() as u32).to_le_bytes()); + if let Some(scope) = scope { + out.extend_from_slice(&scope.pool_index.to_le_bytes()); + out.extend_from_slice(&scope.set_index.to_le_bytes()); + } + out.extend_from_slice(&bucket_len.to_le_bytes()); + out.extend_from_slice(&object_len.to_le_bytes()); out.extend_from_slice(intent.bucket.as_bytes()); out.extend_from_slice(intent.object.as_bytes()); let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); hasher.update(&out[start..]); - out.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes()); + let Ok(checksum) = u32::try_from(hasher.finalize()) else { + out.truncate(start); + return false; + }; + out.extend_from_slice(&checksum.to_le_bytes()); + true } fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { if data.len() < MRF_RECORD_FIXED_HEAD + 8 { return None; } - if data[0] != MRF_JOURNAL_FORMAT || data[1] != MRF_JOURNAL_VERSION { + if data[0] != MRF_JOURNAL_FORMAT || !matches!(data[1], MRF_JOURNAL_VERSION | MRF_JOURNAL_VERSION_SCOPED) { return None; } let kind = match data[2] { @@ -198,24 +283,38 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { _ => return None, }; let attempts = data[3]; - let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().expect("slice length checked")); + let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().ok()?); let has_version = data[12] != 0; let mut cursor = MRF_RECORD_FIXED_HEAD; let version_id = if has_version { if data.len() < cursor + 16 { return None; } - let bytes: [u8; 16] = data[cursor..cursor + 16].try_into().expect("slice length checked"); + let bytes: [u8; 16] = data[cursor..cursor + 16].try_into().ok()?; cursor += 16; Some(bytes) } else { None }; + let scope = if data[1] == MRF_JOURNAL_VERSION_SCOPED { + if data.len() < cursor + 8 { + return None; + } + let pool_index = u32::from_le_bytes(data[cursor..cursor + 4].try_into().ok()?); + let set_index = u32::from_le_bytes(data[cursor + 4..cursor + 8].try_into().ok()?); + cursor += 8; + Some(rustfs_common::mrf_channel::MrfScope { pool_index, set_index }) + } else { + None + }; if data.len() < cursor + 8 { return None; } - let bucket_len = u32::from_le_bytes(data[cursor..cursor + 4].try_into().expect("slice length checked")) as usize; - let object_len = u32::from_le_bytes(data[cursor + 4..cursor + 8].try_into().expect("slice length checked")) as usize; + let bucket_len = usize::try_from(u32::from_le_bytes(data[cursor..cursor + 4].try_into().ok()?)).ok()?; + let object_len = usize::try_from(u32::from_le_bytes(data[cursor + 4..cursor + 8].try_into().ok()?)).ok()?; + if bucket_len > MRF_MAX_IDENTITY_COMPONENT || object_len > MRF_MAX_IDENTITY_COMPONENT { + return None; + } cursor += 8; let body_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?; let record_end = body_end.checked_add(4)?; @@ -224,7 +323,7 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { } let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); hasher.update(&data[..body_end]); - if (hasher.finalize() as u32) != u32::from_le_bytes(data[body_end..record_end].try_into().expect("slice length checked")) { + if u32::try_from(hasher.finalize()).ok()? != u32::from_le_bytes(data[body_end..record_end].try_into().ok()?) { return None; } let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?); @@ -235,6 +334,12 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { object, version_id, kind, + scope: if matches!(kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) { + None + } else { + scope + }, + lease: None, enqueued_at_ms, attempts, }, @@ -269,9 +374,9 @@ async fn journal_disks() -> Vec { map.values().flatten().cloned().collect() } -async fn read_journal() -> Option> { +async fn read_journal(path: &str) -> Option> { for disk in journal_disks().await { - match disk.read_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH).await { + match disk.read_all(super::RUSTFS_META_BUCKET, path).await { Ok(bytes) => return Some(bytes.to_vec()), Err(_) => continue, } @@ -282,35 +387,51 @@ async fn read_journal() -> Option> { /// Write the snapshot to every local disk; returns true when at least one /// disk accepted it, so a total write failure keeps the runtime dirty and /// the next tick retries the persist. -async fn write_journal(data: &[u8]) -> bool { +async fn write_journal(path: &str, data: &[u8]) -> bool { let payload = bytes::Bytes::copy_from_slice(data); let mut any_persisted = false; for disk in journal_disks().await { - match disk - .write_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH, payload.clone()) - .await - { + match disk.write_all(super::RUSTFS_META_BUCKET, path, payload.clone()).await { Ok(()) => any_persisted = true, Err(err) => warn_mrf_journal_write(&err), } } - if !data.is_empty() { - counter!("rustfs_heal_mrf_journal_fsync_total").increment(1); - } - gauge!("rustfs_heal_mrf_journal_bytes").set(data.len() as f64); any_persisted } -async fn delete_journal() { - for disk in journal_disks().await { - let _ = disk +async fn delete_journal(path: &str) -> bool { + let disks = journal_disks().await; + if disks.is_empty() { + counter!("rustfs_heal_mrf_journal_delete_failures_total").increment(1); + return false; + } + let mut all_deleted = true; + for disk in disks { + let result = disk .delete( super::RUSTFS_META_BUCKET, - MRF_JOURNAL_PATH, + path, crate::heal::storage_api::owner::EcstoreDeleteOptions::default(), ) .await; + if let Err(err) = result { + // Delete is idempotent: a compatibility mirror that was never + // written (or was already removed) is clean, not a retry state. + if !matches!(err, super::DiskError::FileNotFound | super::DiskError::VolumeNotFound) { + all_deleted = false; + } + } } + if !all_deleted { + counter!("rustfs_heal_mrf_journal_delete_failures_total").increment(1); + } + all_deleted +} + +async fn delete_journals() -> bool { + let authoritative_deleted = delete_journal(MRF_SCOPED_JOURNAL_PATH).await; + let legacy_deleted = delete_journal(MRF_JOURNAL_PATH).await; + authoritative_deleted && legacy_deleted } fn warn_mrf_journal_write(err: &super::DiskError) { @@ -331,7 +452,10 @@ fn warn_mrf_journal_write(err: &super::DiskError) { pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest { let bucket = intent.bucket.to_string(); let object = intent.object.to_string(); - let version_id = intent.version_id.map(|bytes| Uuid::from_bytes(bytes).to_string()); + let version_id = intent + .version_id + .filter(|bytes| *bytes != [0; 16]) + .map(|bytes| Uuid::from_bytes(bytes).to_string()); let (heal_type, priority) = match intent.kind { rustfs_common::mrf_channel::MrfKind::DecodeFailure => ( HealType::ECDecode { @@ -351,18 +475,28 @@ pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest { HealPriority::Normal, ), }; - let mut request = HealRequest::new(heal_type, HealOptions::default(), priority); + let mut options = HealOptions::default(); + if !matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) + && let Some(scope) = intent.scope + { + options.pool_index = usize::try_from(scope.pool_index).ok(); + options.set_index = usize::try_from(scope.set_index).ok(); + } + let mut request = HealRequest::new(heal_type, options, priority); request.source = rustfs_common::heal_channel::HealRequestSource::Mrf; request } async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> crate::Result { let receipt = manager - .submit_mrf_heal_request_with_receipt( + .submit_mrf_heal_request_with_receipt_and_identity( build_heal_request(intent), intent.bucket.clone(), intent.object.clone(), intent.version_id, + intent.kind, + intent.scope, + intent.lease, ) .await?; Ok(receipt.result) @@ -387,16 +521,38 @@ struct MrfRuntime { } impl MrfRuntime { - fn snapshot(&self) -> Vec { - let mut buf = Vec::new(); + fn snapshot(&self) -> (Vec, Vec) { + let mut authoritative = Vec::new(); + let mut legacy = Vec::new(); for intent in self.queue.intents() { - encode_intent(intent, &mut buf); + let scoped_identity = + !matches!(intent.kind, rustfs_common::mrf_channel::MrfKind::MetadataCorruption) && intent.scope.is_some(); + if !encode_intent(intent, &mut authoritative) { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "journal_identity_oversized").increment(1); + } + if !scoped_identity && !encode_intent(intent, &mut legacy) { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "journal_identity_oversized").increment(1); + } } - buf + (authoritative, legacy) } async fn flush(&mut self) { - let persisted = write_journal(&self.snapshot()).await; + let (authoritative, legacy) = self.snapshot(); + let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await; + if !authoritative.is_empty() { + counter!("rustfs_heal_mrf_journal_fsync_total").increment(1); + } + gauge!("rustfs_heal_mrf_journal_bytes").set(metric_f64(authoritative.len())); + // Publish the compatibility mirror only after the authoritative + // snapshot has reached at least one disk. This ordering prevents an + // old reader from observing a newer epoch that a new reader cannot + // see when the canonical write is unavailable. + let legacy_persisted = authoritative_persisted && write_journal(MRF_JOURNAL_PATH, &legacy).await; + // Keep dirty until both the authoritative snapshot and its + // compatibility mirror have been accepted; otherwise a one-sided + // failure would never retry the missing file. + let persisted = authoritative_persisted && legacy_persisted; self.new_since_flush = 0; // Keep the dirty flag when every disk write failed: a clean backlog // would otherwise never rewrite, losing the periodic persist retry a @@ -404,7 +560,7 @@ impl MrfRuntime { if persisted { self.dirty = false; } - self.journal_on_disk = true; + self.journal_on_disk |= authoritative_persisted || legacy_persisted; } /// Drain pending intents into the heal manager until it is full, the @@ -430,6 +586,7 @@ impl MrfRuntime { intent.attempts = intent.attempts.saturating_add(1); if intent.attempts >= MRF_MAX_ATTEMPTS { counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1); + rustfs_common::mrf_channel::release_mrf_intent(&intent); continue; } self.queue.push_back(intent); @@ -438,11 +595,13 @@ impl MrfRuntime { } Ok(HealAdmissionResult::Dropped(_)) => { counter!("rustfs_heal_mrf_dropped_total", "reason" => "admission_policy").increment(1); + rustfs_common::mrf_channel::release_mrf_intent(&intent); } Err(_) => { intent.attempts = intent.attempts.saturating_add(1); if intent.attempts >= MRF_MAX_ATTEMPTS { counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1); + rustfs_common::mrf_channel::release_mrf_intent(&intent); continue; } self.queue.push_back(intent); @@ -451,8 +610,8 @@ impl MrfRuntime { } } } - gauge!("rustfs_heal_mrf_queue_depth").set(self.queue.depth() as f64); - gauge!("rustfs_heal_mrf_queue_bytes").set(self.queue.bytes() as f64); + gauge!("rustfs_heal_mrf_queue_depth").set(metric_f64(self.queue.depth())); + gauge!("rustfs_heal_mrf_queue_bytes").set(metric_f64(self.queue.bytes())); } } @@ -496,7 +655,12 @@ pub async fn replay_journal_once(manager: &Arc) -> usize { let config = MrfConsumerConfig::default(); let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes); let mut backoff_until: Option = None; - replay_into(manager, &mut queue, &mut backoff_until).await + replay_into(manager, &mut queue, &mut backoff_until).await.replayed +} + +struct ReplayOutcome { + replayed: usize, + journal_on_disk: bool, } /// Shared replay core: read + decode + re-arm + delete, then drain what fits. @@ -504,11 +668,25 @@ async fn replay_into( manager: &Arc, queue: &mut MrfQueue, backoff_until: &mut Option, -) -> usize { - let Some(data) = read_journal().await else { - return 0; +) -> ReplayOutcome { + // The scoped file is a complete authoritative snapshot. Fall back to the + // legacy mirror only when the authoritative path is unavailable; merging + // both files could combine records from different flush epochs. + let data = match read_journal(MRF_SCOPED_JOURNAL_PATH).await { + Some(data) => data, + None => match read_journal(MRF_JOURNAL_PATH).await { + Some(data) => data, + None => { + return ReplayOutcome { + replayed: 0, + journal_on_disk: false, + }; + } + }, }; - let (intents, truncated) = decode_journal(&data); + let mut intents = Vec::new(); + let (decoded, truncated) = decode_journal(&data); + intents.extend(decoded); if truncated > 0 { tracing::warn!( target: "rustfs::heal::mrf", @@ -516,12 +694,15 @@ async fn replay_into( "MRF journal had a torn tail; truncated records were discarded" ); } - counter!("rustfs_heal_mrf_replayed_total").increment(intents.len() as u64); + counter!("rustfs_heal_mrf_replayed_total").increment(u64::try_from(intents.len()).unwrap_or(u64::MAX)); let replayed = intents.len(); for intent in intents { - queue.try_push(intent); + let result = queue.try_push_typed(intent.clone()); + if !matches!(result, MrfQueuePushResult::Enqueued) { + rustfs_common::mrf_channel::release_mrf_intent(&intent); + } } - delete_journal().await; + let journal_on_disk = !delete_journals().await; // Drain the replayed intents immediately; whatever the manager refuses // stays armed in `queue` for the consumer's retry loop. @@ -541,7 +722,10 @@ async fn replay_into( } } } - replayed + ReplayOutcome { + replayed, + journal_on_disk, + } } /// Replay the journal, then keep draining the channel into the heal manager @@ -559,7 +743,8 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive // Replay: read the journal, re-arm intents (duplicates are merged by the // manager's dedup key), then drop the file so the next flush starts clean. - replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await; + let replay = replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await; + runtime.journal_on_disk = replay.journal_on_disk; // The replay deleted the journal file; anything still pending (e.g. the // manager was full and backoff armed) must be re-persisted by the next // flush or a crash before it would lose those intents. @@ -587,9 +772,14 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive return; } for intent in batch.drain(..) { - if runtime.queue.try_push(intent) { - runtime.new_since_flush += 1; - runtime.dirty = true; + match runtime.queue.try_push_typed(intent.clone()) { + MrfQueuePushResult::Enqueued => { + runtime.new_since_flush += 1; + runtime.dirty = true; + } + MrfQueuePushResult::Coalesced | MrfQueuePushResult::Rejected => { + rustfs_common::mrf_channel::release_mrf_intent(&intent); + } } } runtime.dispatch(manager.as_ref()).await; @@ -613,13 +803,14 @@ async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receive TickAction::DeleteJournal => { // All intents consumed: remove the journal so a restart // replays nothing (mirrors MinIO's post-replay unlink). - delete_journal().await; - runtime.journal_on_disk = false; - gauge!("rustfs_heal_mrf_journal_bytes").set(0.0); + if delete_journals().await { + runtime.journal_on_disk = false; + gauge!("rustfs_heal_mrf_journal_bytes").set(0.0); + } } TickAction::Idle => {} } - gauge!("rustfs_heal_mrf_queue_depth").set(runtime.queue.depth() as f64); + gauge!("rustfs_heal_mrf_queue_depth").set(metric_f64(runtime.queue.depth())); } } } @@ -664,6 +855,8 @@ mod tests { object: StdArc::from(object), version_id: Some([7u8; 16]), kind: MrfKind::DecodeFailure, + scope: None, + lease: None, enqueued_at_ms: 1_700_000_000_000, attempts, } @@ -694,17 +887,101 @@ mod tests { fn queue_enforces_count_and_byte_ceilings() { let mut queue = MrfQueue::new(2, usize::MAX); assert!(queue.try_push(intent("b", "o", 0))); - assert!(queue.try_push(intent("b", "o", 0))); - assert!(!queue.try_push(intent("b", "o", 0)), "count ceiling must drop"); + assert!(queue.try_push(intent("b", "o2", 0))); + assert!(!queue.try_push(intent("b", "o3", 0)), "count ceiling must drop"); let mut tiny = MrfQueue::new(usize::MAX, intent("bucket", "object", 0).estimated_bytes()); assert!(tiny.try_push(intent("bucket", "object", 0))); assert!( - !tiny.try_push(intent("bucket", "object", 0)), + !tiny.try_push(intent("bucket", "object2", 0)), "byte budget must drop before the second intent fits" ); } + #[test] + fn duplicate_mrf_intents_coalesce_to_one_execution() { + let mut queue = MrfQueue::new(1000, usize::MAX); + let mut enqueued = 0; + let mut coalesced = 0; + assert_eq!(queue.try_push_typed(intent("bucket", "object", 0)), MrfQueuePushResult::Enqueued); + enqueued += 1; + for _ in 0..999 { + match queue.try_push_typed(intent("bucket", "object", 0)) { + MrfQueuePushResult::Coalesced => coalesced += 1, + other => panic!("duplicate intent was not coalesced: {other:?}"), + } + } + assert_eq!(enqueued, 1); + assert_eq!(coalesced, 999); + assert_eq!(queue.depth(), 1); + } + + #[test] + fn mrf_dedupe_does_not_merge_adjacent_version_pool_or_kind() { + let mut queue = MrfQueue::new(8, usize::MAX); + let mut first = intent("bucket", "object", 0); + first.kind = MrfKind::PartialWrite; + first.scope = Some(rustfs_common::mrf_channel::MrfScope { + pool_index: 1, + set_index: 1, + }); + assert!(queue.try_push(first.clone())); + first.version_id = Some([8u8; 16]); + assert!(queue.try_push(first)); + let mut other_scope = intent("bucket", "object", 0); + other_scope.kind = MrfKind::PartialWrite; + other_scope.scope = Some(rustfs_common::mrf_channel::MrfScope { + pool_index: 2, + set_index: 1, + }); + assert!(queue.try_push(other_scope)); + let mut other_kind = intent("bucket", "object", 0); + other_kind.kind = MrfKind::DecodeFailure; + other_kind.scope = None; + assert!(queue.try_push(other_kind)); + assert_eq!(queue.depth(), 4); + } + + #[test] + fn mrf_dedupe_full_returns_rejected_with_durable_pending() { + let mut queue = MrfQueue::new(1, usize::MAX); + assert_eq!(queue.try_push_typed(intent("bucket", "object", 0)), MrfQueuePushResult::Enqueued); + assert_eq!(queue.try_push_typed(intent("bucket", "other", 0)), MrfQueuePushResult::Rejected); + assert_eq!(queue.depth(), 1); + let mut snapshot = Vec::new(); + assert!(encode_intent(queue.intents().next().expect("resident intent"), &mut snapshot)); + assert!(!snapshot.is_empty(), "the resident intent remains journalable after rejection"); + } + + #[test] + fn mrf_dedupe_failure_releases_key_for_retry() { + let mut queue = MrfQueue::new(1, usize::MAX); + assert_eq!(queue.try_push_typed(intent("bucket", "object", 0)), MrfQueuePushResult::Enqueued); + let _failed = queue.pop_front().expect("queued intent"); + assert_eq!(queue.try_push_typed(intent("bucket", "object", 1)), MrfQueuePushResult::Enqueued); + assert_eq!(queue.depth(), 1); + } + + #[test] + fn mrf_dedupe_key_and_map_are_bounded() { + let mut queue = MrfQueue::new(2, usize::MAX); + assert_eq!(queue.try_push_typed(intent("bucket", "object", 0)), MrfQueuePushResult::Enqueued); + assert_eq!(queue.try_push_typed(intent("bucket", "other", 0)), MrfQueuePushResult::Enqueued); + assert_eq!(queue.pending_keys.len(), 2); + assert_eq!(queue.try_push_typed(intent("bucket", "third", 0)), MrfQueuePushResult::Rejected); + assert_eq!(queue.depth(), 2); + } + + #[test] + fn cross_node_duplicate_execution_remains_idempotent() { + // Node-local ingress maps intentionally do not merge across nodes; + // the manager's existing identity key absorbs the duplicate later. + let mut node_a = MrfQueue::new(8, usize::MAX); + let mut node_b = MrfQueue::new(8, usize::MAX); + assert_eq!(node_a.try_push_typed(intent("bucket", "object", 0)), MrfQueuePushResult::Enqueued); + assert_eq!(node_b.try_push_typed(intent("bucket", "object", 0)), MrfQueuePushResult::Enqueued); + } + #[test] fn journal_roundtrip_preserves_intents() { let intents = vec![ @@ -715,6 +992,8 @@ mod tests { object: StdArc::from("object/c"), version_id: None, kind: MrfKind::MetadataCorruption, + scope: None, + lease: None, enqueued_at_ms: 5, attempts: 1, }, @@ -766,6 +1045,8 @@ mod tests { object: StdArc::from("o"), version_id: None, kind: MrfKind::MetadataCorruption, + scope: None, + lease: None, enqueued_at_ms: 0, attempts: 0, }); @@ -777,6 +1058,8 @@ mod tests { object: StdArc::from("o"), version_id: None, kind: MrfKind::PartialWrite, + scope: None, + lease: None, enqueued_at_ms: 0, attempts: 0, }); diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index 137cdd75d..2fdb4603f 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -35,6 +35,7 @@ use storage_api::endpoint_index::{Endpoint, EndpointServerPools, Endpoints, Pool const META_BUCKET: &str = ".rustfs.sys"; const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin"; +const SCOPED_JOURNAL_REL: &str = "buckets/.heal/mrf/journal-scoped.bin"; async fn heal_env() -> (Vec, Arc) { let env = rustfs_test_utils::TestECStoreEnv::builder() @@ -79,14 +80,18 @@ fn journal_record(kind: u8, bucket: &str, object: &str, version: Option<[u8; 16] body } -fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) { +fn write_journal_path_to_disks(disk_paths: &[std::path::PathBuf], relative_path: &str, data: &[u8]) { for path in disk_paths { - let journal = path.join(META_BUCKET).join(JOURNAL_REL); + let journal = path.join(META_BUCKET).join(relative_path); std::fs::create_dir_all(journal.parent().expect("journal parent")).expect("create journal dir"); std::fs::write(&journal, data).expect("write journal fixture"); } } +fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) { + write_journal_path_to_disks(disk_paths, JOURNAL_REL, data); +} + async fn wait_until(deadline: Duration, mut probe: F) -> bool where F: FnMut() -> Fut, @@ -187,8 +192,73 @@ async fn journal_replay_arms_intents_and_deletes_the_file() { .all(|path| !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()), "the journal file must be removed after a successful replay" ); + assert!( + disk_paths + .iter() + .all(|path| !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()), + "the authoritative journal file must also be removed after replay" + ); let snapshot = manager.operations_snapshot().await; assert_eq!(snapshot.queued_by_priority.urgent, 1, "the decode-failure record must replay as Urgent"); assert!(snapshot.queued_by_priority.normal >= 1, "the partial-write record must replay as Normal"); } + +/// A canonical snapshot and its compatibility mirror may differ after a +/// partial flush. Replay must choose the complete canonical epoch instead of +/// combining records that never coexisted in memory. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn authoritative_journal_is_not_merged_with_legacy_mirror() { + let (disk_paths, storage) = heal_env().await; + let mut endpoints: Vec = disk_paths + .iter() + .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path")) + .collect(); + for (i, endpoint) in endpoints.iter_mut().enumerate() { + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + } + let pool = PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: endpoints.len(), + endpoints: Endpoints::from(endpoints), + cmd_line: "mrf-authoritative-test".to_string(), + platform: String::new(), + }; + init_local_disks(EndpointServerPools::from(vec![pool])) + .await + .expect("local disks should register"); + + let authoritative = journal_record(1, "authoritative-bucket", "authoritative-object", None, 0); + let legacy = journal_record(1, "legacy-bucket", "legacy-object", None, 0); + write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &authoritative); + write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &legacy); + + let manager = make_manager(storage); + let replayed = mrf_queue::replay_journal_once(&manager).await; + assert_eq!(replayed, 1, "only the authoritative snapshot epoch may replay"); + + let snapshot = manager.operations_snapshot().await; + assert_eq!(snapshot.queued_by_source.mrf, 1); + assert!( + disk_paths.iter().all(|path| { + !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + }), + "replay cleanup must remove both journal paths" + ); + + // A scoped-only snapshot is valid during a rollout where no legacy + // compatibility mirror was written. Missing legacy files must not leave + // the runtime in a permanent cleanup-retry state. + let scoped_only = journal_record(1, "scoped-only-bucket", "scoped-only-object", None, 0); + write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &scoped_only); + assert_eq!(mrf_queue::replay_journal_once(&manager).await, 1); + assert!(disk_paths.iter().all(|path| { + !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + })); +} diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 693dd6ebc..81a2a7377 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -1375,13 +1375,14 @@ impl FolderScanner { // Single-flight (backlog#1894 axis A) — the // recording mode and its guarantees are pinned by // corrupt_metadata_recording below. - let mrf_accepted = rustfs_common::mrf_channel::try_send_mrf_intent( + let mrf_result = rustfs_common::mrf_channel::try_send_mrf_intent_typed( rustfs_common::mrf_channel::MrfKind::MetadataCorruption, &item.bucket, &object, None, + None, ); - match corrupt_metadata_recording(mrf_accepted) { + match corrupt_metadata_recording(mrf_result) { CorruptMetadataRecording::LedgerOnly => { // Recorded as Full (retry-later): admission // for this target happens in the MRF diff --git a/crates/scanner/src/scanner_folder/item_actions.rs b/crates/scanner/src/scanner_folder/item_actions.rs index 15bd6736b..0f2d82641 100644 --- a/crates/scanner/src/scanner_folder/item_actions.rs +++ b/crates/scanner/src/scanner_folder/item_actions.rs @@ -50,11 +50,12 @@ pub(super) enum CorruptMetadataRecording { ImmediateAndLedger, } -pub(super) fn corrupt_metadata_recording(mrf_accepted: bool) -> CorruptMetadataRecording { - if mrf_accepted { - CorruptMetadataRecording::LedgerOnly - } else { - CorruptMetadataRecording::ImmediateAndLedger +pub(super) fn corrupt_metadata_recording(result: rustfs_common::mrf_channel::MrfIngressResult) -> CorruptMetadataRecording { + match result { + rustfs_common::mrf_channel::MrfIngressResult::Enqueued | rustfs_common::mrf_channel::MrfIngressResult::Coalesced => { + CorruptMetadataRecording::LedgerOnly + } + rustfs_common::mrf_channel::MrfIngressResult::Dropped(_) => CorruptMetadataRecording::ImmediateAndLedger, } } diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index 503b0e75f..3b01b9be6 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -48,8 +48,20 @@ fn scanner_alert_wire_names_match_canonical_event_names() { /// the backstop survives regardless of delivery. #[test] fn corrupt_metadata_recording_maps_delivery_to_backstop() { - assert_eq!(corrupt_metadata_recording(true), CorruptMetadataRecording::LedgerOnly); - assert_eq!(corrupt_metadata_recording(false), CorruptMetadataRecording::ImmediateAndLedger); + assert_eq!( + corrupt_metadata_recording(rustfs_common::mrf_channel::MrfIngressResult::Enqueued), + CorruptMetadataRecording::LedgerOnly + ); + assert_eq!( + corrupt_metadata_recording(rustfs_common::mrf_channel::MrfIngressResult::Coalesced), + CorruptMetadataRecording::LedgerOnly + ); + assert_eq!( + corrupt_metadata_recording(rustfs_common::mrf_channel::MrfIngressResult::Dropped( + rustfs_common::mrf_channel::MrfDropReason::Full + )), + CorruptMetadataRecording::ImmediateAndLedger + ); } fn cooldown_map_len() -> usize {