ping: Conform to the spec & exclude from connection keep-alive

src/lib.rs

@@ -38,6 +38,7 @@ use crate::{
         mdns::Mdns,
         notification::NotificationProtocol,
         request_response::RequestResponseProtocol,
+        SubstreamKeepAlive,
     },
     transport::{
         manager::{SupportedTransport, TransportManager, TransportManagerBuilder},
@@ -199,6 +200,7 @@ impl Litep2p {
                 config.fallback_names.clone(),
                 config.codec,
                 litep2p_config.keep_alive_timeout,
+                SubstreamKeepAlive::Yes,
             );
             let executor = Arc::clone(&litep2p_config.executor);
             litep2p_config.executor.run(Box::pin(async move {
@@ -219,6 +221,7 @@ impl Litep2p {
                 config.fallback_names.clone(),
                 config.codec,
                 litep2p_config.keep_alive_timeout,
+                SubstreamKeepAlive::Yes,
             );
             litep2p_config.executor.run(Box::pin(async move {
                 RequestResponseProtocol::new(service, config).run().await
@@ -234,6 +237,8 @@ impl Litep2p {
                 Vec::new(),
                 protocol.codec(),
                 litep2p_config.keep_alive_timeout,
+                // TODO: make configurable by user.
+                SubstreamKeepAlive::Yes,
             );
             litep2p_config.executor.run(Box::pin(async move {
                 let _ = protocol.run(service).await;
@@ -253,6 +258,7 @@ impl Litep2p {
                 Vec::new(),
                 ping_config.codec,
                 litep2p_config.keep_alive_timeout,
+                SubstreamKeepAlive::No,
             );
             litep2p_config.executor.run(Box::pin(async move {
                 Ping::new(service, ping_config).run().await
@@ -276,6 +282,7 @@ impl Litep2p {
                 fallback_names,
                 kademlia_config.codec,
                 litep2p_config.keep_alive_timeout,
+                SubstreamKeepAlive::Yes,
             );
             litep2p_config.executor.run(Box::pin(async move {
                 let _ = Kademlia::new(service, kademlia_config).run().await;
@@ -297,6 +304,7 @@ impl Litep2p {
                     Vec::new(),
                     identify_config.codec,
                     litep2p_config.keep_alive_timeout,
+                    SubstreamKeepAlive::No,
                 );
                 identify_config.public = Some(litep2p_config.keypair.public().into());
 
@@ -317,6 +325,7 @@ impl Litep2p {
                 Vec::new(),
                 bitswap_config.codec,
                 litep2p_config.keep_alive_timeout,
+                SubstreamKeepAlive::Yes,
             );
             litep2p_config.executor.run(Box::pin(async move {
                 Bitswap::new(service, bitswap_config).run().await

src/protocol/connection.rs

@@ -158,7 +158,14 @@ impl ConnectionHandle {
     }
 }
 
-/// Type which allows the connection to be kept open.
+/// Type which allows to keep the connection opened and not allow the keep-alive mechanism to close
+/// it.
+///
+/// The [`Permit`] is created when beginning to open a substream and passed on until it reaches
+/// [`TransportService`](crate::protocol::TransportService), where the connection is upgraded (what
+/// means it won't be closed) and the permit is not needed anymore and dropped.
+///
+/// Every use of a permit must be short-lived, make sure to drop it!
 #[derive(Debug, Clone)]
 pub struct Permit {
     /// Active connection.

src/protocol/libp2p/kademlia/mod.rs

@@ -1417,7 +1417,7 @@ mod tests {
     use crate::{
         codec::ProtocolCodec,
         transport::{
-            manager::{TransportManager, TransportManagerBuilder},
+            manager::{SubstreamKeepAlive, TransportManager, TransportManagerBuilder},
             KEEP_ALIVE_TIMEOUT,
         },
         types::protocol::ProtocolName,
@@ -1445,6 +1445,7 @@ mod tests {
             Default::default(),
             manager.transport_manager_handle(),
             KEEP_ALIVE_TIMEOUT,
+            SubstreamKeepAlive::Yes,
         );
         let (event_tx, event_rx) = channel(64);
         let (_cmd_tx, cmd_rx) = channel(64);

src/protocol/libp2p/ping/config.rs

@@ -22,6 +22,7 @@ use crate::{
     codec::ProtocolCodec, protocol::libp2p::ping::PingEvent, types::protocol::ProtocolName,
     DEFAULT_CHANNEL_SIZE,
 };
+use std::time::Duration;
 
 use futures::Stream;
 use tokio::sync::mpsc::{channel, Sender};
@@ -36,6 +37,11 @@ const PING_PAYLOAD_SIZE: usize = 32;
 /// Maximum PING failures.
 const MAX_FAILURES: usize = 3;
 
+/// Ping interval must be set to < 10 secs, because litep2p versions before
+/// <https://github.com/paritytech/litep2p/pull/416> reset the inbound substream if not receive
+/// the payload within 10 seconds of opening the substream.
+pub const PING_INTERVAL: Duration = Duration::from_secs(5);
+
 /// Ping configuration.
 pub struct Config {
     /// Protocol name.
@@ -49,6 +55,8 @@ pub struct Config {
 
     /// TX channel for sending events to the user protocol.
     pub(crate) tx_event: Sender<PingEvent>,
+
+    pub(crate) ping_interval: Duration,
 }
 
 impl Config {
@@ -61,6 +69,7 @@ impl Config {
         (
             Self {
                 tx_event,
+                ping_interval: PING_INTERVAL,
                 max_failures: MAX_FAILURES,
                 protocol: ProtocolName::from(PROTOCOL_NAME),
                 codec: ProtocolCodec::Identity(PING_PAYLOAD_SIZE),
@@ -80,6 +89,9 @@ pub struct ConfigBuilder {
 
     /// Maximum failures before the peer is considered unreachable.
     max_failures: usize,
+
+    /// Interval between outbound pings.
+    ping_interval: Duration,
 }
 
 impl Default for ConfigBuilder {
@@ -92,6 +104,7 @@ impl ConfigBuilder {
     /// Create new default [`Config`] which can be modified by the user.
     pub fn new() -> Self {
         Self {
+            ping_interval: PING_INTERVAL,
             max_failures: MAX_FAILURES,
             protocol: ProtocolName::from(PROTOCOL_NAME),
             codec: ProtocolCodec::Identity(PING_PAYLOAD_SIZE),
@@ -104,13 +117,19 @@ impl ConfigBuilder {
         self
     }
 
+    pub fn with_ping_interval(mut self, ping_interval: Duration) -> Self {
+        self.ping_interval = ping_interval;
+        self
+    }
+
     /// Build [`Config`].
     pub fn build(self) -> (Config, Box<dyn Stream<Item = PingEvent> + Send + Unpin>) {
         let (tx_event, rx_event) = channel(DEFAULT_CHANNEL_SIZE);
 
         (
             Config {
                 tx_event,
+                ping_interval: self.ping_interval,
                 max_failures: self.max_failures,
                 protocol: self.protocol,
                 codec: self.codec,