feat: implement fake DMQ node

.github/workflows/actions/deploy-terraform-infrastructure/action.yml

@@ -63,6 +63,10 @@ inputs:
     description: Mithril use P2P network (experimental, for test only).
     required: false
     default: "false"
+  mithril_p2p_use_dmq_protocol:
+    description: Mithril P2P network use DMQ protocol (experimental, for test only).
+    required: false
+    default: "false"
   mithril_p2p_network_bootstrap_peer:
     description: Mithril P2P network bootstrap peer (experimental, for test only).
     required: false
@@ -247,6 +251,7 @@ runs:
           google_compute_instance_ssh_keys_environment                              = "${{ inputs.google_compute_instance_ssh_keys_environment }}"
           google_service_credentials_json_file                                      = "./google-application-credentials.json"
           mithril_use_p2p_network                                                   = "${{ inputs.mithril_use_p2p_network }}"
+          mithril_p2p_use_dmq_protocol                                              = "${{ inputs.mithril_p2p_use_dmq_protocol }}"
           mithril_p2p_network_bootstrap_peer                                        = "${{ inputs.mithril_p2p_network_bootstrap_peer }}"
           mithril_p2p_signer_relay_signer_registration_mode                         = "${{ inputs.mithril_p2p_signer_relay_signer_registration_mode }}"
           mithril_p2p_signer_relay_signature_registration_mode                      = "${{ inputs.mithril_p2p_signer_relay_signature_registration_mode }}"

.github/workflows/test-deploy-network.yml

@@ -34,6 +34,7 @@ jobs:
             environment_prefix: dev
             cardano_network: preview
             mithril_use_p2p_network: true
+            mithril_p2p_use_dmq_protocol: true
             mithril_p2p_signer_relay_signer_registration_mode: passthrough
             mithril_p2p_signer_relay_signature_registration_mode: p2p
             mithril_api_domain: api.mithril.network
@@ -72,6 +73,7 @@ jobs:
             environment_prefix: dev-follower
             cardano_network: preview
             mithril_use_p2p_network: true
+            mithril_p2p_use_dmq_protocol: true
             mithril_p2p_network_bootstrap_peer: "/dns4/aggregator.dev-preview.api.mithril.network/tcp/6060"
             mithril_p2p_signer_relay_signer_registration_mode: passthrough
             mithril_p2p_signer_relay_signature_registration_mode: p2p
@@ -103,6 +105,7 @@ jobs:
             environment_prefix: dev
             cardano_network: mainnet
             mithril_use_p2p_network: false
+            mithril_p2p_use_dmq_protocol: true
             mithril_api_domain: api.mithril.network
             mithril_era_reader_adapter_type: bootstrap
             mithril_protocol_parameters: |
@@ -160,6 +163,7 @@ jobs:
           google_compute_instance_ssh_keys_environment: testing
           google_application_credentials: ${{ secrets.GOOGLE_APPLICATION_CREDENTIALS }}
           mithril_use_p2p_network: ${{ matrix.mithril_use_p2p_network }}
+          mithril_p2p_use_dmq_protocol: ${{ matrix.mithril_p2p_use_dmq_protocol }}
           mithril_p2p_network_bootstrap_peer: ${{ matrix.mithril_p2p_network_bootstrap_peer }}
           mithril_api_domain: ${{ matrix.mithril_api_domain }}
           mithril_image_id: ${{ inputs.mithril_image_id }}

internal/mithril-dmq/Cargo.toml

@@ -10,17 +10,26 @@ license.workspace = true
 repository.workspace = true
 include = ["**/*.rs", "Cargo.toml", "README.md", ".gitignore"]
 
+[package.metadata.cargo-machete]
+# `serde_bytes` is used for DmqMessage serialization
+ignored = ["serde_bytes"]
+
 [lib]
 crate-type = ["lib", "cdylib", "staticlib"]
 
 [dependencies]
 anyhow = { workspace = true }
 async-trait = { workspace = true }
+bincode = { version = "2.0.1" }
 blake2 = "0.10.6"
 mithril-cardano-node-chain = { path = "../cardano-node/mithril-cardano-node-chain" }
 mithril-common = { path = "../../mithril-common" }
 pallas-network = { git = "https://github.com/txpipe/pallas.git", branch = "main" }
+pallas-codec = { git = "https://github.com/txpipe/pallas.git", branch = "main" }
+serde = { workspace = true }
+serde_bytes = "0.11.17"
 slog = { workspace = true }
+slog-scope = "4.4.0"
 tokio = { workspace = true, features = ["sync"] }
 
 [dev-dependencies]

internal/mithril-dmq/src/consumer/interface.rs → internal/mithril-dmq/src/consumer/client/interface.rs

@@ -2,10 +2,10 @@ use std::fmt::Debug;
 
 use mithril_common::{StdResult, crypto_helper::TryFromBytes, entities::PartyId};
 
-/// Trait for consuming messages from a DMQ node.
+/// Trait for the client side of consuming messages from a DMQ node.
 #[cfg_attr(test, mockall::automock)]
 #[async_trait::async_trait]
-pub trait DmqConsumer<M: TryFromBytes + Debug + Send + Sync>: Send + Sync {
+pub trait DmqConsumerClient<M: TryFromBytes + Debug + Send + Sync>: Send + Sync {
     /// Consume messages from the DMQ node.
     async fn consume_messages(&self) -> StdResult<Vec<(M, PartyId)>>;
 }

internal/mithril-dmq/src/consumer/client/mod.rs

@@ -0,0 +1,5 @@
+mod interface;
+mod pallas;
+
+pub use interface::*;
+pub use pallas::*;

internal/mithril-dmq/src/consumer/pallas.rs → internal/mithril-dmq/src/consumer/client/pallas.rs

@@ -1,7 +1,7 @@
 use std::{fmt::Debug, marker::PhantomData, path::PathBuf};
 
 use anyhow::{Context, anyhow};
-use pallas_network::facades::DmqClient;
+use pallas_network::{facades::DmqClient, miniprotocols::localmsgnotification::State};
 use slog::{Logger, debug, error};
 use tokio::sync::{Mutex, MutexGuard};
 
@@ -12,21 +12,21 @@ use mithril_common::{
     logging::LoggerExtensions,
 };
 
-use crate::DmqConsumer;
+use crate::DmqConsumerClient;
 
-/// A DMQ consumer implementation.
+/// A DMQ client consumer implementation.
 ///
 /// This implementation is built upon the n2c mini-protocols DMQ implementation in Pallas.
-pub struct DmqConsumerPallas<M: TryFromBytes + Debug> {
+pub struct DmqConsumerClientPallas<M: TryFromBytes + Debug> {
     socket: PathBuf,
     network: CardanoNetwork,
     client: Mutex<Option<DmqClient>>,
     logger: Logger,
     phantom: PhantomData<M>,
 }
 
-impl<M: TryFromBytes + Debug> DmqConsumerPallas<M> {
-    /// Creates a new `DmqConsumerPallas` instance.
+impl<M: TryFromBytes + Debug> DmqConsumerClientPallas<M> {
+    /// Creates a new `DmqConsumerClientPallas` instance.
     pub fn new(socket: PathBuf, network: CardanoNetwork, logger: Logger) -> Self {
         Self {
             socket,
@@ -47,7 +47,7 @@ impl<M: TryFromBytes + Debug> DmqConsumerPallas<M> {
         );
         DmqClient::connect(&self.socket, self.network.magic_id())
             .await
-            .with_context(|| "DmqConsumerPallas failed to create a new client")
+            .with_context(|| "DmqConsumerClientPallas failed to create a new client")
     }
 
     /// Gets the cached `DmqClient`, creating a new one if it does not exist.
@@ -94,21 +94,20 @@ impl<M: TryFromBytes + Debug> DmqConsumerPallas<M> {
         debug!(self.logger, "Waiting for messages from DMQ...");
         let mut client_guard = self.get_client().await?;
         let client = client_guard.as_mut().ok_or(anyhow!("DMQ client does not exist"))?;
-        client
-            .msg_notification()
-            .send_request_messages_blocking()
-            .await
-            .with_context(|| "Failed to request notifications from DMQ server: {}")?;
+        if *client.msg_notification().state() == State::Idle {
+            client
+                .msg_notification()
+                .send_request_messages_blocking()
+                .await
+                .with_context(|| "Failed to request notifications from DMQ server: {}")?;
+        }
 
         let reply = client
             .msg_notification()
             .recv_next_reply()
             .await
             .with_context(|| "Failed to receive notifications from DMQ server")?;
         debug!(self.logger, "Received single signatures from DMQ"; "messages" => ?reply);
-        if let Err(e) = client.msg_notification().send_done().await {
-            error!(self.logger, "Failed to send Done"; "error" => ?e);
-        }
 
         reply
             .0
@@ -128,7 +127,7 @@ impl<M: TryFromBytes + Debug> DmqConsumerPallas<M> {
 }
 
 #[async_trait::async_trait]
-impl<M: TryFromBytes + Debug + Sync + Send> DmqConsumer<M> for DmqConsumerPallas<M> {
+impl<M: TryFromBytes + Debug + Sync + Send> DmqConsumerClient<M> for DmqConsumerClientPallas<M> {
     async fn consume_messages(&self) -> StdResult<Vec<(M, PartyId)>> {
         let messages = self.consume_messages_internal().await;
         if messages.is_err() {
@@ -227,10 +226,6 @@ mod tests {
                     // server replies with messages if any
                     server_msg.send_reply_messages_blocking(reply_messages).await.unwrap();
                     assert_eq!(*server_msg.state(), localmsgnotification::State::Idle);
-
-                    // server receives done from client
-                    server_msg.recv_done().await.unwrap();
-                    assert_eq!(*server_msg.state(), localmsgnotification::State::Done);
                 } else {
                     // server waits if no message available
                     future::pending().await
@@ -247,7 +242,7 @@ mod tests {
         let reply_messages = fake_msgs();
         let server = setup_dmq_server(socket_path.clone(), reply_messages);
         let client = tokio::spawn(async move {
-            let consumer = DmqConsumerPallas::new(
+            let consumer = DmqConsumerClientPallas::new(
                 socket_path,
                 CardanoNetwork::TestNet(0),
                 TestLogger::stdout(),
@@ -280,7 +275,7 @@ mod tests {
         let reply_messages = vec![];
         let server = setup_dmq_server(socket_path.clone(), reply_messages);
         let client = tokio::spawn(async move {
-            let consumer = DmqConsumerPallas::<DmqMessageTestPayload>::new(
+            let consumer = DmqConsumerClientPallas::<DmqMessageTestPayload>::new(
                 socket_path,
                 CardanoNetwork::TestNet(0),
                 TestLogger::stdout(),
@@ -304,7 +299,7 @@ mod tests {
         let reply_messages = fake_msgs();
         let server = setup_dmq_server(socket_path.clone(), reply_messages);
         let client = tokio::spawn(async move {
-            let consumer = DmqConsumerPallas::<DmqMessageTestPayload>::new(
+            let consumer = DmqConsumerClientPallas::<DmqMessageTestPayload>::new(
                 socket_path,
                 CardanoNetwork::TestNet(0),
                 TestLogger::stdout(),

internal/mithril-dmq/src/consumer/mod.rs

@@ -1,5 +1,5 @@
-mod interface;
-mod pallas;
+mod client;
+mod server;
 
-pub use interface::*;
-pub use pallas::*;
+pub use client::*;
+pub use server::*;

internal/mithril-dmq/src/consumer/server/interface.rs

@@ -0,0 +1,12 @@
+use mithril_common::StdResult;
+
+/// Trait for the server side of consuming messages from a DMQ node.
+#[cfg_attr(test, mockall::automock)]
+#[async_trait::async_trait]
+pub trait DmqConsumerServer: Send + Sync {
+    /// Processes the next message received from the DMQ network.
+    async fn process_message(&self) -> StdResult<()>;
+
+    /// Runs the DMQ publisher server.
+    async fn run(&self) -> StdResult<()>;
+}

internal/mithril-dmq/src/consumer/server/mod.rs

@@ -0,0 +1,9 @@
+mod interface;
+#[cfg(unix)]
+mod pallas;
+mod queue;
+
+pub use interface::*;
+
+#[cfg(unix)]
+pub use pallas::*;