feat(peersync)!: push local peer record to connected peer on change (#…

…1202)

Description
---
feat(peersync)!: push local peer record to connected peer on change
Add libp2p logs to log4rs

Motivation and Context
---
Connected peers should have an up-to-date and shareable address record
for peers they are connected to. This PR adds this to the peer sync
protocol.

How Has This Been Tested?
---
Manually - looked at logs to observe it working

What process can a PR reviewer use to test or verify this change?
---
Given node A connected to node B prior, a node C can request node B from
Node A and always get a response.

Breaking Changes
---

- [ ] None
- [ ] Requires data directory to be deleted
- [x] Other - Please specify

BREAKING CHANGE: peer sync protocol is not compatible with previous
versions

applications/tari_indexer/log4rs_sample.yml

@@ -72,6 +72,23 @@ appenders:
     encoder:
       pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [{X(node-public-key)},{X(node-id)}] {l:5} {m} // {f}:{L}{n}"
 
+  # An appender named "libp2p" that writes to a file with a custom pattern encoder
+  libp2p:
+    kind: rolling_file
+    path: "{{log_dir}}/log/indexer/libp2p.log"
+    policy:
+      kind: compound
+      trigger:
+        kind: size
+        limit: 10mb
+      roller:
+        kind: fixed_window
+        base: 1
+        count: 5
+        pattern: "{{log_dir}}/log/indexer/libp2p.{}.log"
+    encoder:
+      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [Thread:{I}] {l:5} {m}{n} // {f}:{L} "
+
   # An appender named "other" that writes to a file with a custom pattern encoder
   other:
     kind: rolling_file
@@ -124,13 +141,7 @@ loggers:
       - stdout
     additive: false
 
-  # Route log events sent to the "comms" logger to the "network" appender
-  comms:
-    level: debug
-    appenders:
-      - network
-
-    # Route log events sent to the "yamux" logger to the "network" appender
+  # Route log events sent to the "yamux" logger to the "network" appender
   yamux:
     level: info
     appenders:
@@ -152,3 +163,84 @@ loggers:
     appenders:
       - other
     additive: false
+  # libp2p
+  libp2p_identify:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_noise:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_peersync:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_gossipsub:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  quinn:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  quinn_proto:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_core:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_mdns:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_swarm:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  hickory_proto:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  multistream_select:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_tcp:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_quic:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_kad:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_ping:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_messaging:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false

applications/tari_swarm_daemon/src/process_manager/manager.rs

@@ -101,15 +101,15 @@ impl ProcessManager {
             }
         }
 
-        let num_vns = if self.skip_registration {
+        let num_blocks = if self.skip_registration {
             0
         } else {
-            self.instance_manager.num_validator_nodes()
+            self.instance_manager.num_validator_nodes() +
+                u64::try_from(templates_to_register.len()).expect("impossibly many templates")
         };
-        let num_blocks = num_vns + u64::try_from(templates_to_register.len()).unwrap();
 
-        if !self.skip_registration {
-            // Mine some initial funds, guessing 10 blocks extra to allow for coinbase maturity
+        if num_blocks > 0 {
+            // Mine some initial funds, guessing 10 blocks extra is sufficient for coinbase maturity
             self.mine(num_blocks + 10).await.context("initial mining failed")?;
             self.wait_for_wallet_funds(num_blocks)
                 .await
@@ -121,9 +121,8 @@ impl ProcessManager {
             for templates in templates_to_register {
                 self.register_template(templates).await?;
             }
-        }
 
-        if num_blocks > 0 {
+            // "Mine in" the validators and templates
             self.mine(20).await?;
         }
 

applications/tari_validator_node/log4rs_sample.yml

@@ -89,6 +89,23 @@ appenders:
     encoder:
       pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [{X(node-public-key)},{X(node-id)}] {l:5} {m} // {f}:{L}{n}"
 
+  # An appender named "libp2p" that writes to a file with a custom pattern encoder
+  libp2p:
+    kind: rolling_file
+    path: "{{log_dir}}/log/validator-node/libp2p.log"
+    policy:
+      kind: compound
+      trigger:
+        kind: size
+        limit: 10mb
+      roller:
+        kind: fixed_window
+        base: 1
+        count: 5
+        pattern: "{{log_dir}}/log/validator-node/libp2p.{}.log"
+    encoder:
+      pattern: "{d(%Y-%m-%d %H:%M:%S.%f)} [{t}] [Thread:{I}] {l:5} {m}{n} // {f}:{L} "
+
 
   # An appender named "other" that writes to a file with a custom pattern encoder
   other:
@@ -196,3 +213,84 @@ loggers:
     appenders:
       - other
     additive: false
+  # libp2p
+  libp2p_identify:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_noise:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_peersync:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_gossipsub:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  quinn:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  quinn_proto:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_core:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_mdns:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_swarm:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  hickory_proto:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  multistream_select:
+    level: info
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_tcp:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_quic:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_kad:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_ping:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false
+  libp2p_messaging:
+    level: debug
+    appenders:
+      - libp2p
+    additive: false

networking/core/src/relay_state.rs

@@ -3,11 +3,11 @@
 
 use std::collections::{HashMap, HashSet};
 
-use libp2p::{multiaddr::Protocol, Multiaddr, PeerId};
+use libp2p::{multiaddr::Protocol, swarm::ConnectionId, Multiaddr, PeerId};
 use rand::seq::IteratorRandom;
 
 #[derive(Debug, Clone, Default)]
-pub struct RelayState {
+pub(crate) struct RelayState {
     selected_relay: Option<RelayPeer>,
     possible_relays: HashMap<PeerId, HashSet<Multiaddr>>,
 }
@@ -40,6 +40,23 @@ impl RelayState {
         self.selected_relay.as_mut()
     }
 
+    pub fn set_relay_peer(&mut self, peer_id: PeerId, dialled_address: Option<Multiaddr>) -> bool {
+        if self.selected_relay.as_ref().map_or(false, |p| p.peer_id == peer_id) {
+            return true;
+        }
+
+        if let Some(addrs) = self.possible_relays.get(&peer_id) {
+            self.selected_relay = Some(RelayPeer {
+                peer_id,
+                addresses: addrs.iter().cloned().collect(),
+                circuit_connection_id: None,
+                remote_address: dialled_address,
+            });
+            return true;
+        }
+        false
+    }
+
     pub fn possible_relays(&self) -> impl Iterator<Item = (&PeerId, &HashSet<Multiaddr>)> {
         self.possible_relays.iter()
     }
@@ -49,10 +66,7 @@ impl RelayState {
     }
 
     pub fn has_active_relay(&self) -> bool {
-        self.selected_relay
-            .as_ref()
-            .map(|r| r.is_circuit_established)
-            .unwrap_or(false)
+        self.selected_relay.as_ref().map(|r| r.has_circuit()).unwrap_or(false)
     }
 
     pub fn add_possible_relay(&mut self, peer: PeerId, address: Multiaddr) {
@@ -70,16 +84,22 @@ impl RelayState {
         self.selected_relay = Some(RelayPeer {
             peer_id: *peer,
             addresses: addrs.iter().cloned().collect(),
-            is_circuit_established: false,
-            dialled_address: None,
+            circuit_connection_id: None,
+            remote_address: None,
         });
     }
 }
 
 #[derive(Debug, Clone)]
-pub struct RelayPeer {
+pub(crate) struct RelayPeer {
     pub peer_id: PeerId,
     pub addresses: Vec<Multiaddr>,
-    pub is_circuit_established: bool,
-    pub dialled_address: Option<Multiaddr>,
+    pub circuit_connection_id: Option<ConnectionId>,
+    pub remote_address: Option<Multiaddr>,
+}
+
+impl RelayPeer {
+    pub fn has_circuit(&self) -> bool {
+        self.circuit_connection_id.is_some()
+    }
 }