Merge pull request #964 from input-output-hk/jpraynaud/953-fix-aggreg…

…ator-detect-open-message

Fix retrieve open message from correct epoch

mithril-aggregator/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.3.25"
+version = "0.3.26"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-aggregator/src/certifier_service.rs

@@ -126,7 +126,6 @@ pub struct MithrilCertifierService {
     certificate_verifier: Arc<dyn CertificateVerifier>,
     genesis_verifier: Arc<ProtocolGenesisVerifier>,
     multi_signer: Arc<RwLock<dyn MultiSigner>>,
-    current_epoch: Arc<RwLock<Epoch>>,
     _logger: Logger,
 }
 
@@ -140,7 +139,6 @@ impl MithrilCertifierService {
         certificate_verifier: Arc<dyn CertificateVerifier>,
         genesis_verifier: Arc<ProtocolGenesisVerifier>,
         multi_signer: Arc<RwLock<dyn MultiSigner>>,
-        current_epoch: Epoch,
         logger: Logger,
     ) -> Self {
         Self {
@@ -150,7 +148,6 @@ impl MithrilCertifierService {
             multi_signer,
             certificate_verifier,
             genesis_verifier,
-            current_epoch: Arc::new(RwLock::new(current_epoch)),
             _logger: logger,
         }
     }
@@ -173,19 +170,8 @@ impl MithrilCertifierService {
 impl CertifierService for MithrilCertifierService {
     async fn inform_epoch(&self, epoch: Epoch) -> StdResult<()> {
         debug!("CertifierService::inform_epoch(epoch: {epoch:?})");
-        let mut current_epoch = self.current_epoch.write().await;
-
-        if epoch <= *current_epoch {
-            debug!("CertifierService::inform_epoch: given epoch ({epoch:?}) is older than current epoch ({}), ignoring", *current_epoch);
-
-            return Ok(());
-        }
-        let nb = self
-            .open_message_repository
-            .clean_epoch(*current_epoch)
-            .await?;
-        info!("MithrilCertifierService: Got a new Epoch: {epoch:?}. Cleaned {nb} open messages along with their single signatures.");
-        *current_epoch = epoch;
+        let nb = self.open_message_repository.clean_epoch(epoch).await?;
+        info!("MithrilCertifierService: Informed of a new Epoch: {epoch:?}. Cleaned {nb} open messages along with their single signatures.");
 
         Ok(())
     }
@@ -231,10 +217,13 @@ impl CertifierService for MithrilCertifierService {
         protocol_message: &ProtocolMessage,
     ) -> StdResult<OpenMessage> {
         debug!("CertifierService::create_open_message(signed_entity_type: {signed_entity_type:?}, protocol_message: {protocol_message:?})");
-        let current_epoch = self.current_epoch.read().await;
         let open_message = self
             .open_message_repository
-            .create_open_message(*current_epoch, signed_entity_type, protocol_message)
+            .create_open_message(
+                signed_entity_type.get_epoch(),
+                signed_entity_type,
+                protocol_message,
+            )
             .await?;
         info!("CertifierService::create_open_message: created open message for {signed_entity_type:?}");
         debug!(
@@ -414,43 +403,19 @@ mod tests {
             certificate_verifier,
             genesis_verifier,
             multi_signer,
-            Epoch(0),
             logger,
         )
     }
 
     #[tokio::test]
-    async fn should_not_clean_epoch_when_inform_same_epoch() {
+    async fn should_clean_epoch_when_inform_epoch() {
         let beacon = Beacon::new("devnet".to_string(), 1, 1);
         let signed_entity_type = SignedEntityType::CardanoImmutableFilesFull(beacon.clone());
         let protocol_message = ProtocolMessage::new();
         let epoch = beacon.epoch;
         let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
         let fixture = MithrilFixtureBuilder::default().with_signers(5).build();
-        let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
-        certifier_service
-            .create_open_message(&signed_entity_type, &protocol_message)
-            .await
-            .unwrap();
-        certifier_service.inform_epoch(epoch).await.unwrap();
-        let open_message = certifier_service
-            .get_open_message(&signed_entity_type)
-            .await
-            .unwrap();
-        assert!(open_message.is_some());
-    }
-
-    #[tokio::test]
-    async fn should_clean_epoch_when_inform_new_epoch() {
-        let beacon = Beacon::new("devnet".to_string(), 1, 1);
-        let signed_entity_type = SignedEntityType::CardanoImmutableFilesFull(beacon.clone());
-        let protocol_message = ProtocolMessage::new();
-        let epoch = beacon.epoch;
-        let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
-        let fixture = MithrilFixtureBuilder::default().with_signers(5).build();
-        let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
+        let certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
         certifier_service
             .create_open_message(&signed_entity_type, &protocol_message)
             .await
@@ -468,11 +433,9 @@ mod tests {
         let beacon = Beacon::new("devnet".to_string(), 3, 1);
         let signed_entity_type = SignedEntityType::CardanoImmutableFilesFull(beacon.clone());
         let protocol_message = ProtocolMessage::new();
-        let epoch = beacon.epoch;
         let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
         let fixture = MithrilFixtureBuilder::default().with_signers(1).build();
-        let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
+        let certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
         certifier_service
             .multi_signer
             .write()
@@ -509,11 +472,9 @@ mod tests {
         let beacon = Beacon::new("devnet".to_string(), 3, 1);
         let signed_entity_type = SignedEntityType::CardanoImmutableFilesFull(beacon.clone());
         let mut protocol_message = ProtocolMessage::new();
-        let epoch = beacon.epoch;
         let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
         let fixture = MithrilFixtureBuilder::default().with_signers(1).build();
-        let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
+        let certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
         certifier_service
             .multi_signer
             .write()
@@ -549,11 +510,9 @@ mod tests {
         let beacon = Beacon::new("devnet".to_string(), 3, 1);
         let signed_entity_type = SignedEntityType::CardanoImmutableFilesFull(beacon.clone());
         let protocol_message = ProtocolMessage::new();
-        let epoch = beacon.epoch;
         let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
         let fixture = MithrilFixtureBuilder::default().with_signers(1).build();
-        let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
+        let certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
         let mut open_message = certifier_service
             .open_message_repository
             .create_open_message(beacon.epoch, &signed_entity_type, &protocol_message)
@@ -583,11 +542,9 @@ mod tests {
         let beacon = Beacon::new("devnet".to_string(), 3, 1);
         let signed_entity_type = SignedEntityType::CardanoImmutableFilesFull(beacon.clone());
         let protocol_message = ProtocolMessage::new();
-        let epoch = beacon.epoch;
         let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
         let fixture = MithrilFixtureBuilder::default().with_signers(3).build();
-        let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
+        let certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
         certifier_service
             .multi_signer
             .write()
@@ -664,11 +621,9 @@ mod tests {
     async fn should_not_create_certificate_for_open_message_not_created() {
         let beacon = Beacon::new("devnet".to_string(), 1, 1);
         let signed_entity_type = SignedEntityType::CardanoImmutableFilesFull(beacon.clone());
-        let epoch = beacon.epoch;
         let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
         let fixture = MithrilFixtureBuilder::default().with_signers(5).build();
-        let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
+        let certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
         certifier_service
             .create_certificate(&signed_entity_type)
             .await
@@ -683,8 +638,7 @@ mod tests {
         let epoch = beacon.epoch;
         let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
         let fixture = MithrilFixtureBuilder::default().with_signers(5).build();
-        let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
+        let certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
         certifier_service
             .open_message_repository
             .create_open_message(epoch, &signed_entity_type, &protocol_message)
@@ -705,11 +659,9 @@ mod tests {
         let beacon = Beacon::new("devnet".to_string(), 1, 1);
         let signed_entity_type = SignedEntityType::CardanoImmutableFilesFull(beacon.clone());
         let protocol_message = ProtocolMessage::new();
-        let epoch = beacon.epoch;
         let epochs_with_signers = (1..=5).map(Epoch).collect::<Vec<_>>();
         let fixture = MithrilFixtureBuilder::default().with_signers(5).build();
         let mut certifier_service = setup_certifier_service(&fixture, &epochs_with_signers).await;
-        certifier_service.current_epoch = Arc::new(RwLock::new(epoch));
         certifier_service.multi_signer = Arc::new(RwLock::new(mock_multi_signer));
         certifier_service
             .create_open_message(&signed_entity_type, &protocol_message)

mithril-aggregator/src/database/provider/open_message.rs

@@ -421,6 +421,13 @@ impl<'client> OpenMessageWithSingleSignaturesProvider<'client> {
         Self { connection }
     }
 
+    fn get_epoch_condition(&self, epoch: Epoch) -> WhereCondition {
+        WhereCondition::new(
+            "epoch_setting_id = ?*",
+            vec![Value::Integer(epoch.0 as i64)],
+        )
+    }
+
     fn get_signed_entity_type_condition(
         &self,
         signed_entity_type: &SignedEntityType,
@@ -493,14 +500,16 @@ impl OpenMessageRepository {
         Ok(messages.next())
     }
 
-    /// Return an open message with its associated single signatures if any.
+    /// Return an open message with its associated single signatures for the given Epoch and [SignedEntityType].
     pub async fn get_open_message_with_single_signatures(
         &self,
         signed_entity_type: &SignedEntityType,
     ) -> StdResult<Option<OpenMessageWithSingleSignaturesRecord>> {
         let lock = self.connection.lock().await;
         let provider = OpenMessageWithSingleSignaturesProvider::new(&lock);
-        let filters = provider.get_signed_entity_type_condition(signed_entity_type);
+        let filters = provider
+            .get_epoch_condition(signed_entity_type.get_epoch())
+            .and_where(provider.get_signed_entity_type_condition(signed_entity_type));
         let mut messages = provider.find(filters)?;
 
         Ok(messages.next())

mithril-aggregator/src/dependency_injection/builder.rs

@@ -1104,7 +1104,7 @@ impl DependenciesBuilder {
         ));
         let certificate_verifier = self.get_certificate_verifier().await?;
         let genesis_verifier = self.get_genesis_verifier().await?;
-        let multisigner = self.get_multi_signer().await?;
+        let multi_signer = self.get_multi_signer().await?;
         let logger = self.get_logger().await?;
 
         Ok(Arc::new(MithrilCertifierService::new(
@@ -1113,8 +1113,7 @@ impl DependenciesBuilder {
             certificate_repository,
             certificate_verifier,
             genesis_verifier,
-            multisigner,
-            Epoch(0),
+            multi_signer,
             logger,
         )))
     }

mithril-signer/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-signer"
-version = "0.2.47"
+version = "0.2.48"
 description = "A Mithril Signer"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-signer/src/runtime/runner.rs

@@ -366,11 +366,7 @@ impl Runner for SignerRunner {
             .await?;
 
         // 2 set the next signers keys and stakes in the message
-        let epoch = match signed_entity_type {
-            SignedEntityType::MithrilStakeDistribution(epoch) => epoch,
-            SignedEntityType::CardanoImmutableFilesFull(beacon) => &beacon.epoch,
-            SignedEntityType::CardanoStakeDistribution(epoch) => epoch,
-        };
+        let epoch = signed_entity_type.get_epoch();
         let next_signer_retrieval_epoch = epoch.offset_to_next_signer_retrieval_epoch();
         let next_protocol_initializer = self
             .services

mithril-test-lab/mithril-end-to-end/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-end-to-end"
-version = "0.1.19"
+version = "0.1.20"
 authors = { workspace = true }
 edition = { workspace = true }
 documentation = { workspace = true }

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

@@ -255,7 +255,7 @@ async fn update_protocol_parameters(aggregator: &mut Aggregator) -> Result<(), S
     let protocol_parameters_new = ProtocolParameters {
         k: 150,
         m: 200,
-        phi_f: 0.80,
+        phi_f: 0.85,
     };
     info!(
         "> updating protocol parameters to {:?}...",