diff --git a/crates/common/src/mrf_channel.rs b/crates/common/src/mrf_channel.rs index a4f1b78be..7f782bebf 100644 --- a/crates/common/src/mrf_channel.rs +++ b/crates/common/src/mrf_channel.rs @@ -441,6 +441,27 @@ pub fn try_send_mrf_intent_typed( } } +/// Acquire a fresh process-local lease for one durable replay record. +/// +/// Journal records deliberately do not persist leases. A replay consumer must +/// call this before submitting the record so a later verified repair event can +/// identify the exact replay admission. Replay does not reserve the live +/// producer coalescer key: the replay queue first deduplicates legacy records, +/// then manager admission owns task-level deduplication with live producers. +pub fn try_rearm_mrf_replay_intent(intent: &mut MrfIntent) -> MrfIngressResult { + if intent.lease.is_some() { + return MrfIngressResult::Enqueued; + } + if intent.bucket.len() > MRF_MAX_IDENTITY_COMPONENT || intent.object.len() > MRF_MAX_IDENTITY_COMPONENT { + return MrfIngressResult::Dropped(MrfDropReason::OversizedIdentity); + } + let (version_id, scope) = canonical_identity(intent.kind, intent.version_id, intent.scope); + intent.version_id = version_id; + intent.scope = scope; + intent.lease = Some(MrfIngressLease::new(NEXT_MRF_LEASE.fetch_add(1, Ordering::Relaxed))); + MrfIngressResult::Enqueued +} + /// 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); @@ -697,6 +718,33 @@ mod tests { ); } + #[test] + fn durable_replay_rearm_assigns_a_fresh_dischargeable_lease() { + let unique = Uuid::new_v4(); + let mut intent = MrfIntent { + bucket: Arc::from(format!("replay-{unique}")), + object: Arc::from("object"), + version_id: Some([0; 16]), + kind: MrfKind::PartialWrite, + scope: Some(MrfScope { + pool_index: 2, + set_index: 3, + }), + lease: None, + enqueued_at_ms: 1, + attempts: 0, + }; + + assert_eq!(try_rearm_mrf_replay_intent(&mut intent), MrfIngressResult::Enqueued); + assert_eq!(intent.version_id, None, "nil versions remain canonical during replay"); + assert!(intent.lease.is_some(), "replay admission must carry a fresh lease"); + assert!( + MrfDurableRepairAnchor::from_intent(&intent, Uuid::new_v4()).is_some(), + "a rearmed replay record can participate in exact durable proof matching" + ); + release_mrf_intent(&intent); + } + #[test] fn verified_repair_events_consume_only_exact_durable_anchors() { let bucket = Arc::::from("proof-bucket"); diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 698d2d585..a43f591f6 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -36,7 +36,7 @@ use super::{DiskStore, HealDiskExt as _, local_disk_map_read}; use crate::heal::manager::{HealManager, MrfRepairNoticeTarget}; use metrics::{counter, gauge}; -use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIntent}; +use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIngressResult, MrfIntent}; use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; use std::collections::{HashSet, VecDeque}; use std::sync::Arc; @@ -705,9 +705,9 @@ async fn replay_into( } }, }; - let mut intents = Vec::new(); let (decoded, truncated) = decode_journal(&data); - intents.extend(decoded); + let replayed = decoded.len(); + let intents = decoded; if truncated > 0 { tracing::warn!( target: "rustfs::heal::mrf", @@ -715,8 +715,7 @@ async fn replay_into( "MRF journal had a torn tail; truncated records were discarded" ); } - counter!("rustfs_heal_mrf_replayed_total").increment(u64::try_from(intents.len()).unwrap_or(u64::MAX)); - let replayed = intents.len(); + counter!("rustfs_heal_mrf_replayed_total").increment(u64::try_from(replayed).unwrap_or(u64::MAX)); let replay_bytes = intents .iter() .fold(0usize, |total, intent| total.saturating_add(intent.estimated_bytes())); @@ -743,6 +742,15 @@ async fn replay_into( let mut retained_replay_intents = Vec::new(); if backoff_until.is_none() { while let Some(mut intent) = queue.pop_front() { + if !matches!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut intent), + MrfIngressResult::Enqueued + ) { + queue.push_back(intent); + rearm_incomplete = true; + *backoff_until = Some(tokio::time::Instant::now()); + break; + } match submit_mrf_heal_request(manager, &intent).await { Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => { retained_replay_intents.push(intent); @@ -759,7 +767,9 @@ async fn replay_into( } break; } - Ok(HealAdmissionResult::Dropped(_)) => {} + Ok(HealAdmissionResult::Dropped(_)) => { + rustfs_common::mrf_channel::release_mrf_intent(&intent); + } Err(_) => { intent.attempts = intent.attempts.saturating_add(1); if intent.attempts < MRF_MAX_ATTEMPTS { @@ -975,6 +985,40 @@ mod tests { ); } + #[test] + fn durable_replay_acquires_a_fresh_lease_before_manager_admission() { + let unique = uuid::Uuid::new_v4(); + let original = intent(&format!("replay-{unique}"), "object", 0); + assert!(original.lease.is_none(), "legacy journal records do not persist process leases"); + let mut queue = MrfQueue::new(2, usize::MAX); + assert_eq!(queue.try_push_typed(original.clone()), MrfQueuePushResult::Enqueued); + assert_eq!( + queue.try_push_typed(original), + MrfQueuePushResult::Coalesced, + "legacy duplicates are one durable responsibility before a lease is assigned" + ); + let mut replay = queue.pop_front().expect("one deduplicated replay record"); + assert_eq!( + rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut replay), + MrfIngressResult::Enqueued + ); + assert!(replay.lease.is_some(), "manager admission must receive the replay lease"); + assert!( + rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(&replay, uuid::Uuid::new_v4()).is_some(), + "the replay identity must be usable by the durable proof consumer" + ); + let mut encoded = Vec::new(); + assert!(encode_intent(&replay, &mut encoded)); + let (decoded, truncated) = decode_journal(&encoded); + assert_eq!(truncated, 0); + assert_eq!(decoded.len(), 1); + assert!( + decoded[0].lease.is_none(), + "process-local leases must not enter the durable journal format" + ); + rustfs_common::mrf_channel::release_mrf_intent(&replay); + } + #[test] fn retained_replay_anchor_remains_in_successor_snapshot() { let retained = intent("accepted-replay", "object", 0);