add era reader in the Aggregator

ghubertpalo · ghubertpalo · commit 3fd4d50099bf · 2023-02-01T17:48:41.000+01:00
diff --git a/mithril-aggregator/src/command_args.rs b/mithril-aggregator/src/command_args.rs
@@ -15,9 +15,9 @@ use mithril_common::{
         CardanoImmutableDigester, ImmutableFileSystemObserver,
     },
     entities::{Epoch, HexEncodedGenesisSecretKey},
-    era::{EraChecker, SupportedEra},
+    era::{adapters::BootstrapAdapter as BootstrapEraReaderAdapter, EraChecker, EraReader},
     store::{adapter::SQLiteAdapter, StakeStore},
-    BeaconProviderImpl,
+    BeaconProvider, BeaconProviderImpl,
 };
 
 use crate::{
@@ -399,8 +399,14 @@ impl ServeCommand {
         ));
 
         // TODO: use EraReader when it is implemented to retrieve current era
-        let current_era = SupportedEra::Thales;
-        let era_checker = Arc::new(EraChecker::new(current_era));
+        let era_reader = Arc::new(EraReader::new(Box::new(BootstrapEraReaderAdapter)));
+        let era_epoch_token = era_reader
+            .read_era_epoch_token(beacon_provider.get_current_beacon().await?.epoch)
+            .await?;
+        let era_checker = Arc::new(EraChecker::new(
+            era_epoch_token.get_current_supported_era()?,
+            era_epoch_token.get_current_epoch(),
+        ));
 
         // Init dependency manager
         let dependency_manager = DependencyManager {
@@ -424,6 +430,7 @@ impl ServeCommand {
             signer_registerer: signer_registerer.clone(),
             signer_registration_round_opener: signer_registerer.clone(),
             era_checker: era_checker.clone(),
+            era_reader: era_reader.clone(),
         };
         let dependency_manager = Arc::new(dependency_manager);
 
diff --git a/mithril-aggregator/src/dependency.rs b/mithril-aggregator/src/dependency.rs
@@ -6,7 +6,7 @@ use mithril_common::{
     entities::{
         Certificate, Epoch, ProtocolParameters, Signer, SignerWithStake, StakeDistribution,
     },
-    era::EraChecker,
+    era::{EraChecker, EraReader},
     store::{StakeStore, StakeStorer},
     BeaconProvider,
 };
@@ -87,6 +87,9 @@ pub struct DependencyManager {
 
     /// Era checker service
     pub era_checker: Arc<EraChecker>,
+
+    /// Era reader service
+    pub era_reader: Arc<EraReader>,
 }
 
 #[doc(hidden)]
@@ -269,10 +272,10 @@ pub mod tests {
         chain_observer::FakeObserver,
         crypto_helper::{key_encode_hex, ProtocolGenesisSigner},
         digesters::{DumbImmutableDigester, DumbImmutableFileObserver},
-        era::{EraChecker, SupportedEra},
+        era::{adapters::BootstrapAdapter as BootstrapEraReaderAdapter, EraChecker, EraReader},
         store::{adapter::MemoryAdapter, StakeStore},
         test_utils::fake_data,
-        BeaconProviderImpl, CardanoNetwork,
+        BeaconProvider, BeaconProviderImpl, CardanoNetwork,
     };
     use std::{path::PathBuf, sync::Arc};
     use tokio::sync::RwLock;
@@ -347,7 +350,15 @@ pub mod tests {
             chain_observer.clone(),
             verification_key_store.clone(),
         ));
-        let era_checker = Arc::new(EraChecker::new(SupportedEra::dummy()));
+        let era_reader = Arc::new(EraReader::new(Box::new(BootstrapEraReaderAdapter)));
+        let era_epoch_token = era_reader
+            .read_era_epoch_token(beacon_provider.get_current_beacon().await.unwrap().epoch)
+            .await
+            .unwrap();
+        let era_checker = Arc::new(EraChecker::new(
+            era_epoch_token.get_current_supported_era().unwrap(),
+            era_epoch_token.get_current_epoch(),
+        ));
 
         let dependency_manager = DependencyManager {
             config,
@@ -370,6 +381,7 @@ pub mod tests {
             signer_registerer: signer_registerer.clone(),
             signer_registration_round_opener: signer_registerer,
             era_checker,
+            era_reader,
         };
 
         let config = AggregatorConfig::new(
diff --git a/mithril-aggregator/src/runtime/runner.rs b/mithril-aggregator/src/runtime/runner.rs
@@ -147,6 +147,9 @@ pub trait AggregatorRunnerTrait: Sync + Send {
         ongoing_snapshot: &OngoingSnapshot,
         remote_locations: Vec<String>,
     ) -> Result<Snapshot, RuntimeError>;
+
+    /// Update the EraChecker with EraReader information.
+    async fn update_era_checker(&self, beacon: &Beacon) -> Result<(), RuntimeError>;
 }
 
 /// The runner responsibility is to expose a code API for the state machine. It
@@ -614,6 +617,25 @@ impl AggregatorRunnerTrait for AggregatorRunner {
 
         Ok(snapshot)
     }
+
+    async fn update_era_checker(&self, beacon: &Beacon) -> Result<(), RuntimeError> {
+        let token = self
+            .dependencies
+            .era_reader
+            .read_era_epoch_token(beacon.epoch)
+            .await
+            .map_err(|e| {
+                RuntimeError::General(format!("Could not get Era information ('{e}')").into())
+            })?;
+        self.dependencies.era_checker.change_era(
+            token.get_current_supported_era().map_err(|e| {
+                RuntimeError::General(format!("Could not update EraChecker service ('{e}')").into())
+            })?,
+            token.get_current_epoch(),
+        );
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -1148,4 +1170,19 @@ pub mod tests {
             ongoing_snapshot.get_file_path()
         );
     }
+
+    #[tokio::test]
+    async fn test_update_era_checker() {
+        let (deps, config) = initialize_dependencies().await;
+        let beacon_provider = deps.beacon_provider.clone();
+        let era_checker = deps.era_checker.clone();
+        let mut beacon = beacon_provider.get_current_beacon().await.unwrap();
+
+        assert_eq!(beacon.epoch, era_checker.current_epoch());
+        let runner = AggregatorRunner::new(config, Arc::new(deps));
+        beacon.epoch += 1;
+
+        runner.update_era_checker(&beacon).await.unwrap();
+        assert_eq!(beacon.epoch, era_checker.current_epoch());
+    }
 }
diff --git a/mithril-aggregator/src/runtime/state_machine.rs b/mithril-aggregator/src/runtime/state_machine.rs
@@ -222,6 +222,7 @@ impl AggregatorRuntime {
         if maybe_current_beacon.is_none() || maybe_current_beacon.unwrap().epoch < new_beacon.epoch
         {
             self.runner.close_signer_registration_round().await?;
+            self.runner.update_era_checker(&new_beacon).await?;
             self.runner.update_stake_distribution(&new_beacon).await?;
             self.runner
                 .open_signer_registration_round(&new_beacon)
@@ -388,6 +389,11 @@ mod tests {
             .expect_is_certificate_chain_valid()
             .once()
             .returning(|| Ok(false));
+        runner
+            .expect_update_era_checker()
+            .with(predicate::eq(fake_data::beacon()))
+            .once()
+            .returning(|_| Ok(()));
 
         let mut runtime = init_runtime(
             Some(AggregatorState::Idle(IdleState {
@@ -435,6 +441,11 @@ mod tests {
             .expect_is_certificate_chain_valid()
             .once()
             .returning(|| Ok(true));
+        runner
+            .expect_update_era_checker()
+            .with(predicate::eq(fake_data::beacon()))
+            .once()
+            .returning(|_| Ok(()));
 
         let mut runtime = init_runtime(
             Some(AggregatorState::Idle(IdleState {
diff --git a/mithril-aggregator/tests/test_extensions/dependency.rs b/mithril-aggregator/tests/test_extensions/dependency.rs
@@ -9,10 +9,11 @@ use mithril_common::chain_observer::FakeObserver;
 use mithril_common::crypto_helper::{key_encode_hex, ProtocolGenesisSigner};
 use mithril_common::digesters::{DumbImmutableDigester, DumbImmutableFileObserver};
 use mithril_common::entities::ProtocolParameters;
-use mithril_common::era::{EraChecker, SupportedEra};
+use mithril_common::era::EraReader;
+use mithril_common::era::{adapters::BootstrapAdapter as BootstrapEraReaderAdapter, EraChecker};
 use mithril_common::store::adapter::MemoryAdapter;
 use mithril_common::store::StakeStore;
-use mithril_common::{BeaconProviderImpl, CardanoNetwork};
+use mithril_common::{BeaconProvider, BeaconProviderImpl, CardanoNetwork};
 use std::path::PathBuf;
 use std::sync::Arc;
 use tokio::sync::RwLock;
@@ -91,7 +92,15 @@ pub async fn initialize_dependencies(
         chain_observer.clone(),
         verification_key_store.clone(),
     ));
-    let era_checker = Arc::new(EraChecker::new(SupportedEra::dummy()));
+    let era_reader = Arc::new(EraReader::new(Box::new(BootstrapEraReaderAdapter)));
+    let era_epoch_token = era_reader
+        .read_era_epoch_token(beacon_provider.get_current_beacon().await.unwrap().epoch)
+        .await
+        .unwrap();
+    let era_checker = Arc::new(EraChecker::new(
+        era_epoch_token.get_current_supported_era().unwrap(),
+        era_epoch_token.get_current_epoch(),
+    ));
 
     let dependency_manager = DependencyManager {
         config,
@@ -114,6 +123,7 @@ pub async fn initialize_dependencies(
         signer_registerer: signer_registerer.clone(),
         signer_registration_round_opener: signer_registerer,
         era_checker,
+        era_reader,
     };
 
     let config = AggregatorConfig::new(
diff --git a/mithril-common/src/era/adapters/mod.rs b/mithril-common/src/era/adapters/mod.rs
@@ -1,4 +1,4 @@
-//! Module dedicated to [EraReaderAdapter] implementations.
+//! Module dedicated to EraReaderAdapter implementations.
 mod bootstrap;
 mod dummy;
 
diff --git a/mithril-common/src/era/era_checker.rs b/mithril-common/src/era/era_checker.rs
@@ -1,29 +1,45 @@
 use std::sync::RwLock;
 
+use crate::entities::Epoch;
+
 use super::SupportedEra;
 
+struct SupportedEraStamp {
+    era: SupportedEra,
+    epoch: Epoch,
+}
+
 /// EraChecker allows the verification of the current era
 pub struct EraChecker {
-    current_era: RwLock<SupportedEra>,
+    current_era_stamp: RwLock<SupportedEraStamp>,
 }
 
 impl EraChecker {
     /// Era checker factory
-    pub fn new(era: SupportedEra) -> Self {
+    pub fn new(era: SupportedEra, epoch: Epoch) -> Self {
         Self {
-            current_era: RwLock::new(era),
+            current_era_stamp: RwLock::new(SupportedEraStamp { era, epoch }),
         }
     }
 
     /// Retrieve the current era
     pub fn current_era(&self) -> SupportedEra {
-        self.current_era.read().unwrap().to_owned()
+        self.current_era_stamp.read().unwrap().era
+    }
+
+    /// Retrieve the current Epoch
+    pub fn current_epoch(&self) -> Epoch {
+        self.current_era_stamp.read().unwrap().epoch
     }
 
     /// Change the current era
-    pub fn change_era(&self, new_era: SupportedEra) {
-        let mut era = self.current_era.write().unwrap();
-        *era = new_era;
+    pub fn change_era(&self, new_era: SupportedEra, current_epoch: Epoch) {
+        let new_stamp = SupportedEraStamp {
+            era: new_era,
+            epoch: current_epoch,
+        };
+        let mut stamp = self.current_era_stamp.write().unwrap();
+        *stamp = new_stamp;
     }
 
     /// Check if an era is active
@@ -39,16 +55,11 @@ mod tests {
     #[test]
     fn can_change_era() {
         let expected_era = SupportedEra::dummy();
-        let era_checker = EraChecker::new(expected_era);
-        era_checker.change_era(expected_era);
+        let era_checker = EraChecker::new(expected_era, Epoch(1));
+        era_checker.change_era(expected_era, Epoch(2));
 
         assert_eq!(expected_era, era_checker.current_era());
-    }
-
-    #[test]
-    fn is_era_active_panics_when_current_era_not_set() {
-        let expected_era = SupportedEra::dummy();
-        let era_checker = EraChecker::new(expected_era);
-        era_checker.is_era_active(expected_era);
+        assert_eq!(Epoch(2), era_checker.current_epoch());
+        assert!(era_checker.is_era_active(expected_era));
     }
 }
diff --git a/mithril-common/src/era/era_reader.rs b/mithril-common/src/era/era_reader.rs
@@ -6,11 +6,14 @@ use thiserror::Error;
 
 use super::{supported_era::UnsupportedEraError, SupportedEra};
 
-/// Represents a tag of Era change.
+/// Value object that represents a tag of Era change.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct EraMarker {
-    name: String,
-    epoch: Option<Epoch>,
+    /// Era name
+    pub name: String,
+
+    /// Eventual information that advertises the Epoch of transition.
+    pub epoch: Option<Epoch>,
 }
 
 impl EraMarker {
@@ -25,7 +28,7 @@ impl EraMarker {
 /// Adapters are responsible of technically reading the information of
 /// [EraMarker]s from a backend.
 #[async_trait]
-pub trait EraReaderAdapter {
+pub trait EraReaderAdapter: Sync + Send {
     /// Read era markers from the underlying adapter.
     async fn read(&self) -> Result<Vec<EraMarker>, Box<dyn StdError>>;
 }
@@ -61,6 +64,11 @@ impl EraEpochToken {
         &self.current_era
     }
 
+    /// Return the epoch the Token has been created at
+    pub fn get_current_epoch(&self) -> Epoch {
+        self.current_epoch
+    }
+
     /// Try to cast the next [EraMarker] to a [SupportedEra]. If it fails, that
     /// means the coming Era will not be supported by this version of the
     /// software. This mechanism is used to issue a warning to the user asking

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-//! Module dedicated to [EraReaderAdapter] implementations.`
	`1`	`+//! Module dedicated to EraReaderAdapter implementations.`
`2`	`2`	`mod bootstrap;`
`3`	`3`	`mod dummy;`
`4`	`4`