test(heal): persist MRF rollback mirror boundary (#7520)

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

crates/heal/tests/mrf_pipeline_test.rs

@@ -22,7 +22,7 @@
 //! Under `cargo nextest` each test runs in its own process, which keeps the
 //! process-global MRF channel singleton safe.
 
-use rustfs_common::mrf_channel::{self, MrfKind};
+use rustfs_common::mrf_channel::{self, MrfIngressResult, MrfKind, MrfScope};
 use rustfs_heal::heal::{
     manager::{HealConfig, HealManager},
     mrf_queue,
@@ -237,6 +237,33 @@ 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 journal_contains_on_all_disks(disk_paths: &[PathBuf], relative_path: &str, needle: &[u8]) -> bool {
+    disk_paths.iter().all(|path| {
+        std::fs::read(path.join(META_BUCKET).join(relative_path))
+            .is_ok_and(|actual| actual.windows(needle.len()).any(|window| window == needle))
+    })
+}
+
+fn journal_contains_on_any_disk(disk_paths: &[PathBuf], relative_path: &str, needle: &[u8]) -> bool {
+    disk_paths.iter().any(|path| {
+        std::fs::read(path.join(META_BUCKET).join(relative_path))
+            .is_ok_and(|actual| actual.windows(needle.len()).any(|window| window == needle))
+    })
+}
+
+fn committed_payload_contains_on_all_disks(disk_paths: &[PathBuf], needles: &[&[u8]]) -> bool {
+    disk_paths.iter().all(|path| {
+        let root = path.join(META_BUCKET);
+        COMMITTED_PAYLOAD_RELS.into_iter().any(|payload_rel| {
+            std::fs::read(root.join(payload_rel)).is_ok_and(|payload| {
+                needles
+                    .iter()
+                    .all(|needle| payload.windows(needle.len()).any(|window| window == *needle))
+            })
+        })
+    })
+}
+
 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);
@@ -615,6 +642,73 @@ async fn journal_replay_retains_file_when_manager_is_full() {
     );
 }
 
+/// Rollback mirrors are for v1 readers only: the committed and scoped
+/// snapshots remain authoritative, while the legacy journal omits scoped-only
+/// records that an older binary cannot represent safely.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn rollback_legacy_mirror_persists_only_v1_compatible_records() {
+    let (disk_paths, storage) = heal_env().await;
+    register_local_disks(&disk_paths, "mrf-rollback-mirror-test").await;
+
+    let manager = Arc::new(HealManager::new(
+        storage,
+        Some(HealConfig {
+            queue_size: 0,
+            heal_interval: Duration::from_secs(3600),
+            enable_auto_heal: false,
+            ..Default::default()
+        }),
+    ));
+    mrf_queue::spawn_mrf_consumer(manager.clone());
+
+    let scoped_only = b"rollback-scoped-only-object";
+    let v1_compatible = b"rollback-v1-compatible-object";
+    assert_eq!(
+        mrf_channel::try_send_mrf_intent_typed(
+            MrfKind::PartialWrite,
+            "rollback-bucket",
+            std::str::from_utf8(scoped_only).expect("fixture object is UTF-8"),
+            None,
+            Some(MrfScope {
+                pool_index: 3,
+                set_index: 7,
+            }),
+        ),
+        MrfIngressResult::Enqueued,
+        "scoped-only intent should be accepted by the live consumer"
+    );
+    assert_eq!(
+        mrf_channel::try_send_mrf_intent_typed(
+            MrfKind::PartialWrite,
+            "rollback-bucket",
+            std::str::from_utf8(v1_compatible).expect("fixture object is UTF-8"),
+            None,
+            None,
+        ),
+        MrfIngressResult::Enqueued,
+        "v1-compatible intent should be accepted by the live consumer"
+    );
+
+    let flushed = wait_until(Duration::from_secs(10), || async {
+        committed_payload_contains_on_all_disks(&disk_paths, &[scoped_only, v1_compatible])
+            && journal_contains_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, scoped_only)
+            && journal_contains_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, v1_compatible)
+            && journal_contains_on_all_disks(&disk_paths, JOURNAL_REL, v1_compatible)
+            && !journal_contains_on_any_disk(&disk_paths, JOURNAL_REL, scoped_only)
+    })
+    .await;
+    assert!(
+        flushed,
+        "runtime flush must persist rollback-safe mirrors without leaking scoped-only records into the legacy journal"
+    );
+    assert_eq!(
+        manager.operations_snapshot().await.queued_by_source.mrf,
+        0,
+        "zero-capacity manager keeps both intents in the MRF runtime so the persisted snapshot is observable"
+    );
+}
+
 #[test]
 fn mrf_journal_child_process_fixture() {
     let Ok(root) = std::env::var("RUSTFS_MRF_REPLAY_CHILD_ROOT") else {