feat(identify): make timeout and concurrent streams configurable

Cargo.toml

@@ -84,7 +84,7 @@ libp2p-dcutr = { version = "0.12.0", path = "protocols/dcutr" }
 libp2p-dns = { version = "0.42.0", path = "transports/dns" }
 libp2p-floodsub = { version = "0.45.0", path = "protocols/floodsub" }
 libp2p-gossipsub = { version = "0.48.0", path = "protocols/gossipsub" }
-libp2p-identify = { version = "0.45.1", path = "protocols/identify" }
+libp2p-identify = { version = "0.45.2", path = "protocols/identify" }
 libp2p-identity = { version = "0.2.9" }
 libp2p-kad = { version = "0.47.0", path = "protocols/kad" }
 libp2p-mdns = { version = "0.46.0", path = "protocols/mdns" }

protocols/identify/CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.45.2
+
+- Make stream timeout and maximum number of concurrent streams configurable on identify.
+  See [PR 5654](https://github.com/libp2p/rust-libp2p/pull/5654).
+
 ## 0.45.1
 
 - Add `hide_listen_addrs` option to prevent leaking (local) listen addresses.

protocols/identify/Cargo.toml

@@ -3,7 +3,7 @@ name = "libp2p-identify"
 edition = "2021"
 rust-version = { workspace = true }
 description = "Nodes identification protocol for libp2p"
-version = "0.45.1"
+version = "0.45.2"
 authors = ["Parity Technologies <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

protocols/identify/src/behaviour.rs

@@ -42,6 +42,9 @@ use std::{
     time::Duration,
 };
 
+const STREAM_TIMEOUT: Duration = Duration::from_secs(60);
+const MAX_CONCURRENT_STREAMS_PER_CONNECTION: usize = 10;
+
 /// Whether an [`Multiaddr`] is a valid for the QUIC transport.
 fn is_quic_addr(addr: &Multiaddr, v1: bool) -> bool {
     use Protocol::*;
@@ -153,6 +156,17 @@ pub struct Config {
     ///
     /// Disabled by default.
     pub hide_listen_addrs: bool,
+
+    /// Maximum duration for which an identification request stream may remain
+    /// active before timing out.
+    ///
+    /// Defaults to 1 minute.
+    pub stream_timeout: Duration,
+
+    /// Maximum number of concurrent identification request streams per connection.
+    ///
+    /// Defaults to 10.
+    pub max_concurrent_streams_per_connection: usize,
 }
 
 impl Config {
@@ -167,6 +181,8 @@ impl Config {
             push_listen_addr_updates: false,
             cache_size: 100,
             hide_listen_addrs: false,
+            stream_timeout: STREAM_TIMEOUT,
+            max_concurrent_streams_per_connection: MAX_CONCURRENT_STREAMS_PER_CONNECTION,
         }
     }
 
@@ -202,6 +218,20 @@ impl Config {
         self.hide_listen_addrs = b;
         self
     }
+
+    /// Configures maximum allowed duration for which an active identification request
+    /// stream is allowed to timeout.
+    pub fn with_stream_timeout(mut self, t: Duration) -> Self {
+        self.stream_timeout = t;
+        self
+    }
+
+    /// Configures the maximum number of concurrent identification request streams
+    /// allowed per connection.
+    pub fn with_max_concurrent_streams_per_connection(mut self, s: usize) -> Self {
+        self.max_concurrent_streams_per_connection = s;
+        self
+    }
 }
 
 impl Behaviour {
@@ -343,11 +373,8 @@ impl NetworkBehaviour for Behaviour {
         remote_addr: &Multiaddr,
     ) -> Result<THandler<Self>, ConnectionDenied> {
         Ok(Handler::new(
-            self.config.interval,
+            self.config.clone(),
             peer,
-            self.config.local_public_key.clone(),
-            self.config.protocol_version.clone(),
-            self.config.agent_version.clone(),
             remote_addr.clone(),
             self.all_addresses(),
         ))
@@ -375,11 +402,8 @@ impl NetworkBehaviour for Behaviour {
         }
 
         Ok(Handler::new(
-            self.config.interval,
+            self.config.clone(),
             peer,
-            self.config.local_public_key.clone(),
-            self.config.protocol_version.clone(),
-            self.config.agent_version.clone(),
             addr.clone(), // TODO: This is weird? That is the public address we dialed, shouldn't need to tell the other party?
             self.all_addresses(),
         ))

protocols/identify/src/handler.rs

@@ -19,7 +19,7 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::protocol::{Info, PushInfo, UpgradeError};
-use crate::{protocol, PROTOCOL_NAME, PUSH_PROTOCOL_NAME};
+use crate::{behaviour::Config, protocol, PROTOCOL_NAME, PUSH_PROTOCOL_NAME};
 use either::Either;
 use futures::prelude::*;
 use futures_bounded::Timeout;
@@ -41,9 +41,6 @@ use std::collections::HashSet;
 use std::{task::Context, task::Poll, time::Duration};
 use tracing::Level;
 
-const STREAM_TIMEOUT: Duration = Duration::from_secs(60);
-const MAX_CONCURRENT_STREAMS_PER_CONNECTION: usize = 10;
-
 /// Protocol handler for sending and receiving identification requests.
 ///
 /// Outbound requests are sent periodically. The handler performs expects
@@ -117,27 +114,24 @@ pub enum Event {
 impl Handler {
     /// Creates a new `Handler`.
     pub fn new(
-        interval: Duration,
+        config: Config,
         remote_peer_id: PeerId,
-        public_key: PublicKey,
-        protocol_version: String,
-        agent_version: String,
         observed_addr: Multiaddr,
         external_addresses: HashSet<Multiaddr>,
     ) -> Self {
         Self {
             remote_peer_id,
             events: SmallVec::new(),
             active_streams: futures_bounded::FuturesSet::new(
-                STREAM_TIMEOUT,
-                MAX_CONCURRENT_STREAMS_PER_CONNECTION,
+                config.stream_timeout,
+                config.max_concurrent_streams_per_connection,
             ),
             trigger_next_identify: Delay::new(Duration::ZERO),
             exchanged_one_periodic_identify: false,
-            interval,
-            public_key,
-            protocol_version,
-            agent_version,
+            interval: config.interval,
+            public_key: config.local_public_key.clone(),
+            protocol_version: config.protocol_version,
+            agent_version: config.agent_version,
             observed_addr,
             local_supported_protocols: SupportedProtocols::default(),
             remote_supported_protocols: HashSet::default(),