diff --git a/crates/e2e_test/src/chaos.rs b/crates/e2e_test/src/chaos.rs index f0b2971e9..228a78342 100644 --- a/crates/e2e_test/src/chaos.rs +++ b/crates/e2e_test/src/chaos.rs @@ -374,6 +374,30 @@ pub(crate) fn census_object_version_on_disk( }) } +pub(crate) fn is_cluster_heal_coordination_unavailable(error: &(dyn std::error::Error + Send + Sync)) -> bool { + let message = error.to_string(); + message.contains("500 Internal Server Error") && message.contains("cluster heal coordination unavailable") +} + +/// Wait for the restarted cluster to admit the first root heal request. +pub(crate) async fn start_root_heal_when_control_ready( + heal_url: &str, + heal_body: &str, + access_key: &str, + secret_key: &str, +) -> ChaosResult<()> { + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(45); + loop { + match signed_admin_post(heal_url, Some(heal_body), access_key, secret_key).await { + Ok(_) => return Ok(()), + Err(error) if is_cluster_heal_coordination_unavailable(error.as_ref()) && tokio::time::Instant::now() < deadline => { + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + Err(error) => return Err(error), + } + } +} + /// `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. diff --git a/crates/e2e_test/src/distributed/heal_test.rs b/crates/e2e_test/src/distributed/heal_test.rs index d09c0a32e..9b1d8897b 100644 --- a/crates/e2e_test/src/distributed/heal_test.rs +++ b/crates/e2e_test/src/distributed/heal_test.rs @@ -16,7 +16,7 @@ use super::harness::{ DistCluster, DistLayout, TestResult, assert_inventory, get_object_bytes, payload_for, put_object, sha256_hex, unique_bucket, wait_until, }; -use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post}; +use crate::chaos::{VersionShardCensus, census_object_version_on_disk, start_root_heal_when_control_ready}; use crate::common::init_logging; use aws_sdk_s3::Client; use aws_sdk_s3::primitives::ByteStream; @@ -26,7 +26,6 @@ use std::collections::{BTreeMap, HashSet}; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; use std::time::Duration; -use tokio::time::{Instant, sleep}; const EC84_NODE_COUNT: usize = 3; const EC84_DRIVES_PER_NODE: usize = 4; @@ -34,8 +33,6 @@ const EC84_DATA_BLOCKS: usize = 8; const EC84_PARITY_BLOCKS: usize = 4; const EC84_TARGET_DRIVE_RESTART_CASE: &str = "ec84-target-drive-restart"; const EC84_TARGET_DRIVE_RESTART_ORACLE: &str = "ec84-target-drive-restart.json"; -const EC84_HEAL_CONTROL_READY_TIMEOUT: Duration = Duration::from_secs(45); -const EC84_HEAL_CONTROL_RETRY_DELAY: Duration = Duration::from_millis(250); #[derive(Clone)] struct ExpectedShard { @@ -214,29 +211,6 @@ fn assert_replaced_drive_empty(drive: &Path, bucket: &str, keys: &[String]) -> T Ok(()) } -fn is_cluster_heal_coordination_unavailable(error: &(dyn std::error::Error + Send + Sync)) -> bool { - let message = error.to_string(); - message.contains("500 Internal Server Error") && message.contains("cluster heal coordination unavailable") -} - -async fn start_ec84_root_heal_when_control_ready( - heal_url: &str, - heal_body: &str, - access_key: &str, - secret_key: &str, -) -> TestResult { - let deadline = Instant::now() + EC84_HEAL_CONTROL_READY_TIMEOUT; - loop { - match signed_admin_post(heal_url, Some(heal_body), access_key, secret_key).await { - Ok(_) => return Ok(()), - Err(error) if is_cluster_heal_coordination_unavailable(error.as_ref()) && Instant::now() < deadline => { - sleep(EC84_HEAL_CONTROL_RETRY_DELAY).await; - } - Err(error) => return Err(error), - } - } -} - async fn put_large_inventory(client: &Client, bucket: &str) -> TestResult> { let mut expected = Vec::new(); for index in 0..4 { @@ -330,7 +304,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res 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/{bucket}?forceStart=true", dist.cluster.nodes[0].url); - start_ec84_root_heal_when_control_ready(&heal_url, heal_body, &dist.cluster.access_key, &dist.cluster.secret_key).await?; + start_root_heal_when_control_ready(&heal_url, heal_body, &dist.cluster.access_key, &dist.cluster.secret_key).await?; wait_until( Duration::from_secs(120), @@ -394,7 +368,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res #[cfg(test)] mod tests { - use super::*; + use crate::chaos::is_cluster_heal_coordination_unavailable; #[test] fn cluster_heal_coordination_retry_is_exact() { 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 a4df5f4b2..5d663be9f 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs @@ -16,7 +16,9 @@ #[cfg(test)] mod tests { - use crate::chaos::{VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post}; + use crate::chaos::{ + VersionShardCensus, census_object_version_on_disk, sha256_hex, signed_admin_post, start_root_heal_when_control_ready, + }; use crate::common::{ ClusterTopology, FAST_DATA_USAGE_SCANNER_ENV, RustFSTestClusterEnvironment, RustFSTestEnvironment, admin_request, init_logging, rustfs_binary_path, @@ -960,19 +962,26 @@ mod tests { let online_key = "cluster/online-before-replacement.bin"; let online_body = b"object written while all cluster nodes are online".to_vec(); - clients[0] - .put_object() - .bucket(bucket) - .key(online_key) - .body(ByteStream::from(online_body.clone())) - .send() - .await?; - let replaced_disk = PathBuf::from(&cluster.nodes[1].data_dir); - assert!( - object_metadata_exists_on_disk(&replaced_disk, bucket, online_key), - "node 1 should contain metadata before disk replacement" - ); + // A quorum write need not include the disk this fixture will replace. + // Establish that disk's baseline before testing its reconstruction. + timeout(Duration::from_secs(30), async { + loop { + clients[0] + .put_object() + .bucket(bucket) + .key(online_key) + .body(ByteStream::from(online_body.clone())) + .send() + .await?; + if object_metadata_exists_on_disk(&replaced_disk, bucket, online_key) { + return Ok::<_, Box>(()); + } + sleep(Duration::from_millis(100)).await; + } + }) + .await + .map_err(|_| "node 1 did not store the baseline object before disk replacement")??; cluster.stop_node(1)?; std::fs::remove_dir_all(&replaced_disk)?; @@ -1017,7 +1026,7 @@ mod tests { 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", cluster.nodes[0].url); - signed_admin_post(&heal_url, Some(heal_body), &cluster.access_key, &cluster.secret_key).await?; + start_root_heal_when_control_ready(&heal_url, heal_body, &cluster.access_key, &cluster.secret_key).await?; let expected_objects = [(online_key, online_body.as_slice()), (outage_key, outage_body.as_slice())]; let mut remaining_rebuild_keys: HashSet<&str> = expected_objects.iter().map(|(key, _)| *key).collect();