test(e2e): add disk fault injection harness and reliability tests (#4223)

Add crates/e2e_test/src/chaos.rs with in-process fault-injection
primitives for a single-node multi-disk RustFS server: take a disk
offline (rename to <dir>.offline), bring it back online, replace a
disk with a fresh empty directory, corrupt an object's erasure shard
(part.* byte flips, xl.meta untouched), and SIGKILL/restart the server
with the same volumes and port.

Add reliability tests on a 4-disk (EC 2+2) topology, verified via
sha256 manifests recorded at write time:

- degraded read/write with one disk offline (incl. multipart object)
- bitrot read-through with two corrupted shards per object
- fresh-disk replacement healed via admin deep heal after a SIGKILL
  restart

Also reuse the shared signed_admin_post helper from chaos.rs in the
heal regression suite instead of a duplicated local copy.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

crates/e2e_test/src/chaos.rs

@@ -0,0 +1,259 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! In-process fault-injection primitives for reliability E2E tests.
+//!
+//! `DiskFaultHarness` runs a single-node, multi-disk RustFS server whose
+//! per-disk data directories can be manipulated while the process is alive:
+//!
+//! - [`DiskFaultHarness::take_disk_offline`] renames a disk directory to
+//!   `<dir>.offline`. Every subsequent per-request open on that volume fails
+//!   with "not found", which RustFS treats as a faulted drive. This simulates
+//!   a disk that disappears at the mount-point level (unplugged/unmounted),
+//!   not an EIO-returning half-dead disk.
+//! - [`DiskFaultHarness::bring_disk_online`] restores the renamed directory.
+//!   If the server recreated a skeleton directory at the original path in the
+//!   meantime, the skeleton is discarded first.
+//! - [`DiskFaultHarness::replace_disk_with_empty`] retires the directory and
+//!   creates a fresh empty one, simulating a fresh-disk replacement that the
+//!   server should detect as unformatted and heal.
+//! - [`DiskFaultHarness::corrupt_object_shard`] flips bytes in the middle of
+//!   one erasure shard (`part.*` file) of an object, simulating silent bitrot.
+//!   Only shard data is touched, never `xl.meta`.
+//! - [`DiskFaultHarness::kill_server`] / [`DiskFaultHarness::restart_server`]
+//!   SIGKILL the process and restart it with the same volumes and port so
+//!   heal-after-restart can be exercised.
+
+use crate::common::{RustFSTestEnvironment, local_http_client};
+use http::header::{CONTENT_TYPE, HOST};
+use rustfs_signer::constants::UNSIGNED_PAYLOAD;
+use rustfs_signer::sign_v4;
+use s3s::Body;
+use std::error::Error;
+use std::path::{Path, PathBuf};
+use tracing::info;
+use uuid::Uuid;
+use walkdir::WalkDir;
+
+type ChaosResult<T> = Result<T, Box<dyn Error + Send + Sync>>;
+
+/// Single-node RustFS server with `disk_count` local volume directories that
+/// can be faulted individually while the server is running.
+pub struct DiskFaultHarness {
+    pub env: RustFSTestEnvironment,
+    root: PathBuf,
+    disks: Vec<PathBuf>,
+    extra_env: Vec<(String, String)>,
+}
+
+impl DiskFaultHarness {
+    /// Create the volume directories (`disk0..diskN-1`) under a fresh temp root.
+    /// The server is not started yet; call [`Self::start_server`].
+    pub async fn new(disk_count: usize) -> ChaosResult<Self> {
+        if disk_count < 2 {
+            return Err("disk fault harness needs at least 2 disks".into());
+        }
+
+        let env = RustFSTestEnvironment::new().await?;
+        let root = PathBuf::from(env.temp_dir.clone());
+        let disks: Vec<PathBuf> = (0..disk_count).map(|i| root.join(format!("disk{i}"))).collect();
+        for disk in &disks {
+            std::fs::create_dir_all(disk)?;
+        }
+
+        Ok(Self {
+            env,
+            root,
+            disks,
+            // All disk directories live on one filesystem, so the startup
+            // same-disk check must be bypassed like other multi-disk e2e tests.
+            extra_env: vec![("RUSTFS_UNSAFE_BYPASS_DISK_CHECK".to_string(), "true".to_string())],
+        })
+    }
+
+    /// Add an extra environment variable for the server process (applies to
+    /// subsequent [`Self::start_server`] / [`Self::restart_server`] calls).
+    pub fn set_env<K, V>(&mut self, key: K, value: V)
+    where
+        K: Into<String>,
+        V: Into<String>,
+    {
+        self.extra_env.push((key.into(), value.into()));
+    }
+
+    pub fn disk_path(&self, disk_index: usize) -> &Path {
+        &self.disks[disk_index]
+    }
+
+    fn offline_path(&self, disk_index: usize) -> PathBuf {
+        let disk = &self.disks[disk_index];
+        let mut name = disk.file_name().expect("disk path has a file name").to_os_string();
+        name.push(".offline");
+        disk.with_file_name(name)
+    }
+
+    /// Start (or restart after [`Self::kill_server`]) the RustFS server with
+    /// all volume directories and the same address.
+    pub async fn start_server(&mut self) -> ChaosResult<()> {
+        // The common helper always appends `env.temp_dir` as the final storage
+        // path: point it at the last disk for startup and pass the remaining
+        // disks explicitly, then restore it to the root so Drop cleans up
+        // every disk directory.
+        let (last_disk, head_disks) = self.disks.split_last().expect("at least two disks");
+        self.env.temp_dir = last_disk.to_string_lossy().to_string();
+
+        let disk_args: Vec<String> = head_disks.iter().map(|d| d.to_string_lossy().to_string()).collect();
+        let arg_refs: Vec<&str> = disk_args.iter().map(String::as_str).collect();
+        let env_refs: Vec<(&str, &str)> = self.extra_env.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
+
+        let result = self.env.start_rustfs_server_with_env(arg_refs, &env_refs).await;
+        self.env.temp_dir = self.root.to_string_lossy().to_string();
+        result
+    }
+
+    /// SIGKILL the server process (simulates a crash / power loss for the
+    /// process; no graceful shutdown runs).
+    pub fn kill_server(&mut self) {
+        self.env.stop_server();
+    }
+
+    /// Restart the server with the same volumes and port after [`Self::kill_server`].
+    pub async fn restart_server(&mut self) -> ChaosResult<()> {
+        self.start_server().await
+    }
+
+    /// Make one volume directory unavailable to the running server by
+    /// renaming it to `<dir>.offline`. Path-based opens on the old location
+    /// fail immediately, so the drive turns faulted from the server's view.
+    pub fn take_disk_offline(&self, disk_index: usize) -> ChaosResult<()> {
+        let disk = &self.disks[disk_index];
+        let offline = self.offline_path(disk_index);
+        if offline.exists() {
+            return Err(format!("disk {disk_index} is already offline").into());
+        }
+        std::fs::rename(disk, &offline)?;
+        info!("Took disk {} offline: {:?} -> {:?}", disk_index, disk, offline);
+        Ok(())
+    }
+
+    /// Restore a disk previously taken offline with [`Self::take_disk_offline`].
+    pub fn bring_disk_online(&self, disk_index: usize) -> ChaosResult<()> {
+        let disk = &self.disks[disk_index];
+        let offline = self.offline_path(disk_index);
+        if !offline.exists() {
+            return Err(format!("disk {disk_index} is not offline").into());
+        }
+        // The running server may have recreated a skeleton directory at the
+        // original path (e.g. via degraded writes); discard it before
+        // restoring the original data.
+        if disk.exists() {
+            std::fs::remove_dir_all(disk)?;
+        }
+        std::fs::rename(&offline, disk)?;
+        info!("Brought disk {} back online: {:?}", disk_index, disk);
+        Ok(())
+    }
+
+    /// Retire a disk directory and create a fresh empty one in its place,
+    /// simulating a fresh-disk replacement. Intended to be used while the
+    /// server is stopped; the server should treat the empty directory as an
+    /// unformatted drive and heal it.
+    pub fn replace_disk_with_empty(&self, disk_index: usize) -> ChaosResult<()> {
+        let disk = &self.disks[disk_index];
+        let retired = self.root.join(format!(
+            "{}.retired-{}",
+            disk.file_name().expect("disk path has a file name").to_string_lossy(),
+            Uuid::new_v4()
+        ));
+        std::fs::rename(disk, &retired)?;
+        std::fs::create_dir_all(disk)?;
+        info!("Replaced disk {} with an empty directory (old data at {:?})", disk_index, retired);
+        Ok(())
+    }
+
+    /// Flip bytes in the middle of one erasure shard of `bucket/key` stored on
+    /// the given disk (bitrot simulation). Corrupts a `part.*` data file, never
+    /// `xl.meta`. Fails if the object has no shard file on that disk (e.g. the
+    /// object is small enough to be inlined into `xl.meta`).
+    pub fn corrupt_object_shard(&self, disk_index: usize, bucket: &str, key: &str) -> ChaosResult<PathBuf> {
+        let object_dir = self.disks[disk_index].join(bucket).join(key);
+        let mut part_files: Vec<PathBuf> = WalkDir::new(&object_dir)
+            .into_iter()
+            .filter_map(Result::ok)
+            .filter(|entry| entry.file_type().is_file())
+            .filter(|entry| entry.file_name().to_string_lossy().starts_with("part."))
+            .map(|entry| entry.into_path())
+            .collect();
+        part_files.sort();
+
+        let Some(part_file) = part_files.first() else {
+            return Err(format!("no part.* shard file found under {object_dir:?}; object data may be inlined in xl.meta").into());
+        };
+
+        let mut data = std::fs::read(part_file)?;
+        if data.len() < 32 {
+            return Err(format!("shard file {part_file:?} is too small to corrupt safely ({} bytes)", data.len()).into());
+        }
+        let mid = data.len() / 2;
+        for byte in &mut data[mid..mid + 8] {
+            *byte ^= 0xFF;
+        }
+        std::fs::write(part_file, data)?;
+        info!("Corrupted 8 bytes in the middle of shard {:?}", part_file);
+        Ok(part_file.clone())
+    }
+
+    /// Whether the object's `xl.meta` exists on the given disk. Useful for
+    /// polling heal progress on a replaced disk.
+    pub fn object_metadata_exists_on_disk(&self, disk_index: usize, bucket: &str, key: &str) -> bool {
+        self.disks[disk_index].join(bucket).join(key).join("xl.meta").is_file()
+    }
+}
+
+/// `POST` a signed (SigV4, service `s3`) admin request without relying on the
+/// external `awscurl` binary. Mirrors the admin heal calls used by the heal
+/// regression suite.
+pub async fn signed_admin_post(url: &str, body: Option<&str>, access_key: &str, secret_key: &str) -> ChaosResult<String> {
+    let uri = url.parse::<http::Uri>()?;
+    let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
+    let mut request = http::Request::builder()
+        .method(http::Method::POST)
+        .uri(uri)
+        .header(HOST, authority)
+        .header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
+
+    if body.is_some() {
+        request = request.header(CONTENT_TYPE, "application/json");
+    }
+
+    let content_len = body.map(str::len).unwrap_or_default() as i64;
+    let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1");
+
+    let mut request_builder = local_http_client().post(url);
+    for (name, value) in signed.headers() {
+        request_builder = request_builder.header(name, value);
+    }
+    if let Some(body) = body {
+        request_builder = request_builder.body(body.to_string());
+    }
+
+    let response = request_builder.send().await?;
+    let status = response.status();
+    let body = response.text().await?;
+    if !status.is_success() {
+        return Err(format!("admin POST failed: {status} {body}").into());
+    }
+
+    Ok(body)
+}

crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs

@@ -16,12 +16,9 @@
 
 #[cfg(test)]
 mod tests {
-    use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, local_http_client};
+    use crate::chaos::signed_admin_post;
+    use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging};
     use aws_sdk_s3::primitives::ByteStream;
-    use http::header::{CONTENT_TYPE, HOST};
-    use rustfs_signer::constants::UNSIGNED_PAYLOAD;
-    use rustfs_signer::sign_v4;
-    use s3s::Body;
     use serial_test::serial;
     use std::collections::HashSet;
     use std::error::Error;
@@ -65,45 +62,6 @@ mod tests {
         assert_eq!(body.as_ref(), expected, "object body changed for {key}");
     }
 
-    async fn signed_admin_post(
-        url: &str,
-        body: Option<&str>,
-        access_key: &str,
-        secret_key: &str,
-    ) -> Result<String, Box<dyn Error + Send + Sync>> {
-        let uri = url.parse::<http::Uri>()?;
-        let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
-        let mut request = http::Request::builder()
-            .method(http::Method::POST)
-            .uri(uri)
-            .header(HOST, authority)
-            .header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
-
-        if body.is_some() {
-            request = request.header(CONTENT_TYPE, "application/json");
-        }
-
-        let content_len = body.map(str::len).unwrap_or_default() as i64;
-        let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1");
-
-        let mut request_builder = local_http_client().post(url);
-        for (name, value) in signed.headers() {
-            request_builder = request_builder.header(name, value);
-        }
-        if let Some(body) = body {
-            request_builder = request_builder.body(body.to_string());
-        }
-
-        let response = request_builder.send().await?;
-        let status = response.status();
-        let body = response.text().await?;
-        if !status.is_success() {
-            return Err(format!("admin POST failed: {status} {body}").into());
-        }
-
-        Ok(body)
-    }
-
     #[tokio::test]
     #[serial]
     async fn test_auto_heal_rebuilds_runtime_wiped_disk_without_restart() {

crates/e2e_test/src/lib.rs

@@ -19,6 +19,14 @@ mod storage_api;
 #[cfg(test)]
 pub mod common;
 
+// In-process fault-injection primitives (disk offline/replacement, shard corruption)
+#[cfg(test)]
+pub mod chaos;
+
+// Reliability tests built on the fault-injection harness
+#[cfg(test)]
+mod reliability_disk_fault_test;
+
 #[cfg(test)]
 mod version_id_regression_test;
 

crates/e2e_test/src/reliability_disk_fault_test.rs

@@ -0,0 +1,307 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Reliability tests driven by in-process fault injection (see `chaos.rs`):
+//! degraded reads/writes with an offline disk, bitrot read-through, and
+//! fresh-disk replacement heal after a SIGKILL restart.
+//!
+//! All tests use a single-node 4-disk topology (default erasure coding for
+//! 4 drives is 2 data + 2 parity) and verify object content via sha256
+//! manifests recorded at write time.
+
+#[cfg(test)]
+mod tests {
+    use crate::chaos::{DiskFaultHarness, signed_admin_post};
+    use crate::common::init_logging;
+    use aws_sdk_s3::Client;
+    use aws_sdk_s3::primitives::ByteStream;
+    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+    use serial_test::serial;
+    use sha2::{Digest, Sha256};
+    use std::collections::HashSet;
+    use tokio::time::{Duration, sleep, timeout};
+    use tracing::info;
+
+    const GET_TIMEOUT: Duration = Duration::from_secs(60);
+    const PUT_TIMEOUT: Duration = Duration::from_secs(60);
+
+    fn sha256_hex(data: &[u8]) -> String {
+        let digest = Sha256::digest(data);
+        digest.iter().map(|byte| format!("{byte:02x}")).collect()
+    }
+
+    /// Deterministic pseudo-random payload so tests stay reproducible.
+    fn payload(len: usize, seed: u8) -> Vec<u8> {
+        (0..len)
+            .map(|i| (i as u64).wrapping_mul(2654435761).wrapping_add(seed as u64) as u8)
+            .collect()
+    }
+
+    async fn put_and_record(
+        client: &Client,
+        bucket: &str,
+        key: &str,
+        body: Vec<u8>,
+        manifest: &mut Vec<(String, String)>,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let digest = sha256_hex(&body);
+        timeout(
+            PUT_TIMEOUT,
+            client
+                .put_object()
+                .bucket(bucket)
+                .key(key)
+                .body(ByteStream::from(body))
+                .send(),
+        )
+        .await
+        .map_err(|_| format!("PUT {key} timed out"))??;
+        manifest.push((key.to_string(), digest));
+        Ok(())
+    }
+
+    async fn multipart_put_and_record(
+        client: &Client,
+        bucket: &str,
+        key: &str,
+        parts: Vec<Vec<u8>>,
+        manifest: &mut Vec<(String, String)>,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let full_body: Vec<u8> = parts.iter().flatten().copied().collect();
+        let digest = sha256_hex(&full_body);
+
+        let create = client.create_multipart_upload().bucket(bucket).key(key).send().await?;
+        let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
+
+        let mut completed_parts = Vec::with_capacity(parts.len());
+        for (index, part_body) in parts.into_iter().enumerate() {
+            let part_number = (index + 1) as i32;
+            let uploaded = timeout(
+                PUT_TIMEOUT,
+                client
+                    .upload_part()
+                    .bucket(bucket)
+                    .key(key)
+                    .upload_id(&upload_id)
+                    .part_number(part_number)
+                    .body(ByteStream::from(part_body))
+                    .send(),
+            )
+            .await
+            .map_err(|_| format!("upload_part {part_number} for {key} timed out"))??;
+            completed_parts.push(
+                CompletedPart::builder()
+                    .part_number(part_number)
+                    .e_tag(uploaded.e_tag().ok_or("missing part etag")?)
+                    .build(),
+            );
+        }
+
+        client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(&upload_id)
+            .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
+            .send()
+            .await?;
+
+        manifest.push((key.to_string(), digest));
+        Ok(())
+    }
+
+    async fn verify_manifest(
+        client: &Client,
+        bucket: &str,
+        manifest: &[(String, String)],
+        phase: &str,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        for (key, expected_sha256) in manifest {
+            let response = timeout(GET_TIMEOUT, client.get_object().bucket(bucket).key(key).send())
+                .await
+                .map_err(|_| format!("GET {key} timed out during {phase}"))?
+                .map_err(|err| format!("GET {key} failed during {phase}: {err}"))?;
+            let body = response
+                .body
+                .collect()
+                .await
+                .map_err(|err| format!("GET {key} body collect failed during {phase}: {err}"))?
+                .into_bytes();
+            let actual_sha256 = sha256_hex(&body);
+            if actual_sha256 != *expected_sha256 {
+                return Err(format!(
+                    "sha256 mismatch for {key} during {phase}: expected {expected_sha256}, got {actual_sha256} ({} bytes)",
+                    body.len()
+                )
+                .into());
+            }
+        }
+        info!("Verified {} objects during {}", manifest.len(), phase);
+        Ok(())
+    }
+
+    /// One disk goes offline at runtime: all previously written objects (from
+    /// inline-small to multipart-sized) must stay readable with intact
+    /// content, degraded writes must succeed, and everything must still
+    /// verify after the disk returns.
+    #[tokio::test]
+    #[serial]
+    async fn test_degraded_read_write_with_one_disk_offline() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        init_logging();
+        info!("Reliability: degraded read/write with one of four disks offline");
+
+        let mut harness = DiskFaultHarness::new(4).await?;
+        harness.start_server().await?;
+        let client = harness.env.create_s3_client();
+
+        let bucket = "reliability-degraded-rw";
+        client.create_bucket().bucket(bucket).send().await?;
+
+        let mut manifest: Vec<(String, String)> = Vec::new();
+        put_and_record(&client, bucket, "degraded/small.bin", payload(4 * 1024, 1), &mut manifest).await?;
+        put_and_record(&client, bucket, "degraded/medium.bin", payload(1024 * 1024, 2), &mut manifest).await?;
+        put_and_record(&client, bucket, "degraded/big.bin", payload(3 * 1024 * 1024, 3), &mut manifest).await?;
+        multipart_put_and_record(
+            &client,
+            bucket,
+            "degraded/multipart.bin",
+            vec![payload(5 * 1024 * 1024, 4), payload(1024 * 1024, 5)],
+            &mut manifest,
+        )
+        .await?;
+
+        verify_manifest(&client, bucket, &manifest, "baseline with all disks online").await?;
+
+        harness.take_disk_offline(0)?;
+
+        verify_manifest(&client, bucket, &manifest, "degraded read with disk0 offline").await?;
+
+        // Degraded writes: 3 of 4 disks still satisfy the write quorum for
+        // an EC 2+2 set.
+        put_and_record(
+            &client,
+            bucket,
+            "degraded/written-while-offline.bin",
+            payload(1024 * 1024, 9),
+            &mut manifest,
+        )
+        .await?;
+        verify_manifest(&client, bucket, &manifest, "read-back of degraded write").await?;
+
+        harness.bring_disk_online(0)?;
+
+        verify_manifest(&client, bucket, &manifest, "after disk0 came back online").await?;
+        Ok(())
+    }
+
+    /// Silent bitrot in a single erasure shard must never surface corrupted
+    /// bytes to a reader: per-shard bitrot checksums reject the bad shard and
+    /// the object is reconstructed from the remaining shards.
+    #[tokio::test]
+    #[serial]
+    async fn test_bitrot_corrupted_shard_read_returns_correct_data() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        init_logging();
+        info!("Reliability: GET must read through a bitrot-corrupted shard");
+
+        let mut harness = DiskFaultHarness::new(4).await?;
+        harness.start_server().await?;
+        let client = harness.env.create_s3_client();
+
+        let bucket = "reliability-bitrot";
+        client.create_bucket().bucket(bucket).send().await?;
+
+        let key_a = "bitrot/object-a.bin";
+        let key_b = "bitrot/object-b.bin";
+        let mut manifest: Vec<(String, String)> = Vec::new();
+        // 2 MiB objects are far above the 128 KiB inline threshold, so every
+        // disk holds a real part.1 shard file to corrupt.
+        put_and_record(&client, bucket, key_a, payload(2 * 1024 * 1024, 21), &mut manifest).await?;
+        put_and_record(&client, bucket, key_b, payload(2 * 1024 * 1024, 22), &mut manifest).await?;
+
+        verify_manifest(&client, bucket, &manifest, "baseline before corruption").await?;
+
+        // Corrupt two shards per object (the maximum an EC 2+2 set can lose)
+        // on different disk pairs. Which disks hold data vs parity depends on
+        // the per-object distribution, so corrupting a pair makes it very
+        // likely that at least one data shard is hit; either way the read
+        // must return intact content reconstructed from the clean shards.
+        harness.corrupt_object_shard(0, bucket, key_a)?;
+        harness.corrupt_object_shard(1, bucket, key_a)?;
+        harness.corrupt_object_shard(2, bucket, key_b)?;
+        harness.corrupt_object_shard(3, bucket, key_b)?;
+
+        verify_manifest(&client, bucket, &manifest, "first read after shard corruption").await?;
+        // A second pass ensures repeated reads stay correct as well.
+        verify_manifest(&client, bucket, &manifest, "second read after shard corruption").await?;
+        Ok(())
+    }
+
+    /// Fresh-disk replacement: SIGKILL the server, swap one disk for an empty
+    /// directory, restart with the same volumes/port, trigger an admin deep
+    /// heal, and require the replaced disk to be rebuilt and all content to
+    /// verify against the sha256 manifest.
+    #[tokio::test]
+    #[serial]
+    async fn test_fresh_disk_replacement_heals_after_sigkill_restart() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        init_logging();
+        info!("Reliability: fresh-disk replacement heals after SIGKILL restart");
+
+        let mut harness = DiskFaultHarness::new(4).await?;
+        harness.start_server().await?;
+        let client = harness.env.create_s3_client();
+
+        let bucket = "reliability-fresh-disk";
+        client.create_bucket().bucket(bucket).send().await?;
+
+        let mut manifest: Vec<(String, String)> = Vec::new();
+        put_and_record(&client, bucket, "heal/tiny.bin", payload(4 * 1024, 31), &mut manifest).await?;
+        put_and_record(&client, bucket, "heal/small.bin", payload(256 * 1024, 32), &mut manifest).await?;
+        put_and_record(&client, bucket, "heal/medium.bin", payload(1024 * 1024, 33), &mut manifest).await?;
+        put_and_record(&client, bucket, "heal/nested/large.bin", payload(2 * 1024 * 1024, 34), &mut manifest).await?;
+
+        verify_manifest(&client, bucket, &manifest, "baseline before disk replacement").await?;
+        for (key, _) in &manifest {
+            assert!(
+                harness.object_metadata_exists_on_disk(0, bucket, key),
+                "disk0 should hold xl.meta for {key} before replacement"
+            );
+        }
+
+        harness.kill_server();
+        harness.replace_disk_with_empty(0)?;
+        harness.restart_server().await?;
+
+        let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
+        let heal_url = format!("{}/rustfs/admin/v3/heal/{}?forceStart=true", harness.env.url, bucket);
+        signed_admin_post(&heal_url, Some(heal_body), &harness.env.access_key, &harness.env.secret_key).await?;
+
+        let client = harness.env.create_s3_client();
+        let mut remaining: HashSet<String> = manifest.iter().map(|(key, _)| key.clone()).collect();
+        let heal_timeout_secs = std::env::var("RUSTFS_RELIABILITY_HEAL_TIMEOUT_SECS")
+            .ok()
+            .and_then(|value| value.parse::<u64>().ok())
+            .unwrap_or(120);
+
+        for _ in 0..heal_timeout_secs {
+            remaining.retain(|key| !harness.object_metadata_exists_on_disk(0, bucket, key));
+            if remaining.is_empty() {
+                verify_manifest(&client, bucket, &manifest, "after fresh-disk heal completed").await?;
+                return Ok(());
+            }
+            sleep(Duration::from_secs(1)).await;
+        }
+
+        Err(format!("fresh-disk heal did not rebuild {remaining:?} on the replaced disk within {heal_timeout_secs}s").into())
+    }
+}