protocols/ping: Don't force close conn if not supported by remote (#2149)

Don't close connection if ping protocol is unsupported by remote. Previously, a
failed protocol negotation for ping caused a force close of the connection. As a
result, all nodes in a network had to support ping. To allow networks where some
nodes don't support ping, we now emit `PingFailure::Unsupported` once for every
connection on which ping is not supported.

Co-authored-by: Max Inden <[email protected]>

Author: thomaseizinger

examples/ipfs-private.rs

@@ -223,6 +223,12 @@ fn main() -> Result<(), Box<dyn Error>> {
                 } => {
                     println!("ping: timeout to {}", peer.to_base58());
                 }
+                PingEvent {
+                    peer,
+                    result: Result::Err(PingFailure::Unsupported),
+                } => {
+                    println!("ping: {} does not support ping protocol", peer.to_base58());
+                }
                 PingEvent {
                     peer,
                     result: Result::Err(PingFailure::Other { error }),

protocols/ping/CHANGELOG.md

@@ -2,6 +2,19 @@
 
 - Update dependencies.
 
+- Don't close connection if ping protocol is unsupported by remote.
+  Previously, a failed protocol negotation for ping caused a force close of the connection.
+  As a result, all nodes in a network had to support ping.
+  To allow networks where some nodes don't support ping, we now emit
+  `PingFailure::Unsupported` once for every connection on which ping is not supported.
+
+  In case you want to stick with the old behavior, you need to close the connection
+  manually on `PingFailure::Unsupported`.
+
+  Fixes [#2109](https://github.com/libp2p/rust-libp2p/issues/2109). Also see [PR 2149].
+
+  [PR 2149]: https://github.com/libp2p/rust-libp2p/pull/2149/
+
 # 0.30.0 [2021-07-12]
 
 - Update dependencies.

protocols/ping/src/handler.rs

@@ -21,6 +21,7 @@
 use crate::protocol;
 use futures::prelude::*;
 use futures::future::BoxFuture;
+use libp2p_core::{UpgradeError, upgrade::NegotiationError};
 use libp2p_swarm::{
     KeepAlive,
     NegotiatedSubstream,
@@ -140,6 +141,8 @@ pub enum PingFailure {
     /// The ping timed out, i.e. no response was received within the
     /// configured ping timeout.
     Timeout,
+    /// The peer does not support the ping protocol.
+    Unsupported,
     /// The ping failed for reasons other than a timeout.
     Other { error: Box<dyn std::error::Error + Send + 'static> }
 }
@@ -148,7 +151,8 @@ impl fmt::Display for PingFailure {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             PingFailure::Timeout => f.write_str("Ping timeout"),
-            PingFailure::Other { error } => write!(f, "Ping error: {}", error)
+            PingFailure::Other { error } => write!(f, "Ping error: {}", error),
+            PingFailure::Unsupported => write!(f, "Ping protocol not supported"),
         }
     }
 }
@@ -157,7 +161,8 @@ impl Error for PingFailure {
     fn source(&self) -> Option<&(dyn Error + 'static)> {
         match self {
             PingFailure::Timeout => None,
-            PingFailure::Other { error } => Some(&**error)
+            PingFailure::Other { error } => Some(&**error),
+            PingFailure::Unsupported => None,
         }
     }
 }
@@ -184,6 +189,21 @@ pub struct PingHandler {
     /// substream, this is always a future that waits for the
     /// next inbound ping to be answered.
     inbound: Option<PongFuture>,
+    /// Tracks the state of our handler.
+    state: State
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum State {
+    /// We are inactive because the other peer doesn't support ping.
+    Inactive {
+        /// Whether or not we've reported the missing support yet.
+        ///
+        /// This is used to avoid repeated events being emitted for a specific connection.
+        reported: bool
+    },
+    /// We are actively pinging the other peer.
+    Active,
 }
 
 impl PingHandler {
@@ -196,6 +216,7 @@ impl PingHandler {
             failures: 0,
             outbound: None,
             inbound: None,
+            state: State::Active,
         }
     }
 }
@@ -226,12 +247,22 @@ impl ProtocolsHandler for PingHandler {
 
     fn inject_dial_upgrade_error(&mut self, _info: (), error: ProtocolsHandlerUpgrErr<Void>) {
         self.outbound = None; // Request a new substream on the next `poll`.
-        self.pending_errors.push_front(
-            match error {
-                // Note: This timeout only covers protocol negotiation.
-                ProtocolsHandlerUpgrErr::Timeout => PingFailure::Timeout,
-                e => PingFailure::Other { error: Box::new(e) },
-            })
+
+        let error = match error {
+            ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
+                debug_assert_eq!(self.state, State::Active);
+
+                self.state = State::Inactive {
+                    reported: false
+                };
+                return;
+            },
+            // Note: This timeout only covers protocol negotiation.
+            ProtocolsHandlerUpgrErr::Timeout => PingFailure::Timeout,
+            e => PingFailure::Other { error: Box::new(e) },
+        };
+
+        self.pending_errors.push_front(error);
     }
 
     fn connection_keep_alive(&self) -> KeepAlive {
@@ -243,6 +274,17 @@ impl ProtocolsHandler for PingHandler {
     }
 
     fn poll(&mut self, cx: &mut Context<'_>) -> Poll<ProtocolsHandlerEvent<protocol::Ping, (), PingResult, Self::Error>> {
+        match self.state {
+            State::Inactive { reported: true } => {
+                return Poll::Pending // nothing to do on this connection
+            },
+            State::Inactive { reported: false } => {
+                self.state = State::Inactive { reported: true };
+                return Poll::Ready(ProtocolsHandlerEvent::Custom(Err(PingFailure::Unsupported)));
+            },
+            State::Active => {}
+        }
+
         // Respond to inbound pings.
         if let Some(fut) = self.inbound.as_mut() {
             match fut.poll_unpin(cx) {
@@ -355,4 +397,3 @@ enum PingState {
     /// A ping is being sent and the response awaited.
     Ping(PingFuture),
 }
-

protocols/ping/tests/ping.rs

@@ -31,7 +31,7 @@ use libp2p_core::{
 use libp2p_mplex as mplex;
 use libp2p_noise as noise;
 use libp2p_ping::*;
-use libp2p_swarm::{Swarm, SwarmEvent};
+use libp2p_swarm::{DummyBehaviour, KeepAlive, Swarm, SwarmEvent};
 use libp2p_tcp::TcpConfig;
 use libp2p_yamux as yamux;
 use futures::{prelude::*, channel::mpsc};
@@ -83,18 +83,18 @@ fn ping_pong() {
 
             loop {
                 match swarm2.select_next_some().await {
-                    SwarmEvent::Behaviour(PingEvent { 
-                        peer, 
-                        result: Ok(PingSuccess::Ping { rtt }) 
+                    SwarmEvent::Behaviour(PingEvent {
+                        peer,
+                        result: Ok(PingSuccess::Ping { rtt })
                     }) => {
                         count2 -= 1;
                         if count2 == 0 {
                             return (pid2.clone(), peer, rtt)
                         }
                     },
-                    SwarmEvent::Behaviour(PingEvent { 
-                        result: Err(e), 
-                        .. 
+                    SwarmEvent::Behaviour(PingEvent {
+                        result: Err(e),
+                        ..
                     }) => panic!("Ping failure: {:?}", e),
                     _ => {}
                 }
@@ -189,6 +189,52 @@ fn max_failures() {
     QuickCheck::new().tests(10).quickcheck(prop as fn(_,_))
 }
 
+#[test]
+fn unsupported_doesnt_fail() {
+    let (peer1_id, trans) = mk_transport(MuxerChoice::Mplex);
+    let mut swarm1 = Swarm::new(trans, DummyBehaviour::with_keep_alive(KeepAlive::Yes), peer1_id.clone());
+
+    let (peer2_id, trans) = mk_transport(MuxerChoice::Mplex);
+    let mut swarm2 = Swarm::new(trans, Ping::new(PingConfig::new().with_keep_alive(true)), peer2_id.clone());
+
+    let (mut tx, mut rx) = mpsc::channel::<Multiaddr>(1);
+
+    let addr = "/ip4/127.0.0.1/tcp/0".parse().unwrap();
+    swarm1.listen_on(addr).unwrap();
+
+    async_std::task::spawn(async move {
+        loop {
+            match swarm1.select_next_some().await {
+                SwarmEvent::NewListenAddr { address, .. } => tx.send(address).await.unwrap(),
+                _ => {}
+            }
+        }
+    });
+
+    let result = async_std::task::block_on(async move {
+        swarm2.dial_addr(rx.next().await.unwrap()).unwrap();
+
+        loop {
+            match swarm2.select_next_some().await {
+                SwarmEvent::Behaviour(PingEvent {
+                    result: Err(PingFailure::Unsupported), ..
+                }) => {
+                    swarm2.disconnect_peer_id(peer1_id).unwrap();
+                }
+                SwarmEvent::ConnectionClosed { cause: Some(e), .. } => {
+                    break Err(e);
+                }
+                SwarmEvent::ConnectionClosed { cause: None, .. } => {
+                    break Ok(());
+                }
+                _ => {}
+            }
+        }
+    });
+
+    result.expect("node with ping should not fail connection due to unsupported protocol");
+}
+
 
 fn mk_transport(muxer: MuxerChoice) -> (
     PeerId,