From 7a7b080265e02b127aab1db63294650d53a2bf93 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 13 Sep 2026 14:29:41 +0800 Subject: [PATCH] fix(heal): verify inline shard bitrot during deep scans (#7731) * fix(heal): verify inline shard bitrot during deep scans * test(heal): fix CI lint and topology fixtures --- crates/ecstore/src/set_disk/mod.rs | 147 +++++++++- crates/ecstore/src/set_disk/ops/heal.rs | 366 ++++++++++++++++++++++++ 2 files changed, 507 insertions(+), 6 deletions(-) diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index f2cbafc55..21a6f0ada 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -6179,6 +6179,36 @@ fn join_errs(errs: &[Option]) -> String { errs.join(", ") } +async fn verify_inline_part_bitrot(meta: &FileInfo) -> disk::error::Result<()> { + meta.validate_for_metadata_read()?; + let [part] = meta.parts.as_slice() else { + return Err(DiskError::FileCorrupt); + }; + let data = meta.data.as_deref().ok_or(DiskError::FileCorrupt)?; + let checksum = meta.erasure.get_checksum_info(part.number); + let algo = match checksum.algorithm { + HashAlgorithm::HighwayHash256S if meta.uses_legacy_checksum => HashAlgorithm::HighwayHash256SLegacy, + algo @ (HashAlgorithm::HighwayHash256S | HashAlgorithm::HighwayHash256SLegacy) => algo, + _ => return Err(DiskError::BitrotHashAlgoInvalid), + }; + let shard_size = inline_erasure_shard_size(meta.erasure.block_size, meta.erasure.data_blocks, meta.uses_legacy_checksum); + let part_size = + inline_erasure_shard_file_size(part.size, meta.erasure.block_size, meta.erasure.data_blocks, meta.uses_legacy_checksum); + // Check framing arithmetic and the physical buffer length before the shared + // verifier sizes its scratch buffer from the metadata. + let encoded_size = part_size + .div_ceil(shard_size) + .checked_mul(algo.size()) + .and_then(|hash_size| part_size.checked_add(hash_size)) + .ok_or(DiskError::FileCorrupt)?; + if encoded_size != data.len() { + return Err(DiskError::FileCorrupt); + } + coding::bitrot_verify(Cursor::new(data), encoded_size, part_size, algo, shard_size) + .await + .map_err(|_| DiskError::FileCorrupt) +} + /// disks_with_all_partsv2 is a corrected version based on Go implementation. /// It sets partsMetadata and onlineDisks when xl.meta is inexistant/corrupted or outdated. /// It also checks if the status of each part (corrupted, missing, ok) in each drive. @@ -6350,16 +6380,22 @@ async fn disks_with_all_parts( continue; } - // Inline data is stored inside xl.meta, so there is no separate part file to - // verify here. Treat the shard as present once metadata was read successfully; - // object reads/heal will validate the inline shard through the normal bitrot - // reader path. Running bitrot_verify directly here can falsely mark small - // inline shards corrupt when older metadata has no per-part checksum entries. + // Normal scans only check presence. Deep scans must verify inline bytes + // before deciding whether reconstruction (and its bitrot readers) is needed. if (meta.data.is_some() || meta.size == 0) && !meta.parts.is_empty() { + let part_status = if scan_mode == HealScanMode::Deep && meta.data.is_some() { + match verify_inline_part_bitrot(meta).await { + Ok(()) => CHECK_PART_SUCCESS, + Err(DiskError::FileCorrupt) => CHECK_PART_FILE_CORRUPT, + Err(err) => return Err(err), + } + } else { + CHECK_PART_SUCCESS + }; if let Some(vec) = data_errs_by_part.get_mut(&0) && index < vec.len() { - vec[index] = CHECK_PART_SUCCESS; + vec[index] = part_status; } continue; } @@ -12468,6 +12504,8 @@ mod tests { file.add_object_part(1, "part-etag-inline".to_string(), payload.len(), file.mod_time, file.size, None, None); file.set_inline_data(); file.erasure.index = files.len() + 1; + file.erasure.block_size = erasure.block_size; + file.uses_legacy_checksum = uses_legacy; file.data = Some(Bytes::from(data)); files.push(file); } @@ -12479,6 +12517,103 @@ mod tests { inline_bitrot_files_for_payload_with_mode(payload, false).await } + #[tokio::test] + async fn deep_heal_inline_bitrot_checks_current_and_legacy_frames() { + let payload = vec![0x7b; 4113]; + for legacy in [false, true] { + let (_, files, _, _) = inline_bitrot_files_for_payload_with_mode(&payload, legacy).await; + for mut meta in files { + assert!(meta.erasure.checksums.is_empty(), "legacy metadata may omit per-part checksum entries"); + verify_inline_part_bitrot(&meta) + .await + .expect("healthy data and parity shards should verify"); + let original = meta.data.clone().expect("fixture should retain inline bytes"); + for offset in [0, 32, original.len() - 1] { + let mut damaged = original.to_vec(); + damaged[offset] ^= 1; + meta.data = Some(Bytes::from(damaged)); + assert_eq!(verify_inline_part_bitrot(&meta).await, Err(DiskError::FileCorrupt)); + } + let mut trailing = original.to_vec(); + trailing.push(0x7b); + for damaged in [Vec::new(), original[..original.len() - 1].to_vec(), trailing] { + meta.data = Some(Bytes::from(damaged)); + assert_eq!(verify_inline_part_bitrot(&meta).await, Err(DiskError::FileCorrupt)); + } + } + } + } + + #[tokio::test] + async fn deep_heal_inline_bitrot_checks_empty_objects_and_metadata_bounds() { + let (_, files, _, _) = inline_bitrot_files_for_payload(b"inline metadata bounds").await; + let mut empty = files[0].clone(); + empty.size = 0; + empty.parts[0].size = 0; + empty.parts[0].actual_size = 0; + empty.data = Some(Bytes::new()); + verify_inline_part_bitrot(&empty) + .await + .expect("empty inline objects have no bitrot frames"); + empty.data = Some(Bytes::from_static(b"unexpected")); + assert_eq!(verify_inline_part_bitrot(&empty).await, Err(DiskError::FileCorrupt)); + + let mut invalid = files[0].clone(); + invalid.erasure.block_size = 0; + assert_eq!(verify_inline_part_bitrot(&invalid).await, Err(DiskError::FileCorrupt)); + invalid = files[0].clone(); + invalid.parts.push(invalid.parts[0].clone()); + invalid.parts[1].number = 2; + assert_eq!(verify_inline_part_bitrot(&invalid).await, Err(DiskError::FileCorrupt)); + + let mut oversized = files[0].clone(); + oversized.erasure.block_size = 2; + oversized.size = i64::MAX - 3; + oversized.parts[0].size = usize::try_from(oversized.size).expect("64-bit metadata size should fit"); + oversized + .validate_for_metadata_read() + .expect("logical shard length should fit metadata bounds"); + assert_eq!(verify_inline_part_bitrot(&oversized).await, Err(DiskError::FileCorrupt)); + + oversized.erasure.block_size = 1 << 40; + oversized.size = 1 << 40; + oversized.parts[0].size = usize::try_from(oversized.size).expect("64-bit metadata size should fit"); + oversized + .validate_for_metadata_read() + .expect("large metadata geometry should remain representable"); + assert_eq!(verify_inline_part_bitrot(&oversized).await, Err(DiskError::FileCorrupt)); + } + + #[tokio::test] + async fn deep_heal_inline_bitrot_accepts_pinned_disk_fixtures() { + use rustfs_filemeta::test_data::{create_issue_2265_legacy_meta_v2_object_xlmeta, create_issue_2288_legacy_xlmeta}; + + for (bytes, size, legacy) in [ + (create_issue_2288_legacy_xlmeta().expect("pinned meta v1 fixture"), 35, false), + ( + create_issue_2265_legacy_meta_v2_object_xlmeta().expect("pinned meta v2 fixture"), + 707, + true, + ), + ] { + let file_meta = rustfs_filemeta::FileMeta::load(&bytes).expect("historical xl.meta should decode"); + let mut meta = file_meta + .into_fileinfo("bucket", "object", "", true, false, true) + .expect("historical object metadata should decode"); + assert_eq!(meta.size, size); + assert_eq!(meta.uses_legacy_checksum, legacy); + assert!(meta.inline_data()); + assert!(meta.data.is_some(), "the production decoder should extract the historical inline value"); + verify_inline_part_bitrot(&meta) + .await + .expect("historical inline shard should pass deep verification"); + let mut damaged = meta.data.as_ref().expect("fixture should be inline").to_vec(); + *damaged.last_mut().expect("fixture shard should not be empty") ^= 1; + meta.data = Some(Bytes::from(damaged)); + assert_eq!(verify_inline_part_bitrot(&meta).await, Err(DiskError::FileCorrupt)); + } + } + fn disk_ordered_fileinfos(files: &[FileInfo]) -> Vec { let distribution = &files .first() diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 4f934b458..73e18e4aa 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -3339,6 +3339,372 @@ mod heal_result_report_tests { ); } + #[tokio::test] + async fn deep_heal_inline_bitrot_rebuilds_physical_data_and_parity_shards() { + use crate::storage_api_contracts::range::HTTPRangeSpec; + use tokio::io::AsyncReadExt; + + let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "deep-heal-inline-bitrot"; + for disk in &disks { + disk.make_volume(bucket).await.expect("test bucket should be created"); + } + let payload: Vec = (0..4113) + .map(|index| u8::try_from(index % 251).expect("payload byte")) + .collect(); + let read_options = ReadOptions { + read_data: true, + ..Default::default() + }; + let object_options = ObjectOptions { + no_lock: true, + ..Default::default() + }; + + for parity in [false, true] { + for corrupt_prefix in [false, true] { + let object = format!("parity-{parity}-prefix-{corrupt_prefix}.bin"); + set.put_object(bucket, &object, &mut PutObjReader::from_vec(payload.clone()), &object_options) + .await + .expect("full-fanout PUT should persist every inline shard"); + + let mut target = None; + for (disk_index, disk) in disks.iter().enumerate() { + let metadata = disk + .read_version("", bucket, &object, "", &read_options) + .await + .expect("inline metadata should be readable"); + let target_shard = if parity { metadata.erasure.data_blocks + 1 } else { 1 }; + if metadata.erasure.index == target_shard { + target = Some((disk_index, metadata)); + break; + } + } + let (disk_index, original_meta) = target.expect("requested data/parity shard should exist"); + let original_inline = original_meta.data.as_ref().expect("object should be inline"); + let metadata_path = temp_dirs[disk_index] + .path() + .join(bucket) + .join(&object) + .join(STORAGE_FORMAT_FILE); + let mut damaged_raw = tokio::fs::read(&metadata_path).await.expect("physical xl.meta should exist"); + let offsets: Vec<_> = damaged_raw + .windows(original_inline.len()) + .enumerate() + .filter_map(|(offset, bytes)| (bytes == original_inline.as_ref()).then_some(offset)) + .collect(); + assert_eq!(offsets.len(), 1, "inline shard must have a unique physical byte range"); + let bitflip_offset = if corrupt_prefix { 0 } else { original_inline.len() - 1 }; + damaged_raw[offsets[0] + bitflip_offset] ^= 1; + tokio::fs::write(&metadata_path, &damaged_raw) + .await + .expect("only the encoded inline shard should be damaged"); + let damaged_meta = disks[disk_index] + .read_version("", bucket, &object, "", &read_options) + .await + .expect("payload corruption must leave metadata parseable"); + assert!(damaged_meta.equals(&original_meta)); + assert_eq!(damaged_meta.version_id, original_meta.version_id); + assert_eq!(damaged_meta.data_dir, original_meta.data_dir); + assert_ne!(damaged_meta.data, original_meta.data); + + for scan_mode in [HealScanMode::Normal, HealScanMode::Deep] { + let (result, error) = set + .heal_object( + bucket, + &object, + "", + &HealOpts { + no_lock: true, + scan_mode, + ..Default::default() + }, + ) + .await + .expect("inline heal should complete"); + assert!(error.is_none(), "inline heal should remain recoverable: {error:?}"); + let deep = scan_mode == HealScanMode::Deep; + assert_eq!(result.drives_healed(), Some(usize::from(deep)), "{object}: {scan_mode:?}"); + assert_eq!(result.after.drives[disk_index].state, DriveState::Ok.to_string()); + if deep { + assert_eq!(result.before.drives[disk_index].state, DriveState::Corrupt.to_string()); + let repaired = disks[disk_index] + .read_version("", bucket, &object, "", &read_options) + .await + .expect("repaired physical shard should be readable"); + assert_eq!(repaired.data, original_meta.data, "heal must restore the exact encoded shard"); + assert_eq!(repaired.version_id, original_meta.version_id); + assert_eq!(repaired.data_dir, original_meta.data_dir); + } + + for (range, expected) in [ + (None, payload.as_slice()), + ( + Some(HTTPRangeSpec { + start: 113, + end: 1023, + is_suffix_length: false, + }), + &payload[113..1024], + ), + ] { + let mut reader = set + .get_object_reader(bucket, &object, range, Default::default(), &object_options) + .await + .expect("redundant shards should serve full and range reads"); + let mut body = Vec::new(); + reader + .stream + .read_to_end(&mut body) + .await + .expect("GET should finish without truncation"); + assert_eq!(body, expected); + } + if !deep { + assert_eq!( + tokio::fs::read(&metadata_path) + .await + .expect("damaged shard should remain on disk"), + damaged_raw, + "successful GET and normal heal do not prove the damaged shard was repaired" + ); + } + } + + let (healthy, error) = set + .heal_object( + bucket, + &object, + "", + &HealOpts { + no_lock: true, + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + ) + .await + .expect("healthy inline object should pass a second deep scan"); + assert!(error.is_none()); + assert_eq!(healthy.drives_healed(), Some(0)); + } + } + } + + #[tokio::test] + async fn deep_heal_inline_bitrot_preserves_quorum_and_dry_run_boundaries() { + for corrupt_shards in [2, 3] { + let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "deep-heal-inline-quorum"; + let object = "object.bin"; + for disk in &disks { + disk.make_volume(bucket).await.expect("test bucket should be created"); + } + set.put_object( + bucket, + object, + &mut PutObjReader::from_vec(vec![0x71; 4113]), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("all inline shards should be committed"); + let read_options = ReadOptions { + read_data: true, + ..Default::default() + }; + let mut originals = Vec::new(); + let mut damaged_files = Vec::new(); + for disk_index in 0..corrupt_shards { + let metadata = disks[disk_index] + .read_version("", bucket, object, "", &read_options) + .await + .expect("source shard should be readable"); + let inline = metadata.data.expect("source shard should be inline"); + let path = temp_dirs[disk_index] + .path() + .join(bucket) + .join(object) + .join(STORAGE_FORMAT_FILE); + let mut raw = tokio::fs::read(&path).await.expect("physical metadata should exist"); + let offsets: Vec<_> = raw + .windows(inline.len()) + .enumerate() + .filter_map(|(offset, bytes)| (bytes == inline.as_ref()).then_some(offset)) + .collect(); + assert_eq!(offsets.len(), 1, "shard byte range must be unique"); + raw[offsets[0] + inline.len() - 1] ^= 1; + tokio::fs::write(&path, &raw).await.expect("inline shard should be damaged"); + originals.push(inline); + damaged_files.push((path, raw)); + } + let opts = HealOpts { + no_lock: true, + scan_mode: HealScanMode::Deep, + ..Default::default() + }; + let (dry_run, error) = set + .heal_object(bucket, object, "", &HealOpts { dry_run: true, ..opts }) + .await + .expect("dry-run should report inline corruption"); + assert!(error.is_none()); + assert_eq!(dry_run.drives_healed(), Some(0)); + for (disk_index, (path, raw)) in damaged_files.iter().enumerate() { + assert_eq!(dry_run.after.drives[disk_index].state, DriveState::Corrupt.to_string()); + assert_eq!(tokio::fs::read(path).await.expect("dry-run should preserve metadata"), *raw); + } + + let (result, error) = set + .heal_object(bucket, object, "", &opts) + .await + .expect("heal should report its outcome"); + if corrupt_shards == 2 { + assert!(error.is_none(), "exact read quorum should reconstruct: {error:?}"); + assert_eq!(result.drives_healed(), Some(2)); + for (disk_index, original) in originals.iter().enumerate() { + let repaired = disks[disk_index] + .read_version("", bucket, object, "", &read_options) + .await + .expect("reconstructed shard should be persisted"); + assert_eq!(repaired.data.as_ref(), Some(original)); + } + } else { + assert_eq!(error, Some(DiskError::ErasureReadQuorum)); + assert_eq!(result.drives_healed(), Some(0)); + for (path, raw) in damaged_files { + assert_eq!( + tokio::fs::read(path) + .await + .expect("unrecoverable evidence must remain on disk"), + raw + ); + } + } + } + } + + #[tokio::test] + async fn deep_heal_inline_bitrot_rejects_unsupported_algorithms() { + use crate::set_disk::disks_with_all_parts; + use rustfs_filemeta::ChecksumInfo; + use rustfs_utils::HashAlgorithm; + + let (_temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "deep-heal-inline-algorithm"; + let object = "object.bin"; + for disk in &disks { + disk.make_volume(bucket).await.expect("test bucket should be created"); + } + set.put_object( + bucket, + object, + &mut PutObjReader::from_vec(vec![0x71; 4113]), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("inline fixture should be committed"); + let mut metadata = Vec::new(); + for disk in &disks { + metadata.push( + disk.read_version( + "", + bucket, + object, + "", + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("inline metadata should decode"), + ); + } + let latest = metadata[1].clone(); + let mut online: Vec<_> = disks.into_iter().map(Some).collect(); + for algorithm in [ + HashAlgorithm::SHA256, + HashAlgorithm::HighwayHash256, + HashAlgorithm::BLAKE2b512, + ] { + metadata[0].erasure.checksums = vec![ChecksumInfo { + part_number: 1, + algorithm, + ..Default::default() + }]; + for mode in [HealScanMode::Normal, HealScanMode::Deep] { + let result = disks_with_all_parts( + &mut online, + &mut metadata, + &[None, None, None, None], + &latest, + false, + bucket, + object, + mode, + ) + .await; + if mode == HealScanMode::Deep { + assert_eq!( + result.expect_err("unverifiable inline data must not become healthy"), + DiskError::BitrotHashAlgoInvalid + ); + } else { + let (by_disk, _) = result.expect("normal scan should retain presence-only behavior"); + assert_eq!(by_disk[&0], vec![crate::disk::CHECK_PART_SUCCESS]); + } + } + } + } + + #[tokio::test] + async fn deep_heal_inline_bitrot_accepts_persisted_empty_objects() { + let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "deep-heal-inline-empty"; + let object = "empty.bin"; + for disk in &disks { + disk.make_volume(bucket).await.expect("test bucket should be created"); + } + set.put_object( + bucket, + object, + &mut PutObjReader::from_vec(Vec::new()), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("empty object should be committed"); + let mut originals = Vec::new(); + for dir in &temp_dirs { + let path = dir.path().join(bucket).join(object).join(STORAGE_FORMAT_FILE); + originals.push((path.clone(), tokio::fs::read(path).await.expect("empty object metadata should exist"))); + } + let (result, error) = set + .heal_object( + bucket, + object, + "", + &HealOpts { + no_lock: true, + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + ) + .await + .expect("empty object should pass deep heal"); + assert!(error.is_none()); + assert_eq!(result.drives_healed(), Some(0)); + for (path, raw) in originals { + assert_eq!(tokio::fs::read(path).await.expect("healthy metadata should remain on disk"), raw); + } + } + #[tokio::test] async fn replacement_target_readback_checks_the_requested_historical_version() { let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;