diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs
index 77d52720b..d2738d501 100644
--- a/crates/heal/src/heal/mrf_queue.rs
+++ b/crates/heal/src/heal/mrf_queue.rs
@@ -521,6 +521,8 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
 struct MrfRuntime {
     queue: MrfQueue,
     config: MrfConsumerConfig,
+    checkpoint_owner: Uuid,
+    next_checkpoint_sequence: u64,
     new_since_flush: usize,
     /// True while the in-memory pending set has changed since the last
     /// journal flush (push, pop, or an attempts bump that alters the encoded
@@ -537,6 +539,12 @@ struct MrfRuntime {
     /// 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>,
+    /// Startup replay source to remove after the retained replay
+    /// responsibilities are discharged. `None` means the runtime only needs
+    /// the legacy journal cleanup path for snapshots it wrote itself.
+    replay_cleanup: Option<ReplayCleanup>,
+    /// Last committed checkpoint published by this runtime flush path.
+    runtime_checkpoint: Option<(Uuid, u64)>,
     /// Earliest instant a full-admission retry may proceed.
     backoff_until: Option<tokio::time::Instant>,
 }
@@ -560,6 +568,34 @@ impl MrfRuntime {
 
     async fn flush(&mut self) {
         let (authoritative, legacy) = self.snapshot();
+        let (committed_persisted, committed_on_disk) = if authoritative.is_empty() {
+            (true, false)
+        } else {
+            match snapshot::publish_committed_snapshot(
+                &journal_disks().await,
+                self.checkpoint_owner,
+                self.next_checkpoint_sequence,
+                &authoritative,
+                self.config.journal_max_bytes,
+            )
+            .await
+            {
+                Ok(publication) => {
+                    self.runtime_checkpoint = Some((publication.owner, publication.sequence));
+                    self.next_checkpoint_sequence = publication.sequence.saturating_add(1);
+                    (true, true)
+                }
+                Err(err) => {
+                    tracing::warn!(
+                        target: "rustfs::heal::mrf",
+                        error = %err,
+                        sequence = self.next_checkpoint_sequence,
+                        "MRF committed checkpoint publish failed; retaining previous replay anchor"
+                    );
+                    (false, false)
+                }
+            }
+        };
         let authoritative_persisted = write_journal(MRF_SCOPED_JOURNAL_PATH, &authoritative).await;
         if !authoritative.is_empty() {
             counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
@@ -570,10 +606,10 @@ impl MrfRuntime {
         // 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;
+        // Keep dirty until the committed checkpoint, authoritative snapshot,
+        // and compatibility mirror have all been accepted; otherwise a
+        // one-sided failure would never retry the missing recovery anchor.
+        let persisted = committed_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
@@ -581,7 +617,7 @@ impl MrfRuntime {
         if persisted {
             self.dirty = false;
         }
-        self.journal_on_disk |= authoritative_persisted || legacy_persisted;
+        self.journal_on_disk |= committed_on_disk || authoritative_persisted || legacy_persisted;
     }
 
     /// Drain pending intents into the heal manager until it is full, the
@@ -639,6 +675,45 @@ impl MrfRuntime {
         self.retain_replay_journal || !self.durable_replay_anchors.is_empty()
     }
 
+    fn replay_cleanup_to_delete(&self) -> Option<ReplayCleanup> {
+        if self.journal_on_disk && !self.retained_replay_journal() {
+            Some(self.replay_cleanup.unwrap_or(ReplayCleanup::Legacy))
+        } else {
+            None
+        }
+    }
+
+    async fn delete_idle_recovery_anchors(&mut self) -> bool {
+        let runtime_deleted = match self.runtime_checkpoint {
+            Some((owner, sequence)) => {
+                match snapshot::delete_committed_snapshots_through(owner, sequence, self.config.journal_max_bytes).await {
+                    Ok(deleted) => deleted,
+                    Err(err) => {
+                        tracing::warn!(
+                            target: "rustfs::heal::mrf",
+                            error = %err,
+                            sequence,
+                            "MRF runtime checkpoint cleanup failed"
+                        );
+                        false
+                    }
+                }
+            }
+            None => true,
+        };
+        let replay_deleted = match self.replay_cleanup_to_delete() {
+            Some(cleanup) => delete_replay_source(cleanup, self.config.journal_max_bytes).await,
+            None => true,
+        };
+        if runtime_deleted && replay_deleted {
+            self.runtime_checkpoint = None;
+            self.replay_cleanup = None;
+            true
+        } else {
+            false
+        }
+    }
+
     fn discharge_durable_replay_anchors(&mut self) {
         if self.durable_replay_anchors.is_empty() {
             return;
@@ -708,6 +783,8 @@ struct ReplayOutcome {
     journal_on_disk: bool,
     retain_journal_for_replay: bool,
     durable_replay_anchors: Vec<MrfDurableRepairAnchor>,
+    cleanup: Option<ReplayCleanup>,
+    next_checkpoint_sequence: u64,
 }
 
 fn replay_must_retain_journal(
@@ -719,7 +796,7 @@ fn replay_must_retain_journal(
     rearm_incomplete || pending_depth > 0 || accepted_without_durable_anchor || durable_replay_anchors > 0
 }
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum ReplayCleanup {
     Legacy,
     Committed { owner: Uuid, sequence: u64 },
@@ -793,6 +870,8 @@ async fn replay_into(
                 journal_on_disk: false,
                 retain_journal_for_replay: false,
                 durable_replay_anchors: Vec::new(),
+                cleanup: None,
+                next_checkpoint_sequence: 1,
             };
         }
         Err(err) => {
@@ -806,10 +885,16 @@ async fn replay_into(
                 journal_on_disk: true,
                 retain_journal_for_replay: true,
                 durable_replay_anchors: Vec::new(),
+                cleanup: None,
+                next_checkpoint_sequence: 1,
             };
         }
     };
     let cleanup = source.cleanup;
+    let next_checkpoint_sequence = match cleanup {
+        ReplayCleanup::Legacy => 1,
+        ReplayCleanup::Committed { sequence, .. } => sequence.saturating_add(1),
+    };
     let data = source.data;
     let (decoded, truncated) = decode_journal(&data);
     let replayed = decoded.len();
@@ -913,6 +998,8 @@ async fn replay_into(
         journal_on_disk,
         retain_journal_for_replay,
         durable_replay_anchors,
+        cleanup: journal_on_disk.then_some(cleanup),
+        next_checkpoint_sequence,
     }
 }
 
@@ -923,11 +1010,15 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     let mut runtime = MrfRuntime {
         queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
         config: config.clone(),
+        checkpoint_owner: Uuid::new_v4(),
+        next_checkpoint_sequence: 1,
         new_since_flush: 0,
         dirty: false,
         journal_on_disk: false,
         retain_replay_journal: false,
         durable_replay_anchors: Vec::new(),
+        replay_cleanup: None,
+        runtime_checkpoint: None,
         backoff_until: None,
     };
 
@@ -937,6 +1028,8 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
     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;
+    runtime.replay_cleanup = replay.cleanup;
+    runtime.next_checkpoint_sequence = replay.next_checkpoint_sequence;
     // 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.
@@ -1001,7 +1094,7 @@ async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receive
                     TickAction::DeleteJournal => {
                         // All replayed intents have either been accepted,
                         // merged, or replaced by a pending successor snapshot.
-                        if delete_journals().await {
+                        if runtime.delete_idle_recovery_anchors().await {
                             runtime.journal_on_disk = false;
                             gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
                         }
@@ -1117,14 +1210,23 @@ mod tests {
         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 cleanup_owner = uuid::Uuid::new_v4();
+        let cleanup = ReplayCleanup::Committed {
+            owner: cleanup_owner,
+            sequence: 17,
+        };
         let mut runtime = MrfRuntime {
             queue: MrfQueue::new(2, usize::MAX),
             config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
             new_since_flush: 0,
             dirty: false,
             journal_on_disk: true,
             retain_replay_journal: false,
             durable_replay_anchors: vec![anchor],
+            replay_cleanup: Some(cleanup),
+            runtime_checkpoint: None,
             backoff_until: None,
         };
         rustfs_common::mrf_channel::note_mrf_verified_repair(MrfVerifiedRepairEvent {
@@ -1142,14 +1244,48 @@ mod tests {
             runtime.retained_replay_journal(),
             "anchor must retain the startup journal before proof is consumed"
         );
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            None,
+            "the committed replay source must not be reclaimed before the exact proof"
+        );
         runtime.discharge_durable_replay_anchors();
         assert!(
             !runtime.retained_replay_journal(),
             "matching verified proof discharges the durable replay anchor"
         );
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            Some(cleanup),
+            "proof discharge must preserve the committed owner/sequence cleanup target"
+        );
         rustfs_common::mrf_channel::release_mrf_intent(&intent);
     }
 
+    #[test]
+    fn runtime_cleanup_defaults_to_legacy_for_runtime_written_journals() {
+        let runtime = MrfRuntime {
+            queue: MrfQueue::new(2, usize::MAX),
+            config: MrfConsumerConfig::default(),
+            checkpoint_owner: Uuid::new_v4(),
+            next_checkpoint_sequence: 1,
+            new_since_flush: 0,
+            dirty: false,
+            journal_on_disk: true,
+            retain_replay_journal: false,
+            durable_replay_anchors: Vec::new(),
+            replay_cleanup: None,
+            runtime_checkpoint: None,
+            backoff_until: None,
+        };
+
+        assert_eq!(
+            runtime.replay_cleanup_to_delete(),
+            Some(ReplayCleanup::Legacy),
+            "journals written by the runtime still use the legacy cleanup path"
+        );
+    }
+
     #[test]
     fn durable_replay_acquires_a_fresh_lease_before_manager_admission() {
         let unique = uuid::Uuid::new_v4();
diff --git a/crates/heal/src/heal/mrf_queue/snapshot.rs b/crates/heal/src/heal/mrf_queue/snapshot.rs
index d4cafac10..c00141f27 100644
--- a/crates/heal/src/heal/mrf_queue/snapshot.rs
+++ b/crates/heal/src/heal/mrf_queue/snapshot.rs
@@ -22,8 +22,9 @@
 //! 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 wired into the replay reader before writer activation, but the writer
-//! remains gated on ownership-aware handoff.
+//! The live consumer writes committed checkpoints alongside the scoped and
+//! legacy journal mirrors; cleanup remains gated by replay ownership and exact
+//! verified repair proof handoff.
 //! One surviving committed replica supports process restart recovery only;
 //! this reader does not establish a replication quorum or a power-loss policy.
 
@@ -356,8 +357,8 @@ fn validate_reusable_manifest_slot(existing: Option<&[u8]>, sequence: u64, paylo
 /// The writer is a narrow production primitive for the ownership-aware MRF
 /// handoff: it validates the whole journal payload, preserves the previous
 /// committed slot, and publishes the manifest only after the successor payload
-/// reaches the same disk. It does not delete legacy journals, tombstone older
-/// anchors, or activate the live consumer.
+/// reaches the same disk. It does not delete legacy journals or tombstone older
+/// anchors by itself; the consumer decides cleanup after replay handoff.
 pub async fn publish_committed_snapshot(
     disks: &[EcstoreDiskStore],
     owner: Uuid,
@@ -371,7 +372,10 @@ pub async fn publish_committed_snapshot(
     if owner.is_nil() || sequence == 0 || sequence == u64::MAX {
         return Err(SnapshotError::Corrupt);
     }
-    if payload.len() > limit || decode_journal(payload).1 != 0 {
+    if payload.len() > limit {
+        return Err(SnapshotError::TooLarge);
+    }
+    if decode_journal(payload).1 != 0 {
         return Err(SnapshotError::Corrupt);
     }
     let current = read_committed(disks, limit).await?;
@@ -555,13 +559,13 @@ pub async fn inspect_local_committed_snapshot(max_bytes: usize) -> Result<Option
     read_committed(&super::journal_disks().await, max_bytes).await
 }
 
-/// Remove committed manifests from `owner` whose sequence is no newer than
+/// Remove committed checkpoints from `owner` 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.
+/// Cleanup is manifest-first so readers cannot rediscover an older payload
+/// after the newest replay has been fully discharged. The payload is removed
+/// only after the manifest and body were revalidated as one complete committed
+/// checkpoint; damaged, future, mismatched, or foreign-owner slots are retained.
 pub async fn delete_committed_snapshots_through(
     owner: Uuid,
     committed_through: u64,
@@ -580,11 +584,10 @@ async fn delete_committed_snapshots_through_on(
     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 {
+        for (manifest_path, payload_path) in MANIFEST_PATHS.into_iter().zip(PAYLOAD_PATHS) {
+            let manifest_bytes = match read_bounded(disk, manifest_path, MANIFEST_LEN).await {
                 Ok(Some(existing)) => existing,
                 Ok(None) => continue,
                 Err(error) => {
@@ -594,7 +597,7 @@ async fn delete_committed_snapshots_through_on(
                     continue;
                 }
             };
-            let manifest = match Manifest::decode(&existing, max_bytes) {
+            let manifest = match Manifest::decode(&manifest_bytes, max_bytes) {
                 Ok(manifest) => manifest,
                 Err(error) => {
                     if first_error.is_none() {
@@ -606,16 +609,58 @@ async fn delete_committed_snapshots_through_on(
             if manifest.owner != owner || manifest.sequence > committed_through {
                 continue;
             }
+            let payload_bytes = match read_bounded(disk, payload_path, manifest.payload_len).await {
+                Ok(Some(payload)) => payload,
+                Ok(None) => {
+                    if first_error.is_none() {
+                        first_error = Some(SnapshotError::Corrupt);
+                    }
+                    continue;
+                }
+                Err(error) => {
+                    if first_error.is_none() {
+                        first_error = Some(error);
+                    }
+                    continue;
+                }
+            };
+            if let Err(error) = CommittedSnapshot::decode(0, &manifest_bytes, payload_bytes.clone(), max_bytes) {
+                if first_error.is_none() {
+                    first_error = Some(error);
+                }
+                continue;
+            }
             match EcstoreDiskAPI::compare_and_update_file(
                 disk.as_ref(),
                 RUSTFS_META_BUCKET,
-                path,
-                Some(EcstoreDiskBytes::from(existing)),
+                manifest_path,
+                Some(EcstoreDiskBytes::copy_from_slice(&manifest_bytes)),
                 None,
             )
             .await
             {
-                Ok(EcstoreConditionalFileUpdate::Updated) => any_changed = true,
+                Ok(EcstoreConditionalFileUpdate::Updated) => {
+                    match EcstoreDiskAPI::compare_and_update_file(
+                        disk.as_ref(),
+                        RUSTFS_META_BUCKET,
+                        payload_path,
+                        Some(EcstoreDiskBytes::copy_from_slice(&payload_bytes)),
+                        None,
+                    )
+                    .await
+                    {
+                        Ok(
+                            EcstoreConditionalFileUpdate::Updated
+                            | EcstoreConditionalFileUpdate::Missing
+                            | EcstoreConditionalFileUpdate::Mismatch,
+                        ) => {}
+                        Err(error) => {
+                            if first_error.is_none() {
+                                first_error = Some(SnapshotError::Disk(error));
+                            }
+                        }
+                    }
+                }
                 Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {}
                 Err(error) => {
                     if first_error.is_none() {
@@ -625,13 +670,7 @@ async fn delete_committed_snapshots_through_on(
             }
         }
     }
-    if any_changed {
-        Ok(true)
-    } else if let Some(error) = first_error {
-        Err(error)
-    } else {
-        Ok(true)
-    }
+    if let Some(error) = first_error { Err(error) } else { Ok(true) }
 }
 
 async fn read_recovery_snapshot(disks: &[EcstoreDiskStore], limit: usize) -> Result<Option<RecoverySnapshot>, SnapshotError> {
@@ -1101,6 +1140,45 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn committed_snapshot_writer_capacity_failure_preserves_previous_anchor() {
+        let root = TempDir::new().expect("test directory");
+        let store = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let old = payload("old");
+        let next = payload("next");
+        commit(&store, 0, owner, 1, &old).await;
+
+        let result = publish_committed_snapshot(std::slice::from_ref(&store), owner, 2, &next, next.len() - 1).await;
+
+        assert!(
+            matches!(result, Err(SnapshotError::TooLarge)),
+            "capacity failure must be reported separately from corruption: {result:?}"
+        );
+        let reopened = disk(&root, "disk").await;
+        let recovered = read_committed(std::slice::from_ref(&reopened), 4096)
+            .await
+            .expect("read previous committed snapshot")
+            .expect("old anchor remains committed");
+        assert_eq!(recovered.sequence(), 1);
+        assert_eq!(recovered.slot(), 0);
+        assert_eq!(recovered.payload(), old.as_slice());
+        assert_eq!(
+            EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
+                .await
+                .expect("old manifest retained")
+                .as_ref(),
+            manifest(owner, 1, &old).as_slice()
+        );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "oversized successor payload must not be staged"
+        );
+    }
+
     #[tokio::test]
     async fn committed_snapshot_writer_does_not_overwrite_damaged_inactive_manifest() {
         let root = TempDir::new().expect("test directory");
@@ -1355,7 +1433,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn committed_cleanup_removes_only_manifests_at_or_below_sequence() {
+    async fn committed_cleanup_removes_complete_checkpoint_at_or_below_sequence() {
         let root = TempDir::new().expect("test directory");
         let disk = disk(&root, "disk").await;
         let owner = Uuid::new_v4();
@@ -1377,12 +1455,19 @@ mod tests {
             ),
             "old manifest is gone, so the old payload cannot become visible again"
         );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "old payload should be reclaimed after its manifest is removed"
+        );
         assert_eq!(
-            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
                 .await
-                .expect("old payload orphan may remain")
+                .expect("newer payload retained")
                 .as_ref(),
-            older
+            newer
         );
         let recovered = read_committed(std::slice::from_ref(&disk), 4096)
             .await
@@ -1392,6 +1477,40 @@ mod tests {
         assert_eq!(recovered.payload(), newer);
     }
 
+    #[tokio::test]
+    async fn committed_cleanup_retains_manifest_when_payload_identity_mismatches() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let owner = Uuid::new_v4();
+        let declared = payload("declared");
+        let actual = payload("actual");
+        install(&disk, MANIFEST_PATHS[0], &manifest(owner, 3, &declared)).await;
+        install(&disk, PAYLOAD_PATHS[0], &actual).await;
+
+        assert!(
+            matches!(
+                delete_committed_snapshots_through_on(std::slice::from_ref(&disk), owner, 3, 4096).await,
+                Err(SnapshotError::Corrupt)
+            ),
+            "cleanup must fail closed when the committed body no longer matches its manifest"
+        );
+
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
+                .await
+                .expect("mismatched manifest retained")
+                .as_ref(),
+            manifest(owner, 3, &declared)
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
+                .await
+                .expect("mismatched payload retained")
+                .as_ref(),
+            actual
+        );
+    }
+
     #[tokio::test]
     async fn committed_cleanup_preserves_other_owner_manifests_within_sequence_window() {
         let root = TempDir::new().expect("test directory");
@@ -1417,6 +1536,13 @@ mod tests {
             ),
             "the replay owner's manifest is reclaimed"
         );
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "the replay owner's payload is reclaimed"
+        );
         assert_eq!(
             EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
                 .await
diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs
index bf7ada9b2..2eef2b322 100644
--- a/crates/heal/tests/mrf_pipeline_test.rs
+++ b/crates/heal/tests/mrf_pipeline_test.rs
@@ -51,6 +51,8 @@ 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_PAYLOAD_RELS: [&str; 2] = [".heal-mrf-snapshot.0.bin", ".heal-mrf-snapshot.1.bin"];
+const COMMITTED_MANIFEST_RELS: [&str; 2] = [".heal-mrf-commit.0.bin", ".heal-mrf-commit.1.bin"];
 const COMMITTED_MAGIC: &[u8; 8] = b"RFMRFC01";
 const COMMITTED_MANIFEST_LEN: usize = 8 + 1 + 16 + 8 + 8 + 32 + 32;
 
@@ -235,6 +237,44 @@ fn journal_matches_on_all_disks(disk_paths: &[PathBuf], relative_path: &str, exp
         .all(|path| std::fs::read(path.join(META_BUCKET).join(relative_path)).is_ok_and(|actual| actual == expected))
 }
 
+fn committed_checkpoint_matches_on_all_disks(disk_paths: &[PathBuf], sequence: u64, expected_payload: &[u8]) -> bool {
+    disk_paths.iter().all(|path| {
+        let root = path.join(META_BUCKET);
+        COMMITTED_PAYLOAD_RELS
+            .into_iter()
+            .zip(COMMITTED_MANIFEST_RELS)
+            .any(|(payload_rel, manifest_rel)| {
+                let Ok(payload) = std::fs::read(root.join(payload_rel)) else {
+                    return false;
+                };
+                if payload != expected_payload {
+                    return false;
+                }
+                let Ok(manifest) = std::fs::read(root.join(manifest_rel)) else {
+                    return false;
+                };
+                if manifest.len() != COMMITTED_MANIFEST_LEN || &manifest[..8] != COMMITTED_MAGIC || manifest[8] != 1 {
+                    return false;
+                }
+                let Ok(recorded_sequence) = <[u8; 8]>::try_from(&manifest[25..33]).map(u64::from_le_bytes) else {
+                    return false;
+                };
+                let Ok(recorded_len) = <[u8; 8]>::try_from(&manifest[33..41]).map(u64::from_le_bytes) else {
+                    return false;
+                };
+                let Ok(expected_len) = u64::try_from(expected_payload.len()) else {
+                    return false;
+                };
+                if recorded_sequence != sequence || recorded_len != expected_len {
+                    return false;
+                }
+                let payload_digest: [u8; 32] = Sha256::digest(expected_payload).into();
+                let manifest_digest: [u8; 32] = Sha256::digest(&manifest[..COMMITTED_MANIFEST_LEN - 32]).into();
+                payload_digest.as_slice() == &manifest[41..73] && manifest_digest.as_slice() == &manifest[73..]
+            })
+    })
+}
+
 async fn wait_until<F, Fut>(deadline: Duration, mut probe: F) -> bool
 where
     F: FnMut() -> Fut,
@@ -629,13 +669,14 @@ fn mrf_successor_flush_child_process_fixture() {
         let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2);
         let flushed = wait_until(Duration::from_secs(10), || async {
             manager.operations_snapshot().await.queued_by_source.mrf == 1
+                && committed_checkpoint_matches_on_all_disks(&disk_paths, 2, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor)
         })
         .await;
         assert!(
             flushed,
-            "child process must publish the pending successor snapshot before the delete phase"
+            "child process must publish the committed pending successor before the delete phase"
         );
     });
     std::process::exit(78);
@@ -676,13 +717,14 @@ fn mrf_successor_flush_waiting_child_process_fixture() {
         let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2);
         let flushed = wait_until(Duration::from_secs(10), || async {
             manager.operations_snapshot().await.queued_by_source.mrf == 1
+                && committed_checkpoint_matches_on_all_disks(&disk_paths, 2, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor)
                 && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor)
         })
         .await;
         assert!(
             flushed,
-            "child process must publish the pending successor snapshot before it can be killed"
+            "child process must publish the committed pending successor before it can be killed"
         );
         std::fs::write(&ready_path, b"ready").expect("write ready marker");
         loop {