This enables tracking messages across nodes

Merge pull request #1881 This enables tracking messages across nodes during testnet, e.g. originating `msg tag` as the transaction protocol is executed.
tari-project · May 22, 2020 · b8ad396 · b8ad396
2 parents e1b70d2 + e1e555c
commit b8ad396
Show file tree

Hide file tree

Showing 17 changed files with 227 additions and 88 deletions.
diff --git a/base_layer/p2p/src/services/liveness/service.rs b/base_layer/p2p/src/services/liveness/service.rs
@@ -367,7 +367,7 @@ where
             let excluded = self
                 .neighbours
                 .node_ids()
-                .into_iter()
+                .iter()
                 .chain(vec![node_id])
                 .cloned()
                 .collect();
@@ -441,7 +441,7 @@ where
     }
 
     async fn refresh_random_peer_pool(&mut self) -> Result<(), LivenessError> {
-        let excluded = self.neighbours.node_ids().into_iter().cloned().collect();
+        let excluded = self.neighbours.node_ids().to_vec();
 
         // Select a pool of random peers the same length as neighbouring peers
         let random_peers = self
@@ -565,6 +565,7 @@ mod test {
     use rand::rngs::OsRng;
     use std::time::Duration;
     use tari_comms::{
+        message::MessageTag,
         multiaddr::Multiaddr,
         peer_manager::{NodeId, Peer, PeerFeatures, PeerFlags},
     };
@@ -694,6 +695,7 @@ mod test {
                 message_type: DhtMessageType::None,
                 network: Network::LocalTest,
                 flags: Default::default(),
+                message_tag: MessageTag::new(),
             },
             authenticated_origin: None,
             source_peer,

diff --git a/base_layer/p2p/src/test_utils.rs b/base_layer/p2p/src/test_utils.rs
@@ -67,7 +67,7 @@ pub fn make_node_identity() -> Arc<NodeIdentity> {
     )
 }
 
-pub fn make_dht_header() -> DhtMessageHeader {
+pub fn make_dht_header(trace: MessageTag) -> DhtMessageHeader {
     DhtMessageHeader {
         version: 0,
         destination: NodeDestination::Unknown,
@@ -76,13 +76,15 @@ pub fn make_dht_header() -> DhtMessageHeader {
         message_type: DhtMessageType::None,
         network: Network::LocalTest,
         flags: DhtMessageFlags::NONE,
+        message_tag: trace,
     }
 }
 
 pub fn make_dht_inbound_message(node_identity: &NodeIdentity, message: Vec<u8>) -> DhtInboundMessage {
+    let msg_tag = MessageTag::new();
     DhtInboundMessage::new(
-        MessageTag::new(),
-        make_dht_header(),
+        msg_tag,
+        make_dht_header(msg_tag),
         Arc::new(Peer::new(
             node_identity.public_key().clone(),
             node_identity.node_id().clone(),

diff --git a/base_layer/wallet/src/output_manager_service/service.rs b/base_layer/wallet/src/output_manager_service/service.rs
@@ -189,8 +189,8 @@ where
                 },
                  // Incoming messages from the Comms layer
                 msg = base_node_response_stream.select_next_some() => {
-                    trace!(target: LOG_TARGET, "Handling Base Node Response");
-                    let (origin_public_key, inner_msg) = msg.into_origin_and_inner();
+                    let (origin_public_key, inner_msg) = msg.clone().into_origin_and_inner();
+                    trace!(target: LOG_TARGET, "Handling Base Node Response, Trace: {}", msg.dht_header.message_tag);
                     let result = self.handle_base_node_response(inner_msg).await.or_else(|resp| {
                         error!(target: LOG_TARGET, "Error handling base node service response from {}: {:?}", origin_public_key, resp);
                         Err(resp)
@@ -516,7 +516,7 @@ where
         received_output: &TransactionOutput,
     ) -> Result<(), OutputManagerError>
     {
-        let pending_transaction = self.db.fetch_pending_transaction_outputs(tx_id.clone()).await?;
+        let pending_transaction = self.db.fetch_pending_transaction_outputs(tx_id).await?;
 
         // Assumption: We are only allowing a single output per receiver in the current transaction protocols.
         if pending_transaction.outputs_to_be_received.len() != 1 ||
@@ -529,7 +529,7 @@ where
         }
 
         self.db
-            .confirm_pending_transaction_outputs(pending_transaction.tx_id.clone())
+            .confirm_pending_transaction_outputs(pending_transaction.tx_id)
             .await?;
 
         Ok(())
@@ -623,7 +623,7 @@ where
         outputs: &[TransactionOutput],
     ) -> Result<(), OutputManagerError>
     {
-        let pending_transaction = self.db.fetch_pending_transaction_outputs(tx_id.clone()).await?;
+        let pending_transaction = self.db.fetch_pending_transaction_outputs(tx_id).await?;
 
         // Check that outputs to be spent can all be found in the provided transaction inputs
         let mut inputs_confirmed = true;

diff --git a/base_layer/wallet/src/transaction_service/protocols/transaction_send_protocol.rs b/base_layer/wallet/src/transaction_service/protocols/transaction_send_protocol.rs
@@ -353,19 +353,21 @@ where TBackend: TransactionBackend + Clone + 'static
                 },
                 Some(send_states) => {
                     if send_states.len() == 1 {
+                        let msg_tag = send_states[0].tag;
                         debug!(
                             target: LOG_TARGET,
-                            "Transaction Finalized (TxId: {}) Direct Send to {} queued with Message Tag: {:?}",
+                            "Transaction Finalized (TxId: {}) Direct Send to {} queued with Message Tag: {}",
                             self.id,
                             self.dest_pubkey,
-                            send_states[0].tag,
+                            &msg_tag,
                         );
                         match send_states.wait_single().await {
                             true => {
                                 info!(
                                     target: LOG_TARGET,
-                                    "Direct Send of Transaction Finalized message for TX_ID: {} was successful",
-                                    self.id
+                                    "Direct Send of Transaction Finalized message for TX_ID: {} was successful ({})",
+                                    self.id,
+                                    msg_tag
                                 );
                             },
                             false => {
@@ -507,7 +509,7 @@ where TBackend: TransactionBackend + Clone + 'static
         if send_states.len() == 1 {
             debug!(
                 target: LOG_TARGET,
-                "Transaction (TxId: {}) Direct Send to {} queued with Message Tag: {:?}",
+                "Transaction (TxId: {}) Direct Send to {} queued with Message Tag: {}",
                 self.id,
                 self.dest_pubkey,
                 send_states[0].tag,