diff --git a/crates/e2e_test/src/chaos.rs b/crates/e2e_test/src/chaos.rs new file mode 100644 index 000000000..c61ff5bc5 --- /dev/null +++ b/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 +//! `.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 = Result>; + +/// 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, + 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 { + 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 = (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(&mut self, key: K, value: V) + where + K: Into, + V: Into, + { + 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 = 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 `.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 { + let object_dir = self.disks[disk_index].join(bucket).join(key); + let mut part_files: Vec = 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 { + let uri = url.parse::()?; + 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) +} diff --git a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs index 576331ed9..1b735b3be 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/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> { - let uri = url.parse::()?; - 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() { diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 6afa0c712..bc070d8d0 100644 --- a/crates/e2e_test/src/lib.rs +++ b/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; diff --git a/crates/e2e_test/src/reliability_disk_fault_test.rs b/crates/e2e_test/src/reliability_disk_fault_test.rs new file mode 100644 index 000000000..9ff84ad90 --- /dev/null +++ b/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 { + (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, + manifest: &mut Vec<(String, String)>, + ) -> Result<(), Box> { + 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>, + manifest: &mut Vec<(String, String)>, + ) -> Result<(), Box> { + let full_body: Vec = 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> { + 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> { + 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> { + 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> { + 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 = 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::().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()) + } +}