Merge pull request #2351 from input-output-hk/jpraynaud/2334-aggregator-slave-registration-mode

Feat: implement aggregator `slave` signer registration mode

Author: jpraynaud

CHANGELOG.md

@@ -15,6 +15,8 @@ As a minor extension, we have adopted a slightly different versioning convention
 
 - End support for **macOS x64 pre-built binaries** for the client CLI.
 
+- **UNSTABLE** Implement a slave signer registration mode in the aggregator.
+
 - **UNSTABLE** Cardano database incremental certification:
 
   - Implement the client library for the the signed entity type `CardanoDatabase` (download and prove snapshot).

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.7.10"
+version = "0.7.11"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }
@@ -69,6 +69,8 @@ tikv-jemallocator = { version = "0.6.0", optional = true }
 
 [dev-dependencies]
 criterion = { version = "0.5.1", features = ["html_reports", "async_tokio"] }
+http = "1.2.0"
+httpmock = "0.7.0"
 mithril-common = { path = "../mithril-common", features = [
     "allow_skip_signer_certification",
     "test_tools",

mithril-aggregator/Cargo.toml

@@ -15,16 +15,16 @@ use crate::{dependency_injection::DependenciesBuilder, Configuration};
 pub struct ServeCommand {
     /// Server listening IP
     #[clap(long)]
-    pub server_ip: Option<String>,
+    server_ip: Option<String>,
 
     /// Server TCP port
     #[clap(long)]
-    pub server_port: Option<u16>,
+    server_port: Option<u16>,
 
     /// Directory to store snapshot
     /// Defaults to work folder
     #[clap(long)]
-    pub snapshot_directory: Option<PathBuf>,
+    snapshot_directory: Option<PathBuf>,
 
     /// Disable immutables digests cache.
     #[clap(long)]
@@ -53,6 +53,14 @@ pub struct ServeCommand {
     /// Metrics HTTP server listening port.
     #[clap(long)]
     metrics_server_port: Option<u16>,
+
+    /// Master aggregator endpoint
+    ///
+    /// This is the endpoint of the aggregator that will be used to fetch the latest epoch settings
+    /// and store the signer registrations when the aggregator is running in a slave mode.
+    /// If this is not set, the aggregator will run in a master mode.
+    #[clap(long)]
+    master_aggregator_endpoint: Option<String>,
 }
 
 impl Source for ServeCommand {
@@ -121,6 +129,15 @@ impl Source for ServeCommand {
                 Value::new(Some(&namespace), ValueKind::from(metrics_server_port)),
             );
         }
+        if let Some(master_aggregator_endpoint) = self.master_aggregator_endpoint.clone() {
+            result.insert(
+                "master_aggregator_endpoint".to_string(),
+                Value::new(
+                    Some(&namespace),
+                    ValueKind::from(Some(master_aggregator_endpoint)),
+                ),
+            );
+        }
 
         Ok(result)
     }

mithril-aggregator/src/commands/serve_command.rs

@@ -193,6 +193,13 @@ pub struct Configuration {
 
     /// Time interval at which usage metrics are persisted in event database (in seconds).
     pub persist_usage_report_interval_in_seconds: u64,
+
+    // Master aggregator endpoint
+    ///
+    /// This is the endpoint of the aggregator that will be used to fetch the latest epoch settings
+    /// and store the signer registrations when the aggregator is running in a slave mode.
+    /// If this is not set, the aggregator will run in a master mode.
+    pub master_aggregator_endpoint: Option<String>,
 }
 
 /// Uploader needed to copy the snapshot once computed.
@@ -283,6 +290,7 @@ impl Configuration {
             metrics_server_ip: "0.0.0.0".to_string(),
             metrics_server_port: 9090,
             persist_usage_report_interval_in_seconds: 10,
+            master_aggregator_endpoint: None,
         }
     }
 
@@ -375,6 +383,11 @@ impl Configuration {
             cardano_transactions_signing_config: self.cardano_transactions_signing_config.clone(),
         }
     }
+
+    /// Check if the aggregator is running in slave mode.
+    pub fn is_slave_aggregator(&self) -> bool {
+        self.master_aggregator_endpoint.is_some()
+    }
 }
 
 /// Default configuration with all the default values for configurations.
@@ -726,4 +739,24 @@ mod test {
             "Joined URL `{joined_url}`, does not contain subpath `{subpath_without_trailing_slash}`"
         );
     }
+
+    #[test]
+    fn is_slave_aggregator_returns_true_when_in_slave_mode() {
+        let config = Configuration {
+            master_aggregator_endpoint: Some("some_endpoint".to_string()),
+            ..Configuration::new_sample()
+        };
+
+        assert!(config.is_slave_aggregator());
+    }
+
+    #[test]
+    fn is_slave_aggregator_returns_false_when_in_master_mode() {
+        let config = Configuration {
+            master_aggregator_endpoint: None,
+            ..Configuration::new_sample()
+        };
+
+        assert!(!config.is_slave_aggregator());
+    }
 }

mithril-aggregator/src/configuration.rs

@@ -5,12 +5,15 @@
 //! - redefine the actual categories so those miscellaneous enablers fit into them
 
 use std::sync::Arc;
+use std::time::Duration;
 
 use mithril_signed_entity_lock::SignedEntityTypeLock;
 
 use crate::database::repository::CertificateRepository;
 use crate::dependency_injection::{DependenciesBuilder, Result};
-use crate::services::{MessageService, MithrilMessageService};
+use crate::services::{
+    AggregatorClient, AggregatorHTTPClient, MessageService, MithrilMessageService,
+};
 
 impl DependenciesBuilder {
     async fn build_signed_entity_lock(&mut self) -> Result<Arc<SignedEntityTypeLock>> {
@@ -34,10 +37,12 @@ impl DependenciesBuilder {
         ));
         let signed_entity_storer = self.get_signed_entity_storer().await?;
         let immutable_file_digest_mapper = self.get_immutable_file_digest_mapper().await?;
+        let epoch_service = self.get_epoch_service().await?;
         let service = MithrilMessageService::new(
             certificate_repository,
             signed_entity_storer,
             immutable_file_digest_mapper,
+            epoch_service,
         );
 
         Ok(Arc::new(service))
@@ -51,4 +56,31 @@ impl DependenciesBuilder {
 
         Ok(self.message_service.as_ref().cloned().unwrap())
     }
+
+    /// build an [AggregatorClient]
+    pub async fn build_master_aggregator_client(&mut self) -> Result<Arc<dyn AggregatorClient>> {
+        let master_aggregator_endpoint = self
+            .configuration
+            .master_aggregator_endpoint
+            .to_owned()
+            .unwrap_or_default();
+        let aggregator_client = AggregatorHTTPClient::new(
+            master_aggregator_endpoint,
+            None,
+            self.get_api_version_provider().await?,
+            Some(Duration::from_secs(30)),
+            self.root_logger(),
+        );
+
+        Ok(Arc::new(aggregator_client))
+    }
+
+    /// Returns a master [AggregatorClient]
+    pub async fn get_master_aggregator_client(&mut self) -> Result<Arc<dyn AggregatorClient>> {
+        if self.master_aggregator_client.is_none() {
+            self.master_aggregator_client = Some(self.build_master_aggregator_client().await?);
+        }
+
+        Ok(self.master_aggregator_client.as_ref().cloned().unwrap())
+    }
 }

mithril-aggregator/src/dependency_injection/builder/enablers/misc.rs

@@ -44,12 +44,14 @@ use crate::{
     file_uploaders::FileUploader,
     http_server::routes::router::{self, RouterConfig, RouterState},
     services::{
-        CertifierService, MessageService, ProverService, SignedEntityService, Snapshotter,
+        AggregatorClient, CertifierService, MessageService, MithrilSignerRegistrationSlave,
+        ProverService, SignedEntityService, SignerSynchronizer, Snapshotter,
         StakeDistributionService, UpkeepService,
     },
     tools::GenesisToolsDependency,
     AggregatorConfig, AggregatorRunner, AggregatorRuntime, Configuration, DependencyContainer,
-    ImmutableFileDigestMapper, MetricsService, MithrilSignerRegisterer, MultiSigner,
+    ImmutableFileDigestMapper, MetricsService, MithrilSignerRegistrationMaster, MultiSigner,
+    SignerRegisterer, SignerRegistrationRoundOpener, SignerRegistrationVerifier,
     SingleSignatureAuthenticator, VerificationKeyStorer,
 };
 
@@ -147,8 +149,23 @@ pub struct DependenciesBuilder {
     /// Genesis signature verifier service.
     pub genesis_verifier: Option<Arc<ProtocolGenesisVerifier>>,
 
+    /// Mithril signer registration master service
+    pub mithril_signer_registration_master: Option<Arc<MithrilSignerRegistrationMaster>>,
+
+    /// Mithril signer registration slave service
+    pub mithril_signer_registration_slave: Option<Arc<MithrilSignerRegistrationSlave>>,
+
     /// Signer registerer service
-    pub mithril_registerer: Option<Arc<MithrilSignerRegisterer>>,
+    pub signer_registerer: Option<Arc<dyn SignerRegisterer>>,
+
+    /// Signer synchronizer service
+    pub signer_synchronizer: Option<Arc<dyn SignerSynchronizer>>,
+
+    /// Signer registration verifier
+    pub signer_registration_verifier: Option<Arc<dyn SignerRegistrationVerifier>>,
+
+    /// Signer registration round opener service
+    pub signer_registration_round_opener: Option<Arc<dyn SignerRegistrationRoundOpener>>,
 
     /// Era checker service
     pub era_checker: Option<Arc<EraChecker>>,
@@ -218,6 +235,9 @@ pub struct DependenciesBuilder {
 
     /// Metrics service
     pub metrics_service: Option<Arc<MetricsService>>,
+
+    /// Master aggregator client
+    pub master_aggregator_client: Option<Arc<dyn AggregatorClient>>,
 }
 
 impl DependenciesBuilder {
@@ -249,7 +269,12 @@ impl DependenciesBuilder {
             snapshotter: None,
             certificate_verifier: None,
             genesis_verifier: None,
-            mithril_registerer: None,
+            mithril_signer_registration_master: None,
+            mithril_signer_registration_slave: None,
+            signer_registerer: None,
+            signer_synchronizer: None,
+            signer_registration_verifier: None,
+            signer_registration_round_opener: None,
             era_reader_adapter: None,
             era_checker: None,
             era_reader: None,
@@ -272,6 +297,7 @@ impl DependenciesBuilder {
             upkeep_service: None,
             single_signer_authenticator: None,
             metrics_service: None,
+            master_aggregator_client: None,
         }
     }
 
@@ -316,8 +342,10 @@ impl DependenciesBuilder {
             snapshotter: self.get_snapshotter().await?,
             certificate_verifier: self.get_certificate_verifier().await?,
             genesis_verifier: self.get_genesis_verifier().await?,
-            signer_registerer: self.get_mithril_registerer().await?,
-            signer_registration_round_opener: self.get_mithril_registerer().await?,
+            signer_registerer: self.get_signer_registerer().await?,
+            signer_synchronizer: self.get_signer_synchronizer().await?,
+            signer_registration_verifier: self.get_signer_registration_verifier().await?,
+            signer_registration_round_opener: self.get_signer_registration_round_opener().await?,
             era_checker: self.get_era_checker().await?,
             era_reader: self.get_era_reader().await?,
             event_transmitter: self.get_event_transmitter().await?,
@@ -339,6 +367,7 @@ impl DependenciesBuilder {
             upkeep_service: self.get_upkeep_service().await?,
             single_signer_authenticator: self.get_single_signature_authenticator().await?,
             metrics_service: self.get_metrics_service().await?,
+            master_aggregator_client: self.get_master_aggregator_client().await?,
         };
 
         Ok(dependency_manager)
@@ -348,7 +377,10 @@ impl DependenciesBuilder {
     pub async fn create_aggregator_runner(&mut self) -> Result<AggregatorRuntime> {
         let dependency_container = Arc::new(self.build_dependency_container().await?);
 
-        let config = AggregatorConfig::new(Duration::from_millis(self.configuration.run_interval));
+        let config = AggregatorConfig::new(
+            Duration::from_millis(self.configuration.run_interval),
+            self.configuration.is_slave_aggregator(),
+        );
         let runtime = AggregatorRuntime::new(
             config,
             None,