Fix test_readiness_updates flaky test (#6053)

Author: guilload

quickwit/quickwit-common/src/lib.rs

@@ -188,6 +188,32 @@ pub fn no_color() -> bool {
     matches!(env::var("NO_COLOR"), Ok(value) if !value.is_empty())
 }
 
+#[macro_export]
+macro_rules! assert_eventually {
+    ($cond:expr, $timeout:expr, $interval:expr) => {
+        let start = std::time::Instant::now();
+        loop {
+            if $cond {
+                break;
+            }
+            if start.elapsed() > $timeout {
+                panic!(
+                    "assertion failed: condition `{}` never became true within {} ms",
+                    stringify!($cond),
+                    $timeout.as_millis()
+                );
+            }
+            tokio::time::sleep($interval).await;
+        }
+    };
+    ($cond:expr, $timeout:expr) => {
+        assert_eventually!($cond, $timeout, std::time::Duration::from_millis(50));
+    };
+    ($cond:expr) => {
+        assert_eventually!($cond, std::time::Duration::from_secs(1));
+    };
+}
+
 #[macro_export]
 macro_rules! ignore_error_kind {
     ($kind:path, $expr:expr) => {

quickwit/quickwit-serve/src/lib.rs

@@ -1353,8 +1353,8 @@ async fn check_cluster_configuration(
 #[cfg(test)]
 mod tests {
     use quickwit_cluster::{ChannelTransport, ClusterNode, create_cluster_for_test};
-    use quickwit_common::ServiceStream;
     use quickwit_common::uri::Uri;
+    use quickwit_common::{ServiceStream, assert_eventually};
     use quickwit_config::SearcherConfig;
     use quickwit_metastore::{IndexMetadata, metastore_for_test};
     use quickwit_proto::indexing::IndexingTask;
@@ -1472,16 +1472,14 @@ mod tests {
 
         metastore_readiness_tx.send(true).unwrap();
         ingester_status_tx.send(IngesterStatus::Ready).unwrap();
-        tokio::time::sleep(Duration::from_millis(25)).await;
-        assert!(cluster.is_self_node_ready().await);
+        assert_eventually!(cluster.is_self_node_ready().await);
 
         let request = tonic::Request::new(HealthCheckRequest::default());
         let response = health_client.check(request).await.unwrap().into_inner();
         assert_eq!(response.status(), ServingStatus::Serving.into());
 
         metastore_readiness_tx.send(false).unwrap();
-        tokio::time::sleep(Duration::from_millis(25)).await;
-        assert!(!cluster.is_self_node_ready().await);
+        assert_eventually!(!cluster.is_self_node_ready().await);
 
         let request = tonic::Request::new(HealthCheckRequest::default());
         let response = health_client.check(request).await.unwrap().into_inner();