feat(aggregator): move & rework handle discrepancies

- run it at the end of the serve dependency container build
- retrieve and save data from the network configuration provider instead
  of the local node configuration
- update follower integration test to check that local protocol
  parameter configuration is not read, instead the configuration is read
  through the network configuration provider from the leader

Author: Alenar

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

@@ -61,8 +61,6 @@ impl DependenciesBuilder {
                 ))
             };
 
-        //TODO handle discrepency here
-
         Ok(network_configuration_provider)
     }
 

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

@@ -3,7 +3,7 @@ mod protocol;
 mod support;
 
 use anyhow::Context;
-use slog::Logger;
+use slog::{Logger, debug};
 use std::{path::PathBuf, sync::Arc};
 use tokio::{
     sync::{
@@ -45,10 +45,11 @@ use super::{
     GenesisCommandDependenciesContainer, Result, ToolsCommandDependenciesContainer,
 };
 use crate::{
-    AggregatorConfig, AggregatorRunner, AggregatorRuntime, ImmutableFileDigestMapper,
-    MetricsService, MithrilSignerRegistrationLeader, MultiSigner, ProtocolParametersRetriever,
-    ServeCommandDependenciesContainer, SignerRegisterer, SignerRegistrationRoundOpener,
-    SignerRegistrationVerifier, SingleSignatureAuthenticator, VerificationKeyStorer,
+    AggregatorConfig, AggregatorRunner, AggregatorRuntime, EpochSettingsStorer,
+    ImmutableFileDigestMapper, MetricsService, MithrilSignerRegistrationLeader, MultiSigner,
+    ProtocolParametersRetriever, ServeCommandDependenciesContainer, SignerRegisterer,
+    SignerRegistrationRoundOpener, SignerRegistrationVerifier, SingleSignatureAuthenticator,
+    VerificationKeyStorer,
     configuration::ConfigurationSource,
     database::repository::{
         CertificateRepository, EpochSettingsStore, OpenMessageRepository, SignedEntityStorer,
@@ -411,6 +412,8 @@ impl DependenciesBuilder {
             metrics_service: self.get_metrics_service().await?,
         };
 
+        self.handle_discrepancies_at_startup().await?;
+
         Ok(dependencies_manager)
     }
 
@@ -527,6 +530,53 @@ impl DependenciesBuilder {
         Ok(dependencies)
     }
 
+    /// Look for discrepancies in stored data and fix them.
+    ///
+    /// Fix discrepancies for:
+    /// - epoch settings: ensure that the network configuration parameters for the three working epochs
+    ///   window are stored in the database (see [mithril_protocol_config::model::MithrilNetworkConfiguration])
+    pub async fn handle_discrepancies_at_startup(&mut self) -> Result<()> {
+        let logger = self.root_logger();
+        let current_epoch = self
+            .get_chain_observer()
+            .await?
+            .get_current_epoch()
+            .await
+            .map_err(|e| DependenciesBuilderError::Initialization {
+                message: "cannot handle startup discrepancies: failed to retrieve current epoch."
+                    .to_string(),
+                error: Some(e.into()),
+            })?
+            .ok_or(DependenciesBuilderError::Initialization {
+                message: "cannot handle startup discrepancies: no epoch returned.".to_string(),
+                error: None,
+            })?;
+        let network_configuration = self
+            .get_mithril_network_configuration_provider()
+            .await?
+            .get_network_configuration(current_epoch)
+            .await
+            .map_err(|e| DependenciesBuilderError::Initialization {
+                message: format!("cannot handle startup discrepancies: failed to retrieve network configuration for epoch {current_epoch}"),
+                error: Some(e),
+            })?;
+        let epoch_settings_store = self.get_epoch_settings_store().await?;
+
+        debug!(
+            logger,
+            "Handle discrepancies at startup of epoch settings store, will record epoch settings from the configuration for epoch {current_epoch}";
+            "network_configuration" => ?network_configuration,
+        );
+        epoch_settings_store
+            .handle_discrepancies_at_startup(&network_configuration)
+            .await
+            .map_err(|e| DependenciesBuilderError::Initialization {
+                message: "can not create aggregator runner".to_string(),
+                error: Some(e),
+            })?;
+        Ok(())
+    }
+
     /// Remove the dependencies builder from memory to release Arc instances.
     pub async fn vanish(self) {
         self.drop_sqlite_connections().await;

mithril-aggregator/src/dependency_injection/builder/support/stores.rs

@@ -1,5 +1,4 @@
 use anyhow::Context;
-use slog::debug;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -11,11 +10,11 @@ use crate::database::repository::{
     OpenMessageRepository, SignedEntityStore, SignedEntityStorer, SignerRegistrationStore,
     SignerStore, StakePoolStore,
 };
-use crate::dependency_injection::{DependenciesBuilder, DependenciesBuilderError, Result};
+use crate::dependency_injection::{DependenciesBuilder, Result};
 use crate::get_dependency;
 use crate::{
-    CExplorerSignerRetriever, EpochSettingsStorer, ImmutableFileDigestMapper,
-    ProtocolParametersRetriever, SignersImporter, VerificationKeyStorer,
+    CExplorerSignerRetriever, ImmutableFileDigestMapper, ProtocolParametersRetriever,
+    SignersImporter, VerificationKeyStorer,
 };
 
 impl DependenciesBuilder {
@@ -89,52 +88,15 @@ impl DependenciesBuilder {
     }
 
     async fn build_epoch_settings_store(&mut self) -> Result<Arc<EpochSettingsStore>> {
-        let logger = self.root_logger();
         let epoch_settings_store = EpochSettingsStore::new(
             self.get_sqlite_connection().await?,
             self.configuration.safe_epoch_retention_limit(),
         );
-        let current_epoch = self
-            .get_chain_observer()
-            .await?
-            .get_current_epoch()
-            .await
-            .map_err(|e| DependenciesBuilderError::Initialization {
-                message: "cannot create aggregator runner: failed to retrieve current epoch."
-                    .to_string(),
-                error: Some(e.into()),
-            })?
-            .ok_or(DependenciesBuilderError::Initialization {
-                message: "cannot build aggregator runner: no epoch returned.".to_string(),
-                error: None,
-            })?;
-        let retrieval_epoch = current_epoch
-            .offset_to_signer_retrieval_epoch()
-            .map_err(|e| DependenciesBuilderError::Initialization {
-                message: format!("cannot create aggregator runner: failed to offset current epoch '{current_epoch}' to signer retrieval epoch."),
-                error: Some(e.into()),
-            })?;
-
-        let epoch_settings_configuration = self
-            .configuration
-            .get_leader_aggregator_epoch_settings_configuration()?;
-        debug!(
-            logger,
-            "Handle discrepancies at startup of epoch settings store, will record epoch settings from the configuration for epoch {retrieval_epoch}";
-            "epoch_settings_configuration" => ?epoch_settings_configuration,
-        );
-        epoch_settings_store
-            .handle_discrepancies_at_startup(retrieval_epoch, &epoch_settings_configuration)
-            .await
-            .map_err(|e| DependenciesBuilderError::Initialization {
-                message: "can not create aggregator runner".to_string(),
-                error: Some(e),
-            })?;
 
         Ok(Arc::new(epoch_settings_store))
     }
 
-    /// Get a configured [EpochSettingsStorer].
+    /// Get a configured [EpochSettingsStore].
     pub async fn get_epoch_settings_store(&mut self) -> Result<Arc<EpochSettingsStore>> {
         get_dependency!(self.epoch_settings_store)
     }

mithril-aggregator/src/store/epoch_settings_storer.rs

@@ -1,12 +1,13 @@
+use anyhow::anyhow;
+use async_trait::async_trait;
 #[cfg(test)]
 use std::collections::HashMap;
-
-use async_trait::async_trait;
-use mithril_common::StdResult;
 #[cfg(test)]
 use tokio::sync::RwLock;
 
+use mithril_common::StdResult;
 use mithril_common::entities::{Epoch, ProtocolParameters};
+use mithril_protocol_config::model::MithrilNetworkConfiguration;
 
 use crate::{entities::AggregatorEpochSettings, services::EpochPruningTask};
 
@@ -43,14 +44,37 @@ pub trait EpochSettingsStorer:
     /// call and the epoch service call.
     async fn handle_discrepancies_at_startup(
         &self,
-        current_epoch: Epoch,
-        epoch_settings_configuration: &AggregatorEpochSettings,
+        network_configuration: &MithrilNetworkConfiguration,
     ) -> StdResult<()> {
-        for epoch_offset in 0..=3 {
-            let epoch = current_epoch + epoch_offset;
+        for (epoch, epoch_configuration) in [
+            (
+                network_configuration.epoch.offset_to_signer_retrieval_epoch()?,
+                &network_configuration.configuration_for_aggregation,
+            ),
+            (
+                network_configuration.epoch,
+                &network_configuration.configuration_for_next_aggregation,
+            ),
+            (
+                network_configuration.epoch.offset_to_recording_epoch(),
+                &network_configuration.configuration_for_registration,
+            ),
+        ] {
             if self.get_epoch_settings(epoch).await?.is_none() {
-                self.save_epoch_settings(epoch, epoch_settings_configuration.clone())
-                    .await?;
+                self.save_epoch_settings(
+                    epoch,
+                    AggregatorEpochSettings {
+                        protocol_parameters: epoch_configuration.protocol_parameters.clone(),
+                        cardano_transactions_signing_config: epoch_configuration
+                            .signed_entity_types_config
+                            .cardano_transactions
+                            .clone()
+                            .ok_or(anyhow!(
+                                "missing cardano transactions signing config for epoch {epoch}"
+                            ))?,
+                    },
+                )
+                .await?;
             }
         }
 
@@ -116,7 +140,8 @@ impl EpochPruningTask for FakeEpochSettingsStorer {
 
 #[cfg(test)]
 mod tests {
-    use mithril_common::entities::CardanoTransactionsSigningConfig;
+    use std::collections::BTreeSet;
+
     use mithril_common::test::double::Dummy;
 
     use super::*;
@@ -178,40 +203,43 @@ mod tests {
     #[tokio::test]
     async fn test_handle_discrepancies_at_startup_should_complete_at_least_four_epochs() {
         let epoch_settings = AggregatorEpochSettings::dummy();
-        let epoch_settings_new = AggregatorEpochSettings {
-            protocol_parameters: ProtocolParameters {
-                k: epoch_settings.protocol_parameters.k + 1,
-                ..epoch_settings.protocol_parameters
-            },
-            cardano_transactions_signing_config: CardanoTransactionsSigningConfig {
-                step: epoch_settings.cardano_transactions_signing_config.step + 1,
-                ..epoch_settings.cardano_transactions_signing_config
-            },
-        };
-        let epoch = Epoch(1);
-        let store = FakeEpochSettingsStorer::new(vec![
-            (epoch, epoch_settings.clone()),
-            (epoch + 1, epoch_settings.clone()),
-        ]);
+        let mut aggregation_epoch_settings = epoch_settings.clone();
+        aggregation_epoch_settings.protocol_parameters.k += 15;
 
+        let mut next_aggregation_epoch_settings = epoch_settings.clone();
+        next_aggregation_epoch_settings.protocol_parameters.k += 26;
+
+        let mut registration_epoch_settings = epoch_settings.clone();
+        registration_epoch_settings.protocol_parameters.k += 37;
+
+        let epoch = Epoch(5);
+        let store = FakeEpochSettingsStorer::new(vec![]);
         store
-            .handle_discrepancies_at_startup(epoch, &epoch_settings_new)
+            .handle_discrepancies_at_startup(&MithrilNetworkConfiguration {
+                epoch,
+                configuration_for_aggregation: aggregation_epoch_settings
+                    .clone()
+                    .into_network_configuration_for_epoch(BTreeSet::new()),
+                configuration_for_next_aggregation: next_aggregation_epoch_settings
+                    .clone()
+                    .into_network_configuration_for_epoch(BTreeSet::new()),
+                configuration_for_registration: registration_epoch_settings
+                    .clone()
+                    .into_network_configuration_for_epoch(BTreeSet::new()),
+            })
             .await
             .unwrap();
 
+        let epoch_settings_stored = store.get_epoch_settings(epoch - 1).await.unwrap();
+        assert_eq!(Some(aggregation_epoch_settings), epoch_settings_stored);
+
         let epoch_settings_stored = store.get_epoch_settings(epoch).await.unwrap();
-        assert_eq!(Some(epoch_settings.clone()), epoch_settings_stored);
+        assert_eq!(Some(next_aggregation_epoch_settings), epoch_settings_stored);
 
         let epoch_settings_stored = store.get_epoch_settings(epoch + 1).await.unwrap();
-        assert_eq!(Some(epoch_settings.clone()), epoch_settings_stored);
+        assert_eq!(Some(registration_epoch_settings), epoch_settings_stored);
 
         let epoch_settings_stored = store.get_epoch_settings(epoch + 2).await.unwrap();
-        assert_eq!(Some(epoch_settings_new.clone()), epoch_settings_stored);
-
-        let epoch_settings_stored = store.get_epoch_settings(epoch + 3).await.unwrap();
-        assert_eq!(Some(epoch_settings_new.clone()), epoch_settings_stored);
-
-        let epoch_settings_stored = store.get_epoch_settings(epoch + 4).await.unwrap();
         assert!(epoch_settings_stored.is_none());
     }
 }

mithril-aggregator/tests/create_certificate_follower.rs

@@ -117,6 +117,9 @@ async fn create_certificate_follower() {
             "create_certificate_follower",
         ),
         leader_aggregator_endpoint: Some(leader_aggregator_http_server.url().to_string()),
+        // Follower must retrieve parameters from the network configuration (today through the leader)
+        // so this parameters should not be read
+        protocol_parameters: None,
         ..leader_configuration
     };
     let mut follower_tester = RuntimeTester::build(start_time_point, follower_configuration).await;