feat(kad): configurable outbound_substreams timeout

Then testing `put_record_to` with poor connectioin but large packet size (around 4MB), the `put_record_to` frequently got failed and the PutRecordToError::stats showing always failed at 10s.
```
"kad_event::PutRecordError::QuorumFailed", required quorum 2, stored on [], QueryStats { requests: 5, success: 0, failure: 5, start: Some(Instant { tv_sec: 121323, tv_nsec: 961187829 }), end: Some(Instant { tv_sec: 121334, tv_nsec: 188694869 }) } - ProgressStep { count: 1, last: true }
```
However, this timeout is not defined by request/response protocol timeout (which in our case increased to 30s already).

Further digging through the libp2p code, it turned out this timeout is actually affected by outbound_substreams timeout.
We tested with increase to 30s, do see much less put_record_to failure and QueryStats showing extended completion duration (larger than 10s, and in case of failed, to be 30s).

Hence raised this PR to provide a configurable approach of the timeout.

Pull-Request: libp2p#6015.

Author: maqi

Cargo.lock

@@ -85,7 +85,7 @@ libp2p-floodsub = { version = "0.46.1", path = "protocols/floodsub" }
 libp2p-gossipsub = { version = "0.49.0", path = "protocols/gossipsub" }
 libp2p-identify = { version = "0.47.0", path = "protocols/identify" }
 libp2p-identity = { version = "0.2.11" }
-libp2p-kad = { version = "0.47.0", path = "protocols/kad" }
+libp2p-kad = { version = "0.47.1", path = "protocols/kad" }
 libp2p-mdns = { version = "0.47.0", path = "protocols/mdns" }
 libp2p-memory-connection-limits = { version = "0.4.0", path = "misc/memory-connection-limits" }
 libp2p-metrics = { version = "0.17.0", path = "misc/metrics" }

Cargo.toml

@@ -1,3 +1,8 @@
+## 0.47.1
+
+- Configurable outbound_substreams_timeout.
+  See [PR 6015](https://github.com/libp2p/rust-libp2p/pull/6015).
+
 ## 0.47.0
 
 - Expose a kad query facility allowing specify num_results dynamically.
@@ -15,7 +20,7 @@
 - Remove deprecated default constructor for `ProtocolConfig`.
   See [PR 5774](https://github.com/libp2p/rust-libp2p/pull/5774).
 - Add lazy cleanup for expired provider records in `Behavior::get_providers` and `Behavior::provider_peers`.
-  See [PR 5980](https://github.com/libp2p/rust-libp2p/pull/5980)
+  See [PR 5980](https://github.com/libp2p/rust-libp2p/pull/5980).
 
 <!-- Update to libp2p-core v0.43.0 -->
 

protocols/kad/CHANGELOG.md

@@ -3,7 +3,7 @@ name = "libp2p-kad"
 edition.workspace = true
 rust-version = { workspace = true }
 description = "Kademlia protocol for libp2p"
-version = "0.47.0"
+version = "0.47.1"
 authors = ["Parity Technologies <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

protocols/kad/Cargo.toml

@@ -382,6 +382,16 @@ impl Config {
         self
     }
 
+    /// Modifies the timeout duration of outbount_substreams.
+    ///
+    /// * Default to `10` seconds.
+    /// * May need to increase this value when sending large records with poor connection.
+    pub fn set_outbound_substreams_timeout(&mut self, timeout: Duration) -> &mut Self {
+        self.protocol_config
+            .set_outbound_substreams_timeout(timeout);
+        self
+    }
+
     /// Sets the k-bucket insertion strategy for the Kademlia routing table.
     pub fn set_kbucket_inserts(&mut self, inserts: BucketInserts) -> &mut Self {
         self.kbucket_inserts = inserts;

protocols/kad/src/behaviour.rs

@@ -24,7 +24,6 @@ use std::{
     marker::PhantomData,
     pin::Pin,
     task::{Context, Poll, Waker},
-    time::Duration,
 };
 
 use either::Either;
@@ -454,6 +453,8 @@ impl Handler {
             }
         }
 
+        let outbound_substreams_timeout = protocol_config.outbound_substreams_timeout_s();
+
         Handler {
             protocol_config,
             mode,
@@ -462,7 +463,7 @@ impl Handler {
             next_connec_unique_id: UniqueConnecId(0),
             inbound_substreams: Default::default(),
             outbound_substreams: futures_bounded::FuturesTupleSet::new(
-                Duration::from_secs(10),
+                outbound_substreams_timeout,
                 MAX_NUM_STREAMS,
             ),
             pending_streams: Default::default(),

protocols/kad/src/handler.rs

@@ -49,6 +49,8 @@ use crate::{
 pub(crate) const DEFAULT_PROTO_NAME: StreamProtocol = StreamProtocol::new("/ipfs/kad/1.0.0");
 /// The default maximum size for a varint length-delimited packet.
 pub(crate) const DEFAULT_MAX_PACKET_SIZE: usize = 16 * 1024;
+/// The default timeout of outbound_substreams to be 10 (seconds).
+const DEFAULT_OUTBOUND_SUBSTREAMS_TIMEOUT_S: Duration = Duration::from_secs(10);
 /// Status of our connection to a node reported by the Kademlia protocol.
 #[derive(Copy, Clone, PartialEq, Eq, Debug, Hash)]
 pub enum ConnectionType {
@@ -145,6 +147,8 @@ pub struct ProtocolConfig {
     protocol_names: Vec<StreamProtocol>,
     /// Maximum allowed size of a packet.
     max_packet_size: usize,
+    /// Specifies the outbound_substreams timeout in seconds
+    outbound_substreams_timeout_s: Duration,
 }
 
 impl ProtocolConfig {
@@ -153,6 +157,7 @@ impl ProtocolConfig {
         ProtocolConfig {
             protocol_names: vec![protocol_name],
             max_packet_size: DEFAULT_MAX_PACKET_SIZE,
+            outbound_substreams_timeout_s: DEFAULT_OUTBOUND_SUBSTREAMS_TIMEOUT_S,
         }
     }
 
@@ -165,6 +170,16 @@ impl ProtocolConfig {
     pub fn set_max_packet_size(&mut self, size: usize) {
         self.max_packet_size = size;
     }
+
+    /// Modifies outbount_substreams timeout.
+    pub fn set_outbound_substreams_timeout(&mut self, timeout: Duration) {
+        self.outbound_substreams_timeout_s = timeout;
+    }
+
+    /// Getter of outbount_substreams_timeout_s.
+    pub fn outbound_substreams_timeout_s(&self) -> Duration {
+        self.outbound_substreams_timeout_s
+    }
 }
 
 impl UpgradeInfo for ProtocolConfig {