heal: replay committed MRF checkpoints durably

Prefer committed MRF checkpoints during startup replay, retain accepted replay responsibilities until exact verified repair proofs arrive, and reclaim committed manifests only after discharge.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/heal/src/heal/manager.rs

@@ -1567,6 +1567,18 @@ impl HealManager {
             .await
     }
 
+    pub(crate) async fn durable_mrf_repair_anchor(
+        &self,
+        intent: &rustfs_common::mrf_channel::MrfIntent,
+    ) -> Option<rustfs_common::mrf_channel::MrfDurableRepairAnchor> {
+        match self.storage.mrf_bucket_incarnation_id(intent.bucket.as_ref()).await {
+            Ok(Some(bucket_incarnation_id)) => {
+                rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(intent, bucket_incarnation_id)
+            }
+            Ok(None) | Err(_) => None,
+        }
+    }
+
     async fn submit_heal_request_with_receipt_alias_and_mrf_notice(
         &self,
         request: HealRequest,

crates/heal/src/heal/manager/scheduler.rs

@@ -733,23 +733,30 @@ pub(super) fn mrf_verified_repair_event_for_target(
         HealObjectDisposition::AuthoritativelyAbsent => MrfVerifiedRepairDisposition::AuthoritativelyAbsent,
         _ => return None,
     };
-    if target.kind != MrfKind::PartialWrite {
-        return None;
-    }
-    let expected_kind = HealObjectKind::Object;
+    let expected_kind = match target.kind {
+        MrfKind::DecodeFailure => HealObjectKind::Decode,
+        MrfKind::MetadataCorruption => HealObjectKind::Metadata,
+        MrfKind::PartialWrite => HealObjectKind::Object,
+    };
     if outcome.identity.kind != expected_kind
         || outcome.identity.bucket.as_str() != target.bucket.as_ref()
         || outcome.identity.object.as_str() != target.object.as_ref()
     {
         return None;
     }
-    let version_id = target.version_id.filter(|bytes| *bytes != [0; 16]);
+    let version_id = (!matches!(target.kind, MrfKind::MetadataCorruption))
+        .then_some(target.version_id)
+        .flatten()
+        .filter(|bytes| *bytes != [0; 16]);
     let expected_version = version_id.map(|bytes| uuid::Uuid::from_bytes(bytes).to_string());
     if outcome.identity.version_id != expected_version {
         return None;
     }
-    let expected_pool = target.scope.and_then(|scope| usize::try_from(scope.pool_index).ok());
-    let expected_set = target.scope.and_then(|scope| usize::try_from(scope.set_index).ok());
+    let scope = (!matches!(target.kind, MrfKind::MetadataCorruption))
+        .then_some(target.scope)
+        .flatten();
+    let expected_pool = scope.and_then(|scope| usize::try_from(scope.pool_index).ok());
+    let expected_set = scope.and_then(|scope| usize::try_from(scope.set_index).ok());
     if outcome.identity.pool_index != expected_pool || outcome.identity.set_index != expected_set {
         return None;
     }
@@ -759,7 +766,7 @@ pub(super) fn mrf_verified_repair_event_for_target(
         bucket: target.bucket.clone(),
         object: target.object.clone(),
         version_id,
-        scope: target.scope,
+        scope,
         lease: target.lease,
         bucket_incarnation_id,
         disposition,

crates/heal/src/heal/manager/tests.rs

@@ -1142,18 +1142,48 @@ fn mrf_verified_repair_event_requires_positive_exact_identity() {
     assert_eq!(event.bucket_incarnation_id, incarnation);
     assert_eq!(event.disposition, MrfVerifiedRepairDisposition::Repaired);
 
-    assert!(
-        mrf_verified_repair_event_for_target(
-            &MrfRepairNoticeTarget {
-                kind: MrfKind::DecodeFailure,
-                ..target.clone()
-            },
-            &matching
-        )
-        .is_none(),
-        "only receipt-producing partial-write object heals can publish verified events today"
+    let decode_target = MrfRepairNoticeTarget {
+        kind: MrfKind::DecodeFailure,
+        ..target.clone()
+    };
+    let decode_outcome = HealObjectOutcome {
+        identity: HealObjectIdentity {
+            kind: HealObjectKind::Decode,
+            ..matching.identity.clone()
+        },
+        ..matching.clone()
+    };
+    let decode_event =
+        mrf_verified_repair_event_for_target(&decode_target, &decode_outcome).expect("decode repairs publish exact proofs");
+    assert_eq!(decode_event.kind, MrfKind::DecodeFailure);
+    assert_eq!(
+        decode_event.scope,
+        Some(MrfScope {
+            pool_index: 1,
+            set_index: 2
+        })
     );
 
+    let metadata_target = MrfRepairNoticeTarget {
+        kind: MrfKind::MetadataCorruption,
+        ..target.clone()
+    };
+    let metadata_outcome = HealObjectOutcome {
+        identity: HealObjectIdentity {
+            kind: HealObjectKind::Metadata,
+            version_id: None,
+            pool_index: None,
+            set_index: None,
+            ..matching.identity.clone()
+        },
+        ..matching.clone()
+    };
+    let metadata_event =
+        mrf_verified_repair_event_for_target(&metadata_target, &metadata_outcome).expect("metadata repairs publish exact proofs");
+    assert_eq!(metadata_event.kind, MrfKind::MetadataCorruption);
+    assert_eq!(metadata_event.version_id, None);
+    assert_eq!(metadata_event.scope, None);
+
     for rejected in [
         HealObjectOutcome {
             disposition: HealObjectDisposition::Unknown,

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, MrfIngressResult, MrfIntent};
+use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfDurableRepairAnchor, MrfIngressResult, MrfIntent};
 use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
 use std::collections::{HashSet, VecDeque};
 use std::sync::Arc;
@@ -527,9 +527,12 @@ struct MrfRuntime {
     /// True while a journal snapshot exists on disk that may still be needed
     /// for replay or cleanup.
     journal_on_disk: bool,
-    /// True after replay admitted records into the manager and the startup
-    /// journal must stay until a durable replay proof is persisted.
+    /// True when replay observed a responsibility that cannot be discharged by
+    /// a complete verified repair proof in this process.
     retain_replay_journal: bool,
+    /// Partial-write responsibilities accepted from replay and waiting for an
+    /// exact storage-owned proof before the startup journal can be deleted.
+    durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
     /// Earliest instant a full-admission retry may proceed.
     backoff_until: Option<tokio::time::Instant>,
 }
@@ -627,6 +630,29 @@ impl MrfRuntime {
         gauge!("rustfs_heal_mrf_queue_depth").set(metric_f64(self.queue.depth()));
         gauge!("rustfs_heal_mrf_queue_bytes").set(metric_f64(self.queue.bytes()));
     }
+
+    fn retained_replay_journal(&self) -> bool {
+        self.retain_replay_journal || !self.durable_replay_anchors.is_empty()
+    }
+
+    fn discharge_durable_replay_anchors(&mut self) {
+        if self.durable_replay_anchors.is_empty() {
+            return;
+        }
+        let mut buckets: Vec<Arc<str>> = self
+            .durable_replay_anchors
+            .iter()
+            .map(|anchor| anchor.bucket.clone())
+            .collect();
+        buckets.sort_unstable();
+        buckets.dedup();
+        for bucket in buckets {
+            rustfs_common::mrf_channel::consume_recorded_verified_mrf_repair_events_for(
+                bucket.as_ref(),
+                &mut self.durable_replay_anchors,
+            );
+        }
+    }
 }
 
 /// Initialize the global MRF channel (honoring `RUSTFS_HEAL_MRF_ENABLE`) and
@@ -677,10 +703,71 @@ struct ReplayOutcome {
     replayed: usize,
     journal_on_disk: bool,
     retain_journal_for_replay: bool,
+    durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
 }
 
-fn replay_must_retain_journal(rearm_incomplete: bool, pending_depth: usize, accepted_or_merged: bool) -> bool {
-    rearm_incomplete || pending_depth > 0 || accepted_or_merged
+fn replay_must_retain_journal(
+    rearm_incomplete: bool,
+    pending_depth: usize,
+    accepted_without_durable_anchor: bool,
+    durable_replay_anchors: usize,
+) -> bool {
+    rearm_incomplete || pending_depth > 0 || accepted_without_durable_anchor || durable_replay_anchors > 0
+}
+
+#[derive(Clone, Copy)]
+enum ReplayCleanup {
+    Legacy,
+    Committed { sequence: u64 },
+}
+
+struct ReplaySource {
+    data: Vec<u8>,
+    cleanup: ReplayCleanup,
+}
+
+async fn read_replay_source(max_bytes: usize) -> Result<Option<ReplaySource>, snapshot::SnapshotError> {
+    if let Some(committed) = snapshot::inspect_local_committed_snapshot(max_bytes).await? {
+        return Ok(Some(ReplaySource {
+            data: committed.payload().to_vec(),
+            cleanup: ReplayCleanup::Committed {
+                sequence: committed.sequence(),
+            },
+        }));
+    }
+    // The scoped file is a complete authoritative legacy snapshot. Fall back
+    // to the v1 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 Ok(None),
+        },
+    };
+    Ok(Some(ReplaySource {
+        data,
+        cleanup: ReplayCleanup::Legacy,
+    }))
+}
+
+async fn delete_replay_source(cleanup: ReplayCleanup, max_bytes: usize) -> bool {
+    let committed_deleted = match cleanup {
+        ReplayCleanup::Legacy => true,
+        ReplayCleanup::Committed { sequence } => match snapshot::delete_committed_snapshots_through(sequence, max_bytes).await {
+            Ok(deleted) => deleted,
+            Err(err) => {
+                tracing::warn!(
+                    target: "rustfs::heal::mrf",
+                    error = %err,
+                    sequence,
+                    "MRF committed replay checkpoint cleanup failed"
+                );
+                false
+            }
+        },
+    };
+    committed_deleted && delete_journals().await
 }
 
 /// Shared replay core: read + decode + re-arm, then drain what fits. The
@@ -691,22 +778,32 @@ async fn replay_into(
     queue: &mut MrfQueue,
     backoff_until: &mut Option<tokio::time::Instant>,
 ) -> 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,
-                    retain_journal_for_replay: false,
-                };
-            }
-        },
+    let source = match read_replay_source(queue.byte_budget).await {
+        Ok(Some(source)) => source,
+        Ok(None) => {
+            return ReplayOutcome {
+                replayed: 0,
+                journal_on_disk: false,
+                retain_journal_for_replay: false,
+                durable_replay_anchors: Vec::new(),
+            };
+        }
+        Err(err) => {
+            tracing::warn!(
+                target: "rustfs::heal::mrf",
+                error = %err,
+                "MRF committed replay checkpoint could not be inspected"
+            );
+            return ReplayOutcome {
+                replayed: 0,
+                journal_on_disk: true,
+                retain_journal_for_replay: true,
+                durable_replay_anchors: Vec::new(),
+            };
+        }
     };
+    let cleanup = source.cleanup;
+    let data = source.data;
     let (decoded, truncated) = decode_journal(&data);
     let replayed = decoded.len();
     let intents = decoded;
@@ -727,7 +824,8 @@ async fn replay_into(
     // prefix.
     queue.raise_limits_for_replay(intents.len(), replay_bytes);
     let mut rearm_incomplete = false;
-    let mut accepted_or_merged = false;
+    let mut accepted_without_durable_anchor = false;
+    let mut durable_replay_anchors = Vec::new();
     for intent in intents {
         let result = queue.try_push_typed(intent.clone());
         match result {
@@ -755,7 +853,11 @@ async fn replay_into(
             }
             match submit_mrf_heal_request(manager, &intent).await {
                 Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
-                    accepted_or_merged = true;
+                    if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await {
+                        durable_replay_anchors.push(anchor);
+                    } else {
+                        accepted_without_durable_anchor = true;
+                    }
                 }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
@@ -787,16 +889,23 @@ async fn replay_into(
             }
         }
     }
-    let retain_journal_for_replay = replay_must_retain_journal(rearm_incomplete, queue.depth(), accepted_or_merged);
-    let journal_on_disk = if retain_journal_for_replay {
+    let must_retain_journal = replay_must_retain_journal(
+        rearm_incomplete,
+        queue.depth(),
+        accepted_without_durable_anchor,
+        durable_replay_anchors.len(),
+    );
+    let retain_journal_for_replay = rearm_incomplete || accepted_without_durable_anchor;
+    let journal_on_disk = if must_retain_journal {
         true
     } else {
-        !delete_journals().await
+        !delete_replay_source(cleanup, queue.byte_budget).await
     };
     ReplayOutcome {
         replayed,
         journal_on_disk,
         retain_journal_for_replay,
+        durable_replay_anchors,
     }
 }
 
@@ -811,6 +920,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
         dirty: false,
         journal_on_disk: false,
         retain_replay_journal: false,
+        durable_replay_anchors: Vec::new(),
         backoff_until: None,
     };
 
@@ -819,6 +929,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     let replay = replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await;
     runtime.journal_on_disk = replay.journal_on_disk;
     runtime.retain_replay_journal = replay.retain_journal_for_replay;
+    runtime.durable_replay_anchors = replay.durable_replay_anchors;
     // Anything still pending (e.g. the manager was full and backoff armed)
     // must be re-persisted by the next flush before replay can delete the
     // startup anchor.
@@ -862,11 +973,12 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                 }
             }
             _ = flush_tick.tick() => {
+                runtime.discharge_durable_replay_anchors();
                 match tick_action(
                     runtime.dirty,
                     runtime.queue.depth(),
                     runtime.journal_on_disk,
-                    runtime.retain_replay_journal,
+                    runtime.retained_replay_journal(),
                 ) {
                     TickAction::Flush => {
                         runtime.flush().await;
@@ -925,7 +1037,7 @@ fn tick_action(dirty: bool, depth: usize, journal_on_disk: bool, retain_replay_j
 #[cfg(test)]
 mod tests {
     use super::*;
-    use rustfs_common::mrf_channel::{MrfIntent, MrfKind};
+    use rustfs_common::mrf_channel::{MrfIntent, MrfKind, MrfVerifiedRepairDisposition, MrfVerifiedRepairEvent};
     use std::sync::Arc as StdArc;
 
     fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent {
@@ -966,21 +1078,69 @@ mod tests {
     #[test]
     fn replay_cleanup_retains_journal_for_unarmed_or_refused_records() {
         assert!(
-            replay_must_retain_journal(true, 0, false),
+            replay_must_retain_journal(true, 0, false, 0),
             "a rejected replay record still needs its disk anchor"
         );
         assert!(
-            replay_must_retain_journal(false, 1, false),
+            replay_must_retain_journal(false, 1, false, 0),
             "a Full admission retry must keep the startup journal until the next snapshot"
         );
         assert!(
-            !replay_must_retain_journal(false, 0, false),
+            !replay_must_retain_journal(false, 0, false, 0),
             "a fully consumed replay snapshot without accepts may be deleted"
         );
         assert!(
-            replay_must_retain_journal(false, 0, true),
-            "accepted or merged replay records still need a durable successor"
+            replay_must_retain_journal(false, 0, true, 0),
+            "accepted or merged replay records without a proof identity still need a durable successor"
         );
+        assert!(
+            replay_must_retain_journal(false, 0, false, 1),
+            "accepted replay records with a durable proof anchor must retain the journal until proof arrives"
+        );
+    }
+
+    #[test]
+    fn runtime_releases_retained_replay_journal_after_verified_repair_proof() {
+        let mut intent = intent("proof-bucket", "proof-object", 0);
+        intent.kind = MrfKind::PartialWrite;
+        assert_eq!(
+            rustfs_common::mrf_channel::try_rearm_mrf_replay_intent(&mut intent),
+            MrfIngressResult::Enqueued
+        );
+        let bucket_incarnation_id = uuid::Uuid::new_v4();
+        let anchor = rustfs_common::mrf_channel::MrfDurableRepairAnchor::from_intent(&intent, bucket_incarnation_id)
+            .expect("fresh replay lease and bucket incarnation build a durable anchor");
+        let mut runtime = MrfRuntime {
+            queue: MrfQueue::new(2, usize::MAX),
+            config: MrfConsumerConfig::default(),
+            new_since_flush: 0,
+            dirty: false,
+            journal_on_disk: true,
+            retain_replay_journal: false,
+            durable_replay_anchors: vec![anchor],
+            backoff_until: None,
+        };
+        rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
+            kind: intent.kind,
+            bucket: intent.bucket.clone(),
+            object: intent.object.clone(),
+            version_id: intent.version_id,
+            scope: intent.scope,
+            lease: intent.lease,
+            bucket_incarnation_id,
+            disposition: MrfVerifiedRepairDisposition::Repaired,
+        });
+
+        assert!(
+            runtime.retained_replay_journal(),
+            "anchor must retain the startup journal before proof is consumed"
+        );
+        runtime.discharge_durable_replay_anchors();
+        assert!(
+            !runtime.retained_replay_journal(),
+            "matching verified proof discharges the durable replay anchor"
+        );
+        rustfs_common::mrf_channel::release_mrf_intent(&intent);
     }
 
     #[test]

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -17,11 +17,13 @@
 //! Each of two slots has a payload and a commit manifest. The manifest binds
 //! the writer identity, persistent sequence, length and whole-payload digest.
 //! Replacing the inactive slot must leave the previous committed slot intact.
-//! Production publication and reclamation are deliberately not enabled here.
+//! Production publication is deliberately reader-first; reclamation only
+//! removes manifest entries after the owning replay path has discharged every
+//! responsibility through a newer durable snapshot or a verified repair proof.
 //! An unreadable commit path cannot prove that only legacy data exists. This
 //! explicit inspection API fails closed and never mutates recovery anchors.
-//! It is not wired into the legacy consumer: that transition requires the
-//! ownership-aware replay and producer handoff before writer activation.
+//! It is wired into the replay reader before writer activation, but the writer
+//! remains gated on ownership-aware handoff.
 //! One surviving committed replica supports process restart recovery only;
 //! this reader does not establish a replication quorum or a power-loss policy.
 
@@ -542,6 +544,91 @@ pub async fn inspect_local_recovery_snapshot(max_bytes: usize) -> Result<Option<
     read_recovery_snapshot(&super::journal_disks().await, max_bytes).await
 }
 
+/// Inspect only committed MRF checkpoints.
+///
+/// The legacy replay path keeps its historical torn-tail behavior, so startup
+/// replay uses this narrower API to avoid turning a legacy torn tail into a
+/// committed-format failure. A corrupt or conflicting committed checkpoint is
+/// still authoritative: callers must fail closed instead of falling back to a
+/// stale legacy mirror.
+pub async fn inspect_local_committed_snapshot(max_bytes: usize) -> Result<Option<CommittedSnapshot>, SnapshotError> {
+    read_committed(&super::journal_disks().await, max_bytes).await
+}
+
+/// Remove committed manifests whose sequence is no newer than
+/// `committed_through`.
+///
+/// Payload files are intentionally left as orphans after their manifest is
+/// removed. Readers cannot discover a payload without its matching manifest,
+/// and deleting manifests first prevents an older retained slot from becoming
+/// visible again after the newest replay has been fully discharged.
+pub async fn delete_committed_snapshots_through(committed_through: u64, max_bytes: usize) -> Result<bool, SnapshotError> {
+    let disks = super::journal_disks().await;
+    delete_committed_snapshots_through_on(&disks, committed_through, max_bytes).await
+}
+
+async fn delete_committed_snapshots_through_on(
+    disks: &[EcstoreDiskStore],
+    committed_through: u64,
+    max_bytes: usize,
+) -> Result<bool, SnapshotError> {
+    if disks.is_empty() {
+        return Err(SnapshotError::NoWritableReplica);
+    }
+    let mut any_changed = false;
+    let mut first_error = None;
+    for disk in disks {
+        for path in MANIFEST_PATHS {
+            let existing = match read_bounded(disk, path, MANIFEST_LEN).await {
+                Ok(Some(existing)) => existing,
+                Ok(None) => continue,
+                Err(error) => {
+                    if first_error.is_none() {
+                        first_error = Some(error);
+                    }
+                    continue;
+                }
+            };
+            let manifest = match Manifest::decode(&existing, max_bytes) {
+                Ok(manifest) => manifest,
+                Err(error) => {
+                    if first_error.is_none() {
+                        first_error = Some(error);
+                    }
+                    continue;
+                }
+            };
+            if manifest.sequence > committed_through {
+                continue;
+            }
+            match EcstoreDiskAPI::compare_and_update_file(
+                disk.as_ref(),
+                RUSTFS_META_BUCKET,
+                path,
+                Some(EcstoreDiskBytes::from(existing)),
+                None,
+            )
+            .await
+            {
+                Ok(EcstoreConditionalFileUpdate::Updated) => any_changed = true,
+                Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {}
+                Err(error) => {
+                    if first_error.is_none() {
+                        first_error = Some(SnapshotError::Disk(error));
+                    }
+                }
+            }
+        }
+    }
+    if any_changed {
+        Ok(true)
+    } else if let Some(error) = first_error {
+        Err(error)
+    } else {
+        Ok(true)
+    }
+}
+
 async fn read_recovery_snapshot(disks: &[EcstoreDiskStore], limit: usize) -> Result<Option<RecoverySnapshot>, SnapshotError> {
     if let Some(snapshot) = read_committed(disks, limit).await? {
         return Ok(Some(RecoverySnapshot::Committed(snapshot)));
@@ -1185,6 +1272,44 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn committed_cleanup_removes_only_manifests_at_or_below_sequence() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let older = payload("older");
+        let newer = payload("newer");
+        commit(&disk, 0, owner, 3, &older).await;
+        commit(&disk, 1, owner, 4, &newer).await;
+
+        assert!(
+            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), 3, 4096)
+                .await
+                .expect("delete old manifest"),
+            "old committed manifest should be removed"
+        );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "old manifest is gone, so the old payload cannot become visible again"
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
+                .await
+                .expect("old payload orphan may remain")
+                .as_ref(),
+            older
+        );
+        let recovered = read_committed(std::slice::from_ref(&disk), 4096)
+            .await
+            .expect("read newer commit")
+            .expect("newer commit remains visible");
+        assert_eq!(recovered.sequence(), 4);
+        assert_eq!(recovered.payload(), newer);
+    }
+
     #[tokio::test]
     async fn legacy_import_requires_complete_consistent_replicas() {
         let root = TempDir::new().expect("test directory");

crates/heal/src/heal/storage.rs

@@ -356,6 +356,14 @@ pub trait HealStorageAPI: Send + Sync {
     /// Get bucket info
     async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>>;
 
+    /// Return the current bucket incarnation for exact MRF durable proof
+    /// matching. Alternate backends that cannot expose this must return
+    /// `None`, leaving replay anchors retained instead of acknowledged with an
+    /// incomplete identity.
+    async fn mrf_bucket_incarnation_id(&self, _bucket: &str) -> Result<Option<Uuid>> {
+        Ok(None)
+    }
+
     /// Aggregate usage-cache baselines for the requested buckets.
     async fn erasure_set_usage_baseline(&self, _buckets: &[String]) -> Result<Option<HealBucketUsageBaseline>> {
         Ok(None)
@@ -815,6 +823,14 @@ impl HealStorageAPI for ECStoreHealStorage {
         }
     }
 
+    async fn mrf_bucket_incarnation_id(&self, bucket: &str) -> Result<Option<Uuid>> {
+        self.ecstore
+            .bucket_incarnation_id(bucket)
+            .await
+            .map(Some)
+            .map_err(Error::Storage)
+    }
+
     async fn erasure_set_usage_baseline(&self, buckets: &[String]) -> Result<Option<HealBucketUsageBaseline>> {
         if buckets.is_empty() {
             return Ok(None);

crates/heal/tests/mrf_pipeline_test.rs

@@ -29,6 +29,7 @@ use rustfs_heal::heal::{
     storage::{ECStoreHealStorage, HealStorageAPI},
 };
 use serial_test::serial;
+use sha2::{Digest, Sha256};
 #[cfg(unix)]
 use std::{
     fs::{File, OpenOptions},
@@ -48,6 +49,10 @@ 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";
+const COMMITTED_PAYLOAD_REL: &str = ".heal-mrf-snapshot.0.bin";
+const COMMITTED_MANIFEST_REL: &str = ".heal-mrf-commit.0.bin";
+const COMMITTED_MAGIC: &[u8; 8] = b"RFMRFC01";
+const COMMITTED_MANIFEST_LEN: usize = 8 + 1 + 16 + 8 + 8 + 32 + 32;
 
 async fn heal_env() -> (Vec<std::path::PathBuf>, Arc<dyn HealStorageAPI>) {
     heal_env_at(None).await
@@ -195,6 +200,29 @@ fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) {
     write_journal_path_to_disks(disk_paths, JOURNAL_REL, data);
 }
 
+fn committed_manifest(owner: uuid::Uuid, sequence: u64, payload: &[u8]) -> Vec<u8> {
+    let mut manifest = Vec::with_capacity(COMMITTED_MANIFEST_LEN);
+    manifest.extend_from_slice(COMMITTED_MAGIC);
+    manifest.push(1);
+    manifest.extend_from_slice(owner.as_bytes());
+    manifest.extend_from_slice(&sequence.to_le_bytes());
+    manifest.extend_from_slice(
+        &u64::try_from(payload.len())
+            .expect("fixture payload length fits")
+            .to_le_bytes(),
+    );
+    manifest.extend_from_slice(&Sha256::digest(payload));
+    manifest.extend_from_slice(&Sha256::digest(&manifest));
+    assert_eq!(manifest.len(), COMMITTED_MANIFEST_LEN, "committed fixture manifest length");
+    manifest
+}
+
+fn write_committed_snapshot_to_disks(disk_paths: &[std::path::PathBuf], sequence: u64, payload: &[u8]) {
+    let manifest = committed_manifest(uuid::Uuid::new_v4(), sequence, payload);
+    write_journal_path_to_disks(disk_paths, COMMITTED_PAYLOAD_REL, payload);
+    write_journal_path_to_disks(disk_paths, COMMITTED_MANIFEST_REL, &manifest);
+}
+
 fn journal_exists_on_all_disks(disk_paths: &[std::path::PathBuf], relative_path: &str) -> bool {
     disk_paths
         .iter()
@@ -255,11 +283,12 @@ async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() {
 }
 
 /// A journal left behind by a previous process must be replayed into the
-/// manager queue and then removed, and a torn tail must not block replay of
-/// the intact records.
+/// manager queue, and a torn tail must not block replay of the intact records.
+/// The partial-write record keeps the legacy journal as the durable anchor
+/// until an exact verified repair proof can discharge it.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
-async fn journal_replay_arms_intents_and_deletes_the_file() {
+async fn journal_replay_arms_intents_and_retains_unproven_partial_write_anchor() {
     let (disk_paths, storage) = heal_env().await;
 
     // The journal reader resolves disks through the process-local disk map;
@@ -285,14 +314,14 @@ async fn journal_replay_arms_intents_and_deletes_the_file() {
     assert!(
         disk_paths
             .iter()
-            .all(|path| !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()),
-        "the journal file must be removed after a successful replay"
+            .all(|path| Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()),
+        "partial-write replay must retain the legacy journal until durable proof"
     );
     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"
+        "missing authoritative journal remains absent"
     );
 
     let snapshot = manager.operations_snapshot().await;
@@ -300,6 +329,41 @@ async fn journal_replay_arms_intents_and_deletes_the_file() {
     assert!(snapshot.queued_by_priority.normal >= 1, "the partial-write record must replay as Normal");
 }
 
+/// A committed checkpoint published by the new two-slot writer is the
+/// authoritative startup snapshot. Legacy mirrors are fallback-only and must
+/// not be merged with or preferred over the committed epoch.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn committed_snapshot_replay_takes_precedence_over_stale_legacy_mirror() {
+    let (disk_paths, storage) = heal_env().await;
+    register_local_disks(&disk_paths, "mrf-committed-replay-test").await;
+
+    let committed = scoped_journal_record(3, "committed-bucket", "committed-object", Some([9u8; 16]), 0, 0, 0);
+    let stale_legacy = journal_record(1, "legacy-bucket", "legacy-object", None, 0);
+    write_committed_snapshot_to_disks(&disk_paths, 7, &committed);
+    write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &stale_legacy);
+    write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &stale_legacy);
+
+    let manager = make_manager(storage);
+    let replayed = mrf_queue::replay_journal_once(&manager).await;
+    assert_eq!(replayed, 1, "only the committed snapshot epoch may replay");
+
+    let snapshot = manager.operations_snapshot().await;
+    assert_eq!(snapshot.queued_by_source.mrf, 1);
+    assert_eq!(
+        snapshot.queued_by_priority.normal, 1,
+        "the committed partial-write record must replay instead of the stale legacy decode-failure"
+    );
+    assert_eq!(
+        snapshot.queued_by_priority.urgent, 0,
+        "stale legacy decode-failure records must not be mixed into committed replay"
+    );
+    assert!(
+        journal_exists_on_all_disks(&disk_paths, COMMITTED_MANIFEST_REL),
+        "the committed checkpoint remains until the accepted partial-write has proof"
+    );
+}
+
 /// 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.
@@ -322,21 +386,24 @@ async fn authoritative_journal_is_not_merged_with_legacy_mirror() {
     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()
+            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"
+        "accepted replay responsibilities remain anchored until a verified repair proof"
     );
 
     // 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 (disk_paths, storage) = heal_env().await;
+    register_local_disks(&disk_paths, "mrf-scoped-authoritative-test").await;
+    let manager = make_manager(storage);
     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()
+            && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()
     }));
 
     let scoped_v2 = scoped_journal_record(1, "scoped-v2-bucket", "scoped-v2-object", None, 0, 3, 7);
@@ -350,12 +417,12 @@ async fn authoritative_journal_is_not_merged_with_legacy_mirror() {
     );
     assert_eq!(
         manager.operations_snapshot().await.queued_by_source.mrf,
-        3,
-        "only the three authoritative/scoped-only epochs should have reached the manager"
+        2,
+        "only the scoped-only and scoped-v2 authoritative epochs should have reached the manager"
     );
     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()
+        Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()
+            && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()
     }));
 }
 
@@ -400,10 +467,13 @@ async fn authoritative_journal_replay_preserves_kind_and_scope_identity() {
         snapshot.queued_by_priority.urgent, 1,
         "decode-failure repair must not merge with object repair responsibility"
     );
-    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()
-    }));
+    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()
+        }),
+        "partial-write responsibilities keep both replay anchors until proof"
+    );
 }
 
 /// If replay reaches a full heal-manager queue, the old journal remains the
@@ -698,10 +768,10 @@ async fn journal_replay_survives_successor_flush_before_delete() {
     );
     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()
+            Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()
+                && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()
         }),
-        "a fully consumed successor snapshot may be deleted after restart replay"
+        "the accepted successor remains anchored until a verified repair proof"
     );
 }
 
@@ -751,10 +821,10 @@ async fn journal_replay_survives_service_kill_after_successor_flush() {
     );
     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()
+            Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()
+                && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()
         }),
-        "a fully consumed successor snapshot may be deleted after service-kill restart replay"
+        "the accepted successor remains anchored until a verified repair proof after service-kill restart"
     );
 }
 
@@ -814,9 +884,9 @@ async fn journal_replay_survives_sigkill_after_authoritative_successor_fsync_bef
     );
     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()
+            Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()
+                && Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()
         }),
-        "a fully consumed authoritative successor may clean both epochs after restart replay"
+        "the accepted authoritative successor remains anchored until a verified repair proof"
     );
 }