test(heal): wait for replacement disk and coordinator readiness

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.

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<Vec<ExpectedShard>> {
     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() {

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<dyn Error + Send + Sync>>(());
+                }
+                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();