Merge pull request #1129 from input-output-hk/djo/1097/add-signers-re…

…gistered-route-to-aggregator

Add signers registered route to aggregator

mithril-aggregator/Cargo.toml

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

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

@@ -8,7 +8,7 @@ use mithril_common::{
     crypto_helper::KESPeriod,
     entities::{
         Epoch, HexEncodedOpCert, HexEncodedVerificationKey, HexEncodedVerificationKeySignature,
-        PartyId, Signer, SignerWithStake, Stake,
+        PartyId, Signer, SignerWithStake, Stake, StakeDistribution,
     },
     sqlite::{
         EntityCursor, HydrationError, Projection, Provider, SourceAlias, SqLiteEntity,
@@ -458,6 +458,26 @@ impl VerificationKeyStorer for SignerRegistrationStore {
 
         Ok(())
     }
+
+    async fn get_stake_distribution_for_epoch(
+        &self,
+        epoch: Epoch,
+    ) -> Result<Option<StakeDistribution>, StoreError> {
+        let connection = &*self.connection.lock().await;
+        let provider = SignerRegistrationRecordProvider::new(connection);
+        let cursor = provider
+            .get_by_epoch(&epoch)
+            .map_err(|e| AdapterError::GeneralError(format!("{e}")))?;
+
+        let stake_distribution = StakeDistribution::from_iter(
+            cursor.map(|r| (r.signer_id, r.stake.unwrap_or_default())),
+        );
+
+        match stake_distribution.is_empty() {
+            true => Ok(None),
+            false => Ok(Some(stake_distribution)),
+        }
+    }
 }
 
 #[cfg(test)]

mithril-aggregator/src/entities/mod.rs

@@ -2,5 +2,9 @@
 //!
 //! This module provide domain entities for the services & state machine.
 mod open_message;
+mod signer_registration_message;
 
 pub use open_message::OpenMessage;
+pub use signer_registration_message::{
+    SignerRegistrationsListItemMessage, SignerRegistrationsMessage,
+};

mithril-aggregator/src/entities/signer_registration_message.rs

@@ -0,0 +1,41 @@
+use mithril_common::entities::{Epoch, PartyId, Stake, StakeDistribution};
+use serde::{Deserialize, Serialize};
+
+/// Message structure of signer registrations for an epoch.
+#[derive(Clone, Debug, PartialEq, Eq, Default, Serialize, Deserialize)]
+pub struct SignerRegistrationsMessage {
+    /// The epoch at which the registration was sent.
+    pub registered_at: Epoch,
+
+    /// The epoch at which the registration was able to send signatures.
+    pub signing_at: Epoch,
+
+    /// The signer registrations
+    pub registrations: Vec<SignerRegistrationsListItemMessage>,
+}
+
+/// Message structure of a signer registration
+#[derive(Clone, Debug, PartialEq, Eq, Default, Serialize, Deserialize)]
+pub struct SignerRegistrationsListItemMessage {
+    /// The registered signer party id
+    pub party_id: PartyId,
+
+    /// The registered signer stake
+    pub stake: Stake,
+}
+
+impl SignerRegistrationsMessage {
+    /// Build a [SignerRegistrationsMessage] from a [stake distribution][StakeDistribution].
+    pub fn new(registered_at: Epoch, stake_distribution: StakeDistribution) -> Self {
+        let registrations: Vec<SignerRegistrationsListItemMessage> = stake_distribution
+            .into_iter()
+            .map(|(party_id, stake)| SignerRegistrationsListItemMessage { party_id, stake })
+            .collect();
+
+        Self {
+            registered_at,
+            signing_at: registered_at.offset_to_signer_signing_offset(),
+            registrations,
+        }
+    }
+}

mithril-aggregator/src/http_server/routes/middlewares.rs

@@ -4,7 +4,7 @@ use crate::{
     services::CertifierService,
     services::{SignedEntityService, TickerService},
     CertificatePendingStore, Configuration, DependencyContainer, ProtocolParametersStorer,
-    SignerRegisterer,
+    SignerRegisterer, VerificationKeyStorer,
 };
 
 use mithril_common::BeaconProvider;
@@ -81,3 +81,10 @@ pub fn with_signed_entity_service(
 ) -> impl Filter<Extract = (Arc<dyn SignedEntityService>,), Error = Infallible> + Clone {
     warp::any().map(move || dependency_manager.signed_entity_service.clone())
 }
+
+/// With verification key store
+pub fn with_verification_key_store(
+    dependency_manager: Arc<DependencyContainer>,
+) -> impl Filter<Extract = (Arc<dyn VerificationKeyStorer>,), Error = Infallible> + Clone {
+    warp::any().map(move || dependency_manager.verification_key_store.clone())
+}

mithril-aggregator/src/http_server/routes/signer_routes.rs

@@ -8,7 +8,7 @@ const MITHRIL_SIGNER_VERSION_HEADER: &str = "signer-node-version";
 pub fn routes(
     dependency_manager: Arc<DependencyContainer>,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
-    register_signer(dependency_manager)
+    register_signer(dependency_manager.clone()).or(registered_signers(dependency_manager))
 }
 
 /// POST /register-signer
@@ -31,10 +31,22 @@ fn register_signer(
         .and_then(handlers::register_signer)
 }
 
+/// Get /signers/registered/:epoch
+fn registered_signers(
+    dependency_manager: Arc<DependencyContainer>,
+) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
+    warp::path!("signers" / "registered" / String)
+        .and(warp::get())
+        .and(middlewares::with_verification_key_store(dependency_manager))
+        .and_then(handlers::registered_signers)
+}
+
 mod handlers {
+    use crate::entities::SignerRegistrationsMessage;
     use crate::event_store::{EventMessage, TransmitterService};
-    use crate::FromRegisterSignerAdapter;
     use crate::{http_server::routes::reply, SignerRegisterer, SignerRegistrationError};
+    use crate::{FromRegisterSignerAdapter, VerificationKeyStorer};
+    use mithril_common::entities::Epoch;
     use mithril_common::messages::{RegisterSignerMessage, TryFromMessageAdapter};
     use mithril_common::BeaconProvider;
     use slog_scope::{debug, warn};
@@ -139,21 +151,68 @@ mod handlers {
             }
         }
     }
+
+    /// Get Registered Signers for a given epoch
+    pub async fn registered_signers(
+        registered_at: String,
+        verification_key_store: Arc<dyn VerificationKeyStorer>,
+    ) -> Result<impl warp::Reply, Infallible> {
+        debug!("⇄ HTTP SERVER: signers/registered/{:?}", registered_at);
+
+        let registered_at = match registered_at.parse::<u64>() {
+            Ok(epoch) => Epoch(epoch),
+            Err(err) => {
+                warn!("registered_signers::invalid_epoch"; "error" => ?err);
+                return Ok(reply::bad_request(
+                    "invalid_epoch".to_string(),
+                    err.to_string(),
+                ));
+            }
+        };
+
+        // The given epoch is the epoch at which the signer registered, the store works on
+        // the recording epoch so we need to offset.
+        match verification_key_store
+            .get_stake_distribution_for_epoch(registered_at.offset_to_recording_epoch())
+            .await
+        {
+            Ok(Some(stake_distribution)) => {
+                let message = SignerRegistrationsMessage::new(registered_at, stake_distribution);
+                Ok(reply::json(&message, StatusCode::OK))
+            }
+            Ok(None) => {
+                warn!("registered_signers::not_found");
+                Ok(reply::empty(StatusCode::NOT_FOUND))
+            }
+            Err(err) => {
+                warn!("registered_signers::error"; "error" => ?err);
+                Ok(reply::internal_server_error(err.to_string()))
+            }
+        }
+    }
 }
 
 #[cfg(test)]
 mod tests {
-    use mithril_common::crypto_helper::ProtocolRegistrationError;
-    use mithril_common::messages::RegisterSignerMessage;
-    use mithril_common::test_utils::apispec::APISpec;
-    use mithril_common::test_utils::fake_data;
-    use warp::http::Method;
-    use warp::test::request;
+    use mithril_common::entities::Epoch;
+    use mithril_common::{
+        crypto_helper::ProtocolRegistrationError,
+        entities::StakeDistribution,
+        messages::RegisterSignerMessage,
+        store::adapter::AdapterError,
+        test_utils::{apispec::APISpec, fake_data},
+    };
+    use mockall::predicate::eq;
+    use serde_json::Value::Null;
+    use warp::{http::Method, test::request};
+
+    use crate::{
+        http_server::SERVER_BASE_PATH, initialize_dependencies,
+        signer_registerer::MockSignerRegisterer, store::MockVerificationKeyStorer,
+        SignerRegistrationError,
+    };
 
     use super::*;
-    use crate::http_server::SERVER_BASE_PATH;
-    use crate::signer_registerer::MockSignerRegisterer;
-    use crate::{initialize_dependencies, SignerRegistrationError};
 
     fn setup_router(
         dependency_manager: Arc<DependencyContainer>,
@@ -316,4 +375,128 @@ mod tests {
             &response,
         );
     }
+
+    #[tokio::test]
+    async fn test_registered_signers_get_offset_given_epoch_to_registration_epoch() {
+        let asked_epoch = Epoch(1);
+        let expected_retrieval_epoch = asked_epoch.offset_to_recording_epoch();
+        let stake_distribution = StakeDistribution::from_iter(
+            fake_data::signers_with_stakes(3)
+                .into_iter()
+                .map(|s| (s.party_id, s.stake)),
+        );
+        let mut mock_verification_key_store = MockVerificationKeyStorer::new();
+        mock_verification_key_store
+            .expect_get_stake_distribution_for_epoch()
+            .with(eq(expected_retrieval_epoch))
+            .return_once(|_| Ok(Some(stake_distribution)))
+            .once();
+        let mut dependency_manager = initialize_dependencies().await;
+        dependency_manager.verification_key_store = Arc::new(mock_verification_key_store);
+
+        let method = Method::GET.as_str();
+        let base_path = "/signers/registered";
+
+        let response = request()
+            .method(method)
+            .path(&format!("/{SERVER_BASE_PATH}{base_path}/{}", asked_epoch))
+            .reply(&setup_router(Arc::new(dependency_manager)))
+            .await;
+
+        assert!(
+            response.status().is_success(),
+            "expected the response to succeed, was: {response:#?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_registered_signers_get_ok() {
+        let stake_distribution = StakeDistribution::from_iter(
+            fake_data::signers_with_stakes(3)
+                .into_iter()
+                .map(|s| (s.party_id, s.stake)),
+        );
+        let mut mock_verification_key_store = MockVerificationKeyStorer::new();
+        mock_verification_key_store
+            .expect_get_stake_distribution_for_epoch()
+            .return_once(|_| Ok(Some(stake_distribution)))
+            .once();
+        let mut dependency_manager = initialize_dependencies().await;
+        dependency_manager.verification_key_store = Arc::new(mock_verification_key_store);
+
+        let base_path = "/signers/registered";
+        let method = Method::GET.as_str();
+
+        let response = request()
+            .method(method)
+            .path(&format!("/{SERVER_BASE_PATH}{base_path}/1"))
+            .reply(&setup_router(Arc::new(dependency_manager)))
+            .await;
+
+        APISpec::verify_conformity(
+            APISpec::get_all_spec_files(),
+            method,
+            &format!("{base_path}/{{epoch}}"),
+            "application/json",
+            &Null,
+            &response,
+        );
+    }
+
+    #[tokio::test]
+    async fn test_registered_signers_get_ok_noregistration() {
+        let mut mock_verification_key_store = MockVerificationKeyStorer::new();
+        mock_verification_key_store
+            .expect_get_stake_distribution_for_epoch()
+            .return_once(|_| Ok(None))
+            .once();
+        let mut dependency_manager = initialize_dependencies().await;
+        dependency_manager.verification_key_store = Arc::new(mock_verification_key_store);
+
+        let method = Method::GET.as_str();
+        let base_path = "/signers/registered";
+
+        let response = request()
+            .method(method)
+            .path(&format!("/{SERVER_BASE_PATH}{base_path}/3"))
+            .reply(&setup_router(Arc::new(dependency_manager)))
+            .await;
+
+        APISpec::verify_conformity(
+            APISpec::get_all_spec_files(),
+            method,
+            &format!("{base_path}/{{epoch}}"),
+            "application/json",
+            &Null,
+            &response,
+        );
+    }
+
+    #[tokio::test]
+    async fn test_registered_signers_get_ko() {
+        let mut mock_verification_key_store = MockVerificationKeyStorer::new();
+        mock_verification_key_store
+            .expect_get_stake_distribution_for_epoch()
+            .return_once(|_| Err(AdapterError::GeneralError("invalid query".to_string()).into()));
+        let mut dependency_manager = initialize_dependencies().await;
+        dependency_manager.verification_key_store = Arc::new(mock_verification_key_store);
+
+        let method = Method::GET.as_str();
+        let base_path = "/signers/registered";
+
+        let response = request()
+            .method(method)
+            .path(&format!("/{SERVER_BASE_PATH}{base_path}/1"))
+            .reply(&setup_router(Arc::new(dependency_manager)))
+            .await;
+
+        APISpec::verify_conformity(
+            APISpec::get_all_spec_files(),
+            method,
+            &format!("{base_path}/{{epoch}}"),
+            "application/json",
+            &Null,
+            &response,
+        );
+    }
 }

mithril-aggregator/src/store/mod.rs

@@ -10,3 +10,5 @@ pub use verification_key_store::{VerificationKeyStore, VerificationKeyStorer};
 pub use verification_key_store::test_suite as verification_key_store_test_suite;
 #[cfg(test)]
 pub(crate) use verification_key_store::test_verification_key_storer;
+#[cfg(test)]
+pub use verification_key_store::MockVerificationKeyStorer;