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

@@ -6179,6 +6179,36 @@ fn join_errs(errs: &[Option<DiskError>]) -> 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<FileInfo> {
         let distribution = &files
             .first()

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<u8> = (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;