fix(gossipsub): gracefully disable handler on stream errors

protocols/gossipsub/CHANGELOG.md

@@ -2,8 +2,11 @@
 
 - Signed messages now use sequential integers in the sequence number field.
   See [PR 3551].
+- Gracefully disable handler on stream errors. Deprecate a few variants of `HandlerError`.
+  See [PR 3625].
 
 [PR 3551]: https://github.com/libp2p/rust-libp2p/pull/3551
+[PR 3625]: https://github.com/libp2p/rust-libp2p/pull/3325
 
 # 0.44.1
 

protocols/gossipsub/Cargo.toml

@@ -11,7 +11,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
-libp2p-swarm = { version = "0.42.0", path = "../../swarm" }
+libp2p-swarm = { version = "0.42.1", path = "../../swarm" }
 libp2p-core = { version = "0.39.0", path = "../../core" }
 libp2p-identity = { version = "0.1.0", path = "../../identity" }
 bytes = "1.4"
@@ -33,6 +33,7 @@ serde = { version = "1", optional = true, features = ["derive"] }
 thiserror = "1.0"
 wasm-timer = "0.2.5"
 instant = "0.1.11"
+void = "1.0.2"
 # Metrics dependencies
 prometheus-client = "0.19.0"
 

protocols/gossipsub/src/error.rs

@@ -30,16 +30,10 @@ pub type PublishError = crate::error_priv::PublishError;
 )]
 pub type SubscriptionError = crate::error_priv::SubscriptionError;
 
-#[deprecated(
-    since = "0.44.0",
-    note = "Use re-exports that omit `Gossipsub` prefix, i.e. `libp2p::gossipsub::HandlerError"
-)]
+#[deprecated(note = "This error will no longer be emitted")]
 pub type GossipsubHandlerError = crate::error_priv::HandlerError;
 
-#[deprecated(
-    since = "0.44.0",
-    note = "Use `libp2p::gossipsub::HandlerError` instead, as the `error` module will become crate-private in the future."
-)]
+#[deprecated(note = "This error will no longer be emitted")]
 pub type HandlerError = crate::error_priv::HandlerError;
 
 #[deprecated(

protocols/gossipsub/src/error_priv.rs

@@ -134,12 +134,6 @@ impl std::fmt::Display for ValidationError {
 
 impl std::error::Error for ValidationError {}
 
-impl From<std::io::Error> for HandlerError {
-    fn from(error: std::io::Error) -> HandlerError {
-        HandlerError::Codec(quick_protobuf_codec::Error::from(error))
-    }
-}
-
 impl From<std::io::Error> for PublishError {
     fn from(error: std::io::Error) -> PublishError {
         PublishError::TransformFailed(error)

protocols/gossipsub/src/handler.rs

@@ -21,7 +21,7 @@
 use crate::protocol::{GossipsubCodec, ProtocolConfig};
 use crate::rpc_proto::proto;
 use crate::types::{PeerKind, RawMessage, Rpc};
-use crate::{HandlerError, ValidationError};
+use crate::ValidationError;
 use asynchronous_codec::Framed;
 use futures::prelude::*;
 use futures::StreamExt;
@@ -33,11 +33,8 @@ use libp2p_swarm::handler::{
     SubstreamProtocol,
 };
 use libp2p_swarm::NegotiatedSubstream;
-use log::{error, trace, warn};
 use smallvec::SmallVec;
 use std::{
-    collections::VecDeque,
-    io,
     pin::Pin,
     task::{Context, Poll},
     time::Duration,
@@ -124,9 +121,6 @@ pub struct Handler {
     /// The amount of time we allow idle connections before disconnecting.
     idle_timeout: Duration,
 
-    /// Collection of errors from attempting an upgrade.
-    upgrade_errors: VecDeque<ConnectionHandlerUpgrErr<HandlerError>>,
-
     /// Flag determining whether to maintain the connection to the peer.
     keep_alive: KeepAlive,
 
@@ -174,7 +168,6 @@ impl Handler {
             peer_kind_sent: false,
             protocol_unsupported: false,
             idle_timeout,
-            upgrade_errors: VecDeque::new(),
             keep_alive: KeepAlive::Until(Instant::now() + Duration::from_secs(INITIAL_KEEP_ALIVE)),
             in_mesh: false,
         }
@@ -202,7 +195,7 @@ impl Handler {
         }
 
         // new inbound substream. Replace the current one, if it exists.
-        trace!("New inbound substream request");
+        log::trace!("New inbound substream request");
         self.inbound_substream = Some(InboundSubstreamState::WaitingInput(substream));
     }
 
@@ -223,7 +216,6 @@ impl Handler {
             return;
         }
 
-        self.outbound_substream_establishing = false;
         self.outbound_substreams_created += 1;
 
         // update the known kind of peer
@@ -234,7 +226,7 @@ impl Handler {
         // Should never establish a new outbound substream if one already exists.
         // If this happens, an outbound message is not sent.
         if self.outbound_substream.is_some() {
-            warn!("Established an outbound substream with one already available");
+            log::warn!("Established an outbound substream with one already available");
             // Add the message back to the send queue
             self.send_queue.push(message);
         } else {
@@ -246,7 +238,7 @@ impl Handler {
 impl ConnectionHandler for Handler {
     type InEvent = HandlerIn;
     type OutEvent = HandlerEvent;
-    type Error = HandlerError;
+    type Error = crate::error_priv::HandlerError; // TODO: Replace this with `Void`.
     type InboundOpenInfo = ();
     type InboundProtocol = ProtocolConfig;
     type OutboundOpenInfo = proto::RPC;
@@ -289,44 +281,15 @@ impl ConnectionHandler for Handler {
             Self::Error,
         >,
     > {
-        // Handle any upgrade errors
-        if let Some(error) = self.upgrade_errors.pop_front() {
-            let reported_error = match error {
-                // Timeout errors get mapped to NegotiationTimeout and we close the connection.
-                ConnectionHandlerUpgrErr::Timeout | ConnectionHandlerUpgrErr::Timer => {
-                    Some(HandlerError::NegotiationTimeout)
-                }
-                // There was an error post negotiation, close the connection.
-                ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)) => Some(e),
-                ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(negotiation_error)) => {
-                    match negotiation_error {
-                        NegotiationError::Failed => {
-                            // The protocol is not supported
-                            self.protocol_unsupported = true;
-                            if !self.peer_kind_sent {
-                                self.peer_kind_sent = true;
-                                // clear all substreams so the keep alive returns false
-                                self.inbound_substream = None;
-                                self.outbound_substream = None;
-                                self.keep_alive = KeepAlive::No;
-                                return Poll::Ready(ConnectionHandlerEvent::Custom(
-                                    HandlerEvent::PeerKind(PeerKind::NotSupported),
-                                ));
-                            } else {
-                                None
-                            }
-                        }
-                        NegotiationError::ProtocolError(e) => {
-                            Some(HandlerError::NegotiationProtocolError(e))
-                        }
-                    }
-                }
-            };
-
-            // If there was a fatal error, close the connection.
-            if let Some(error) = reported_error {
-                return Poll::Ready(ConnectionHandlerEvent::Close(error));
-            }
+        if self.protocol_unsupported && !self.peer_kind_sent {
+            self.peer_kind_sent = true;
+            // clear all substreams so the keep alive returns false
+            self.inbound_substream = None;
+            self.outbound_substream = None;
+            self.keep_alive = KeepAlive::No;
+            return Poll::Ready(ConnectionHandlerEvent::Custom(HandlerEvent::PeerKind(
+                PeerKind::NotSupported,
+            )));
         }
 
         if !self.peer_kind_sent {
@@ -338,23 +301,14 @@ impl ConnectionHandler for Handler {
             }
         }
 
-        if self.inbound_substreams_created > MAX_SUBSTREAM_CREATION {
-            // Too many inbound substreams have been created, end the connection.
-            return Poll::Ready(ConnectionHandlerEvent::Close(
-                HandlerError::MaxInboundSubstreams,
-            ));
-        }
+        // Invariant: `self.inbound_substreams_created < MAX_SUBSTREAM_CREATION`.
 
         // determine if we need to create the stream
         if !self.send_queue.is_empty()
             && self.outbound_substream.is_none()
             && !self.outbound_substream_establishing
         {
-            if self.outbound_substreams_created >= MAX_SUBSTREAM_CREATION {
-                return Poll::Ready(ConnectionHandlerEvent::Close(
-                    HandlerError::MaxOutboundSubstreams,
-                ));
-            }
+            // Invariant: `self.outbound_substreams_created < MAX_SUBSTREAM_CREATION`.
             let message = self.send_queue.remove(0);
             self.send_queue.shrink_to_fit();
             self.outbound_substream_establishing = true;
@@ -381,25 +335,16 @@ impl ConnectionHandler for Handler {
                             return Poll::Ready(ConnectionHandlerEvent::Custom(message));
                         }
                         Poll::Ready(Some(Err(error))) => {
-                            match error {
-                                HandlerError::MaxTransmissionSize => {
-                                    warn!("Message exceeded the maximum transmission size");
-                                    self.inbound_substream =
-                                        Some(InboundSubstreamState::WaitingInput(substream));
-                                }
-                                _ => {
-                                    warn!("Inbound stream error: {}", error);
-                                    // More serious errors, close this side of the stream. If the
-                                    // peer is still around, they will re-establish their
-                                    // connection
-                                    self.inbound_substream =
-                                        Some(InboundSubstreamState::Closing(substream));
-                                }
-                            }
+                            log::warn!("Inbound stream error: {}", error);
+                            // Close this side of the stream. If the
+                            // peer is still around, they will re-establish their
+                            // connection
+                            self.inbound_substream =
+                                Some(InboundSubstreamState::Closing(substream));
                         }
                         // peer closed the stream
                         Poll::Ready(None) => {
-                            warn!("Peer closed their outbound stream");
+                            log::warn!("Peer closed their outbound stream");
                             self.inbound_substream =
                                 Some(InboundSubstreamState::Closing(substream));
                         }
@@ -417,7 +362,7 @@ impl ConnectionHandler for Handler {
                                 // Don't close the connection but just drop the inbound substream.
                                 // In case the remote has more to send, they will open up a new
                                 // substream.
-                                warn!("Inbound substream error while closing: {:?}", e);
+                                log::warn!("Inbound substream error while closing: {e}");
                             }
                             self.inbound_substream = None;
                             if self.outbound_substream.is_none() {
@@ -469,20 +414,19 @@ impl ConnectionHandler for Handler {
                                     self.outbound_substream =
                                         Some(OutboundSubstreamState::PendingFlush(substream))
                                 }
-                                Err(HandlerError::MaxTransmissionSize) => {
-                                    error!("Message exceeded the maximum transmission size and was not sent.");
-                                    self.outbound_substream =
-                                        Some(OutboundSubstreamState::WaitingOutput(substream));
-                                }
                                 Err(e) => {
-                                    error!("Error sending message: {}", e);
-                                    return Poll::Ready(ConnectionHandlerEvent::Close(e));
+                                    log::debug!(
+                                        "Outbound substream error while sending output: {e}"
+                                    );
+                                    self.outbound_substream = None;
+                                    break;
                                 }
                             }
                         }
                         Poll::Ready(Err(e)) => {
-                            error!("Outbound substream error while sending output: {:?}", e);
-                            return Poll::Ready(ConnectionHandlerEvent::Close(e));
+                            log::debug!("Outbound substream error while sending output: {e}");
+                            self.outbound_substream = None;
+                            break;
                         }
                         Poll::Pending => {
                             self.keep_alive = KeepAlive::Yes;
@@ -504,7 +448,9 @@ impl ConnectionHandler for Handler {
                                 Some(OutboundSubstreamState::WaitingOutput(substream))
                         }
                         Poll::Ready(Err(e)) => {
-                            return Poll::Ready(ConnectionHandlerEvent::Close(e))
+                            log::debug!("Outbound substream error while flushing output: {e}");
+                            self.outbound_substream = None;
+                            break;
                         }
                         Poll::Pending => {
                             self.keep_alive = KeepAlive::Yes;
@@ -525,14 +471,9 @@ impl ConnectionHandler for Handler {
                             break;
                         }
                         Poll::Ready(Err(e)) => {
-                            warn!("Outbound substream error while closing: {:?}", e);
-                            return Poll::Ready(ConnectionHandlerEvent::Close(
-                                io::Error::new(
-                                    io::ErrorKind::BrokenPipe,
-                                    "Failed to close outbound substream",
-                                )
-                                .into(),
-                            ));
+                            log::debug!("Outbound substream error while closing: {e}");
+                            self.outbound_substream = None;
+                            break;
                         }
                         Poll::Pending => {
                             self.keep_alive = KeepAlive::No;
@@ -564,17 +505,58 @@ impl ConnectionHandler for Handler {
             Self::OutboundOpenInfo,
         >,
     ) {
+        if event.is_outbound() {
+            self.outbound_substream_establishing = false;
+        }
+
+        if event.is_inbound() && self.inbound_substreams_created == MAX_SUBSTREAM_CREATION {
+            // Too many inbound substreams have been created, disable the handler.
+            self.keep_alive = KeepAlive::No;
+            log::info!("The maximum number of inbound substreams created has been exceeded.");
+            return;
+        }
+
+        if event.is_outbound() && self.outbound_substreams_created == MAX_SUBSTREAM_CREATION {
+            // Too many outbound substreams have been created, disable the handler.
+            self.keep_alive = KeepAlive::No;
+            log::info!("The maximum number of outbound substreams created has been exceeded.");
+            return;
+        }
+
         match event {
             ConnectionEvent::FullyNegotiatedInbound(fully_negotiated_inbound) => {
                 self.on_fully_negotiated_inbound(fully_negotiated_inbound)
             }
             ConnectionEvent::FullyNegotiatedOutbound(fully_negotiated_outbound) => {
                 self.on_fully_negotiated_outbound(fully_negotiated_outbound)
             }
-            ConnectionEvent::DialUpgradeError(DialUpgradeError { error: e, .. }) => {
-                self.outbound_substream_establishing = false;
-                warn!("Dial upgrade error {:?}", e);
-                self.upgrade_errors.push_back(e);
+            ConnectionEvent::DialUpgradeError(DialUpgradeError {
+                error: ConnectionHandlerUpgrErr::Timeout | ConnectionHandlerUpgrErr::Timer,
+                ..
+            }) => {
+                log::debug!("Dial upgrade error: Protocol negotiation timeout.");
+            }
+            ConnectionEvent::DialUpgradeError(DialUpgradeError {
+                error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)),
+                ..
+            }) => void::unreachable(e),
+            ConnectionEvent::DialUpgradeError(DialUpgradeError {
+                error:
+                    ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)),
+                ..
+            }) => {
+                // The protocol is not supported
+                self.protocol_unsupported = true;
+                log::debug!("The remote peer does not support gossipsub on this connection");
+            }
+            ConnectionEvent::DialUpgradeError(DialUpgradeError {
+                error:
+                    ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(
+                        NegotiationError::ProtocolError(e),
+                    )),
+                ..
+            }) => {
+                log::debug!("Protocol negotiation failed: {e}")
             }
             ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
         }

protocols/gossipsub/src/lib.rs

@@ -158,9 +158,12 @@ mod types;
 
 mod rpc_proto;
 
+#[deprecated(note = "This error will no longer be emitted")]
+pub type HandlerError = error_priv::HandlerError;
+
 pub use self::behaviour::{Behaviour, Event, MessageAuthenticity};
 pub use self::config::{Config, ConfigBuilder, ValidationMode, Version};
-pub use self::error_priv::{HandlerError, PublishError, SubscriptionError, ValidationError};
+pub use self::error_priv::{PublishError, SubscriptionError, ValidationError};
 pub use self::peer_score::{
     score_parameter_decay, score_parameter_decay_with_base, PeerScoreParams, PeerScoreThresholds,
     TopicScoreParams,