From 609abd4928bff576beee8a397a050c735bb15110 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 02:32:30 +0800 Subject: [PATCH] test(heal): cover MRF snapshot recovery bounds Cover committed MRF snapshot reader resource accounting and reopen recovery when successor publication is incomplete or stale. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/heal/src/heal/mrf_queue/snapshot.rs | 126 ++++++++++++++++++++- 1 file changed, 124 insertions(+), 2 deletions(-) diff --git a/crates/heal/src/heal/mrf_queue/snapshot.rs b/crates/heal/src/heal/mrf_queue/snapshot.rs index d39e0fafa..653ab0215 100644 --- a/crates/heal/src/heal/mrf_queue/snapshot.rs +++ b/crates/heal/src/heal/mrf_queue/snapshot.rs @@ -123,6 +123,13 @@ pub struct CommittedSnapshot { payload: Vec, } +#[derive(Default)] +struct SnapshotReadStats { + file_reads: usize, + bytes_read: usize, + peak_file_bytes: usize, +} + impl CommittedSnapshot { /// Persistent single-writer sequence, not a process UUID ordering. pub fn sequence(&self) -> u64 { @@ -162,6 +169,15 @@ pub enum RecoverySnapshot { } async fn read_bounded(disk: &EcstoreDiskStore, path: &str, limit: usize) -> Result>, SnapshotError> { + read_bounded_with_stats(disk, path, limit, None).await +} + +async fn read_bounded_with_stats( + disk: &EcstoreDiskStore, + path: &str, + limit: usize, + mut stats: Option<&mut SnapshotReadStats>, +) -> Result>, SnapshotError> { let reader = match EcstoreDiskAPI::read_file(disk.as_ref(), RUSTFS_META_BUCKET, path).await { Ok(reader) => reader, Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound) => return Ok(None), @@ -178,6 +194,11 @@ async fn read_bounded(disk: &EcstoreDiskStore, path: &str, limit: usize) -> Resu if bytes.len() > limit { return Err(SnapshotError::TooLarge); } + if let Some(stats) = stats.as_mut() { + stats.file_reads += 1; + stats.bytes_read = stats.bytes_read.checked_add(bytes.len()).ok_or(SnapshotError::TooLarge)?; + stats.peak_file_bytes = stats.peak_file_bytes.max(bytes.len()); + } Ok(Some(bytes)) } @@ -197,17 +218,26 @@ fn select_snapshot(selected: &mut Option, candidate: Committe } async fn read_committed(disks: &[EcstoreDiskStore], limit: usize) -> Result, SnapshotError> { + read_committed_with_stats(disks, limit, None).await +} + +async fn read_committed_with_stats( + disks: &[EcstoreDiskStore], + limit: usize, + mut stats: Option<&mut SnapshotReadStats>, +) -> Result, SnapshotError> { let mut selected = None; let mut damaged = None; let mut identities = HashMap::new(); for disk in disks { for (manifest_path, payload_path) in MANIFEST_PATHS.into_iter().zip(PAYLOAD_PATHS) { let candidate = async { - let Some(manifest) = read_bounded(disk, manifest_path, MANIFEST_LEN).await? else { + let Some(manifest) = read_bounded_with_stats(disk, manifest_path, MANIFEST_LEN, stats.as_deref_mut()).await? + else { return Ok(None); }; let header = Manifest::decode(&manifest, limit)?; - let payload = read_bounded(disk, payload_path, header.payload_len) + let payload = read_bounded_with_stats(disk, payload_path, header.payload_len, stats.as_deref_mut()) .await? .ok_or(SnapshotError::Corrupt)?; CommittedSnapshot::decode(&manifest, payload, limit).map(Some) @@ -492,6 +522,69 @@ mod tests { assert_eq!(recovered.manifest.sequence, 1); } + #[tokio::test] + async fn committed_reader_reopens_previous_anchor_across_publication_boundaries() { + let owner = Uuid::new_v4(); + let old = payload("old"); + let next = payload("next"); + let boundaries = [ + ("payload-only", next.clone(), None), + ("torn-manifest", next.clone(), Some(manifest(owner, 2, &next)[..20].to_vec())), + ("stale-payload", old.clone(), Some(manifest(owner, 2, &next))), + ]; + for (case, successor_payload, successor_manifest) in boundaries { + let root = TempDir::new().expect("test directory"); + let store = disk(&root, "disk").await; + commit(&store, 0, owner, 1, &old).await; + install(&store, PAYLOAD_PATHS[1], &successor_payload).await; + if let Some(manifest) = &successor_manifest { + install(&store, MANIFEST_PATHS[1], manifest).await; + } + + let reopened = disk(&root, "disk").await; + let recovered = read_committed(std::slice::from_ref(&reopened), 4096) + .await + .unwrap_or_else(|error| panic!("{case}: old anchor must remain readable after reopen: {error:?}")) + .unwrap_or_else(|| panic!("{case}: previous committed anchor missing after reopen")); + assert_eq!(recovered.manifest.sequence, 1, "{case}: successor must not become authoritative"); + assert_eq!(recovered.payload, old, "{case}: previous payload must survive"); + assert_eq!( + EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0]) + .await + .expect("old payload retained") + .as_ref(), + old.as_slice(), + "{case}: previous payload bytes changed" + ); + 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(), + "{case}: previous manifest bytes changed" + ); + assert_eq!( + EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1]) + .await + .expect("successor payload retained") + .as_ref(), + successor_payload.as_slice(), + "{case}: successor evidence changed" + ); + if let Some(manifest) = &successor_manifest { + assert_eq!( + EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1]) + .await + .expect("successor manifest retained") + .as_ref(), + manifest.as_slice(), + "{case}: successor manifest evidence changed" + ); + } + } + } + #[tokio::test] async fn stale_manifest_cas_cannot_replace_committed_anchor() { let root = TempDir::new().expect("test directory"); @@ -571,6 +664,35 @@ mod tests { ); } + #[tokio::test] + async fn committed_reader_resource_bounds_are_measured() { + let root = TempDir::new().expect("test directory"); + let first = disk(&root, "first").await; + let second = disk(&root, "second").await; + let owner = Uuid::new_v4(); + let old = [payload("old-0"), payload("old-1")].concat(); + let new = [payload("new-0"), payload("new-1"), payload("new-2")].concat(); + commit(&first, 0, owner, 1, &old).await; + commit(&second, 1, owner, 2, &new).await; + + let mut stats = SnapshotReadStats::default(); + let recovered = read_committed_with_stats(&[first, second], 4096, Some(&mut stats)) + .await + .expect("read committed replicas") + .expect("committed snapshot"); + + assert_eq!(recovered.sequence(), 2); + assert_eq!(recovered.payload(), new.as_slice()); + assert_eq!(recovered.manifest.payload_len, new.len()); + assert_eq!(stats.file_reads, 4, "only committed manifests and their payloads are materialized"); + assert_eq!(stats.bytes_read, (MANIFEST_LEN * 2) + old.len() + new.len()); + assert_eq!( + stats.peak_file_bytes, + new.len().max(MANIFEST_LEN), + "reader peak allocation remains bounded by one manifest or payload file" + ); + } + #[tokio::test] async fn legacy_inspection_rejects_complete_subsets_and_scope_ambiguity() { let scoped = |set_index| {