fix(dmq): gate code behind 'future_dmq' feature

jpraynaud · jpraynaud · commit 56ac1b3614b5 · 2025-07-22T11:38:58.000+02:00
diff --git a/internal/mithril-dmq/src/consumer/server/pallas.rs b/internal/mithril-dmq/src/consumer/server/pallas.rs
@@ -48,9 +48,15 @@ impl DmqConsumerServerPallas {
 
     /// Creates and returns a new `DmqServer` connected to the specified socket.
     async fn new_server(&self) -> StdResult<DmqServer> {
-        let magic = self.network.code();
+        info!(
+            self.logger,
+            "Creating a new DMQ consumer server";
+            "socket" => ?self.socket,
+            "network" => ?self.network
+        );
+        let magic = self.network.magic_id();
         if self.socket.exists() {
-            fs::remove_file(self.socket.clone()).unwrap();
+            fs::remove_file(self.socket.clone())?;
         }
         let listener = UnixListener::bind(&self.socket)
             .map_err(|err| anyhow!(err))
@@ -177,7 +183,7 @@ impl DmqConsumerServer for DmqConsumerServerPallas {
             Some(ref mut receiver) => loop {
                 select! {
                     _ = stop_rx.changed() => {
-                        warn!(self.logger, "Stopping signature processor...");
+                        warn!(self.logger, "Stopping DMQ consumer server...");
 
                         return Ok(());
                     }
diff --git a/internal/mithril-dmq/src/publisher/server/pallas.rs b/internal/mithril-dmq/src/publisher/server/pallas.rs
@@ -50,9 +50,15 @@ impl DmqPublisherServerPallas {
 
     /// Creates and returns a new `DmqServer` connected to the specified socket.
     async fn new_server(&self) -> StdResult<DmqServer> {
-        let magic = self.network.code();
+        info!(
+            self.logger,
+            "Creating a new DMQ publisher server";
+            "socket" => ?self.socket,
+            "network" => ?self.network
+        );
+        let magic = self.network.magic_id();
         if self.socket.exists() {
-            fs::remove_file(self.socket.clone()).unwrap();
+            fs::remove_file(self.socket.clone())?;
         }
         let listener = UnixListener::bind(&self.socket)
             .map_err(|err| anyhow!(err))
@@ -184,7 +190,7 @@ impl DmqPublisherServer for DmqPublisherServerPallas {
         loop {
             select! {
                 _ = stop_rx.changed() => {
-                    warn!(self.logger, "Stopping signature processor...");
+                    warn!(self.logger, "Stopping DMQ publisher server...");
 
                     return Ok(());
                 }
diff --git a/mithril-relay/src/commands/aggregator.rs b/mithril-relay/src/commands/aggregator.rs
@@ -1,10 +1,13 @@
+#[cfg(feature = "future_dmq")]
 use std::path::PathBuf;
 
 use clap::Parser;
 use libp2p::Multiaddr;
 use slog::error;
 
-use mithril_common::{CardanoNetwork, StdResult};
+#[cfg(feature = "future_dmq")]
+use mithril_common::CardanoNetwork;
+use mithril_common::StdResult;
 
 use crate::AggregatorRelay;
 
@@ -21,6 +24,7 @@ pub struct AggregatorCommand {
     dial_to: Option<Multiaddr>,
 
     /// Path to the DMQ socket file
+    #[cfg(feature = "future_dmq")]
     #[clap(
         long,
         env = "DMQ_NODE_SOCKET_PATH",
@@ -30,11 +34,13 @@ pub struct AggregatorCommand {
     dmq_node_socket_path: PathBuf,
 
     /// Cardano network
+    #[cfg(feature = "future_dmq")]
     #[clap(long, env = "NETWORK")]
     pub network: String,
 
     /// Cardano Network Magic number
     /// useful for TestNet & DevNet
+    #[cfg(feature = "future_dmq")]
     #[clap(long, env = "NETWORK_MAGIC")]
     pub network_magic: Option<u64>,
 
@@ -50,12 +56,15 @@ impl AggregatorCommand {
         let dial_to = self.dial_to.to_owned();
         let addr: Multiaddr = format!("/ip4/0.0.0.0/tcp/{}", self.listen_port).parse()?;
         let aggregator_endpoint = self.aggregator_endpoint.to_owned();
+        #[cfg(feature = "future_dmq")]
         let cardano_network =
             CardanoNetwork::from_code(self.network.to_owned(), self.network_magic)?;
 
         let mut relay = AggregatorRelay::start(
             &addr,
+            #[cfg(feature = "future_dmq")]
             &self.dmq_node_socket_path,
+            #[cfg(feature = "future_dmq")]
             &cardano_network,
             &aggregator_endpoint,
             logger,
diff --git a/mithril-relay/src/commands/signer.rs b/mithril-relay/src/commands/signer.rs
@@ -1,13 +1,19 @@
-use std::{path::PathBuf, time::Duration};
+#[cfg(feature = "future_dmq")]
+use std::path::PathBuf;
+use std::time::Duration;
 
 use clap::Parser;
 use libp2p::Multiaddr;
-use mithril_common::{CardanoNetwork, StdResult};
 use slog::error;
 
-use super::CommandContext;
+#[cfg(feature = "future_dmq")]
+use mithril_common::CardanoNetwork;
+use mithril_common::StdResult;
+
 use crate::{SignerRelay, SignerRelayMode};
 
+use super::CommandContext;
+
 #[derive(Parser, Debug, Clone)]
 pub struct SignerCommand {
     /// HTTP Server listening port
@@ -23,6 +29,7 @@ pub struct SignerCommand {
     dial_to: Option<Multiaddr>,
 
     /// Path to the DMQ socket file
+    #[cfg(feature = "future_dmq")]
     #[clap(
         long,
         env = "DMQ_NODE_SOCKET_PATH",
@@ -32,11 +39,13 @@ pub struct SignerCommand {
     dmq_node_socket_path: PathBuf,
 
     /// Cardano network
+    #[cfg(feature = "future_dmq")]
     #[clap(long, env = "NETWORK")]
     pub network: String,
 
     /// Cardano Network Magic number
     /// useful for TestNet & DevNet
+    #[cfg(feature = "future_dmq")]
     #[clap(long, env = "NETWORK_MAGIC")]
     pub network_magic: Option<u64>,
 
@@ -68,13 +77,16 @@ impl SignerCommand {
         let signature_registration_mode = &self.signature_registration_mode;
         let aggregator_endpoint = self.aggregator_endpoint.to_owned();
         let signer_repeater_delay = Duration::from_millis(self.signer_repeater_delay);
+        #[cfg(feature = "future_dmq")]
         let cardano_network =
             CardanoNetwork::from_code(self.network.to_owned(), self.network_magic)?;
 
         let mut relay = SignerRelay::start(
             &addr,
             &server_port,
+            #[cfg(feature = "future_dmq")]
             &self.dmq_node_socket_path,
+            #[cfg(feature = "future_dmq")]
             &cardano_network,
             signer_registration_mode,
             signature_registration_mode,
diff --git a/mithril-relay/src/p2p/peer.rs b/mithril-relay/src/p2p/peer.rs
@@ -15,6 +15,7 @@ use mithril_common::{
     logging::LoggerExtensions,
     messages::{RegisterSignatureMessageHttp, RegisterSignerMessage},
 };
+#[cfg(feature = "future_dmq")]
 use mithril_dmq::DmqMessage;
 use serde::{Deserialize, Serialize};
 use slog::{Logger, debug, info};
@@ -72,6 +73,7 @@ pub enum BroadcastMessage {
     RegisterSignatureHttp(RegisterSignatureMessageHttp),
 
     /// A DMQ signature registration message received from the Gossip sub
+    #[cfg(feature = "future_dmq")]
     RegisterSignatureDmq(DmqMessage),
 }
 
@@ -257,6 +259,7 @@ impl Peer {
     }
 
     /// Publish a DMQ signature on the P2P pubsub
+    #[cfg(feature = "future_dmq")]
     pub fn publish_signature_dmq(
         &mut self,
         message: &DmqMessage,
diff --git a/mithril-relay/src/relay/aggregator.rs b/mithril-relay/src/relay/aggregator.rs
@@ -1,17 +1,22 @@
+#[cfg(feature = "future_dmq")]
 use std::{path::Path, sync::Arc};
 
 use anyhow::anyhow;
 use libp2p::Multiaddr;
+#[cfg(feature = "future_dmq")]
 use mithril_dmq::{DmqConsumerServer, DmqConsumerServerPallas, DmqMessage};
 use reqwest::StatusCode;
 use slog::{Logger, error, info};
+#[cfg(feature = "future_dmq")]
 use tokio::sync::{
     mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
-    watch::{self, Receiver},
+    watch::{self, Receiver, Sender},
 };
 
+#[cfg(feature = "future_dmq")]
+use mithril_common::CardanoNetwork;
 use mithril_common::{
-    CardanoNetwork, StdResult,
+    StdResult,
     logging::LoggerExtensions,
     messages::{RegisterSignatureMessageHttp, RegisterSignerMessage},
 };
@@ -22,38 +27,56 @@ use crate::p2p::{BroadcastMessage, Peer, PeerEvent};
 pub struct AggregatorRelay {
     aggregator_endpoint: String,
     peer: Peer,
+    #[cfg(feature = "future_dmq")]
     signature_dmq_tx: UnboundedSender<DmqMessage>,
+    #[cfg(feature = "future_dmq")]
+    #[allow(unused)]
+    stop_tx: Sender<()>,
     logger: Logger,
 }
 
 impl AggregatorRelay {
     /// Start a relay for a Mithril aggregator
     pub async fn start(
         addr: &Multiaddr,
-        dmq_node_socket_path: &Path,
-        cardano_network: &CardanoNetwork,
+        #[cfg(feature = "future_dmq")] dmq_node_socket_path: &Path,
+        #[cfg(feature = "future_dmq")] cardano_network: &CardanoNetwork,
         aggregator_endpoint: &str,
         logger: &Logger,
     ) -> StdResult<Self> {
-        let (_stop_tx, stop_rx) = watch::channel(());
-        let (signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
-        let _dmq_consumer_server = Self::start_dmq_consumer_server(
-            dmq_node_socket_path,
-            cardano_network,
-            signature_dmq_rx,
-            stop_rx,
-            logger.clone(),
-        )
-        .await?;
+        let peer = Peer::new(addr).with_logger(logger).start().await?;
+        let logger = logger.new_with_component_name::<Self>();
+        #[cfg(feature = "future_dmq")]
+        {
+            let (stop_tx, stop_rx) = watch::channel(());
+            let (signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
+            #[cfg(unix)]
+            let _dmq_consumer_server = Self::start_dmq_consumer_server(
+                dmq_node_socket_path,
+                cardano_network,
+                signature_dmq_rx,
+                stop_rx,
+                logger.clone(),
+            )
+            .await?;
 
+            Ok(Self {
+                aggregator_endpoint: aggregator_endpoint.to_owned(),
+                peer,
+                signature_dmq_tx,
+                stop_tx,
+                logger,
+            })
+        }
+        #[cfg(not(feature = "future_dmq"))]
         Ok(Self {
             aggregator_endpoint: aggregator_endpoint.to_owned(),
-            peer: Peer::new(addr).with_logger(logger).start().await?,
-            signature_dmq_tx,
-            logger: logger.new_with_component_name::<Self>(),
+            peer,
+            logger,
         })
     }
 
+    #[cfg(feature = "future_dmq")]
     async fn start_dmq_consumer_server(
         socket: &Path,
         cardano_network: &CardanoNetwork,
@@ -174,6 +197,7 @@ impl AggregatorRelay {
                         }
                     }
                 }
+                #[cfg(feature = "future_dmq")]
                 Ok(Some(BroadcastMessage::RegisterSignatureDmq(signature_message_received))) => {
                     self.signature_dmq_tx.send(signature_message_received).map_err(|e| {
                         anyhow!("Failed to send signature message to DMQ consumer server: {e}")
@@ -235,7 +259,9 @@ mod tests {
         let addr: Multiaddr = "/ip4/0.0.0.0/tcp/0".parse().unwrap();
         let relay = AggregatorRelay::start(
             &addr,
-            &Path::new("test"),
+            #[cfg(feature = "future_dmq")]
+            Path::new("test"),
+            #[cfg(feature = "future_dmq")]
             &CardanoNetwork::TestNet(123),
             &server.url(""),
             &TestLogger::stdout(),
diff --git a/mithril-relay/src/relay/passive.rs b/mithril-relay/src/relay/passive.rs
@@ -38,6 +38,7 @@ impl PassiveRelay {
                 Ok(Some(BroadcastMessage::RegisterSignatureHttp(signature_message_received))) => {
                     info!(self.logger, "Received HTTP signature message from P2P network"; "signature_message" => #?signature_message_received);
                 }
+                #[cfg(feature = "future_dmq")]
                 Ok(Some(BroadcastMessage::RegisterSignatureDmq(signature_message_received))) => {
                     info!(self.logger, "Received DMQ signature message from P2P network"; "signature_message" => #?signature_message_received);
                 }
diff --git a/mithril-relay/src/relay/signer.rs b/mithril-relay/src/relay/signer.rs
diff --git a/mithril-relay/tests/register_signer_signature.rs b/mithril-relay/tests/register_signer_signature.rs

Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,7 @@ impl PassiveRelay {`
`38`	`38`	`Ok(Some(BroadcastMessage::RegisterSignatureHttp(signature_message_received))) => {`
`39`	`39`	`info!(self.logger, "Received HTTP signature message from P2P network"; "signature_message" => #?signature_message_received);`
`40`	`40`	`}`
	`41`	`+ #[cfg(feature = "future_dmq")]`
`41`	`42`	`Ok(Some(BroadcastMessage::RegisterSignatureDmq(signature_message_received))) => {`
`42`	`43`	`info!(self.logger, "Received DMQ signature message from P2P network"; "signature_message" => #?signature_message_received);`
`43`	`44`	`}`