feat(connection): cap gateway courtesy links

Author: sanity

crates/core/src/node/network_bridge/p2p_protoc.rs

@@ -1598,11 +1598,34 @@ impl P2pConnManager {
                 .connection_manager
                 .prune_in_transit_connection(&peer_id);
             let loc = pending_loc.unwrap_or_else(|| Location::from_address(&peer_id.addr));
-            self.bridge
+            let eviction_candidate = self
+                .bridge
                 .op_manager
                 .ring
-                .add_connection(loc, peer_id.clone(), false)
+                .add_connection(loc, peer_id.clone(), false, courtesy)
                 .await;
+            if let Some(victim) = eviction_candidate {
+                if victim == peer_id {
+                    tracing::debug!(
+                        %peer_id,
+                        "Courtesy eviction candidate matched current connection; skipping drop"
+                    );
+                } else {
+                    tracing::info!(
+                        %victim,
+                        %peer_id,
+                        courtesy_limit = true,
+                        "Courtesy connection budget exceeded; dropping oldest courtesy peer"
+                    );
+                    if let Err(error) = self.bridge.drop_connection(&victim).await {
+                        tracing::warn!(
+                            %victim,
+                            ?error,
+                            "Failed to drop courtesy connection after hitting budget"
+                        );
+                    }
+                }
+            }
         }
         Ok(())
     }

crates/core/src/ring/connection_manager.rs

@@ -1,6 +1,7 @@
 use parking_lot::Mutex;
 use rand::prelude::IndexedRandom;
-use std::collections::{btree_map::Entry, BTreeMap};
+use std::collections::{btree_map::Entry, BTreeMap, VecDeque};
+use std::time::Instant;
 
 use crate::topology::{Limits, TopologyManager};
 
@@ -22,8 +23,17 @@ pub(crate) struct ConnectionManager {
     pub max_connections: usize,
     pub rnd_if_htl_above: usize,
     pub pub_key: Arc<TransportPublicKey>,
+    courtesy_links: Arc<Mutex<VecDeque<CourtesyLink>>>,
+    max_courtesy_links: usize,
 }
 
+#[derive(Clone)]
+struct CourtesyLink {
+    peer: PeerId,
+}
+
+const MAX_COURTESY_LINKS: usize = 10;
+
 impl ConnectionManager {
     pub fn new(config: &NodeConfig) -> Self {
         let min_connections = if let Some(v) = config.min_number_conn {
@@ -111,9 +121,43 @@ impl ConnectionManager {
             max_connections,
             rnd_if_htl_above,
             pub_key: Arc::new(pub_key),
+            courtesy_links: Arc::new(Mutex::new(VecDeque::new())),
+            max_courtesy_links: if is_gateway { MAX_COURTESY_LINKS } else { 0 },
         }
     }
 
+    fn register_courtesy_connection(&self, peer: &PeerId) -> Option<PeerId> {
+        if !self.is_gateway || self.max_courtesy_links == 0 {
+            return None;
+        }
+        let mut links = self.courtesy_links.lock();
+        if links.len() == self.max_courtesy_links && links.iter().all(|entry| entry.peer != *peer) {
+            tracing::debug!(
+                %peer,
+                max = self.max_courtesy_links,
+                "register_courtesy_connection: budget full before inserting"
+            );
+        }
+        links.retain(|entry| entry.peer != *peer);
+        links.push_back(CourtesyLink { peer: peer.clone() });
+        if links.len() > self.max_courtesy_links {
+            links.pop_front().map(|entry| entry.peer)
+        } else {
+            None
+        }
+    }
+
+    fn unregister_courtesy_connection(&self, peer: &PeerId) {
+        if !self.is_gateway {
+            return;
+        }
+        let mut links = self.courtesy_links.lock();
+        if links.is_empty() {
+            return;
+        }
+        links.retain(|entry| entry.peer != *peer);
+    }
+
     /// Whether a node should accept a new node connection or not based
     /// on the relative location and other conditions.
     ///
@@ -200,7 +244,7 @@ impl ConnectionManager {
             return true;
         }
 
-        const GATEWAY_DIRECT_ACCEPT_LIMIT: usize = 2;
+        const GATEWAY_DIRECT_ACCEPT_LIMIT: usize = 10;
         if self.is_gateway {
             let direct_total = open + reserved_before;
             if direct_total >= GATEWAY_DIRECT_ACCEPT_LIMIT {
@@ -350,8 +394,20 @@ impl ConnectionManager {
         self.prune_connection(peer, false)
     }
 
-    pub fn add_connection(&self, loc: Location, peer: PeerId, was_reserved: bool) {
-        tracing::info!(%peer, %loc, %was_reserved, "Adding connection to topology");
+    pub fn add_connection(
+        &self,
+        loc: Location,
+        peer: PeerId,
+        was_reserved: bool,
+        courtesy: bool,
+    ) -> Option<PeerId> {
+        tracing::info!(
+            %peer,
+            %loc,
+            %was_reserved,
+            courtesy,
+            "Adding connection to topology"
+        );
         debug_assert!(self.get_peer_key().expect("should be set") != peer);
         if was_reserved {
             let old = self
@@ -381,6 +437,13 @@ impl ConnectionManager {
         self.open_connections
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
         std::mem::drop(lop);
+
+        if courtesy {
+            self.register_courtesy_connection(&peer)
+        } else {
+            self.unregister_courtesy_connection(&peer);
+            None
+        }
     }
 
     pub fn update_peer_identity(&self, old_peer: &PeerId, new_peer: PeerId) -> bool {
@@ -452,6 +515,7 @@ impl ConnectionManager {
         }
 
         if is_alive {
+            self.unregister_courtesy_connection(peer);
             self.open_connections
                 .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
         } else {

crates/core/src/ring/mod.rs

@@ -228,14 +228,28 @@ impl Ring {
             .record_request(recipient, target, request_type);
     }
 
-    pub async fn add_connection(&self, loc: Location, peer: PeerId, was_reserved: bool) {
-        tracing::info!(%peer, this = ?self.connection_manager.get_peer_key(), %was_reserved, "Adding connection to peer");
-        self.connection_manager
-            .add_connection(loc, peer.clone(), was_reserved);
+    pub async fn add_connection(
+        &self,
+        loc: Location,
+        peer: PeerId,
+        was_reserved: bool,
+        courtesy: bool,
+    ) -> Option<PeerId> {
+        tracing::info!(
+            %peer,
+            this = ?self.connection_manager.get_peer_key(),
+            %was_reserved,
+            courtesy,
+            "Adding connection to peer"
+        );
+        let eviction_candidate =
+            self.connection_manager
+                .add_connection(loc, peer.clone(), was_reserved, courtesy);
         self.event_register
             .register_events(Either::Left(NetEventLog::connected(self, peer, loc)))
             .await;
-        self.refresh_density_request_cache()
+        self.refresh_density_request_cache();
+        eviction_candidate
     }
 
     pub fn update_connection_identity(&self, old_peer: &PeerId, new_peer: PeerId) {