feat: add ability to increase relay circuits (#159)

tari-project · Nov 15, 2024 · bfe2cef · bfe2cef
1 parent d8399be
commit bfe2cef
Show file tree

Hide file tree

Showing 7 changed files with 75 additions and 22 deletions.
diff --git a/src/cli/args.rs b/src/cli/args.rs
@@ -95,6 +95,14 @@ pub(crate) struct StartArgs {
     #[arg(long, value_name = "relay-server-disabled", default_value_t = false)]
     pub relay_server_disabled: bool,
 
+    /// Relay Server Max Circuits
+    #[arg(long, value_name = "relay-server-max-circuits")]
+    pub relay_server_max_circuits: Option<usize>,
+
+    /// Relay Server Max Circuits per peer
+    #[arg(long, value_name = "relay-server-max-circuits-per-peer")]
+    pub relay_server_max_circuits_per_peer: Option<usize>,
+
     /// HTTP server disabled
     ///
     /// If set, local HTTP server (stats, health-check, status etc...) is disabled.

diff --git a/src/cli/commands/util.rs b/src/cli/commands/util.rs
@@ -93,6 +93,8 @@ pub async fn server(
     config_builder.with_is_seed_peer(args.is_seed_peer);
     config_builder.with_mdns_enabled(!args.mdns_disabled);
     config_builder.with_relay_disabled(args.relay_server_disabled);
+    config_builder.with_relay_max_circuits(args.relay_server_max_circuits);
+    config_builder.with_relay_max_circuits_per_peer(args.relay_server_max_circuits_per_peer);
     config_builder.with_http_server_enabled(!args.http_server_disabled);
     config_builder.with_user_agent(
         args.user_agent

diff --git a/src/server/config.rs b/src/server/config.rs
@@ -24,6 +24,8 @@ pub struct Config {
     pub max_incoming_connections: Option<u32>,
     pub max_outgoing_connections: Option<u32>,
     pub network_silence_delay: u64,
+    pub max_relay_circuits: Option<usize>,
+    pub max_relay_circuits_per_peer: Option<usize>,
 }
 
 impl Default for Config {
@@ -39,6 +41,8 @@ impl Default for Config {
             max_incoming_connections: Some(100),
             max_outgoing_connections: Some(16),
             network_silence_delay: 300,
+            max_relay_circuits: None,
+            max_relay_circuits_per_peer: None,
         }
     }
 }
@@ -142,6 +146,16 @@ impl ConfigBuilder {
         self
     }
 
+    pub fn with_relay_max_circuits(&mut self, config: Option<usize>) -> &mut Self {
+        self.config.max_relay_circuits = config;
+        self
+    }
+
+    pub fn with_relay_max_circuits_per_peer(&mut self, config: Option<usize>) -> &mut Self {
+        self.config.max_relay_circuits_per_peer = config;
+        self
+    }
+
     pub fn with_http_server_enabled(&mut self, config: bool) -> &mut Self {
         self.config.http_server.enabled = config;
         self

diff --git a/src/server/http/stats_collector.rs b/src/server/http/stats_collector.rs
@@ -169,7 +169,7 @@ impl StatsCollector {
                             let formatter = Formatter::new();
 
                             info!(target: LOG_TARGET,
-                                    "========= Uptime: {}. V{} Chains:  Rx {}..{}, Sha3 {}..{}. Difficulty (Target/Network): Rx: {}/{} Sha3x: {}/{} Miner(A/R): {}/{}. Pool(A/R) {}/{}. Peers(a/g/b) {}/{}/{} libp2p (i/o) {}/{} Last gossip: {}==== ",
+                                    "========= Uptime: {}. v{} Chains:  Rx {}..{}, Sha3 {}..{}. Difficulty (Target/Network): Rx: {}/{} Sha3x: {}/{} Miner(A/R): {}/{}. Pool(A/R) {}/{}. Peers(a/g/b) {}/{}/{} libp2p (i/o) {}/{} Last gossip: {}==== ",
                                     humantime::format_duration(Duration::from_secs(
                                         EpochTime::now().as_u64().checked_sub(
                                             self.first_stat_received.unwrap_or(EpochTime::now()).as_u64())

diff --git a/src/server/p2p/network.rs b/src/server/p2p/network.rs
@@ -625,15 +625,22 @@ where S: ShareChain
                             }
                             info!(target: NEW_TIP_NOTIFY_LOGGING_LOG_TARGET, "[SQUAD_NEW_BLOCK_TOPIC] New block from gossip: {source_peer:?} -> {payload:?}");
 
+                            if self.network_peer_store.is_whitelisted(&source_peer) {
+                                warn!(target: LOG_TARGET, squad = &self.config.squad; "Received a block from a peer {}, but it is not whitelisted. Will process anyway, but may not be able to switch to this chain. Heights:{}", source_peer, &payload.new_blocks.iter().map(|b| b.0.to_string()).join(","));
+                                // return Ok(MessageAcceptance::Accept);
+                            }
                             self.network_peer_store
                                 .add_last_new_tip_notify(&source_peer, payload.clone());
-                            // let _ = self.swarm
-                            //     .behaviour_mut()
-                            //     .peer_sync
-                            //     .add_want_peers(vec![source_peer.clone()])
-                            //     .await.inspect_err(|error| {
-                            //     info!(target: LOG_TARGET, squad = &self.config.squad; "Failed to add want peers:
-                            // {error:?}");     });
+                            let _ = self
+                                .swarm
+                                .behaviour_mut()
+                                .peer_sync
+                                .add_want_peers(vec![source_peer.clone()])
+                                .await
+                                .inspect_err(|error| {
+                                    info!(target: LOG_TARGET, squad = &self.config.squad; "Failed to add want peers:
+                            {error:?}");
+                                });
                             // If we don't have this peer, try do peer exchange
                             // if !self.network_peer_store.exists(message_peer) {
                             //     self.initiate_direct_peer_exchange(message_peer).await;
@@ -1360,6 +1367,8 @@ where S: ShareChain
                 },
                 ServerNetworkBehaviourEvent::Ping(event) => {
                     info!(target: LOG_TARGET, "[PING]: {event:?}");
+                    // Remove a peer from the greylist if we are in contact with them
+                    self.network_peer_store.set_last_ping(&event.peer, EpochTime::now());
                 },
             },
             _ => {},

diff --git a/src/server/p2p/peer_store.rs b/src/server/p2p/peer_store.rs
@@ -50,6 +50,7 @@ pub(crate) struct PeerStoreRecord {
     pub last_grey_list_reason: Option<String>,
     pub catch_up_attempts: u64,
     pub last_new_tip_notify: Option<Arc<NotifyNewTipBlock>>,
+    pub last_ping: Option<EpochTime>,
 }
 
 impl PeerStoreRecord {
@@ -64,6 +65,7 @@ impl PeerStoreRecord {
             last_grey_list_reason: None,
             catch_up_attempts: 0,
             last_new_tip_notify: None,
+            last_ping: None,
         }
     }
 
@@ -73,19 +75,6 @@ impl PeerStoreRecord {
     }
 }
 
-/// Tip of height from known peers.
-// #[derive(Copy, Clone, Debug)]
-// pub struct PeerStoreBlockHeightTip {
-//     pub peer_id: PeerId,
-//     pub height: u64,
-// }
-
-// impl PeerStoreBlockHeightTip {
-//     pub fn new(peer_id: PeerId, height: u64) -> Self {
-//         Self { peer_id, height }
-//     }
-// }
-
 pub enum AddPeerStatus {
     NewPeer,
     Existing,
@@ -115,6 +104,30 @@ impl PeerStore {
         }
     }
 
+    pub fn set_last_ping(&mut self, peer_id: &PeerId, timestamp: EpochTime) {
+        if let Some(entry) = self.whitelist_peers.get_mut(&peer_id.to_base58()) {
+            let mut new_record = entry.clone();
+            new_record.last_ping = Some(timestamp);
+            *entry = new_record;
+        }
+        if let Some(entry) = self.greylist_peers.get_mut(&peer_id.to_base58()) {
+            let mut new_record = entry.clone();
+            new_record.last_ping = Some(timestamp);
+            *entry = new_record;
+            // Move it to the whitelist
+            let mut record = self.greylist_peers.remove(&peer_id.to_base58()).unwrap();
+            record.num_grey_listings = 0;
+
+            self.whitelist_peers.insert(peer_id.to_base58(), record);
+        }
+
+        if let Some(entry) = self.blacklist_peers.get_mut(&peer_id.to_base58()) {
+            let mut new_record = entry.clone();
+            new_record.last_ping = Some(timestamp);
+            *entry = new_record;
+        }
+    }
+
     pub fn add_seed_peer(&mut self, peer_id: PeerId) {
         self.seed_peers.push(peer_id);
     }

diff --git a/src/server/p2p/setup.rs b/src/server/p2p/setup.rs
@@ -111,9 +111,16 @@ pub(crate) async fn new_swarm(config: &config::Config) -> Result<Swarm<ServerNet
             }
 
             // relay server
-      let relay_config =  relay::Config{
+      let mut relay_config =  relay::Config{
         ..Default::default()
     };
+    if let Some(max) = config.max_relay_circuits  {
+        relay_config.max_circuits = max;
+    }
+    if let Some(max) = config.max_relay_circuits_per_peer {
+        relay_config.max_circuits_per_peer = max;
+    }
+
 
     let peer_sync =
     libp2p_peersync::Behaviour::new(key_pair.clone(), MemoryPeerStore::new(), libp2p_peersync::Config::default());