Merge pull request #3436 from IntersectMBO/jpraynaud/handle-aggregator-epoch-zero

jpraynaud · web-flow · commit bc9d8c10fe07 · 2026-07-27T17:48:08.000+02:00
feat: accelerate aggregator startup by one epoch in e2e tests
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/internal/cardano-node/mithril-cardano-node-internal-database/Cargo.toml b/internal/cardano-node/mithril-cardano-node-internal-database/Cargo.toml
@@ -15,7 +15,7 @@ anyhow = { workspace = true }
 async-trait = { workspace = true }
 digest = { workspace = true }
 hex = { workspace = true }
-mithril-common = { path = "../../../mithril-common", version = "0.7.13" }
+mithril-common = { path = "../../../mithril-common", version = "0.7.14" }
 serde = { workspace = true }
 serde_json = { workspace = true }
 sha2 = "0.10.9"
diff --git a/internal/mithril-aggregator-client/Cargo.toml b/internal/mithril-aggregator-client/Cargo.toml
@@ -13,7 +13,7 @@ include = ["**/*.rs", "Cargo.toml", "README.md"]
 [dependencies]
 anyhow = { workspace = true }
 async-trait = { workspace = true }
-mithril-common = { path = "../../mithril-common", version = "0.7.13" }
+mithril-common = { path = "../../mithril-common", version = "0.7.14" }
 reqwest = { workspace = true }
 semver = { workspace = true }
 serde = { workspace = true }
diff --git a/internal/mithril-aggregator-discovery/Cargo.toml b/internal/mithril-aggregator-discovery/Cargo.toml
@@ -14,7 +14,7 @@ include = ["**/*.rs", "Cargo.toml", "README.md", ".gitignore"]
 anyhow = { workspace = true }
 async-trait = { workspace = true }
 mithril-aggregator-client = { path = "../mithril-aggregator-client", version = "0.2.4" }
-mithril-common = { path = "../../mithril-common", version = "0.7.13" }
+mithril-common = { path = "../../mithril-common", version = "0.7.14" }
 rand = { version = "0.10.2" }
 reqwest = { workspace = true }
 serde = { workspace = true }
diff --git a/internal/mithril-protocol-config/Cargo.toml b/internal/mithril-protocol-config/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "mithril-protocol-config"
-version = "0.1.11"
+version = "0.1.12"
 description = "Configuraton parameters for Mithril network"
 authors = { workspace = true }
 edition = { workspace = true }
diff --git a/internal/mithril-protocol-config/src/http.rs b/internal/mithril-protocol-config/src/http.rs
@@ -64,10 +64,7 @@ impl MithrilNetworkConfigurationProvider for HttpMithrilNetworkConfigurationProv
         &self,
         epoch: Epoch,
     ) -> StdResult<MithrilNetworkConfiguration> {
-        let aggregation_epoch =
-            epoch.offset_to_signer_retrieval_epoch().with_context(|| {
-                format!("MithrilNetworkConfigurationProvider could not compute aggregation epoch from epoch: {epoch}")
-            })?;
+        let aggregation_epoch = epoch.offset_to_signer_retrieval_epoch_saturating();
         let next_aggregation_epoch = epoch.offset_to_next_signer_retrieval_epoch();
         let registration_epoch = epoch.offset_to_next_signer_retrieval_epoch().next();
 
diff --git a/mithril-aggregator/Cargo.toml b/mithril-aggregator/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.9.17"
+version = "0.9.18"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }
diff --git a/mithril-aggregator/src/services/epoch_service.rs b/mithril-aggregator/src/services/epoch_service.rs
@@ -296,10 +296,7 @@ impl EpochService for MithrilEpochService {
 
         let mithril_era = self.era_checker.current_era();
 
-        let signer_retrieval_epoch =
-            epoch.offset_to_signer_retrieval_epoch().with_context(|| {
-                format!("EpochService could not compute signer retrieval epoch from epoch: {epoch}")
-            })?;
+        let signer_retrieval_epoch = epoch.offset_to_signer_retrieval_epoch_saturating();
         let next_signer_retrieval_epoch = epoch.offset_to_next_signer_retrieval_epoch();
         let signer_registration_epoch = epoch.offset_to_recording_epoch();
 
@@ -957,7 +954,7 @@ mod tests {
 
         async fn build(self) -> MithrilEpochService {
             let signer_retrieval_epoch =
-                self.current_epoch.offset_to_signer_retrieval_epoch().unwrap();
+                self.current_epoch.offset_to_signer_retrieval_epoch_saturating();
             let next_signer_retrieval_epoch =
                 self.current_epoch.offset_to_next_signer_retrieval_epoch();
 
@@ -1098,6 +1095,29 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn inform_epoch_at_epoch_zero_retrieves_signers_at_epoch_zero() {
+        let epoch = Epoch(0);
+        let epoch_fixture = MithrilFixtureBuilder::default().with_signers(3).build();
+
+        let mut service = EpochServiceBuilder::new(epoch, epoch_fixture.clone()).build().await;
+
+        service
+            .inform_epoch(epoch)
+            .await
+            .expect("inform_epoch should not fail at epoch zero");
+
+        assert_eq!(epoch, service.epoch_of_current_data().unwrap());
+        assert_eq!(
+            epoch_fixture.signers(),
+            service.current_signers().unwrap().clone()
+        );
+        assert_eq!(
+            epoch_fixture.signers(),
+            service.next_signers().unwrap().clone()
+        );
+    }
+
     #[tokio::test]
     async fn inform_epoch_get_signed_entity_config_from_its_dependencies_and_store() {
         let epoch = Epoch(5);
diff --git a/mithril-aggregator/src/services/network_configuration_provider.rs b/mithril-aggregator/src/services/network_configuration_provider.rs
@@ -1,7 +1,6 @@
 use std::collections::BTreeSet;
 use std::sync::Arc;
 
-use anyhow::Context;
 use async_trait::async_trait;
 use slog::{Logger, warn};
 
@@ -96,10 +95,7 @@ impl MithrilNetworkConfigurationProvider for LocalMithrilNetworkConfigurationPro
         &self,
         epoch: Epoch,
     ) -> StdResult<MithrilNetworkConfiguration> {
-        let aggregation_epoch =
-            epoch.offset_to_signer_retrieval_epoch().with_context(|| {
-                format!("MithrilNetworkConfigurationProvider could not compute aggregation epoch from epoch: {epoch}")
-            })?;
+        let aggregation_epoch = epoch.offset_to_signer_retrieval_epoch_saturating();
         let next_aggregation_epoch = epoch.offset_to_next_signer_retrieval_epoch();
         let registration_epoch = epoch.offset_to_next_signer_retrieval_epoch().next();
 
diff --git a/mithril-aggregator/src/store/epoch_settings_storer.rs b/mithril-aggregator/src/store/epoch_settings_storer.rs
@@ -47,7 +47,9 @@ pub trait EpochSettingsStorer:
     ) -> StdResult<()> {
         for (epoch, epoch_configuration) in [
             (
-                network_configuration.epoch.offset_to_signer_retrieval_epoch()?,
+                network_configuration
+                    .epoch
+                    .offset_to_signer_retrieval_epoch_saturating(),
                 &network_configuration.configuration_for_aggregation,
             ),
             (
@@ -242,4 +244,43 @@ mod tests {
         let epoch_settings_stored = store.get_epoch_settings(epoch + 2).await.unwrap();
         assert!(epoch_settings_stored.is_none());
     }
+
+    #[tokio::test]
+    async fn test_handle_discrepancies_at_startup_at_epoch_zero_clamps_the_aggregation_epoch() {
+        let epoch_settings = AggregatorEpochSettings::dummy();
+        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(0);
+        let store = FakeEpochSettingsStorer::new(vec![]);
+        store
+            .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
+                    .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(0)).await.unwrap();
+        assert_eq!(Some(aggregation_epoch_settings), epoch_settings_stored);
+
+        let epoch_settings_stored = store.get_epoch_settings(Epoch(1)).await.unwrap();
+        assert_eq!(Some(registration_epoch_settings), epoch_settings_stored);
+
+        let epoch_settings_stored = store.get_epoch_settings(Epoch(2)).await.unwrap();
+        assert!(epoch_settings_stored.is_none());
+    }
 }
diff --git a/mithril-client/Cargo.toml b/mithril-client/Cargo.toml
@@ -64,7 +64,7 @@ flate2 = { version = "1.1.9", optional = true }
 flume = { version = "0.12.0", optional = true }
 futures = "0.3.32"
 mithril-aggregator-client = { path = "../internal/mithril-aggregator-client", version = "0.2.4" }
-mithril-common = { path = "../mithril-common", version = "0.7.13", default-features = false }
+mithril-common = { path = "../mithril-common", version = "0.7.14", default-features = false }
 reqwest = { workspace = true, default-features = false, features = ["charset", "http2", "stream", "system-proxy"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
diff --git a/mithril-common/Cargo.toml b/mithril-common/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "mithril-common"
-version = "0.7.13"
+version = "0.7.14"
 description = "Common types, interfaces, and utilities for Mithril nodes."
 authors = { workspace = true }
 edition = { workspace = true }
diff --git a/mithril-common/src/entities/epoch.rs b/mithril-common/src/entities/epoch.rs
@@ -48,20 +48,29 @@ impl Epoch {
 
     /// Computes a new Epoch by applying an epoch offset.
     ///
-    /// Will fail if the computed epoch is negative.
+    /// Will fail if overflow occurred (see [u64] for bounds).
     pub fn offset_by(&self, epoch_offset: i64) -> Result<Self, EpochError> {
-        let epoch_new = self.0 as i64 + epoch_offset;
-        if epoch_new < 0 {
-            return Err(EpochError::EpochOffset(self.0, epoch_offset));
-        }
-        Ok(Epoch(epoch_new as u64))
+        self.0
+            .checked_add_signed(epoch_offset)
+            .map(Epoch)
+            .ok_or(EpochError::EpochOffset(self.0, epoch_offset))
     }
 
     /// Apply the [retrieval offset][Self::SIGNER_RETRIEVAL_OFFSET] to this epoch
     pub fn offset_to_signer_retrieval_epoch(&self) -> Result<Self, EpochError> {
         self.offset_by(Self::SIGNER_RETRIEVAL_OFFSET)
     }
 
+    /// Apply the [retrieval offset][Self::SIGNER_RETRIEVAL_OFFSET] to this epoch, saturating at
+    /// epoch zero.
+    ///
+    /// Unlike [offset_to_signer_retrieval_epoch][Self::offset_to_signer_retrieval_epoch], this
+    /// returns epoch zero instead of failing when the offset would yield a negative epoch
+    /// (i.e. at epoch zero itself).
+    pub fn offset_to_signer_retrieval_epoch_saturating(&self) -> Self {
+        Epoch(self.0.saturating_add_signed(Self::SIGNER_RETRIEVAL_OFFSET))
+    }
+
     /// Apply the [next signer retrieval offset][Self::NEXT_SIGNER_RETRIEVAL_OFFSET] to this epoch
     pub fn offset_to_next_signer_retrieval_epoch(&self) -> Self {
         *self + Self::NEXT_SIGNER_RETRIEVAL_OFFSET
@@ -302,12 +311,45 @@ mod tests {
         assert_eq!(Epoch(0), Epoch(1) - 5_u64);
     }
 
+    #[test]
+    fn offset_by_covers_the_whole_epoch_range() {
+        assert_eq!(Epoch(2), Epoch(3).offset_by(-1).unwrap());
+        assert_eq!(Epoch(4), Epoch(3).offset_by(1).unwrap());
+        assert_eq!(Epoch(u64::MAX - 1), Epoch(u64::MAX).offset_by(-1).unwrap());
+    }
+
+    #[test]
+    fn offset_by_fails_when_overflow_occurred() {
+        assert!(Epoch(0).offset_by(-1).is_err());
+        assert!(Epoch(u64::MAX).offset_by(1).is_err());
+    }
+
     #[test]
     fn test_previous() {
         assert_eq!(Epoch(2), Epoch(3).previous().unwrap());
         assert!(Epoch(0).previous().is_err());
     }
 
+    #[test]
+    fn offset_to_signer_retrieval_epoch_saturating_clamps_at_epoch_zero() {
+        assert_eq!(
+            Epoch(2),
+            Epoch(3).offset_to_signer_retrieval_epoch_saturating()
+        );
+        assert_eq!(
+            Epoch(0),
+            Epoch(1).offset_to_signer_retrieval_epoch_saturating()
+        );
+        assert_eq!(
+            Epoch(0),
+            Epoch(0).offset_to_signer_retrieval_epoch_saturating()
+        );
+        assert_eq!(
+            Epoch(u64::MAX - 1),
+            Epoch(u64::MAX).offset_to_signer_retrieval_epoch_saturating()
+        );
+    }
+
     #[test]
     fn test_next() {
         assert_eq!(Epoch(4), Epoch(3).next());
diff --git a/mithril-test-lab/mithril-end-to-end/Cargo.toml b/mithril-test-lab/mithril-end-to-end/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "mithril-end-to-end"
-version = "0.5.6"
+version = "0.5.7"
 authors = { workspace = true }
 edition = { workspace = true }
 documentation = { workspace = true }
diff --git a/mithril-test-lab/mithril-end-to-end/src/mithril/infrastructure.rs b/mithril-test-lab/mithril-end-to-end/src/mithril/infrastructure.rs
@@ -123,6 +123,14 @@ impl MithrilInfrastructure {
                 .await?;
 
         Self::register_startup_era(&toolkit, &leader_aggregator, config).await?;
+        toolkit
+            .wait
+            .for_aggregator_at_target_epoch(
+                &leader_aggregator,
+                Epoch(1),
+                "minimal epoch for the aggregator to handle startup discrepancies".to_string(),
+            )
+            .await?;
         leader_aggregator.serve().await?;
 
         let follower_aggregator_endpoints = follower_aggregators
