Merge pull request #776 from input-output-hk/ensemble/665/runtime_error_handling

#665 handling errors better

Author: ghubertpalo

Cargo.lock

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

mithril-aggregator/Cargo.toml

@@ -1,9 +1,13 @@
 use clap::{Parser, Subcommand};
 use config::{builder::DefaultState, ConfigBuilder, Map, Source, Value, ValueKind};
 use slog::Level;
-use slog_scope::debug;
-use std::{error::Error, fs, path::PathBuf, sync::Arc};
-use tokio::{sync::RwLock, time::Duration};
+use slog_scope::{crit, debug, info};
+use std::{error::Error, fs, net::IpAddr, path::PathBuf, sync::Arc};
+use tokio::{
+    sync::{oneshot, RwLock},
+    task::JoinSet,
+    time::Duration,
+};
 
 use mithril_common::{
     certificate_chain::MithrilCertificateVerifier,
@@ -24,11 +28,12 @@ use mithril_common::{
 
 use crate::{
     event_store::{self, TransmitterService},
+    http_server::routes::router,
     tools::{EraTools, GenesisTools, GenesisToolsDependency},
     AggregatorConfig, AggregatorRunner, AggregatorRuntime, CertificatePendingStore,
     CertificateStore, Configuration, DefaultConfiguration, DependencyManager, GenesisConfiguration,
     GzipSnapshotter, MithrilSignerRegisterer, MultiSignerImpl, ProtocolParametersStore,
-    ProtocolParametersStorer, Server, SingleSignatureStore, VerificationKeyStore,
+    ProtocolParametersStorer, SingleSignatureStore, VerificationKeyStore,
 };
 
 const SQLITE_MONITORING_FILE: &str = "monitoring.sqlite3";
@@ -454,51 +459,66 @@ impl ServeCommand {
         .await?;
 
         let network = config.get_network()?;
-        let runtime_dependencies = dependency_manager.clone();
+        let runtime_dependency_manager = dependency_manager.clone();
 
-        // start the monitoring thread
-        let event_store_thread = tokio::spawn(async move {
-            event_store
-                .run(Some(
-                    config.data_stores_directory.join(SQLITE_MONITORING_FILE),
-                ))
-                .await
-                .unwrap()
-        });
+        // start servers
+        println!("Starting server...");
+        println!("Press Ctrl+C to stop");
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
 
-        // Start Aggregator state machine
-        let handle = tokio::spawn(async move {
+        let mut join_set = JoinSet::new();
+        join_set.spawn(async move {
             let config =
                 AggregatorConfig::new(config.run_interval, network, &config.db_directory.clone());
             let mut runtime = AggregatorRuntime::new(
                 Duration::from_millis(config.interval),
                 None,
-                Arc::new(AggregatorRunner::new(config, runtime_dependencies)),
+                Arc::new(AggregatorRunner::new(config, runtime_dependency_manager)),
             )
             .await
             .unwrap();
-            runtime.run().await
+            runtime.run().await.map_err(|e| e.to_string())
         });
+        join_set.spawn(async move {
+            let routes = router::routes(dependency_manager);
+            let (_, server) = warp::serve(routes).bind_with_graceful_shutdown(
+                (
+                    config.server_ip.clone().parse::<IpAddr>().unwrap(),
+                    config.server_port,
+                ),
+                async {
+                    shutdown_rx.await.ok();
+                },
+            );
+            server.await;
 
-        // Start REST server
-        println!("Starting server...");
-        println!("Press Ctrl+C to stop...");
-        let shutdown_signal = async {
-            tokio::signal::ctrl_c()
+            Ok(())
+        });
+        join_set.spawn(async { tokio::signal::ctrl_c().await.map_err(|e| e.to_string()) });
+
+        // start the monitoring thread
+        let event_store_thread = tokio::spawn(async move {
+            event_store
+                .run(Some(
+                    config.data_stores_directory.join(SQLITE_MONITORING_FILE),
+                ))
                 .await
-                .expect("failed to install CTRL+C signal handler");
-        };
-        let http_server = Server::new(
-            config.server_ip.clone(),
-            config.server_port,
-            dependency_manager.clone(),
-        );
-        http_server.start(shutdown_signal).await;
+                .unwrap()
+        });
+
+        let res = join_set.join_next().await.unwrap()?;
+        if let Err(e) = res {
+            crit!("A critical error occurred: {e}");
+        }
+
+        // stop servers
+        join_set.shutdown().await;
+        let _ = shutdown_tx.send(());
 
-        handle.abort();
+        info!("Event store is finishing...");
         event_store_thread.await?;
+        println!("Services stopped, exiting.");
 
-        println!("Exiting...");
         Ok(())
     }
 

mithril-aggregator/src/command_args.rs

@@ -161,7 +161,7 @@ impl Configuration {
     pub fn build_era_reader_adapter(
         &self,
         chain_observer: Arc<dyn ChainObserver>,
-    ) -> Result<Box<dyn EraReaderAdapter>, Box<dyn Error>> {
+    ) -> Result<Arc<dyn EraReaderAdapter>, Box<dyn Error>> {
         Ok(EraReaderAdapterBuilder::new(
             &self.era_reader_adapter_type,
             &self.era_reader_adapter_params,

mithril-aggregator/src/configuration.rs

@@ -361,7 +361,7 @@ pub mod tests {
             chain_observer.clone(),
             verification_key_store.clone(),
         ));
-        let era_reader = Arc::new(EraReader::new(Box::new(EraReaderBootstrapAdapter)));
+        let era_reader = Arc::new(EraReader::new(Arc::new(EraReaderBootstrapAdapter)));
         let era_epoch_token = era_reader
             .read_era_epoch_token(beacon_provider.get_current_beacon().await.unwrap().epoch)
             .await

mithril-aggregator/src/dependency.rs

@@ -1,4 +1,3 @@
-mod routes;
-mod server;
+pub mod routes;
 
-pub use server::{Server, SERVER_BASE_PATH};
+pub const SERVER_BASE_PATH: &str = "aggregator";

mithril-aggregator/src/http_server/mod.rs

@@ -36,11 +36,12 @@ pub use crate::snapshot_stores::{LocalSnapshotStore, RemoteSnapshotStore, Snapsh
 pub use certificate_creator::{CertificateCreator, MithrilCertificateCreator};
 pub use command_args::MainOpts;
 pub use dependency::DependencyManager;
-pub use http_server::Server;
 pub use message_adapters::{
     FromRegisterSignerAdapter, ToCertificatePendingMessageAdapter, ToEpochSettingsMessageAdapter,
 };
-pub use runtime::{AggregatorConfig, AggregatorRunner, AggregatorRunnerTrait, AggregatorRuntime};
+pub use runtime::{
+    AggregatorConfig, AggregatorRunner, AggregatorRunnerTrait, AggregatorRuntime, RuntimeError,
+};
 pub use signer_registerer::{
     MithrilSignerRegisterer, SignerRegisterer, SignerRegistrationError, SignerRegistrationRound,
     SignerRegistrationRoundOpener,

mithril-aggregator/src/http_server/server.rs

@@ -1,65 +1,80 @@
-use crate::certificate_creator::CertificateCreationError;
-use crate::snapshot_stores::SnapshotStoreError;
-use crate::{ProtocolError, SignerRegistrationError, SnapshotError};
-
-use mithril_common::certificate_chain::CertificateVerifierError;
-use mithril_common::chain_observer::ChainObserverError;
-use mithril_common::digesters::{ImmutableDigesterError, ImmutableFileListingError};
-use mithril_common::entities::BeaconComparisonError;
-use mithril_common::entities::Epoch;
-use mithril_common::store::StoreError;
-use mithril_common::BeaconProviderError;
 use std::error::Error as StdError;
-use std::io;
 use thiserror::Error;
 
+/// Error encountered or produced by the Runtime.
+/// This enum represents the faith of the errors produced during the state
+/// transitions.
 #[derive(Error, Debug)]
 pub enum RuntimeError {
-    #[error("multi signer error: {0}")]
-    MultiSigner(#[from] ProtocolError),
-
-    #[error("snapshotter error: {0}")]
-    Snapshotter(#[from] SnapshotError),
-
-    #[error("digester error: {0}")]
-    Digester(#[from] ImmutableDigesterError),
-
-    #[error("store error: {0}")]
-    StoreError(#[from] StoreError),
-
-    #[error("snapshot uploader error: {0}")]
-    SnapshotUploader(String),
-
-    #[error("snapshot build error: {0}")]
-    SnapshotBuild(#[from] io::Error),
-
-    #[error("immutable file scanning error: {0}")]
-    ImmutableFile(#[from] ImmutableFileListingError),
-
-    #[error("chain observer error: {0}")]
-    ChainObserver(#[from] ChainObserverError),
-
-    #[error("beacon provider error: {0}")]
-    BeaconProvider(#[from] BeaconProviderError),
-
-    #[error("certificate verifier error: {0}")]
-    CertificateVerifier(#[from] CertificateVerifierError),
-
-    #[error("certificate chain gap error: {0} vs {1}")]
-    CertificateChainEpochGap(Epoch, Epoch),
-
-    #[error("snapshot store error: {0}")]
-    SnapshotStore(#[from] SnapshotStoreError),
-
-    #[error("certificate creation error: {0}")]
-    CertificateCreation(#[from] CertificateCreationError),
+    /// Errors that need the runtime to try again without changing its state.
+    #[error("An error occured: {message}. This runtime cycle will be skipped. Nested error: {nested_error:#?}.")]
+    KeepState {
+        /// error message
+        message: String,
+
+        /// Eventual caught error
+        nested_error: Option<Box<dyn StdError + Sync + Send>>,
+    },
+    /// A Critical error means the Runtime stops and the software exits with an
+    /// error code.
+    #[error("Critical error:'{message}'. Nested error: {nested_error:#?}.")]
+    Critical {
+        /// error message
+        message: String,
+
+        /// Eventual caught error
+        nested_error: Option<Box<dyn StdError + Sync + Send>>,
+    },
+}
 
-    #[error("beacon comparison error: {0}")]
-    BeaconComparisonError(#[from] BeaconComparisonError),
+impl RuntimeError {
+    /// Create a new KeepState error
+    pub fn keep_state(message: &str, error: Option<Box<dyn StdError + Sync + Send>>) -> Self {
+        Self::KeepState {
+            message: message.to_string(),
+            nested_error: error,
+        }
+    }
+
+    /// Create a new Critical error
+    pub fn critical(message: &str, error: Option<Box<dyn StdError + Sync + Send>>) -> Self {
+        Self::Critical {
+            message: message.to_string(),
+            nested_error: error,
+        }
+    }
+}
 
-    #[error("signer registration error: {0}")]
-    SignerRegistration(#[from] SignerRegistrationError),
+impl From<Box<dyn StdError + Sync + Send>> for RuntimeError {
+    fn from(value: Box<dyn StdError + Sync + Send>) -> Self {
+        Self::KeepState {
+            message: "Error caught, state preserved, will retry to cycle.".to_string(),
+            nested_error: Some(value),
+        }
+    }
+}
 
-    #[error("general error: {0}")]
-    General(Box<dyn StdError + Sync + Send>),
+/// Errors returned when the runner cannot fulfil its missions with no subsystem
+/// to fail.
+#[derive(Debug, Error)]
+pub enum RunnerError {
+    /// Protocol message part is missing
+    #[error("Missing protocol message: '{0}'.")]
+    MissingProtocolMessage(String),
+
+    /// Epoch out of bounds
+    #[error("Epoch out of bounds: '{0}'.")]
+    EpochOutOfBounds(String),
+
+    /// No stack distribution found
+    #[error("Missing stack distribution: '{0}'.")]
+    MissingStakeDistribution(String),
+
+    /// Missing protocol parameters
+    #[error("Missing protocol parameters: '{0}'.")]
+    MissingProtocolParameters(String),
+
+    /// No AVK issued by the multisigner
+    #[error("No MultiSignature issued: '{0}'.")]
+    NoComputedMultiSignature(String),
 }