fix(heal): rearm MRF replay leases before admission (#7435)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

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::<str>::from("proof-bucket");

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);