refactor(mithril-end-to-end): decouple polling interval from timeout in waits

Replace fixed attempt counts and epoch-fraction delays with a deadline-based
poll_until! macro and bounded backoff, keeping detection responsive on long epochs.

Author: jpraynaud

mithril-test-lab/mithril-end-to-end/src/toolkit/check/cardano_database.rs

@@ -8,7 +8,7 @@ use mithril_common::{
 };
 
 use crate::{
-    Aggregator, CardanoDbV2Command, Client, ClientCommand, attempt,
+    Aggregator, CardanoDbV2Command, Client, ClientCommand, poll_until,
     toolkit::{CheckCertificateToolkit, ScenarioToolkitContext},
     utils::AttemptResult,
 };
@@ -99,7 +99,7 @@ impl CheckCardanoDatabaseToolkit {
             }
         }
 
-        match attempt!(10, self.context.tenth_epoch_delay(), {
+        match poll_until!(self.context.appearance_timeout(), self.context.poll_backoff(), {
             fetch_cardano_database_digests_map(url.clone()).await
         }) {
             AttemptResult::Ok(cardano_database_digests_map) => {

mithril-test-lab/mithril-end-to-end/src/toolkit/check/certificate.rs

@@ -7,7 +7,7 @@ use mithril_common::{
     messages::{CertificateListItemMessage, CertificateMessage},
 };
 
-use crate::{Aggregator, attempt, toolkit::ScenarioToolkitContext, utils::AttemptResult};
+use crate::{Aggregator, poll_until, toolkit::ScenarioToolkitContext, utils::AttemptResult};
 
 use super::utils;
 
@@ -58,9 +58,11 @@ impl CheckCertificateToolkit {
             }
         }
 
-        match attempt!(10, self.context.tenth_epoch_delay(), {
-            fetch_certificate_message(url.clone()).await
-        }) {
+        match poll_until!(
+            self.context.appearance_timeout(),
+            self.context.poll_backoff(),
+            { fetch_certificate_message(url.clone()).await }
+        ) {
             AttemptResult::Ok(certificate) => {
                 info!("Aggregator produced a certificate"; "certificate" => ?certificate);
                 if certificate.metadata.signers.len() == total_signers_expected {

mithril-test-lab/mithril-end-to-end/src/toolkit/check/utils.rs

@@ -6,7 +6,7 @@ use slog_scope::info;
 use mithril_common::{StdResult, entities::Epoch};
 
 use crate::utils::TimeoutReason;
-use crate::{Aggregator, attempt, toolkit::ScenarioToolkitContext, utils::AttemptResult};
+use crate::{Aggregator, poll_until, toolkit::ScenarioToolkitContext, utils::AttemptResult};
 
 pub async fn get_json_response<T: DeserializeOwned>(url: String) -> StdResult<reqwest::Result<T>> {
     match reqwest::get(url.clone()).await {
@@ -74,7 +74,7 @@ where
         }
     }
 
-    match attempt!(20, context.tenth_epoch_delay(), {
+    match poll_until!(context.appearance_timeout(), context.poll_backoff(), {
         fetch_last_artifact(artifact_name, url.clone()).await
     }, until &condition) {
         AttemptResult::Ok(last_artifact) => {

mithril-test-lab/mithril-end-to-end/src/toolkit/context.rs

@@ -1,15 +1,25 @@
 use std::time::Duration;
 
+use crate::utils::Backoff;
+
+/// Context shared across scenario toolkits, carrying the polling policy derived from the Cardano
+/// epoch duration.
 #[derive(Debug, Clone)]
 pub struct ScenarioToolkitContext {
+    /// Policy used to derive polling timeouts from the epoch duration.
     attempt_policy: AttemptPolicy,
 }
 
 impl ScenarioToolkitContext {
+    /// Number of epochs to wait for an aggregator artifact (or readiness) to appear.
+    const ARTIFACT_APPEARANCE_TIMEOUT_EPOCHS: u32 = 5;
+
+    /// Builds a context from the given attempt policy.
     pub fn new(attempt_policy: AttemptPolicy) -> Self {
         Self { attempt_policy }
     }
 
+    /// Builds a context from the Cardano slot length and number of slots per epoch.
     pub fn new_from_cardano_epoch(slot_length_in_s: f64, number_of_slot_per_epoch: f64) -> Self {
         Self {
             attempt_policy: AttemptPolicy::from_cardano_epoch(
@@ -19,43 +29,48 @@ impl ScenarioToolkitContext {
         }
     }
 
-    pub fn attempt_policy(&self) -> AttemptPolicy {
-        self.attempt_policy
-    }
-
-    pub fn tenth_epoch_delay(&self) -> Duration {
-        self.attempt_policy.delay(0.10)
+    /// Backoff spacing out polling attempts while waiting for a condition.
+    pub fn poll_backoff(&self) -> Backoff {
+        Backoff::default()
     }
 
-    pub fn half_epoch_delay(&self) -> Duration {
-        self.attempt_policy.delay(0.5)
+    /// Timeout to wait for an aggregator artifact (or readiness) to appear.
+    pub fn appearance_timeout(&self) -> Duration {
+        self.attempt_policy
+            .timeout_for_epochs(Self::ARTIFACT_APPEARANCE_TIMEOUT_EPOCHS)
     }
 
-    pub fn full_epoch_delay(&self) -> Duration {
-        self.attempt_policy.epoch_duration
+    /// Timeout covering the given number of Cardano epochs.
+    pub fn timeout_for_epochs(&self, epochs: u32) -> Duration {
+        self.attempt_policy.timeout_for_epochs(epochs)
     }
 }
 
+/// Policy deriving polling timeouts from the Cardano epoch duration.
 #[derive(Debug, Clone, Copy)]
 pub struct AttemptPolicy {
+    /// Wall-clock duration of a single Cardano epoch.
     epoch_duration: Duration,
 }
 
 impl AttemptPolicy {
+    /// Builds a policy from the given epoch duration.
     pub const fn new(base_duration: Duration) -> Self {
         Self {
             epoch_duration: base_duration,
         }
     }
 
+    /// Builds a policy from the Cardano slot length and number of slots per epoch.
     pub fn from_cardano_epoch(slot_length_in_s: f64, number_of_slot_per_epoch: f64) -> Self {
         Self::new(Duration::from_secs_f64(
             slot_length_in_s * number_of_slot_per_epoch,
         ))
     }
 
-    pub fn delay(self, multiplier: f32) -> Duration {
-        self.epoch_duration.mul_f32(multiplier)
+    /// Returns a timeout covering the given number of epochs.
+    pub fn timeout_for_epochs(self, epochs: u32) -> Duration {
+        self.epoch_duration * epochs
     }
 }
 
@@ -72,9 +87,21 @@ mod tests {
     }
 
     #[test]
-    fn delay_calculation() {
+    fn timeout_for_epochs_scales_with_epoch_duration() {
         let policy = AttemptPolicy::new(Duration::from_secs(10));
-        assert_eq!(policy.delay(0.5), Duration::from_secs(5));
-        assert_eq!(policy.delay(2.0), Duration::from_secs(20));
+
+        assert_eq!(policy.timeout_for_epochs(0), Duration::from_secs(0));
+        assert_eq!(policy.timeout_for_epochs(1), Duration::from_secs(10));
+        assert_eq!(policy.timeout_for_epochs(5), Duration::from_secs(50));
+    }
+
+    #[test]
+    fn appearance_timeout_covers_configured_epochs() {
+        let context = ScenarioToolkitContext::new(AttemptPolicy::new(Duration::from_secs(10)));
+
+        assert_eq!(
+            context.appearance_timeout(),
+            Duration::from_secs(10) * ScenarioToolkitContext::ARTIFACT_APPEARANCE_TIMEOUT_EPOCHS
+        );
     }
 }

mithril-test-lab/mithril-end-to-end/src/toolkit/wait.rs

@@ -5,14 +5,17 @@ use slog_scope::{info, warn};
 use mithril_cardano_node_internal_database::entities::ImmutableFile;
 use mithril_common::{StdResult, entities::Epoch, messages::EpochSettingsMessage};
 
-use crate::{Aggregator, attempt, toolkit::ScenarioToolkitContext, utils::AttemptResult};
+use crate::{Aggregator, poll_until, toolkit::ScenarioToolkitContext, utils::AttemptResult};
 
 #[derive(Debug, Clone)]
 pub struct WaitToolkit {
     context: ScenarioToolkitContext,
 }
 
 impl WaitToolkit {
+    /// Extra epoch added on top of the epoch gap when waiting for a target epoch.
+    const TARGET_EPOCH_SLACK: u64 = 1;
+
     pub fn new(context: ScenarioToolkitContext) -> Self {
         Self { context }
     }
@@ -21,20 +24,24 @@ impl WaitToolkit {
         info!("Waiting that enough immutable have been written in the devnet"; "aggregator" => aggregator.name());
 
         let db_directory = aggregator.db_directory();
-        match attempt!(20, self.context.half_epoch_delay(), {
-            match ImmutableFile::list_completed_in_dir(db_directory)
-                .with_context(|| {
-                    format!(
-                        "Immutable file listing failed in dir `{}`",
-                        db_directory.display(),
-                    )
-                })?
-                .last()
+        match poll_until!(
+            self.context.appearance_timeout(),
+            self.context.poll_backoff(),
             {
-                Some(_) => Ok(Some(())),
-                None => Ok(None),
+                match ImmutableFile::list_completed_in_dir(db_directory)
+                    .with_context(|| {
+                        format!(
+                            "Immutable file listing failed in dir `{}`",
+                            db_directory.display(),
+                        )
+                    })?
+                    .last()
+                {
+                    Some(_) => Ok(Some(())),
+                    None => Ok(None),
+                }
             }
-        }) {
+        ) {
             AttemptResult::Ok(_) => Ok(()),
             AttemptResult::Err(error) => Err(error),
             AttemptResult::Timeout(..) => Err(anyhow!(
@@ -52,26 +59,30 @@ impl WaitToolkit {
         let url = format!("{aggregator_endpoint}/epoch-settings");
         info!("Waiting for the aggregator to expose epoch settings"; "aggregator" => aggregator.name());
 
-        match attempt!(20, self.context.half_epoch_delay(), {
-            match reqwest::get(url.clone()).await {
-                Ok(response) => match response.status() {
-                    StatusCode::OK => {
-                        let epoch_settings = response
-                            .json::<EpochSettingsMessage>()
-                            .await
-                            .with_context(|| "Invalid EpochSettings body")?;
-                        info!("Aggregator ready"; "epoch_settings"  => ?epoch_settings);
-                        Ok(Some(epoch_settings))
-                    }
-                    s if s.is_server_error() => {
-                        warn!( "Server error while waiting for the Aggregator, http code: {s}"; "aggregator" => aggregator.name());
-                        Ok(None)
-                    }
-                    _ => Ok(None),
-                },
-                Err(_) => Ok(None),
+        match poll_until!(
+            self.context.appearance_timeout(),
+            self.context.poll_backoff(),
+            {
+                match reqwest::get(url.clone()).await {
+                    Ok(response) => match response.status() {
+                        StatusCode::OK => {
+                            let epoch_settings = response
+                                .json::<EpochSettingsMessage>()
+                                .await
+                                .with_context(|| "Invalid EpochSettings body")?;
+                            info!("Aggregator ready"; "epoch_settings"  => ?epoch_settings);
+                            Ok(Some(epoch_settings))
+                        }
+                        s if s.is_server_error() => {
+                            warn!( "Server error while waiting for the Aggregator, http code: {s}"; "aggregator" => aggregator.name());
+                            Ok(None)
+                        }
+                        _ => Ok(None),
+                    },
+                    Err(_) => Ok(None),
+                }
             }
-        }) {
+        ) {
             AttemptResult::Ok(epoch_settings) => Ok(epoch_settings),
             AttemptResult::Err(error) => Err(error),
             AttemptResult::Timeout(..) => Err(anyhow!(
@@ -92,7 +103,20 @@ impl WaitToolkit {
             "target_epoch" => ?target_epoch
         );
 
-        match attempt!(90, self.context.half_epoch_delay(), {
+        let current_epoch = aggregator
+            .chain_observer()
+            .get_current_epoch()
+            .await
+            .ok()
+            .flatten()
+            .unwrap_or(Epoch(0));
+        let epochs_to_wait =
+            (*target_epoch).saturating_sub(*current_epoch) + Self::TARGET_EPOCH_SLACK;
+        let timeout = self
+            .context
+            .timeout_for_epochs(u32::try_from(epochs_to_wait).unwrap_or(u32::MAX));
+
+        match poll_until!(timeout, self.context.poll_backoff(), {
             match aggregator
                 .chain_observer()
                 .get_current_epoch()

mithril-test-lab/mithril-end-to-end/src/utils/mod.rs

@@ -11,7 +11,7 @@ pub use compatibility_checker::*;
 pub use formatting::*;
 pub use immutable_files_utils::*;
 pub use mithril_command::MithrilCommand;
-pub use spec_utils::{AttemptResult, TimeoutReason};
+pub use spec_utils::{AttemptResult, Backoff, TimeoutReason};
 pub use version_req::NodeVersion;
 
 pub fn is_running_in_github_actions() -> bool {