feat(connection-pool): Add connection limits

crates/connected-client/src/client.rs

@@ -27,7 +27,7 @@ use futures::{
 };
 use libp2p::core::either::EitherError;
 use libp2p::core::Multiaddr;
-use libp2p::swarm::{ConnectionHandlerUpgrErr, SwarmEvent};
+use libp2p::swarm::{ConnectionHandlerUpgrErr, SwarmBuilder, SwarmEvent};
 use libp2p::{identity::Keypair, PeerId, Swarm};
 
 use fluence_libp2p::{
@@ -111,7 +111,7 @@ impl Client {
             let behaviour = ClientBehaviour::new(protocol_config);
 
             let transport = build_transport(transport, self.key_pair.clone(), transport_timeout);
-            Swarm::with_threadpool_executor(transport, behaviour, self.peer_id)
+            SwarmBuilder::with_async_std_executor(transport, behaviour, self.peer_id).build()
         };
 
         match Swarm::dial(&mut swarm, node.clone()) {

crates/kademlia/src/behaviour.rs

@@ -626,6 +626,7 @@ mod tests {
     use libp2p::core::Multiaddr;
     use libp2p::identity::Keypair;
     use libp2p::multiaddr::Protocol;
+    use libp2p::swarm::SwarmBuilder;
     use libp2p::PeerId;
     use libp2p::Swarm;
 
@@ -656,8 +657,12 @@ mod tests {
         let (kad, _) = Kademlia::new(config, None);
         let timeout = Duration::from_secs(20);
 
-        let mut swarm =
-            Swarm::with_threadpool_executor(build_memory_transport(kp, timeout), kad, peer_id);
+        let mut swarm = SwarmBuilder::with_async_std_executor(
+            build_memory_transport(kp, timeout),
+            kad,
+            peer_id,
+        )
+        .build();
 
         let mut maddr = create_memory_maddr();
         maddr.push(Protocol::P2p(peer_id.into()));

crates/server-config/src/defaults.rs

@@ -40,25 +40,35 @@ pub fn default_config_path() -> PathBuf {
 pub fn default_tcp_port() -> u16 {
     7777
 }
+
 pub fn default_listen_ip() -> IpAddr {
     "0.0.0.0".parse().unwrap()
 }
+
 pub fn default_socket_timeout() -> Duration {
     Duration::from_secs(20)
 }
+
+pub fn default_max_established_per_peer_limit() -> Option<u32> {
+    Some(5)
+}
+
 pub fn default_auto_particle_ttl() -> Duration {
     Duration::from_secs(200)
 }
 
 pub fn default_bootstrap_nodes() -> Vec<Multiaddr> {
     vec![]
 }
+
 pub fn default_websocket_port() -> u16 {
     9999
 }
+
 pub fn default_metrics_port() -> u16 {
     18080
 }
+
 pub fn default_metrics_enabled() -> bool {
     true
 }

crates/server-config/src/network_config.rs

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+use libp2p::swarm::ConnectionLimits;
 use libp2p::{core::Multiaddr, identity::Keypair, PeerId};
 use libp2p_metrics::Metrics;
 
@@ -37,6 +38,7 @@ pub struct NetworkConfig {
     pub allow_local_addresses: bool,
     pub connectivity_metrics: Option<ConnectivityMetrics>,
     pub connection_pool_metrics: Option<ConnectionPoolMetrics>,
+    pub connection_limits: ConnectionLimits,
 }
 
 impl NetworkConfig {
@@ -47,6 +49,7 @@ impl NetworkConfig {
         key_pair: Keypair,
         config: &ResolvedConfig,
         node_version: &'static str,
+        connection_limits: ConnectionLimits,
     ) -> Self {
         Self {
             node_version,
@@ -62,6 +65,7 @@ impl NetworkConfig {
             allow_local_addresses: config.allow_local_addresses,
             connectivity_metrics,
             connection_pool_metrics,
+            connection_limits,
         }
     }
 }

crates/server-config/src/node_config.rs

@@ -150,6 +150,19 @@ pub struct TransportConfig {
     #[serde(default = "default_socket_timeout")]
     #[serde(with = "humantime_serde")]
     pub socket_timeout: Duration,
+
+    pub max_pending_incoming: Option<u32>,
+
+    pub max_pending_outgoing: Option<u32>,
+
+    pub max_established_incoming: Option<u32>,
+
+    pub max_established_outgoing: Option<u32>,
+
+    #[serde(default = "default_max_established_per_peer_limit")]
+    pub max_established_per_peer: Option<u32>,
+
+    pub max_established: Option<u32>,
 }
 
 #[derive(Clone, Deserialize, Derivative)]
@@ -192,6 +205,7 @@ pub struct ListenConfig {
 #[derive(Clone, Deserialize, Debug, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
 #[repr(transparent)]
 pub struct PeerIdSerializable(#[serde(with = "peerid_serializer")] PeerId);
+
 impl Deref for PeerIdSerializable {
     type Target = PeerId;
 

particle-node/src/node.rs

@@ -34,6 +34,7 @@ use libp2p::{
     PeerId, Swarm, TransportError,
 };
 use libp2p_metrics::{Metrics, Recorder};
+use libp2p_swarm::{ConnectionLimits, SwarmBuilder};
 use prometheus_client::registry::Registry;
 
 use aquamarine::{
@@ -136,13 +137,28 @@ impl<RT: AquaRuntime> Node<RT> {
         let plumber_metrics = metrics_registry.as_mut().map(ParticleExecutorMetrics::new);
         let vm_pool_metrics = metrics_registry.as_mut().map(VmPoolMetrics::new);
 
+        let connection_limits = ConnectionLimits::default()
+            .with_max_pending_incoming(config.node_config.transport_config.max_pending_incoming)
+            .with_max_pending_outgoing(config.node_config.transport_config.max_pending_outgoing)
+            .with_max_established_incoming(
+                config.node_config.transport_config.max_established_incoming,
+            )
+            .with_max_established_outgoing(
+                config.node_config.transport_config.max_established_outgoing,
+            )
+            .with_max_established_per_peer(
+                config.node_config.transport_config.max_established_per_peer,
+            )
+            .with_max_established(config.node_config.transport_config.max_established);
+
         let network_config = NetworkConfig::new(
             libp2p_metrics,
             connectivity_metrics,
             connection_pool_metrics,
             key_pair,
             &config,
             node_version,
+            connection_limits,
         );
 
         let (swarm, connectivity, particle_stream) = Self::swarm(
@@ -279,9 +295,12 @@ impl<RT: AquaRuntime> Node<RT> {
         Connectivity,
         BackPressuredInlet<Particle>,
     ) {
+        let connection_limits = network_config.connection_limits.clone();
         let (behaviour, connectivity, particle_stream) =
             FluenceNetworkBehaviour::new(network_config);
-        let mut swarm = Swarm::with_threadpool_executor(transport, behaviour, local_peer_id);
+        let mut swarm = SwarmBuilder::with_async_std_executor(transport, behaviour, local_peer_id)
+            .connection_limits(connection_limits)
+            .build();
 
         // Add external addresses to Swarm
         external_addresses.iter().cloned().for_each(|addr| {
@@ -322,7 +341,6 @@ impl<RT: AquaRuntime> Node<RT> {
         particle_stream: BackPressuredInlet<Particle>,
         effects_stream: Inlet<Result<RoutingEffects, AquamarineApiError>>,
         swarm: Swarm<FluenceNetworkBehaviour>,
-
         connectivity: Connectivity,
         aquamarine_api: AquamarineApi,
         dispatcher: Dispatcher,
@@ -332,11 +350,9 @@ impl<RT: AquaRuntime> Node<RT> {
         spell_event_bus: SpellEventBus,
         spell_events_stream: Inlet<TriggerEvent>,
         sorcerer: Sorcerer,
-
         registry: Option<Registry>,
         services_metrics_backend: ServicesMetricsBackend,
         metrics_listen_addr: SocketAddr,
-
         builtins_management_peer_id: PeerId,
         key_manager: KeyManager,
     ) -> Box<Self> {

rust-toolchain.toml

@@ -1,4 +1,4 @@
 [toolchain]
 channel = "nightly-2022-12-06"
-targets = [ "x86_64-unknown-linux-gnu" ]
-components = [ "rustfmt", "clippy" ]
+targets = ["x86_64-unknown-linux-gnu"]
+components = ["rustfmt", "clippy"]