From 7c7b48aaca200b9f8a9224420fdc3db3742f42f3 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 14:16:31 +0800 Subject: [PATCH] test(heal): cover MRF replay across process restart (#7355) Add an ECStore-backed child-process fixture that publishes the MRF journal in one OS process, then verifies a restarted manager can replay the complete snapshot while retaining the journal when bounded admission accepts only a prefix. Co-authored-by: zhi22915 --- crates/heal/tests/mrf_pipeline_test.rs | 87 ++++++++++++++++++++++++-- 1 file changed, 82 insertions(+), 5 deletions(-) diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index fdcf7848c..1b54b4899 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -29,7 +29,7 @@ use rustfs_heal::heal::{ storage::{ECStoreHealStorage, HealStorageAPI}, }; use serial_test::serial; -use std::{path::Path, sync::Arc, time::Duration}; +use std::{path::Path, process::Command, sync::Arc, time::Duration}; mod storage_api; @@ -40,10 +40,15 @@ const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin"; const SCOPED_JOURNAL_REL: &str = "buckets/.heal/mrf/journal-scoped.bin"; async fn heal_env() -> (Vec, Arc) { - let env = rustfs_test_utils::TestECStoreEnv::builder() - .prefix("rustfs_heal_mrf_test") - .build() - .await; + heal_env_at(None).await +} + +async fn heal_env_at(base_dir: Option<&Path>) -> (Vec, Arc) { + let mut builder = rustfs_test_utils::TestECStoreEnv::builder().prefix("rustfs_heal_mrf_test"); + if let Some(base_dir) = base_dir { + builder = builder.base_dir(base_dir); + } + let env = builder.build().await; let heal_storage: Arc = Arc::new(ECStoreHealStorage::new(env.ecstore.clone())); (env.disk_paths, heal_storage) } @@ -117,6 +122,12 @@ fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) { write_journal_path_to_disks(disk_paths, JOURNAL_REL, data); } +fn journal_exists_on_all_disks(disk_paths: &[std::path::PathBuf], relative_path: &str) -> bool { + disk_paths + .iter() + .all(|path| Path::new(path).join(META_BUCKET).join(relative_path).exists()) +} + async fn wait_until(deadline: Duration, mut probe: F) -> bool where F: FnMut() -> Fut, @@ -313,3 +324,69 @@ async fn journal_replay_retains_file_when_manager_is_full() { "the anchor remains until a successor snapshot can safely replace it" ); } + +#[test] +fn mrf_journal_child_process_fixture() { + let Ok(root) = std::env::var("RUSTFS_MRF_REPLAY_CHILD_ROOT") else { + return; + }; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("child runtime should build"); + runtime.block_on(async { + let (disk_paths, _storage) = heal_env_at(Some(Path::new(&root))).await; + let mut journal = journal_record(1, "child-restart-bucket", "first-object", None, 0); + journal.extend(journal_record(1, "child-restart-bucket", "second-object", None, 0)); + write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &journal); + write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &journal); + assert!( + journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL), + "child process must publish the authoritative MRF journal before exiting" + ); + }); + std::process::exit(77); +} + +/// A journal published by a different OS process must remain a durable anchor +/// when the restarted process can only admit a prefix of the replayed intents. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn journal_replay_retains_child_process_anchor_when_manager_is_full() { + let temp_dir = tempfile::tempdir().expect("child process MRF root"); + let status = Command::new(std::env::current_exe().expect("test binary path")) + .arg("mrf_journal_child_process_fixture") + .arg("--exact") + .arg("--nocapture") + .env("RUSTFS_MRF_REPLAY_CHILD_ROOT", temp_dir.path()) + .status() + .expect("child MRF fixture should start"); + assert_eq!(status.code(), Some(77), "child process did not reach the MRF journal boundary"); + + let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await; + assert!( + journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL), + "restarted process must see the authoritative MRF journal left by the child" + ); + + let restarted = Arc::new(HealManager::new( + storage, + Some(HealConfig { + queue_size: 1, + heal_interval: Duration::from_secs(3600), + enable_auto_heal: false, + ..Default::default() + }), + )); + let replayed = mrf_queue::replay_journal_once(&restarted).await; + assert_eq!(replayed, 2, "the restarted process must decode the complete child journal"); + assert_eq!( + restarted.operations_snapshot().await.queued_by_source.mrf, + 1, + "bounded admission may accept only the prefix, but must not lose the replayed tail" + ); + assert!( + journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL), + "replay must retain the child-published journal until a successor snapshot can replace it" + ); +}