fix: debug mempool lock contention in tx submit peer client

examples/client/src/main.rs

@@ -8,7 +8,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let mut client = submit::submit_service_client::SubmitServiceClient::new(channel);
 
-    let tx = "84a300d901028182582071e8d419538ec00b262d6d625d02fec577ecad60feae0ad0238e74f97a0162b500018182581d606e2e6d54e1a27ad640786852e20c22fb982ebcee4773a2926aae4b391b00000001a1113ab4021a000292b1a100d9010281825820524e506f6a872c4ee3ee6d4b9913670c4b441860b3aa5438de92a676e20f527b5840ebb051997a9b920f366526c87e877ab525014257480ad22f78338f662068bd899850386ab2abb714bbc5c7e719bb0462e2de1c60000f33e81c043fc9db949a02f5f6";
+    let tx = "84a300d9010281825820a2aafdbbfd255fa1ca2372c24b8e1021a8c7bc20360c23d8e1d5ba1de0ccffc400018182581d603f79e7eab3ab95c1f78824872ac6fd65f79d120868057f2bd19306f81b000000012a61c6dd021a000292b1a100d90102818258205d4b008e92a42846add4d060e49d7427700ced0ab8eb73e559acc14d228ca5475840369982092dc0eba4d873bba58169334063d945cceca0616f83980889dacb6ccb18406ff6f7b17fa04cee51de26517946c015486ba51e5ab4ea39b0ede3b92a05f5f6";
 
     let bytes = hex::decode(tx)?;
 

src/pipeline/fanout/mock_ouroboros_tx_submit_server.rs

@@ -82,7 +82,6 @@ impl MockOuroborosTxSubmitPeerServer {
             }
         }
 
-
         // Recieve TxIds for the blocking request
         let txids_reply = match tx_server.receive_next_reply().await {
             Ok(reply) => reply,
@@ -152,7 +151,6 @@ impl MockOuroborosTxSubmitPeerServer {
             panic!("SERVER: error replying TxIds => {err:?}");
         }
 
-
         // Recieve Tx bodies
         let txs_reply = match tx_server.receive_next_reply().await {
             Ok(reply) => reply,
@@ -172,15 +170,17 @@ impl MockOuroborosTxSubmitPeerServer {
             let to_acknowledge_ids = collected_tx_ids.drain(..bodies.len()).collect::<Vec<_>>();
             let mut acknowledge_txs = self.acknowledge_txs.lock().unwrap();
             to_acknowledge_ids.iter().for_each(|EraTxId(_, hash)| {
-                acknowledge_txs.push(pallas::crypto::hash::Hash::new(hash.as_slice().try_into().unwrap()));
+                acknowledge_txs.push(pallas::crypto::hash::Hash::new(
+                    hash.as_slice().try_into().unwrap(),
+                ));
             });
             info!(
                 "SERVER: ack so far => {acknowledge}, State => {:?}, agency is theirs",
                 tx_server.state()
             );
             info!("SERVER: got Tx bodies => {}", bodies.len());
         }
-        
+
         info!("SERVER: done, closing connection");
         peer_server.abort().await;
         *self.is_done.write().unwrap() = true;

src/pipeline/fanout/mod.rs

@@ -1,4 +1,7 @@
-use std::{sync::Arc, time::Duration};
+use std::{
+    sync::Arc,
+    time::Duration
+};
 
 use gasket::framework::*;
 use serde::Deserialize;