test: update e2e to follower auto-sync

Author: Alenar

mithril-test-lab/mithril-end-to-end/src/assertions/exec.rs

@@ -7,17 +7,6 @@ use slog_scope::info;
 
 pub async fn bootstrap_genesis_certificate(aggregator: &Aggregator) -> StdResult<()> {
     info!("Bootstrap genesis certificate"; "aggregator" => &aggregator.name());
-
-    // A follower aggregator needs to wait few cycles of the state machine to be able to bootstrap
-    // This should be removed when the aggregator is able to synchronize its certificate chain from another aggregator
-    if !aggregator.is_first() {
-        const CYCLES_WAIT_FOLLOWER: u64 = 3;
-        tokio::time::sleep(std::time::Duration::from_millis(
-            CYCLES_WAIT_FOLLOWER * aggregator.mithril_run_interval() as u64,
-        ))
-        .await;
-    }
-
     info!("> stopping aggregator"; "aggregator" => &aggregator.name());
     aggregator.stop().await?;
     info!("> bootstrapping genesis using signers registered two epochs ago..."; "aggregator" => &aggregator.name());

mithril-test-lab/mithril-end-to-end/src/end_to_end_spec.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use slog_scope::info;
 use tokio::task::JoinSet;
 
 use mithril_common::{
@@ -54,6 +55,15 @@ impl Spec {
         // As we get closer to the tip of the chain when signing, we'll be able to relax this constraint.
         assertions::transfer_funds(spec.infrastructure.devnet()).await?;
 
+        info!("bootstrapping leader aggregator");
+        spec.bootstrap_leader_aggregator(&spec.infrastructure).await?;
+
+        info!("Starting followers");
+        for follower_aggregator in spec.infrastructure.follower_aggregators() {
+            follower_aggregator.serve().await?;
+        }
+
+        info!("Running scenarios");
         for index in 0..spec.infrastructure.aggregators().len() {
             let spec_clone = spec.clone();
             join_set.spawn(async move {
@@ -72,44 +82,54 @@ impl Spec {
         Ok(())
     }
 
-    pub async fn run_scenario(
+    pub async fn bootstrap_leader_aggregator(
         &self,
-        aggregator: &Aggregator,
         infrastructure: &MithrilInfrastructure,
     ) -> StdResult<()> {
-        assertions::wait_for_enough_immutable(aggregator).await?;
-        let chain_observer = aggregator.chain_observer();
+        let leader_aggregator = infrastructure.leader_aggregator();
+
+        assertions::wait_for_enough_immutable(leader_aggregator).await?;
+        let chain_observer = leader_aggregator.chain_observer();
         let start_epoch = chain_observer.get_current_epoch().await?.unwrap_or_default();
 
         // Wait 4 epochs after start epoch for the aggregator to be able to bootstrap a genesis certificate
         let mut target_epoch = start_epoch + 4;
         assertions::wait_for_aggregator_at_target_epoch(
-            aggregator,
+            leader_aggregator,
             target_epoch,
             "minimal epoch for the aggregator to be able to bootstrap genesis certificate"
                 .to_string(),
         )
         .await?;
-        assertions::bootstrap_genesis_certificate(aggregator).await?;
-        assertions::wait_for_epoch_settings(aggregator).await?;
+        assertions::bootstrap_genesis_certificate(leader_aggregator).await?;
+        assertions::wait_for_epoch_settings(leader_aggregator).await?;
 
         // Wait 2 epochs before changing stake distribution, so that we use at least one original stake distribution
         target_epoch += 2;
         assertions::wait_for_aggregator_at_target_epoch(
-            aggregator,
+            leader_aggregator,
             target_epoch,
             "epoch after which the stake distribution will change".to_string(),
         )
         .await?;
 
-        if aggregator.is_first() {
-            // Delegate some stakes to pools
-            let delegation_round = 1;
-            assertions::delegate_stakes_to_pools(infrastructure.devnet(), delegation_round).await?;
-        }
+        // Delegate some stakes to pools
+        let delegation_round = 1;
+        assertions::delegate_stakes_to_pools(infrastructure.devnet(), delegation_round).await?;
+
+        Ok(())
+    }
+
+    pub async fn run_scenario(
+        &self,
+        aggregator: &Aggregator,
+        infrastructure: &MithrilInfrastructure,
+    ) -> StdResult<()> {
+        let chain_observer = aggregator.chain_observer();
+        let start_epoch = chain_observer.get_current_epoch().await?.unwrap_or_default();
 
         // Wait 2 epochs before changing protocol parameters
-        target_epoch += 2;
+        let mut target_epoch = start_epoch + 2;
         assertions::wait_for_aggregator_at_target_epoch(
             aggregator,
             target_epoch,

mithril-test-lab/mithril-end-to-end/src/mithril/infrastructure.rs

@@ -103,25 +103,30 @@ impl MithrilInfrastructure {
         let relay_signer_registration_mode = &config.relay_signer_registration_mode;
         let relay_signature_registration_mode = &config.relay_signature_registration_mode;
 
-        let aggregators =
-            Self::start_aggregators(config, aggregator_cardano_nodes, chain_observer_type).await?;
-        let aggregator_endpoints = aggregators
+        let (leader_aggregator, follower_aggregators) =
+            Self::prepare_aggregators(config, aggregator_cardano_nodes, chain_observer_type)
+                .await?;
+
+        Self::register_startup_era(&leader_aggregator, config).await?;
+        leader_aggregator.serve().await?;
+
+        let follower_aggregator_endpoints = follower_aggregators
             .iter()
             .map(|aggregator| aggregator.endpoint())
             .collect::<Vec<_>>();
-        let leader_aggregator_endpoint = aggregator_endpoints[0].to_owned();
 
         let (relay_aggregators, relay_signers, relay_passives) = Self::start_relays(
             config,
-            &aggregator_endpoints,
+            leader_aggregator.endpoint(),
+            &follower_aggregator_endpoints,
             &signer_party_ids,
             relay_signer_registration_mode.to_owned(),
             relay_signature_registration_mode.to_owned(),
         )?;
 
         let signers = Self::start_signers(
             config,
-            leader_aggregator_endpoint,
+            leader_aggregator.endpoint(),
             signer_cardano_nodes,
             &relay_signers,
         )
@@ -132,11 +137,14 @@ impl MithrilInfrastructure {
             CardanoNetwork::DevNet(DEVNET_MAGIC_ID),
         ));
 
+        let mut all_aggregators = vec![leader_aggregator];
+        all_aggregators.extend(follower_aggregators);
+
         Ok(Self {
             bin_dir: config.bin_dir.to_path_buf(),
             artifacts_dir: config.artifacts_dir.to_path_buf(),
             devnet: config.devnet.clone(),
-            aggregators,
+            aggregators: all_aggregators,
             signers,
             relay_aggregators,
             relay_signers,
@@ -176,90 +184,110 @@ impl MithrilInfrastructure {
             + 1;
         if self.era_reader_adapter == "cardano-chain" {
             let devnet = self.devnet.clone();
-            assertions::register_era_marker(self.aggregator(0), &devnet, next_era, next_era_epoch)
-                .await?;
+            assertions::register_era_marker(
+                self.leader_aggregator(),
+                &devnet,
+                next_era,
+                next_era_epoch,
+            )
+            .await?;
         }
         let mut current_era = self.current_era.write().await;
         *current_era = next_era.to_owned();
 
         Ok(())
     }
 
-    async fn start_aggregators(
+    async fn prepare_aggregators(
         config: &MithrilInfrastructureConfig,
-        pool_nodes: &[FullNode],
+        full_nodes: &[FullNode],
         chain_observer_type: &str,
-    ) -> StdResult<Vec<Aggregator>> {
-        let mut aggregators = vec![];
-        let mut leader_aggregator_endpoint: Option<String> = None;
-        for (index, full_node) in pool_nodes.iter().enumerate() {
-            let aggregator_name = Aggregator::name_suffix(index);
-            let aggregator_artifacts_dir = config
-                .artifacts_dir
-                .join(format!("mithril-aggregator-{aggregator_name}"));
-            let aggregator_store_dir =
-                config.store_dir.join(format!("aggregator-{aggregator_name}"));
-            let aggregator = Aggregator::new(&AggregatorConfig {
-                index,
-                name: &aggregator_name,
-                server_port: config.server_port + index as u64,
+    ) -> StdResult<(Aggregator, Vec<Aggregator>)> {
+        let [leader_node, follower_nodes @ ..] = full_nodes else {
+            panic!("Can't prepare Aggregators: No full nodes found");
+        };
+        let leader_aggregator =
+            Self::prepare_aggregator(0, leader_node, config, chain_observer_type, None).await?;
+
+        let mut follower_aggregators = vec![];
+        for (index, full_node) in follower_nodes.iter().enumerate() {
+            let aggregator = Self::prepare_aggregator(
+                index + 1,
                 full_node,
-                cardano_cli_path: &config.devnet.cardano_cli_path(),
-                work_dir: &config.work_dir,
-                store_dir: &aggregator_store_dir,
-                artifacts_dir: &aggregator_artifacts_dir,
-                bin_dir: &config.bin_dir,
-                cardano_node_version: &config.cardano_node_version,
-                mithril_run_interval: config.mithril_run_interval,
-                mithril_era: &config.mithril_era,
-                mithril_era_reader_adapter: &config.mithril_era_reader_adapter,
-                mithril_era_marker_address: &config.devnet.mithril_era_marker_address()?,
-                signed_entity_types: &config.signed_entity_types,
+                config,
                 chain_observer_type,
-                leader_aggregator_endpoint: &leader_aggregator_endpoint.clone(),
-            })?;
-
-            aggregator
-                .set_protocol_parameters(&ProtocolParameters {
-                    k: 70,
-                    m: 105,
-                    phi_f: 0.95,
-                })
-                .await;
-
-            if leader_aggregator_endpoint.is_none()
-                && config.has_leader_follower_signer_registration()
-            {
-                leader_aggregator_endpoint = Some(aggregator.endpoint());
-            }
-
-            aggregators.push(aggregator);
+                Some(leader_aggregator.endpoint()),
+            )
+            .await?;
+            follower_aggregators.push(aggregator);
         }
 
-        Self::register_startup_era(&aggregators[0], config).await?;
-
-        for aggregator in &aggregators {
-            aggregator.serve().await?;
-        }
+        Ok((leader_aggregator, follower_aggregators))
+    }
 
-        Ok(aggregators)
+    async fn prepare_aggregator(
+        index: usize,
+        full_node: &FullNode,
+        config: &MithrilInfrastructureConfig,
+        chain_observer_type: &str,
+        leader_aggregator_endpoint: Option<String>,
+    ) -> StdResult<Aggregator> {
+        let aggregator_name = Aggregator::name_suffix(index);
+        let aggregator_artifacts_dir = config
+            .artifacts_dir
+            .join(format!("mithril-aggregator-{aggregator_name}"));
+        let aggregator_store_dir = config.store_dir.join(format!("aggregator-{aggregator_name}"));
+        let aggregator = Aggregator::new(&AggregatorConfig {
+            index,
+            name: &aggregator_name,
+            server_port: config.server_port + index as u64,
+            full_node,
+            cardano_cli_path: &config.devnet.cardano_cli_path(),
+            work_dir: &config.work_dir,
+            store_dir: &aggregator_store_dir,
+            artifacts_dir: &aggregator_artifacts_dir,
+            bin_dir: &config.bin_dir,
+            cardano_node_version: &config.cardano_node_version,
+            mithril_run_interval: config.mithril_run_interval,
+            mithril_era: &config.mithril_era,
+            mithril_era_reader_adapter: &config.mithril_era_reader_adapter,
+            mithril_era_marker_address: &config.devnet.mithril_era_marker_address()?,
+            signed_entity_types: &config.signed_entity_types,
+            chain_observer_type,
+            leader_aggregator_endpoint: &leader_aggregator_endpoint,
+        })?;
+
+        aggregator
+            .set_protocol_parameters(&ProtocolParameters {
+                k: 70,
+                m: 105,
+                phi_f: 0.95,
+            })
+            .await;
+
+        Ok(aggregator)
     }
 
     fn start_relays(
         config: &MithrilInfrastructureConfig,
-        aggregator_endpoints: &[String],
+        leader_aggregator_endpoint: String,
+        follower_aggregator_endpoints: &[String],
         signers_party_ids: &[PartyId],
         relay_signer_registration_mode: String,
         relay_signature_registration_mode: String,
     ) -> StdResult<(Vec<RelayAggregator>, Vec<RelaySigner>, Vec<RelayPassive>)> {
         if !config.use_relays {
             return Ok((vec![], vec![], vec![]));
         }
+        let aggregator_endpoints = [
+            vec![leader_aggregator_endpoint.clone()],
+            follower_aggregator_endpoints.to_vec(),
+        ]
+        .concat();
 
         let mut relay_aggregators: Vec<RelayAggregator> = vec![];
         let mut relay_signers: Vec<RelaySigner> = vec![];
         let mut relay_passives: Vec<RelayPassive> = vec![];
-        let leader_aggregator_endpoint = &aggregator_endpoints[0];
 
         info!("Starting the Mithril infrastructure in P2P mode (experimental)");
 
@@ -287,7 +315,7 @@ impl MithrilInfrastructure {
                 dial_to: bootstrap_peer_addr.clone(),
                 relay_signer_registration_mode: relay_signer_registration_mode.clone(),
                 relay_signature_registration_mode: relay_signature_registration_mode.clone(),
-                aggregator_endpoint: leader_aggregator_endpoint,
+                aggregator_endpoint: &leader_aggregator_endpoint,
                 party_id: party_id.clone(),
                 work_dir: &config.work_dir,
                 bin_dir: &config.bin_dir,
@@ -377,7 +405,7 @@ impl MithrilInfrastructure {
             signer.stop().await?;
         }
 
-        for aggregator in &self.aggregators {
+        for aggregator in self.aggregators() {
             aggregator.stop().await?;
         }
 
@@ -396,6 +424,18 @@ impl MithrilInfrastructure {
         &self.aggregators[index]
     }
 
+    pub fn leader_aggregator(&self) -> &Aggregator {
+        &self.aggregators[0]
+    }
+
+    pub fn follower_aggregators(&self) -> &[Aggregator] {
+        &self.aggregators[1..]
+    }
+
+    pub fn follower_aggregator(&self, index: usize) -> &Aggregator {
+        &self.aggregators[index + 1]
+    }
+
     pub fn signers(&self) -> &[Signer] {
         &self.signers
     }