feat: integrate Haskell DMQ network

CHANGELOG.md

@@ -24,6 +24,9 @@ As a minor extension, we have adopted a slightly different versioning convention
 - Client CLI:
   - Added `--epoch` optional parameter to `cardano-database snapshot list` commands snapshots, the given value can be a number, `latest`, or `latest-{offset}`.
 
+- **UNSTABLE** :
+  - Support for Haskell DMQ node and modifications of the DMQ protocol.
+
 - Crates versions:
 
 | Crate | Version |

internal/mithril-dmq/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "mithril-dmq"
 description = "Mechanisms to publish and consume messages of a 'Decentralized Message Queue network' through a DMQ node"
-version = "0.1.12"
+version = "0.1.13"
 authors.workspace = true
 documentation.workspace = true
 edition.workspace = true
@@ -30,6 +30,7 @@ serde = { workspace = true }
 serde_bytes = "0.11.19"
 slog = { workspace = true }
 slog-scope = "4.4.0"
+thiserror = { workspace = true }
 tokio = { workspace = true, features = ["sync","rt-multi-thread"] }
 
 [dev-dependencies]

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

@@ -6,30 +6,30 @@ use slog::{Logger, debug, error};
 use tokio::sync::{Mutex, MutexGuard};
 
 use mithril_common::{
-    CardanoNetwork, StdResult,
+    StdResult,
     crypto_helper::{
         OpCert, OpCertWithoutColdVerificationKey, TryFromBytes, ed25519::Ed25519VerificationKey,
     },
     entities::PartyId,
     logging::LoggerExtensions,
 };
 
-use crate::DmqConsumerClient;
+use crate::{DmqConsumerClient, model::DmqNetwork};
 
 /// A DMQ client consumer implementation.
 ///
 /// This implementation is built upon the n2c mini-protocols DMQ implementation in Pallas.
 pub struct DmqConsumerClientPallas<M: TryFromBytes + Debug> {
     socket: PathBuf,
-    network: CardanoNetwork,
+    network: DmqNetwork,
     client: Mutex<Option<DmqClient>>,
     logger: Logger,
     phantom: PhantomData<M>,
 }
 
 impl<M: TryFromBytes + Debug> DmqConsumerClientPallas<M> {
     /// Creates a new `DmqConsumerClientPallas` instance.
-    pub fn new(socket: PathBuf, network: CardanoNetwork, logger: Logger) -> Self {
+    pub fn new(socket: PathBuf, network: DmqNetwork, logger: Logger) -> Self {
         Self {
             socket,
             network,
@@ -115,11 +115,13 @@ impl<M: TryFromBytes + Debug> DmqConsumerClientPallas<M> {
             .0
             .into_iter()
             .map(|dmq_message| {
-                let opcert_without_verification_key =
-                    OpCertWithoutColdVerificationKey::try_from_bytes(
-                        &dmq_message.operational_certificate,
-                    )
-                    .with_context(|| "Failed to parse operational certificate")?;
+                let opcert_without_verification_key = OpCertWithoutColdVerificationKey::try_new(
+                    &dmq_message.operational_certificate.kes_vk,
+                    dmq_message.operational_certificate.issue_number,
+                    dmq_message.operational_certificate.start_kes_period,
+                    &dmq_message.operational_certificate.cert_sig,
+                )
+                .with_context(|| "Failed to parse operational certificate")?;
                 let cold_verification_key =
                     Ed25519VerificationKey::from_bytes(&dmq_message.cold_verification_key)
                         .with_context(|| "Failed to parse cold verification key")?
@@ -171,7 +173,7 @@ mod tests {
         facades::DmqServer,
         miniprotocols::{
             localmsgnotification,
-            localmsgsubmission::{DmqMsg, DmqMsgPayload},
+            localmsgsubmission::{DmqMsg, DmqMsgOperationalCertificate, DmqMsgPayload},
         },
     };
     use tokio::{net::UnixListener, task::JoinHandle, time::sleep};
@@ -194,14 +196,20 @@ mod tests {
                     expires_at: 100,
                 },
                 kes_signature: vec![0, 1, 2, 3],
-                operational_certificate: vec![
-                    132, 88, 32, 230, 80, 215, 83, 21, 9, 187, 108, 255, 215, 153, 140, 40, 198,
-                    142, 78, 200, 250, 98, 26, 9, 82, 32, 110, 161, 30, 176, 63, 205, 125, 203, 41,
-                    0, 0, 88, 64, 212, 171, 206, 39, 218, 5, 255, 3, 193, 52, 44, 198, 171, 83, 19,
-                    80, 114, 225, 186, 191, 156, 192, 84, 146, 245, 159, 31, 240, 9, 247, 4, 87,
-                    170, 168, 98, 199, 21, 139, 19, 190, 12, 251, 65, 215, 169, 26, 86, 37, 137,
-                    188, 17, 14, 178, 205, 175, 93, 39, 86, 4, 138, 187, 234, 95, 5,
-                ],
+                operational_certificate: DmqMsgOperationalCertificate {
+                    kes_vk: vec![
+                        50, 45, 160, 42, 80, 78, 184, 20, 210, 77, 140, 152, 63, 49, 165, 168, 5,
+                        131, 101, 152, 110, 242, 144, 157, 176, 210, 5, 10, 166, 91, 196, 168,
+                    ],
+                    issue_number: 0,
+                    start_kes_period: 0,
+                    cert_sig: vec![
+                        207, 135, 144, 168, 238, 41, 179, 216, 245, 74, 164, 231, 4, 158, 234, 141,
+                        5, 19, 166, 11, 78, 34, 210, 211, 183, 72, 127, 83, 185, 156, 107, 55, 160,
+                        190, 73, 251, 204, 47, 197, 86, 174, 231, 13, 49, 7, 83, 173, 177, 27, 53,
+                        209, 66, 24, 203, 226, 152, 3, 91, 66, 56, 244, 206, 79, 0,
+                    ],
+                },
                 cold_verification_key: vec![
                     32, 253, 186, 201, 177, 11, 117, 135, 187, 167, 181, 188, 22, 59, 206, 105,
                     231, 150, 215, 30, 78, 212, 76, 16, 252, 180, 72, 134, 137, 247, 161, 68,
@@ -215,14 +223,20 @@ mod tests {
                     expires_at: 101,
                 },
                 kes_signature: vec![1, 2, 3, 4],
-                operational_certificate: vec![
-                    132, 88, 32, 230, 80, 215, 83, 21, 9, 187, 108, 255, 215, 153, 140, 40, 198,
-                    142, 78, 200, 250, 98, 26, 9, 82, 32, 110, 161, 30, 176, 63, 205, 125, 203, 41,
-                    0, 0, 88, 64, 212, 171, 206, 39, 218, 5, 255, 3, 193, 52, 44, 198, 171, 83, 19,
-                    80, 114, 225, 186, 191, 156, 192, 84, 146, 245, 159, 31, 240, 9, 247, 4, 87,
-                    170, 168, 98, 199, 21, 139, 19, 190, 12, 251, 65, 215, 169, 26, 86, 37, 137,
-                    188, 17, 14, 178, 205, 175, 93, 39, 86, 4, 138, 187, 234, 95, 5,
-                ],
+                operational_certificate: DmqMsgOperationalCertificate {
+                    kes_vk: vec![
+                        50, 45, 160, 42, 80, 78, 184, 20, 210, 77, 140, 152, 63, 49, 165, 168, 5,
+                        131, 101, 152, 110, 242, 144, 157, 176, 210, 5, 10, 166, 91, 196, 168,
+                    ],
+                    issue_number: 0,
+                    start_kes_period: 0,
+                    cert_sig: vec![
+                        207, 135, 144, 168, 238, 41, 179, 216, 245, 74, 164, 231, 4, 158, 234, 141,
+                        5, 19, 166, 11, 78, 34, 210, 211, 183, 72, 127, 83, 185, 156, 107, 55, 160,
+                        190, 73, 251, 204, 47, 197, 86, 174, 231, 13, 49, 7, 83, 173, 177, 27, 53,
+                        209, 66, 24, 203, 226, 152, 3, 91, 66, 56, 244, 206, 79, 0,
+                    ],
+                },
                 cold_verification_key: vec![
                     77, 75, 24, 6, 47, 133, 2, 89, 141, 224, 69, 202, 123, 105, 240, 103, 245, 159,
                     147, 177, 110, 58, 248, 115, 58, 152, 138, 220, 35, 65, 245, 200,
@@ -278,7 +292,7 @@ mod tests {
 
             let consumer = DmqConsumerClientPallas::new(
                 socket_path,
-                CardanoNetwork::TestNet(0),
+                DmqNetwork::TestNet(0),
                 TestLogger::stdout(),
             );
 
@@ -314,7 +328,7 @@ mod tests {
 
             let consumer = DmqConsumerClientPallas::<DmqMessageTestPayload>::new(
                 socket_path,
-                CardanoNetwork::TestNet(0),
+                DmqNetwork::TestNet(0),
                 TestLogger::stdout(),
             );
 
@@ -341,7 +355,7 @@ mod tests {
 
             let consumer = DmqConsumerClientPallas::<DmqMessageTestPayload>::new(
                 socket_path,
-                CardanoNetwork::TestNet(0),
+                DmqNetwork::TestNet(0),
                 TestLogger::stdout(),
             );
 

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

@@ -11,16 +11,16 @@ use tokio::{
 
 use slog::{Logger, debug, error, info, warn};
 
-use mithril_common::{CardanoNetwork, StdResult, logging::LoggerExtensions};
+use mithril_common::{StdResult, logging::LoggerExtensions};
 
-use crate::{DmqConsumerServer, DmqMessage};
+use crate::{DmqConsumerServer, DmqMessage, DmqNetwork};
 
 use super::queue::MessageQueue;
 
 /// A DMQ server implementation for messages notification from a DMQ node.
 pub struct DmqConsumerServerPallas {
     socket: PathBuf,
-    network: CardanoNetwork,
+    network: DmqNetwork,
     server: Mutex<Option<DmqServer>>,
     messages_receiver: Mutex<Option<UnboundedReceiver<DmqMessage>>>,
     messages_buffer: MessageQueue,
@@ -32,7 +32,7 @@ impl DmqConsumerServerPallas {
     /// Creates a new instance of [DmqConsumerServerPallas].
     pub fn new(
         socket: PathBuf,
-        network: CardanoNetwork,
+        network: DmqNetwork,
         stop_rx: Receiver<()>,
         logger: Logger,
     ) -> Self {
@@ -303,10 +303,10 @@ mod tests {
         let (stop_tx, stop_rx) = watch::channel(());
         let (signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
         let socket_path = create_temp_dir(current_function_name).join("node.socket");
-        let cardano_network = CardanoNetwork::TestNet(0);
+        let dmq_network = DmqNetwork::TestNet(0);
         let dmq_consumer_server = Arc::new(DmqConsumerServerPallas::new(
             socket_path.to_path_buf(),
-            cardano_network.to_owned(),
+            dmq_network.to_owned(),
             stop_rx,
             TestLogger::stdout(),
         ));
@@ -363,10 +363,10 @@ mod tests {
         let (stop_tx, stop_rx) = watch::channel(());
         let (signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
         let socket_path = create_temp_dir(current_function_name).join("node.socket");
-        let cardano_network = CardanoNetwork::TestNet(0);
+        let dmq_network = DmqNetwork::TestNet(0);
         let dmq_consumer_server = Arc::new(DmqConsumerServerPallas::new(
             socket_path.to_path_buf(),
-            cardano_network.to_owned(),
+            dmq_network.to_owned(),
             stop_rx,
             TestLogger::stdout(),
         ));
@@ -422,10 +422,10 @@ mod tests {
         let (_stop_tx, stop_rx) = watch::channel(());
         let (_signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
         let socket_path = create_temp_dir(current_function!()).join("node.socket");
-        let cardano_network = CardanoNetwork::TestNet(0);
+        let dmq_network = DmqNetwork::TestNet(0);
         let dmq_consumer_server = Arc::new(DmqConsumerServerPallas::new(
             socket_path.to_path_buf(),
-            cardano_network.to_owned(),
+            dmq_network.to_owned(),
             stop_rx,
             TestLogger::stdout(),
         ));

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

@@ -122,7 +122,9 @@ mod tests {
     use std::{ops::RangeInclusive, time::Duration};
 
     use anyhow::anyhow;
-    use pallas_network::miniprotocols::localmsgsubmission::{DmqMsg, DmqMsgPayload};
+    use pallas_network::miniprotocols::localmsgsubmission::{
+        DmqMsg, DmqMsgOperationalCertificate, DmqMsgPayload,
+    };
     use tokio::time::sleep;
 
     use crate::model::MockUnixTimestampProvider;
@@ -138,7 +140,12 @@ mod tests {
                 expires_at: 100,
             },
             kes_signature: vec![0, 1, 2, 3],
-            operational_certificate: vec![0, 1, 2, 3, 4],
+            operational_certificate: DmqMsgOperationalCertificate {
+                kes_vk: vec![12, 13, 14],
+                issue_number: 15,
+                start_kes_period: 16,
+                cert_sig: vec![17],
+            },
             cold_verification_key: vec![0, 1, 2, 3, 4, 5],
         }
     }

internal/mithril-dmq/src/lib.rs

@@ -9,7 +9,7 @@ pub mod test;
 #[cfg(unix)]
 pub use consumer::DmqConsumerServerPallas;
 pub use consumer::{DmqConsumerClient, DmqConsumerClientPallas, DmqConsumerServer};
-pub use model::{DmqMessage, DmqMessageBuilder};
+pub use model::{DmqMessage, DmqMessageBuilder, DmqNetwork};
 #[cfg(unix)]
 pub use publisher::DmqPublisherServerPallas;
 pub use publisher::{DmqPublisherClient, DmqPublisherClientPallas, DmqPublisherServer};