feat(aggregator): add follower sync in state machine + wire synchroniser

Author: Alenar

mithril-aggregator/src/dependency_injection/builder/mod.rs

@@ -56,9 +56,9 @@ use crate::{
     file_uploaders::FileUploader,
     http_server::routes::router::{self, RouterConfig, RouterState},
     services::{
-        AggregatorHTTPClient, CertifierService, MessageService, MithrilSignerRegistrationFollower,
-        ProverService, SignedEntityService, SignerSynchronizer, Snapshotter,
-        StakeDistributionService, UpkeepService,
+        AggregatorHTTPClient, CertificateChainSynchronizer, CertifierService, MessageService,
+        MithrilSignerRegistrationFollower, ProverService, SignedEntityService, SignerSynchronizer,
+        Snapshotter, StakeDistributionService, UpkeepService,
     },
     tools::file_archiver::FileArchiver,
 };
@@ -185,6 +185,9 @@ pub struct DependenciesBuilder {
     /// Genesis signature verifier service.
     pub genesis_verifier: Option<Arc<ProtocolGenesisVerifier>>,
 
+    /// Certificate chain synchroniser service
+    pub certificate_chain_synchroniser: Option<Arc<dyn CertificateChainSynchronizer>>,
+
     /// Mithril signer registration leader service
     pub mithril_signer_registration_leader: Option<Arc<MithrilSignerRegistrationLeader>>,
 
@@ -312,6 +315,7 @@ impl DependenciesBuilder {
             snapshotter: None,
             certificate_verifier: None,
             genesis_verifier: None,
+            certificate_chain_synchroniser: None,
             mithril_signer_registration_leader: None,
             mithril_signer_registration_follower: None,
             signer_registerer: None,
@@ -374,6 +378,7 @@ impl DependenciesBuilder {
             verification_key_store: self.get_verification_key_store().await?,
             epoch_settings_storer: self.get_epoch_settings_store().await?,
             chain_observer: self.get_chain_observer().await?,
+            certificate_chain_synchroniser: self.get_certificate_chain_synchroniser().await?,
             signer_registerer: self.get_signer_registerer().await?,
             signer_synchronizer: self.get_signer_synchronizer().await?,
             signer_registration_round_opener: self.get_signer_registration_round_opener().await?,

mithril-aggregator/src/dependency_injection/builder/protocol/certificates.rs

@@ -10,8 +10,9 @@ use crate::database::repository::{BufferedSingleSignatureRepository, SingleSigna
 use crate::dependency_injection::{DependenciesBuilder, DependenciesBuilderError, Result};
 use crate::get_dependency;
 use crate::services::{
-    BufferedCertifierService, CertifierService, MithrilCertifierService,
-    MithrilSignerRegistrationFollower, SignerSynchronizer,
+    BufferedCertifierService, CertificateChainSynchronizer, CertifierService,
+    MithrilCertificateChainSynchroniserNoop, MithrilCertificateChainSynchronizer,
+    MithrilCertifierService, MithrilSignerRegistrationFollower, SignerSynchronizer,
 };
 use crate::{
     ExecutionEnvironment, MithrilSignerRegistrationLeader, MithrilSignerRegistrationVerifier,
@@ -84,6 +85,39 @@ impl DependenciesBuilder {
         get_dependency!(self.multi_signer)
     }
 
+    async fn build_certificate_chain_synchroniser(
+        &mut self,
+    ) -> Result<Arc<dyn CertificateChainSynchronizer>> {
+        let synchroniser: Arc<dyn CertificateChainSynchronizer> =
+            if self.configuration.is_follower_aggregator() {
+                let leader_aggregator_client = self.get_leader_aggregator_client().await?;
+                let verifier = Arc::new(MithrilCertificateVerifier::new(
+                    self.root_logger(),
+                    leader_aggregator_client.clone(),
+                ));
+
+                Arc::new(MithrilCertificateChainSynchronizer::new(
+                    leader_aggregator_client,
+                    self.get_certificate_repository().await?,
+                    verifier,
+                    self.get_genesis_verifier().await?,
+                    self.get_open_message_repository().await?,
+                    self.root_logger(),
+                ))
+            } else {
+                Arc::new(MithrilCertificateChainSynchroniserNoop)
+            };
+
+        Ok(synchroniser)
+    }
+
+    /// [CertificateChainSynchronizer] service
+    pub async fn get_certificate_chain_synchroniser(
+        &mut self,
+    ) -> Result<Arc<dyn CertificateChainSynchronizer>> {
+        get_dependency!(self.certificate_chain_synchroniser)
+    }
+
     async fn build_certificate_verifier(&mut self) -> Result<Arc<dyn CertificateVerifier>> {
         let verifier = Arc::new(MithrilCertificateVerifier::new(
             self.root_logger(),

mithril-aggregator/src/dependency_injection/containers/serve.rs

@@ -27,8 +27,9 @@ use crate::{
     entities::AggregatorEpochSettings,
     event_store::{EventMessage, TransmitterService},
     services::{
-        CertifierService, EpochService, MessageService, ProverService, SignedEntityService,
-        SignerRecorder, SignerSynchronizer, StakeDistributionService, UpkeepService,
+        CertificateChainSynchronizer, CertifierService, EpochService, MessageService,
+        ProverService, SignedEntityService, SignerRecorder, SignerSynchronizer,
+        StakeDistributionService, UpkeepService,
     },
 };
 
@@ -57,6 +58,9 @@ pub struct ServeCommandDependenciesContainer {
     /// Chain observer service.
     pub(crate) chain_observer: Arc<dyn ChainObserver>,
 
+    /// Certificate chain synchroniser service
+    pub(crate) certificate_chain_synchroniser: Arc<dyn CertificateChainSynchronizer>,
+
     /// Signer registerer service
     pub signer_registerer: Arc<dyn SignerRegisterer>,
 

mithril-aggregator/src/runtime/runner.rs

@@ -72,6 +72,12 @@ pub trait AggregatorRunnerTrait: Sync + Send {
     /// Synchronize the follower aggregator signer registration.
     async fn synchronize_follower_aggregator_signer_registration(&self) -> StdResult<()>;
 
+    /// Synchronise the follower aggregator certificate chain
+    async fn synchronize_follower_aggregator_certificate_chain(
+        &self,
+        force_sync: bool,
+    ) -> StdResult<()>;
+
     /// Ask the EpochService to update the epoch settings.
     async fn update_epoch_settings(&self) -> StdResult<()>;
 
@@ -506,6 +512,20 @@ impl AggregatorRunnerTrait for AggregatorRunner {
             .get_runtime_cycle_total_since_startup()
             .increment();
     }
+
+    async fn synchronize_follower_aggregator_certificate_chain(
+        &self,
+        force_sync: bool,
+    ) -> StdResult<()> {
+        debug!(
+            self.logger,
+            ">> synchronize_follower_aggregator_certificate_chain(force_sync:{force_sync})"
+        );
+        self.dependencies
+            .certificate_chain_synchroniser
+            .synchronize_certificate_chain(force_sync)
+            .await
+    }
 }
 
 #[cfg(test)]

mithril-aggregator/src/runtime/state_machine.rs

@@ -277,13 +277,21 @@ impl AggregatorRuntime {
             self.runner.precompute_epoch_data().await?;
         }
 
-        self.runner
+        let chain_validity_result = self
+            .runner
             .is_certificate_chain_valid(&new_time_point)
             .await
             .map_err(|e| RuntimeError::KeepState {
                 message: "certificate chain is invalid".to_string(),
                 nested_error: e.into(),
-            })?;
+            });
+        if self.config.is_follower {
+            let force_sync = chain_validity_result.is_err();
+            self.runner
+                .synchronize_follower_aggregator_certificate_chain(force_sync)
+                .await?;
+        }
+        chain_validity_result?;
 
         Ok(())
     }
@@ -834,6 +842,8 @@ mod tests {
     }
 
     mod follower {
+        use mockall::predicate::eq;
+
         use super::*;
 
         #[tokio::test]
@@ -911,6 +921,11 @@ mod tests {
                 .expect_is_certificate_chain_valid()
                 .once()
                 .returning(|_| Ok(()));
+            runner
+                .expect_synchronize_follower_aggregator_certificate_chain()
+                .once()
+                .with(eq(false)) // Certificate chain valid so force_sync must be false
+                .returning(|_| Ok(()));
             runner
                 .expect_update_era_checker()
                 .with(predicate::eq(new_time_point_clone.clone().epoch))

mithril-aggregator/src/services/aggregator_client.rs

@@ -11,6 +11,7 @@ use thiserror::Error;
 use mithril_common::{
     MITHRIL_AGGREGATOR_VERSION_HEADER, MITHRIL_API_VERSION_HEADER, StdError, StdResult,
     api_version::APIVersionProvider,
+    certificate_chain::{CertificateRetriever, CertificateRetrieverError},
     entities::{Certificate, ClientError, ServerError},
     logging::LoggerExtensions,
     messages::{
@@ -334,6 +335,27 @@ impl LeaderAggregatorClient for AggregatorHTTPClient {
     }
 }
 
+#[async_trait]
+impl CertificateRetriever for AggregatorHTTPClient {
+    async fn get_certificate_details(
+        &self,
+        certificate_hash: &str,
+    ) -> Result<Certificate, CertificateRetrieverError> {
+        let message = self
+            .certificates_details(certificate_hash)
+            .await
+            .with_context(|| {
+                format!("Failed to retrieve certificate with hash: '{certificate_hash}'")
+            })
+            .map_err(CertificateRetrieverError)?
+            .ok_or(CertificateRetrieverError(anyhow!(
+                "Certificate does not exist: '{certificate_hash}'"
+            )))?;
+
+        message.try_into().map_err(CertificateRetrieverError)
+    }
+}
+
 #[async_trait]
 impl RemoteCertificateRetriever for AggregatorHTTPClient {
     async fn get_latest_certificate_details(&self) -> StdResult<Option<Certificate>> {

mithril-aggregator/src/services/certificate_chain_synchroniser/mod.rs

@@ -1,5 +1,7 @@
 mod interface;
+mod noop;
 mod synchroniser_service;
 
 pub use interface::*;
+pub use noop::*;
 pub use synchroniser_service::*;

mithril-aggregator/src/services/certificate_chain_synchroniser/noop.rs

@@ -0,0 +1,13 @@
+use mithril_common::StdResult;
+
+use crate::services::CertificateChainSynchronizer;
+
+/// A noop [CertificateChainSynchronizer] for leader aggregators
+pub struct MithrilCertificateChainSynchroniserNoop;
+
+#[async_trait::async_trait]
+impl CertificateChainSynchronizer for MithrilCertificateChainSynchroniserNoop {
+    async fn synchronize_certificate_chain(&self, _force: bool) -> StdResult<()> {
+        Ok(())
+    }
+}