feat(signer): wire 'SignaturePublisherDmq' in DI

Author: jpraynaud

mithril-signer/src/configuration.rs

@@ -35,11 +35,14 @@ pub struct Configuration {
     #[example = "`cardano-cli`"]
     pub cardano_cli_path: PathBuf,
 
-    /// Path of the socket used by the Cardano CLI tool
-    /// to communicate with the Cardano node
-    #[example = "`/tmp/cardano.sock`"]
+    /// Path of the socket opened by the Cardano node
+    #[example = "`/ipc/node.socket`"]
     pub cardano_node_socket_path: PathBuf,
 
+    /// Path of the socket opened by the DMQ node
+    #[example = "`/ipc/dmq.socket`"]
+    pub dmq_node_socket_path: Option<PathBuf>,
+
     /// Cardano network
     #[example = "`testnet` or `mainnet` or `devnet`"]
     pub network: String,
@@ -150,6 +153,7 @@ impl Configuration {
             relay_endpoint: None,
             cardano_cli_path: PathBuf::new(),
             cardano_node_socket_path: PathBuf::new(),
+            dmq_node_socket_path: None,
             db_directory: PathBuf::new(),
             network: "devnet".to_string(),
             network_magic: Some(42),

mithril-signer/src/dependency_injection/builder.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::{anyhow, Context};
+use mithril_dmq_node::{DmqMessageBuilder, DmqPublisherPallas};
 use slog::Logger;
 use tokio::sync::{Mutex, RwLock};
 
@@ -19,14 +20,16 @@ use mithril_cardano_node_internal_database::{
     signable_builder::{CardanoDatabaseSignableBuilder, CardanoImmutableFilesFullSignableBuilder},
     ImmutableFileObserver, ImmutableFileSystemObserver,
 };
-use mithril_common::api_version::APIVersionProvider;
-use mithril_common::crypto_helper::{OpCert, ProtocolPartyId, SerDeShelleyFileFormat};
+use mithril_common::crypto_helper::{
+    KesSigner, KesSignerStandard, OpCert, ProtocolPartyId, SerDeShelleyFileFormat,
+};
 use mithril_common::signable_builder::{
     CardanoStakeDistributionSignableBuilder, CardanoTransactionsSignableBuilder,
     MithrilSignableBuilderService, MithrilStakeDistributionSignableBuilder,
     SignableBuilderServiceDependencies,
 };
 use mithril_common::StdResult;
+use mithril_common::{api_version::APIVersionProvider, messages::RegisterSignatureMessageDmq};
 
 use mithril_era::{EraChecker, EraReader};
 use mithril_signed_entity_lock::SignedEntityTypeLock;
@@ -37,10 +40,6 @@ use mithril_persistence::database::repository::CardanoTransactionRepository;
 use mithril_persistence::database::{ApplicationNodeType, SqlMigration};
 use mithril_persistence::sqlite::{ConnectionBuilder, SqliteConnection, SqliteConnectionPool};
 
-use crate::database::repository::{
-    ProtocolInitializerRepository, SignedBeaconRepository, StakePoolStore,
-};
-use crate::dependency_injection::SignerDependencyContainer;
 use crate::services::{
     AggregatorHTTPClient, CardanoTransactionsImporter,
     CardanoTransactionsPreloaderActivationSigner, MithrilEpochService, MithrilSingleSigner,
@@ -50,6 +49,11 @@ use crate::services::{
     TransactionsImporterWithPruner, TransactionsImporterWithVacuum,
 };
 use crate::store::MKTreeStoreSqlite;
+use crate::{
+    database::repository::{ProtocolInitializerRepository, SignedBeaconRepository, StakePoolStore},
+    services::SignaturePublisher,
+};
+use crate::{dependency_injection::SignerDependencyContainer, services::SignaturePublisherDmq};
 use crate::{
     Configuration, MetricsService, HTTP_REQUEST_TIMEOUT_DURATION, SQLITE_FILE,
     SQLITE_FILE_CARDANO_TRANSACTION,
@@ -390,10 +394,46 @@ impl<'a> DependenciesBuilder<'a> {
             self.root_logger(),
         ));
 
+        let kes_signer = match (
+            &self.config.kes_secret_key_path,
+            &self.config.operational_certificate_path,
+        ) {
+            (Some(kes_secret_key_path), Some(operational_certificate_path)) => {
+                Some(Arc::new(KesSignerStandard::new(
+                    kes_secret_key_path.clone(),
+                    operational_certificate_path.clone(),
+                )) as Arc<dyn KesSigner>)
+            }
+            (Some(_), None) | (None, Some(_)) => {
+                return Err(anyhow!(
+                    "kes_secret_key and operational_certificate are both mandatory".to_string(),
+                ))
+            }
+            _ => None,
+        };
+
+        let cardano_network = &self.config.get_network()?;
+
         let signature_publisher = {
-            // Temporary no-op publisher before a DMQ-based implementation is available.
             let first_publisher = SignaturePublisherRetrier::new(
-                Arc::new(SignaturePublisherNoop {}),
+                if let Some(dmq_node_socket_path) = &self.config.dmq_node_socket_path {
+                    let dmq_message_builder = DmqMessageBuilder::new_with_default_ttl(
+                        kes_signer
+                            .clone()
+                            .ok_or(anyhow!("A KES signer is mandatory to sign DMQ messages"))?,
+                        chain_observer.clone(),
+                    );
+                    Arc::new(SignaturePublisherDmq::new(Arc::new(DmqPublisherPallas::<
+                        RegisterSignatureMessageDmq,
+                    >::new(
+                        dmq_node_socket_path.to_owned(),
+                        *cardano_network,
+                        dmq_message_builder,
+                        self.root_logger(),
+                    )))) as Arc<dyn SignaturePublisher>
+                } else {
+                    Arc::new(SignaturePublisherNoop) as Arc<dyn SignaturePublisher>
+                },
                 SignaturePublishRetryPolicy::never(),
             );
 
@@ -442,6 +482,7 @@ impl<'a> DependenciesBuilder<'a> {
             upkeep_service,
             epoch_service,
             certifier,
+            kes_signer,
         };
 
         Ok(services)

mithril-signer/src/dependency_injection/containers.rs

@@ -1,3 +1,4 @@
+use mithril_common::crypto_helper::KesSigner;
 use std::sync::Arc;
 use tokio::sync::RwLock;
 
@@ -81,4 +82,7 @@ pub struct SignerDependencyContainer {
 
     /// Certifier service
     pub certifier: Arc<dyn CertifierService>,
+
+    /// Kes signer service
+    pub kes_signer: Option<Arc<dyn KesSigner>>,
 }

mithril-signer/src/runtime/runner.rs

@@ -1,14 +1,10 @@
-use std::sync::Arc;
-
 use anyhow::Context;
 use async_trait::async_trait;
 use slog::{debug, warn, Logger};
 use thiserror::Error;
 use tokio::sync::RwLockReadGuard;
 
-use mithril_common::crypto_helper::{
-    KesPeriod, KesSigner, KesSignerStandard, OpCert, ProtocolOpCert, SerDeShelleyFileFormat,
-};
+use mithril_common::crypto_helper::{KesPeriod, OpCert, ProtocolOpCert, SerDeShelleyFileFormat};
 use mithril_common::entities::{
     Epoch, PartyId, ProtocolMessage, SignedEntityType, Signer, TimePoint,
 };
@@ -186,28 +182,10 @@ impl Runner for SignerRunner {
             ),
             None => None,
         };
-        let kes_signer = match (
-            &self.config.kes_secret_key_path,
-            &self.config.operational_certificate_path,
-        ) {
-            (Some(kes_secret_key_path), Some(operational_certificate_path)) => {
-                Some(Arc::new(KesSignerStandard::new(
-                    kes_secret_key_path.clone(),
-                    operational_certificate_path.clone(),
-                )) as Arc<dyn KesSigner>)
-            }
-            (Some(_), None) | (None, Some(_)) => {
-                return Err(RunnerError::NoValueError(
-                    "kes_secret_key and operational_certificate are both mandatory".to_string(),
-                )
-                .into())
-            }
-            _ => None,
-        };
         let protocol_initializer = MithrilProtocolInitializerBuilder::build(
             stake,
             &protocol_parameters,
-            kes_signer,
+            self.services.kes_signer.clone(),
             kes_period,
         )?;
         let signer = Signer::new(
@@ -528,6 +506,7 @@ mod tests {
             aggregator_client.clone(),
             logger.clone(),
         ));
+        let kes_signer = None;
 
         SignerDependencyContainer {
             stake_store,
@@ -547,6 +526,7 @@ mod tests {
             upkeep_service,
             epoch_service,
             certifier,
+            kes_signer,
         }
     }
 

mithril-signer/tests/test_extensions/state_machine_tester.rs

@@ -27,6 +27,7 @@ use mithril_cardano_node_internal_database::{
 };
 use mithril_common::{
     api_version::APIVersionProvider,
+    crypto_helper::{KesSigner, KesSignerStandard},
     entities::{
         BlockNumber, CardanoTransactionsSigningConfig, ChainPoint, Epoch, SignedEntityConfig,
         SignedEntityType, SignedEntityTypeDiscriminants, SignerWithStake, SlotNumber, SupportedEra,
@@ -287,6 +288,10 @@ impl StateMachineTester {
             certificate_handler.clone(),
             logger.clone(),
         ));
+        let kes_signer = Some(Arc::new(KesSignerStandard::new(
+            config.kes_secret_key_path.clone().unwrap(),
+            config.operational_certificate_path.clone().unwrap(),
+        )) as Arc<dyn KesSigner>);
 
         let services = SignerDependencyContainer {
             certificate_handler: certificate_handler.clone(),
@@ -306,6 +311,7 @@ impl StateMachineTester {
             upkeep_service,
             epoch_service,
             certifier,
+            kes_signer,
         };
         // set up stake distribution
         chain_observer