Create multi signature in Mithril Aggregator runtime

mithril-aggregator/.gitignore

@@ -1,4 +1,5 @@
 target/
 mithril-aggregator
 .DS_Store
-*.tar.gz
+*.tar.gz
+snapshots.json

mithril-aggregator/src/beacon_store.rs

@@ -1,21 +1,29 @@
 use async_trait::async_trait;
+use thiserror::Error;
+
 use mithril_common::entities::Beacon;
 
 #[cfg(test)]
 use mockall::automock;
 
+#[derive(Error, Debug)]
+pub enum BeaconStoreError {
+    #[error("Generic error")]
+    GenericError(),
+}
+
 /// BeaconStore represents a beacon store interactor
 #[cfg_attr(test, automock)]
 #[async_trait]
 pub trait BeaconStore: Sync + Send {
     /// Get the current beacon
-    async fn get_current_beacon(&self) -> Result<Option<Beacon>, String>;
+    async fn get_current_beacon(&self) -> Result<Option<Beacon>, BeaconStoreError>;
 
     /// Set the current beacon
-    async fn set_current_beacon(&mut self, beacon: Beacon) -> Result<(), String>;
+    async fn set_current_beacon(&mut self, beacon: Beacon) -> Result<(), BeaconStoreError>;
 
     /// Reset the current beacon
-    async fn reset_current_beacon(&mut self) -> Result<(), String>;
+    async fn reset_current_beacon(&mut self) -> Result<(), BeaconStoreError>;
 }
 
 /// MemoryBeaconStore is in memory [`BeaconStore`]
@@ -34,16 +42,16 @@ impl Default for MemoryBeaconStore {
 
 #[async_trait]
 impl BeaconStore for MemoryBeaconStore {
-    async fn get_current_beacon(&self) -> Result<Option<Beacon>, String> {
+    async fn get_current_beacon(&self) -> Result<Option<Beacon>, BeaconStoreError> {
         Ok(self.current_beacon.clone())
     }
 
-    async fn set_current_beacon(&mut self, beacon: Beacon) -> Result<(), String> {
+    async fn set_current_beacon(&mut self, beacon: Beacon) -> Result<(), BeaconStoreError> {
         self.current_beacon = Some(beacon);
         Ok(())
     }
 
-    async fn reset_current_beacon(&mut self) -> Result<(), String> {
+    async fn reset_current_beacon(&mut self) -> Result<(), BeaconStoreError> {
         self.current_beacon = None;
         Ok(())
     }

mithril-aggregator/src/lib.rs

@@ -10,11 +10,11 @@ mod snapshotter;
 pub use crate::entities::Config;
 pub use crate::http_server::Server;
 pub use crate::multi_signer::{
-    key_decode_hex, MultiSigner, MultiSignerImpl, ProtocolParameters, ProtocolPartyId,
-    ProtocolSignerVerificationKey, ProtocolStake,
+    key_decode_hex, MultiSigner, MultiSignerImpl, ProtocolError, ProtocolParameters,
+    ProtocolPartyId, ProtocolSignerVerificationKey, ProtocolStake,
 };
 pub use crate::snapshot_store::SnapshotStoreHTTPClient;
-pub use beacon_store::MemoryBeaconStore;
+pub use beacon_store::{BeaconStoreError, MemoryBeaconStore};
 pub use dependency::DependencyManager;
 pub use runtime::AggregatorRuntime;
-pub use snapshotter::Snapshotter;
+pub use snapshotter::{SnapshotError, Snapshotter};

mithril-aggregator/src/multi_signer.rs

@@ -261,23 +261,6 @@ impl MultiSigner for MultiSignerImpl {
                     }
                 };
 
-                // Create a multi signature
-                // TODO: Should be triggered separately
-                match self.create_multi_signature() {
-                    Ok(Some(multi_signature)) => {
-                        debug!(
-                            "A multi signature has been created: {}",
-                            key_encode_hex(&multi_signature).unwrap()
-                        );
-                    }
-                    Ok(None) => {
-                        warn!("Not ready to create a multi signature: quorum is not reached yet");
-                    }
-                    Err(e) => {
-                        warn!("Error while creating a multi signature: {}", e);
-                    }
-                }
-
                 Ok(())
             }
             Err(e) => Err(ProtocolError::Core(e.to_string())),
@@ -475,6 +458,9 @@ mod tests {
 
         let quorum_split = protocol_parameters.k as usize - 1;
         assert!(quorum_split > 1, "quorum should be greater");
+        multi_signer
+            .create_multi_signature()
+            .expect("create multi sgnature should not fail");
         assert!(multi_signer
             .get_multi_signature(message.encode_hex::<String>())
             .expect("get multi signature should not fail")
@@ -486,6 +472,9 @@ mod tests {
                     .register_single_signature(*party_id, signature, *index)
                     .expect("register single signature should not fail");
             });
+        multi_signer
+            .create_multi_signature()
+            .expect("create multi sgnature should not fail");
         assert!(multi_signer
             .get_multi_signature(message.encode_hex::<String>())
             .expect("get multi signature should not fail")
@@ -497,6 +486,9 @@ mod tests {
                     .register_single_signature(*party_id, signature, *index)
                     .expect("register single signature should not fail");
             });
+        multi_signer
+            .create_multi_signature()
+            .expect("create multi sgnature should not fail");
         assert!(multi_signer
             .get_multi_signature(message.encode_hex::<String>())
             .expect("get multi signature should not fail")

mithril-aggregator/src/runtime.rs

@@ -1,10 +1,31 @@
-use crate::dependency::{BeaconStoreWrapper, MultiSignerWrapper};
-use crate::Snapshotter;
+use super::dependency::{BeaconStoreWrapper, MultiSignerWrapper};
+use super::{BeaconStoreError, ProtocolError, SnapshotError, Snapshotter};
+
+use mithril_common::crypto_helper::Bytes;
+use mithril_common::entities::Beacon;
 use mithril_common::fake_data;
-use mithril_common::immutable_digester::ImmutableDigester;
-use slog_scope::{debug, error, info};
+use mithril_common::immutable_digester::{ImmutableDigester, ImmutableDigesterError};
+
+use hex::ToHex;
+use slog_scope::{debug, error, info, warn};
+use thiserror::Error;
 use tokio::time::{sleep, Duration};
 
+#[derive(Error, Debug)]
+pub enum RuntimeError {
+    #[error("multi signer error")]
+    MultiSignerError(#[from] ProtocolError),
+
+    #[error("beacon store error")]
+    BeaconStoreError(#[from] BeaconStoreError),
+
+    #[error("snapshotter error")]
+    SnapshotterError(#[from] SnapshotError),
+
+    #[error("immutable digester error")]
+    ImmutableDigesterError(#[from] ImmutableDigesterError),
+}
+
 /// AggregatorRuntime
 pub struct AggregatorRuntime {
     /// Interval between each snapshot, in seconds
@@ -52,34 +73,70 @@ impl AggregatorRuntime {
         }
     }
 
-    async fn do_work(&self) -> Result<(), String> {
+    async fn do_work(&self) -> Result<(), RuntimeError> {
         let snapshotter = Snapshotter::new(self.db_directory.clone());
         let digester = ImmutableDigester::new(self.db_directory.clone(), slog_scope::logger());
         debug!("Making snapshot"; "directory" => &self.db_directory);
 
         match digester.compute_digest() {
             Ok(digest_result) => {
-                let mut beacon_store = self.beacon_store.write().await;
                 let mut beacon = fake_data::beacon();
                 beacon.immutable_file_number = digest_result.last_immutable_file_number;
-                beacon_store.set_current_beacon(beacon.clone()).await?;
-
-                let mut multi_signer = self.multi_signer.write().await;
                 let message = fake_data::digest(&beacon);
-                multi_signer
-                    .update_current_message(message)
-                    .map_err(|e| e.to_string())?;
-
-                snapshotter
-                    .snapshot(digest_result.digest)
-                    .await
-                    .map_err(|e| e.to_string())?;
-                Ok(())
+                match self.manage_trigger_snapshot(&message, &beacon).await {
+                    Ok(true) => {
+                        snapshotter.snapshot(digest_result.digest).await?;
+                        Ok(())
+                    }
+                    Ok(false) => Ok(()),
+                    Err(err) => Err(err),
+                }
             }
-            Err(error) => {
+            Err(err) => {
                 let mut beacon_store = self.beacon_store.write().await;
                 beacon_store.reset_current_beacon().await?;
-                Err(error.to_string())
+                Err(RuntimeError::ImmutableDigesterError(err))
+            }
+        }
+    }
+
+    async fn manage_trigger_snapshot(
+        &self,
+        message: &Bytes,
+        beacon: &Beacon,
+    ) -> Result<bool, RuntimeError> {
+        let mut multi_signer = self.multi_signer.write().await;
+        let mut beacon_store = self.beacon_store.write().await;
+        match multi_signer.get_multi_signature(message.encode_hex::<String>()) {
+            Ok(None) => {
+                beacon_store.set_current_beacon(beacon.clone()).await?;
+                multi_signer.update_current_message(message.to_owned())?;
+                match multi_signer.create_multi_signature() {
+                    Ok(Some(_)) => {
+                        debug!(
+                            "A multi signature has been created for message: {}",
+                            multi_signer
+                                .get_current_message()
+                                .unwrap()
+                                .encode_hex::<String>()
+                        );
+                    }
+                    Ok(None) => {
+                        warn!("Not ready to create a multi signature: quorum is not reached yet");
+                    }
+                    Err(e) => {
+                        warn!("Error while creating a multi signature: {}", e);
+                    }
+                }
+                Ok(true)
+            }
+            Ok(_) => {
+                beacon_store.reset_current_beacon().await?;
+                Ok(false)
+            }
+            Err(err) => {
+                beacon_store.reset_current_beacon().await?;
+                Err(RuntimeError::MultiSignerError(err))
             }
         }
     }

mithril-test-lab/mithril-end-to-end/src/Mithril/Aggregator.hs

@@ -100,4 +100,4 @@ aggregatorProcess cwd port = do
           ]
             <> baseEnv
   unlessM (doesFileExist aggregator) $ failure $ "cannot find mithril-aggregator executable in expected location (" <> binDir <> ")"
-  pure $ (proc aggregator ["--db-directory", "db", "--server-port", show port, "-vvv"]) {cwd, env}
+  pure $ (proc aggregator ["--db-directory", "db", "--server-port", show port, "--snapshot-interval", "5", "-vvv"]) {cwd, env}

mithril-test-lab/mithril-end-to-end/src/Mithril/Signer.hs

@@ -56,7 +56,7 @@ signerProcess cwd aggregatorEndpoint = do
           [ ("AGGREGATOR_ENDPOINT", toString aggregatorEndpoint),
             ("NETWORK", "testnet"),
             ("PARTY_ID", "0"),
-            ("RUN_INTERVAL", "5")
+            ("RUN_INTERVAL", "2")
           ]
             <> baseEnv
 

mithril-test-lab/mithril-end-to-end/test/Test/EndToEndSpec.hs

@@ -54,8 +54,8 @@ spec =
                 -- Start aggregator service on some random port
                 withAggregator (takeDirectory nodeSocket) (contramap AggregatorLog tr) $ \aggregator@Aggregator {aggregatorPort} -> do
                   threadDelay 2
-                  digest <- assertNodeIsProducingSnapshot tr node aggregatorPort
                   withSigner tmp (contramap SignerLog tr) aggregatorPort node $ \signer -> do
+                    digest <- assertNodeIsProducingSnapshot tr node aggregatorPort
                     assertSignerIsSigningSnapshot signer aggregatorPort digest
                     assertClientCanVerifySnapshot tmp aggregator digest
               _ -> failure "No nodes in the cluster"