Merge pull request #2477 from input-output-hk/jpraynaud/2470-signature-processor-dmq

Feat: implement a signature processor for DMQ

Author: jpraynaud

CHANGELOG.md

@@ -9,6 +9,8 @@ As a minor extension, we have adopted a slightly different versioning convention
 
 ## Mithril Distribution [XXXX] - UNRELEASED
 
+- Support for DMQ signature consumer and processor in the aggregator.
+
 - Crates versions:
 
 | Crate | Version |

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-metric"
-version = "0.1.13"
+version = "0.1.14"
 description = "Common tools to expose metrics."
 authors = { workspace = true }
 edition = { workspace = true }

internal/mithril-metric/Cargo.toml

@@ -8,7 +8,7 @@ use axum::{
 };
 use slog::{error, info, warn, Logger};
 use std::sync::Arc;
-use tokio::sync::oneshot::Receiver;
+use tokio::sync::watch::Receiver;
 
 use mithril_common::logging::LoggerExtensions;
 use mithril_common::StdResult;
@@ -93,9 +93,10 @@ impl<T: MetricsServiceExporter + 'static> MetricsServer<T> {
                 .await?;
 
         let serve_logger = self.logger.clone();
+        let mut shutdown_rx = shutdown_rx;
         axum::serve(listener, app)
             .with_graceful_shutdown(async move {
-                shutdown_rx.await.ok();
+                shutdown_rx.changed().await.ok();
                 warn!(
                     serve_logger,
                     "shutting down HTTP server after receiving signal"
@@ -112,7 +113,7 @@ mod tests {
     use anyhow::anyhow;
     use reqwest::StatusCode;
     use std::time::Duration;
-    use tokio::{sync::oneshot, task::yield_now, time::sleep};
+    use tokio::{sync::watch, task::yield_now, time::sleep};
 
     use crate::helper::test_tools::TestLogger;
 
@@ -135,7 +136,7 @@ mod tests {
     #[tokio::test]
     async fn test_metrics_server() {
         let logger = TestLogger::stdout();
-        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        let (shutdown_tx, shutdown_rx) = watch::channel(());
         let metrics_service = Arc::new(PseudoMetricsService::new());
         let metrics_server = Arc::new(MetricsServer::new(
             "0.0.0.0",

internal/mithril-metric/src/server.rs

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

mithril-aggregator/Cargo.toml

@@ -13,7 +13,7 @@ use clap::Parser;
 use config::{builder::DefaultState, ConfigBuilder, Map, Source, Value};
 
 use slog::{crit, debug, info, warn, Logger};
-use tokio::{sync::oneshot, task::JoinSet};
+use tokio::task::JoinSet;
 
 use mithril_cli_helper::{
     register_config_value, register_config_value_bool, register_config_value_option,
@@ -132,18 +132,23 @@ impl ServeCommand {
         let mut dependencies_builder =
             DependenciesBuilder::new(root_logger.clone(), Arc::new(config.clone()));
 
-        // start servers
         println!("Starting server...");
         println!("Press Ctrl+C to stop");
 
-        // start the monitoring thread
+        // Create the stop signal channel
+        let (stop_tx, stop_rx) = dependencies_builder
+            .get_stop_signal_channel()
+            .await
+            .with_context(|| "Dependencies Builder can not create stop signal channel")?;
+
+        // Start the monitoring thread
         let mut event_store = dependencies_builder
             .create_event_store()
             .await
             .with_context(|| "Dependencies Builder can not create event store")?;
         let event_store_thread = tokio::spawn(async move { event_store.run().await.unwrap() });
 
-        // start the database vacuum operation, if needed
+        // Start the database vacuum operation, if needed
         self.perform_database_vacuum_if_needed(
             &config.data_stores_directory,
             &mut dependencies_builder,
@@ -152,15 +157,15 @@ impl ServeCommand {
         )
         .await?;
 
-        // start the aggregator runtime
+        // Start the aggregator runtime
         let mut runtime = dependencies_builder
             .create_aggregator_runner()
             .await
             .with_context(|| "Dependencies Builder can not create aggregator runner")?;
         let mut join_set = JoinSet::new();
         join_set.spawn(async move { runtime.run().await.map_err(|e| e.to_string()) });
 
-        // start the cardano transactions preloader
+        // Start the cardano transactions preloader
         let cardano_transactions_preloader = dependencies_builder
             .create_cardano_transactions_preloader()
             .await
@@ -170,27 +175,41 @@ impl ServeCommand {
         let preload_task =
             tokio::spawn(async move { cardano_transactions_preloader.preload().await });
 
-        // start the HTTP server
-        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        // Start the HTTP server
         let routes = dependencies_builder
             .create_http_routes()
             .await
             .with_context(|| "Dependencies Builder can not create http routes")?;
+        let mut stop_rx_clone = stop_rx.clone();
         join_set.spawn(async move {
             let (_, server) = warp::serve(routes).bind_with_graceful_shutdown(
                 (
                     config.server_ip.clone().parse::<IpAddr>().unwrap(),
                     config.server_port,
                 ),
-                async {
-                    shutdown_rx.await.ok();
+                async move {
+                    stop_rx_clone.changed().await.ok();
                 },
             );
             server.await;
 
             Ok(())
         });
 
+        let signature_processor = dependencies_builder
+            .create_signature_processor()
+            .await
+            .with_context(|| "Dependencies Builder can not create signature processor")?;
+        let signature_processor_clone = signature_processor.clone();
+        join_set.spawn(async move {
+            signature_processor_clone
+                .run()
+                .await
+                .map_err(|e| e.to_string())?;
+
+            Ok(())
+        });
+
         // Create a SignersImporter only if the `cexplorer_pools_url` is provided in the config.
         if let Some(cexplorer_pools_url) = config.cexplorer_pools_url {
             match dependencies_builder
@@ -236,7 +255,7 @@ impl ServeCommand {
             .get_metrics_service()
             .await
             .with_context(|| "Metrics service initialization error")?;
-        let (metrics_server_shutdown_tx, metrics_server_shutdown_rx) = oneshot::channel();
+        let stop_rx_clone = stop_rx.clone();
         if config.enable_metrics_server {
             let metrics_logger = root_logger.clone();
             join_set.spawn(async move {
@@ -246,7 +265,7 @@ impl ServeCommand {
                     metrics_service,
                     metrics_logger.clone(),
                 )
-                .start(metrics_server_shutdown_rx)
+                .start(stop_rx_clone)
                 .await
                 .map_err(|e| anyhow!(e));
 
@@ -261,13 +280,13 @@ impl ServeCommand {
             crit!(root_logger, "A critical error occurred"; "error" => e);
         }
 
-        metrics_server_shutdown_tx
-            .send(())
-            .map_err(|e| anyhow!("Metrics server shutdown signal could not be sent: {e:?}"))?;
-
-        // stop servers
+        // Stop servers
         join_set.shutdown().await;
-        let _ = shutdown_tx.send(());
+
+        // Send the stop signal
+        stop_tx
+            .send(())
+            .map_err(|e| anyhow!("Stop signal could not be sent: {e:?}"))?;
 
         if !preload_task.is_finished() {
             preload_task.abort();

mithril-aggregator/src/commands/serve_command.rs

@@ -14,6 +14,7 @@ use crate::dependency_injection::{DependenciesBuilder, Result};
 use crate::get_dependency;
 use crate::services::{
     AggregatorClient, AggregatorHTTPClient, MessageService, MithrilMessageService,
+    SequentialSignatureProcessor, SignatureConsumer, SignatureConsumerNoop, SignatureProcessor,
 };
 impl DependenciesBuilder {
     async fn build_signed_entity_type_lock(&mut self) -> Result<Arc<SignedEntityTypeLock>> {
@@ -26,7 +27,7 @@ impl DependenciesBuilder {
         get_dependency!(self.signed_entity_type_lock)
     }
 
-    /// build HTTP message service
+    /// Builds HTTP message service
     pub async fn build_message_service(&mut self) -> Result<Arc<dyn MessageService>> {
         let certificate_repository = Arc::new(CertificateRepository::new(
             self.get_sqlite_connection().await?,
@@ -49,7 +50,7 @@ impl DependenciesBuilder {
         get_dependency!(self.message_service)
     }
 
-    /// build an [AggregatorClient]
+    /// Builds an [AggregatorClient]
     pub async fn build_leader_aggregator_client(&mut self) -> Result<Arc<dyn AggregatorClient>> {
         let leader_aggregator_endpoint = self
             .configuration
@@ -70,4 +71,24 @@ impl DependenciesBuilder {
     pub async fn get_leader_aggregator_client(&mut self) -> Result<Arc<dyn AggregatorClient>> {
         get_dependency!(self.leader_aggregator_client)
     }
+
+    /// Builds a [SignatureConsumer]
+    pub async fn build_signature_consumer(&mut self) -> Result<Arc<dyn SignatureConsumer>> {
+        let signature_consumer = SignatureConsumerNoop;
+
+        Ok(Arc::new(signature_consumer))
+    }
+
+    /// Builds a [SignatureProcessor]
+    pub async fn create_signature_processor(&mut self) -> Result<Arc<dyn SignatureProcessor>> {
+        let (_stop_tx, stop_rx) = self.get_stop_signal_channel().await?;
+        let signature_processor = SequentialSignatureProcessor::new(
+            self.build_signature_consumer().await?,
+            self.get_certifier_service().await?,
+            stop_rx,
+            self.root_logger(),
+        );
+
+        Ok(Arc::new(signature_processor))
+    }
 }

mithril-aggregator/src/dependency_injection/builder/enablers/misc.rs

@@ -8,7 +8,7 @@ use std::{path::PathBuf, sync::Arc};
 use tokio::{
     sync::{
         mpsc::{UnboundedReceiver, UnboundedSender},
-        Mutex,
+        watch, Mutex,
     },
     time::Duration,
 };
@@ -273,6 +273,9 @@ pub struct DependenciesBuilder {
 
     /// Protocol parameters retriever
     pub protocol_parameters_retriever: Option<Arc<dyn ProtocolParametersRetriever>>,
+
+    /// Stop signal channel
+    pub stop_signal_channel: Option<(watch::Sender<()>, watch::Receiver<()>)>,
 }
 
 impl DependenciesBuilder {
@@ -335,6 +338,7 @@ impl DependenciesBuilder {
             metrics_service: None,
             leader_aggregator_client: None,
             protocol_parameters_retriever: None,
+            stop_signal_channel: None,
         }
     }
 

mithril-aggregator/src/dependency_injection/builder/mod.rs

@@ -4,6 +4,7 @@
 
 mod compatibility;
 mod observability;
+mod signal;
 mod sqlite;
 mod stores;
 mod upkeep;

mithril-aggregator/src/dependency_injection/builder/support/mod.rs

@@ -0,0 +1,20 @@
+use tokio::sync::watch;
+
+use crate::dependency_injection::{DependenciesBuilder, Result};
+use crate::get_dependency;
+
+impl DependenciesBuilder {
+    /// Builds a stop signal channel
+    pub async fn build_stop_signal_channel(
+        &mut self,
+    ) -> Result<(watch::Sender<()>, watch::Receiver<()>)> {
+        Ok(watch::channel(()))
+    }
+
+    /// Get the stop signal channel
+    pub async fn get_stop_signal_channel(
+        &mut self,
+    ) -> Result<(watch::Sender<()>, watch::Receiver<()>)> {
+        get_dependency!(self.stop_signal_channel)
+    }
+}

mithril-aggregator/src/dependency_injection/builder/support/signal.rs

@@ -16,6 +16,8 @@ mod epoch_service;
 mod message;
 mod prover;
 mod signable_builder;
+mod signature_consumer;
+mod signature_processor;
 mod signed_entity;
 mod signer_registration;
 mod snapshotter;
@@ -30,6 +32,8 @@ pub use epoch_service::*;
 pub use message::*;
 pub use prover::*;
 pub use signable_builder::*;
+pub use signature_consumer::*;
+pub use signature_processor::*;
 pub use signed_entity::*;
 pub use signer_registration::*;
 pub use snapshotter::*;