feat(e2e): support for fake DMQ node

Author: jpraynaud

mithril-test-lab/mithril-end-to-end/src/main.rs

@@ -123,6 +123,12 @@ pub struct Args {
     #[clap(long, default_value = "false")]
     use_p2p_passive_relays: bool,
 
+    /// Use DMQ protocol (used to broadcast signatures)
+    ///
+    /// Requires the Mithril nodes to be compiled with the 'future_dmq' feature
+    #[clap(long)]
+    use_dmq: bool,
+
     /// Skip cardano binaries download
     #[clap(long)]
     skip_cardano_bin_download: bool,
@@ -327,6 +333,7 @@ impl App {
         let use_relays = args.use_relays;
         let relay_signer_registration_mode = args.relay_signer_registration_mode;
         let relay_signature_registration_mode = args.relay_signature_registration_mode;
+        let use_dmq = args.use_dmq;
 
         let use_p2p_passive_relays = args.use_p2p_passive_relays;
 
@@ -360,6 +367,7 @@ impl App {
                 mithril_era_reader_adapter: args.mithril_era_reader_adapter,
                 signed_entity_types: args.signed_entity_types.clone(),
                 run_only_mode,
+                use_dmq,
                 use_relays,
                 relay_signer_registration_mode,
                 relay_signature_registration_mode,

mithril-test-lab/mithril-end-to-end/src/mithril/aggregator.rs

@@ -39,6 +39,7 @@ pub struct AggregatorConfig<'a> {
     pub signed_entity_types: &'a [String],
     pub chain_observer_type: &'a str,
     pub leader_aggregator_endpoint: &'a Option<String>,
+    pub use_dmq: bool,
 }
 
 pub struct Aggregator {
@@ -75,6 +76,7 @@ impl Aggregator {
         let public_server_url = format!("http://localhost:{server_port_parameter}/aggregator");
         let mut env = HashMap::from([
             ("NETWORK", "devnet"),
+            ("NETWORK_MAGIC", &magic_id),
             ("RUN_INTERVAL", &mithril_run_interval),
             ("SERVER_IP", "0.0.0.0"),
             ("SERVER_PORT", &server_port_parameter),
@@ -85,7 +87,6 @@ impl Aggregator {
                 "SNAPSHOT_DIRECTORY",
                 aggregator_config.artifacts_dir.to_str().unwrap(),
             ),
-            ("NETWORK_MAGIC", &magic_id),
             (
                 "DATA_STORES_DIRECTORY",
                 aggregator_config.store_dir.to_str().unwrap(),
@@ -129,6 +130,15 @@ impl Aggregator {
         if let Some(leader_aggregator_endpoint) = aggregator_config.leader_aggregator_endpoint {
             env.insert("LEADER_AGGREGATOR_ENDPOINT", leader_aggregator_endpoint);
         }
+        let dmq_node_socket_path = aggregator_config
+            .work_dir
+            .join(format!("dmq-aggregator-{}.socket", aggregator_config.index));
+        if aggregator_config.use_dmq {
+            env.insert(
+                "DMQ_NODE_SOCKET_PATH",
+                dmq_node_socket_path.to_str().unwrap(),
+            );
+        }
         let args = vec![
             "--db-directory",
             aggregator_config.full_node.db_path.to_str().unwrap(),

mithril-test-lab/mithril-end-to-end/src/mithril/infrastructure.rs

@@ -35,6 +35,7 @@ pub struct MithrilInfrastructureConfig {
     pub relay_signer_registration_mode: String,
     pub relay_signature_registration_mode: String,
     pub use_p2p_passive_relays: bool,
+    pub use_dmq: bool,
     pub use_era_specific_work_dir: bool,
 }
 
@@ -68,6 +69,7 @@ impl MithrilInfrastructureConfig {
             relay_signer_registration_mode: "passthrough".to_string(),
             relay_signature_registration_mode: "passthrough".to_string(),
             use_p2p_passive_relays: false,
+            use_dmq: false,
             use_era_specific_work_dir: false,
         }
     }
@@ -255,6 +257,7 @@ impl MithrilInfrastructure {
             signed_entity_types: &config.signed_entity_types,
             chain_observer_type,
             leader_aggregator_endpoint: &leader_aggregator_endpoint,
+            use_dmq: config.use_dmq,
         })?;
 
         aggregator
@@ -294,12 +297,13 @@ impl MithrilInfrastructure {
         let mut bootstrap_peer_addr = None;
         for (index, aggregator_endpoint) in aggregator_endpoints.iter().enumerate() {
             let mut relay_aggregator = RelayAggregator::new(
-                Aggregator::name_suffix(index),
+                index,
                 config.server_port + index as u64 + 100,
                 bootstrap_peer_addr.clone(),
                 aggregator_endpoint,
                 &config.work_dir,
                 &config.bin_dir,
+                config.use_dmq,
             )?;
             if bootstrap_peer_addr.is_none() {
                 bootstrap_peer_addr = Some(relay_aggregator.peer_addr().to_owned());
@@ -310,6 +314,7 @@ impl MithrilInfrastructure {
 
         for (index, party_id) in signers_party_ids.iter().enumerate() {
             let mut relay_signer = RelaySigner::new(&RelaySignerConfiguration {
+                signer_number: index + 1,
                 listen_port: config.server_port + index as u64 + 200,
                 server_port: config.server_port + index as u64 + 300,
                 dial_to: bootstrap_peer_addr.clone(),
@@ -319,6 +324,7 @@ impl MithrilInfrastructure {
                 party_id: party_id.clone(),
                 work_dir: &config.work_dir,
                 bin_dir: &config.bin_dir,
+                use_dmq: config.use_dmq,
             })?;
             relay_signer.start()?;
 
@@ -389,6 +395,7 @@ impl MithrilInfrastructure {
                 mithril_era_reader_adapter: &config.mithril_era_reader_adapter,
                 mithril_era_marker_address: &config.devnet.mithril_era_marker_address()?,
                 enable_certification,
+                use_dmq: config.use_dmq,
             })?;
             signer.start().await?;
 

mithril-test-lab/mithril-end-to-end/src/mithril/relay_aggregator.rs

@@ -1,4 +1,5 @@
 use crate::utils::MithrilCommand;
+use crate::{Aggregator, DEVNET_MAGIC_ID};
 use mithril_common::StdResult;
 use std::collections::HashMap;
 use std::path::Path;
@@ -14,21 +15,33 @@ pub struct RelayAggregator {
 
 impl RelayAggregator {
     pub fn new(
-        name: String,
+        index: usize,
         listen_port: u64,
         dial_to: Option<String>,
         aggregator_endpoint: &str,
         work_dir: &Path,
         bin_dir: &Path,
+        use_dmq: bool,
     ) -> StdResult<Self> {
+        let name = Aggregator::name_suffix(index);
         let listen_port_str = format!("{listen_port}");
+        let magic_id = DEVNET_MAGIC_ID.to_string();
         let mut env = HashMap::from([
             ("LISTEN_PORT", listen_port_str.as_str()),
+            ("NETWORK", "devnet"),
+            ("NETWORK_MAGIC", &magic_id),
             ("AGGREGATOR_ENDPOINT", aggregator_endpoint),
         ]);
         if let Some(dial_to) = &dial_to {
             env.insert("DIAL_TO", dial_to);
         }
+        let dmq_node_socket_path = work_dir.join(format!("dmq-aggregator-{index}.socket"));
+        if use_dmq {
+            env.insert(
+                "DMQ_NODE_SOCKET_PATH",
+                dmq_node_socket_path.to_str().unwrap(),
+            );
+        }
         let args = vec!["-vvv", "aggregator"];
 
         let mut command = MithrilCommand::new("mithril-relay", work_dir, bin_dir, env, &args)?;

mithril-test-lab/mithril-end-to-end/src/mithril/relay_signer.rs

@@ -1,3 +1,4 @@
+use crate::DEVNET_MAGIC_ID;
 use crate::utils::MithrilCommand;
 use mithril_common::StdResult;
 use mithril_common::entities::PartyId;
@@ -6,6 +7,7 @@ use std::path::Path;
 use tokio::process::Child;
 
 pub struct RelaySignerConfiguration<'a> {
+    pub signer_number: usize,
     pub listen_port: u64,
     pub server_port: u64,
     pub dial_to: Option<String>,
@@ -15,6 +17,7 @@ pub struct RelaySignerConfiguration<'a> {
     pub party_id: PartyId,
     pub work_dir: &'a Path,
     pub bin_dir: &'a Path,
+    pub use_dmq: bool,
 }
 
 #[derive(Debug)]
@@ -28,15 +31,20 @@ pub struct RelaySigner {
 
 impl RelaySigner {
     pub fn new(configuration: &RelaySignerConfiguration) -> StdResult<Self> {
+        let party_id = configuration.party_id.to_owned();
         let listen_port_str = format!("{}", configuration.listen_port);
         let server_port_str = format!("{}", configuration.server_port);
+
+        let magic_id = DEVNET_MAGIC_ID.to_string();
         let relay_signer_registration_mode =
             configuration.relay_signer_registration_mode.to_string();
         let relay_signature_registration_mode =
             configuration.relay_signature_registration_mode.to_string();
         let mut env = HashMap::from([
             ("LISTEN_PORT", listen_port_str.as_str()),
             ("SERVER_PORT", server_port_str.as_str()),
+            ("NETWORK", "devnet"),
+            ("NETWORK_MAGIC", &magic_id),
             ("AGGREGATOR_ENDPOINT", configuration.aggregator_endpoint),
             ("SIGNER_REPEATER_DELAY", "100"),
             (
@@ -51,6 +59,15 @@ impl RelaySigner {
         if let Some(dial_to) = &configuration.dial_to {
             env.insert("DIAL_TO", dial_to);
         }
+        let dmq_node_socket_path = configuration
+            .work_dir
+            .join(format!("dmq-signer-{}.socket", configuration.signer_number));
+        if configuration.use_dmq {
+            env.insert(
+                "DMQ_NODE_SOCKET_PATH",
+                dmq_node_socket_path.to_str().unwrap(),
+            );
+        }
         let args = vec!["-vvv", "signer"];
 
         let mut command = MithrilCommand::new(
@@ -65,7 +82,7 @@ impl RelaySigner {
         Ok(Self {
             listen_port: configuration.listen_port,
             server_port: configuration.server_port,
-            party_id: configuration.party_id.to_owned(),
+            party_id,
             command,
             process: None,
         })

mithril-test-lab/mithril-end-to-end/src/mithril/signer.rs

@@ -25,6 +25,7 @@ pub struct SignerConfig<'a> {
     pub mithril_era_reader_adapter: &'a str,
     pub mithril_era_marker_address: &'a str,
     pub enable_certification: bool,
+    pub use_dmq: bool,
 }
 
 #[derive(Debug)]
@@ -54,6 +55,7 @@ impl Signer {
         let mithril_run_interval = format!("{}", signer_config.mithril_run_interval);
         let mut env = HashMap::from([
             ("NETWORK", "devnet"),
+            ("NETWORK_MAGIC", &magic_id),
             ("RUN_INTERVAL", &mithril_run_interval),
             ("AGGREGATOR_ENDPOINT", &signer_config.aggregator_endpoint),
             (
@@ -65,7 +67,6 @@ impl Signer {
                 signer_config.store_dir.to_str().unwrap(),
             ),
             ("STORE_RETENTION_LIMIT", "10"),
-            ("NETWORK_MAGIC", &magic_id),
             (
                 "CARDANO_NODE_SOCKET_PATH",
                 signer_config.pool_node.socket_path.to_str().unwrap(),
@@ -96,6 +97,15 @@ impl Signer {
         } else {
             env.insert("PARTY_ID", &party_id);
         }
+        let dmq_node_socket_path = signer_config
+            .work_dir
+            .join(format!("dmq-signer-{}.socket", signer_config.signer_number));
+        if signer_config.use_dmq {
+            env.insert(
+                "DMQ_NODE_SOCKET_PATH",
+                dmq_node_socket_path.to_str().unwrap(),
+            );
+        }
         let args = vec!["-vvv"];
 
         let mut command = MithrilCommand::new(

mithril-test-lab/mithril-end-to-end/src/stress_test/aggregator_helpers.rs

@@ -36,6 +36,7 @@ pub async fn bootstrap_aggregator(
         signed_entity_types: &signed_entity_types,
         chain_observer_type,
         leader_aggregator_endpoint: &None,
+        use_dmq: false,
     })
     .unwrap();