fix(heal): harden MRF replay boundaries (#7483)

Reject journal records with unknown version-presence flags even when their CRC is valid, so rollback/future payloads cannot be accepted as known records.

Gate committed checkpoint cleanup by the writer owner captured from the replay source, preserving retained manifests from other owners inside the same sequence window.

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

crates/heal/src/heal/mrf_queue.rs

@@ -294,7 +294,11 @@ fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
     };
     let attempts = data[3];
     let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().ok()?);
-    let has_version = data[12] != 0;
+    let has_version = match data[12] {
+        0 => false,
+        1 => true,
+        _ => return None,
+    };
     let mut cursor = MRF_RECORD_FIXED_HEAD;
     let version_id = if has_version {
         if data.len() < cursor + 16 {
@@ -718,7 +722,7 @@ fn replay_must_retain_journal(
 #[derive(Clone, Copy)]
 enum ReplayCleanup {
     Legacy,
-    Committed { sequence: u64 },
+    Committed { owner: Uuid, sequence: u64 },
 }
 
 struct ReplaySource {
@@ -731,6 +735,7 @@ async fn read_replay_source(max_bytes: usize) -> Result<Option<ReplaySource>, sn
         return Ok(Some(ReplaySource {
             data: committed.payload().to_vec(),
             cleanup: ReplayCleanup::Committed {
+                owner: committed.owner(),
                 sequence: committed.sequence(),
             },
         }));
@@ -754,18 +759,20 @@ async fn read_replay_source(max_bytes: usize) -> Result<Option<ReplaySource>, sn
 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
+        ReplayCleanup::Committed { owner, sequence } => {
+            match snapshot::delete_committed_snapshots_through(owner, 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
 }
@@ -1344,6 +1351,25 @@ mod tests {
         assert_eq!(truncated, corrupt.len());
     }
 
+    #[test]
+    fn journal_rejects_unknown_version_presence_flag_even_with_valid_crc() {
+        let mut versioned = intent("rollback-bucket", "object", 0);
+        versioned.version_id = Some([9; 16]);
+        let mut buf = Vec::new();
+        assert!(encode_intent(&versioned, &mut buf));
+
+        buf[12] = 2;
+        let crc_offset = buf.len() - 4;
+        let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
+        hasher.update(&buf[..crc_offset]);
+        let checksum = u32::try_from(hasher.finalize()).expect("CRC32 fits");
+        buf[crc_offset..].copy_from_slice(&checksum.to_le_bytes());
+
+        let (decoded, truncated) = decode_journal(&buf);
+        assert!(decoded.is_empty(), "unknown boolean encodings are not rollback-compatible payloads");
+        assert_eq!(truncated, buf.len());
+    }
+
     #[test]
     fn heal_request_mapping_follows_priority_matrix() {
         let decode = build_heal_request(&intent("b", "o", 0));

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

@@ -555,20 +555,25 @@ pub async fn inspect_local_committed_snapshot(max_bytes: usize) -> Result<Option
     read_committed(&super::journal_disks().await, max_bytes).await
 }
 
-/// Remove committed manifests whose sequence is no newer than
+/// Remove committed manifests 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.
-pub async fn delete_committed_snapshots_through(committed_through: u64, max_bytes: usize) -> Result<bool, SnapshotError> {
+pub async fn delete_committed_snapshots_through(
+    owner: Uuid,
+    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
+    delete_committed_snapshots_through_on(&disks, owner, committed_through, max_bytes).await
 }
 
 async fn delete_committed_snapshots_through_on(
     disks: &[EcstoreDiskStore],
+    owner: Uuid,
     committed_through: u64,
     max_bytes: usize,
 ) -> Result<bool, SnapshotError> {
@@ -598,7 +603,7 @@ async fn delete_committed_snapshots_through_on(
                     continue;
                 }
             };
-            if manifest.sequence > committed_through {
+            if manifest.owner != owner || manifest.sequence > committed_through {
                 continue;
             }
             match EcstoreDiskAPI::compare_and_update_file(
@@ -1360,7 +1365,7 @@ mod tests {
         commit(&disk, 1, owner, 4, &newer).await;
 
         assert!(
-            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), 3, 4096)
+            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), owner, 3, 4096)
                 .await
                 .expect("delete old manifest"),
             "old committed manifest should be removed"
@@ -1387,6 +1392,47 @@ mod tests {
         assert_eq!(recovered.payload(), newer);
     }
 
+    #[tokio::test]
+    async fn committed_cleanup_preserves_other_owner_manifests_within_sequence_window() {
+        let root = TempDir::new().expect("test directory");
+        let disk = disk(&root, "disk").await;
+        let replay_owner = Uuid::new_v4();
+        let other_owner = Uuid::new_v4();
+        let replay_payload = payload("replay-owner");
+        let other_payload = payload("other-owner");
+        commit(&disk, 0, replay_owner, 9, &replay_payload).await;
+        commit(&disk, 1, other_owner, 4, &other_payload).await;
+
+        assert!(
+            delete_committed_snapshots_through_on(std::slice::from_ref(&disk), replay_owner, 9, 4096)
+                .await
+                .expect("delete replay-owner manifest"),
+            "the matched owner manifest should be removed"
+        );
+
+        assert!(
+            matches!(
+                EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0]).await,
+                Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound)
+            ),
+            "the replay owner's manifest is reclaimed"
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
+                .await
+                .expect("other owner manifest retained")
+                .as_ref(),
+            manifest(other_owner, 4, &other_payload)
+        );
+        assert_eq!(
+            EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
+                .await
+                .expect("other owner payload retained")
+                .as_ref(),
+            other_payload
+        );
+    }
+
     #[tokio::test]
     async fn legacy_import_requires_complete_consistent_replicas() {
         let root = TempDir::new().expect("test directory");