feat(swarm): make stream uprade errors more ergonomic

examples/file-sharing/src/network.rs

@@ -16,7 +16,7 @@ use libp2p::{
     multiaddr::Protocol,
     noise,
     request_response::{self, ProtocolSupport, RequestId, ResponseChannel},
-    swarm::{ConnectionHandlerUpgrErr, NetworkBehaviour, Swarm, SwarmBuilder, SwarmEvent},
+    swarm::{NetworkBehaviour, StreamUpgradeError, Swarm, SwarmBuilder, SwarmEvent},
     tcp, yamux, PeerId, Transport,
 };
 
@@ -216,7 +216,7 @@ impl EventLoop {
 
     async fn handle_event(
         &mut self,
-        event: SwarmEvent<ComposedEvent, Either<ConnectionHandlerUpgrErr<io::Error>, io::Error>>,
+        event: SwarmEvent<ComposedEvent, Either<StreamUpgradeError<io::Error>, io::Error>>,
     ) {
         match event {
             SwarmEvent::Behaviour(ComposedEvent::Kademlia(

misc/metrics/src/relay.rs

@@ -54,7 +54,6 @@ enum EventType {
     ReservationReqDenied,
     ReservationReqDenyFailed,
     ReservationTimedOut,
-    CircuitReqReceiveFailed,
     CircuitReqDenied,
     CircuitReqDenyFailed,
     CircuitReqOutboundConnectFailed,
@@ -75,9 +74,6 @@ impl From<&libp2p_relay::Event> for EventType {
                 EventType::ReservationReqDenyFailed
             }
             libp2p_relay::Event::ReservationTimedOut { .. } => EventType::ReservationTimedOut,
-            libp2p_relay::Event::CircuitReqReceiveFailed { .. } => {
-                EventType::CircuitReqReceiveFailed
-            }
             libp2p_relay::Event::CircuitReqDenied { .. } => EventType::CircuitReqDenied,
             libp2p_relay::Event::CircuitReqOutboundConnectFailed { .. } => {
                 EventType::CircuitReqOutboundConnectFailed

protocols/dcutr/src/behaviour_impl.rs

@@ -30,7 +30,7 @@ use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailu
 use libp2p_swarm::dial_opts::{self, DialOpts};
 use libp2p_swarm::{dummy, ConnectionDenied, ConnectionId, THandler, THandlerOutEvent};
 use libp2p_swarm::{
-    ConnectionHandlerUpgrErr, ExternalAddresses, NetworkBehaviour, NotifyHandler, PollParameters,
+    ExternalAddresses, NetworkBehaviour, NotifyHandler, PollParameters, StreamUpgradeError,
     THandlerInEvent, ToSwarm,
 };
 use std::collections::{HashMap, HashSet, VecDeque};
@@ -65,7 +65,7 @@ pub enum Error {
     #[error("Failed to dial peer.")]
     Dial,
     #[error("Failed to establish substream: {0}.")]
-    Handler(ConnectionHandlerUpgrErr<Void>),
+    Handler(StreamUpgradeError<Void>),
 }
 
 pub struct Behaviour {

protocols/dcutr/src/handler/direct.rs

@@ -23,8 +23,7 @@
 use libp2p_core::upgrade::DeniedUpgrade;
 use libp2p_swarm::handler::ConnectionEvent;
 use libp2p_swarm::{
-    ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr, KeepAlive,
-    SubstreamProtocol,
+    ConnectionHandler, ConnectionHandlerEvent, KeepAlive, StreamUpgradeError, SubstreamProtocol,
 };
 use std::task::{Context, Poll};
 use void::Void;
@@ -42,7 +41,7 @@ pub struct Handler {
 impl ConnectionHandler for Handler {
     type InEvent = void::Void;
     type OutEvent = Event;
-    type Error = ConnectionHandlerUpgrErr<std::io::Error>;
+    type Error = StreamUpgradeError<std::io::Error>;
     type InboundProtocol = DeniedUpgrade;
     type OutboundProtocol = DeniedUpgrade;
     type OutboundOpenInfo = Void;

protocols/dcutr/src/handler/relayed.rs

@@ -26,15 +26,14 @@ use futures::future;
 use futures::future::{BoxFuture, FutureExt};
 use instant::Instant;
 use libp2p_core::multiaddr::Multiaddr;
-use libp2p_core::upgrade::{DeniedUpgrade, NegotiationError, UpgradeError};
+use libp2p_core::upgrade::DeniedUpgrade;
 use libp2p_core::ConnectedPoint;
 use libp2p_swarm::handler::{
     ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
     ListenUpgradeError,
 };
 use libp2p_swarm::{
-    ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr, KeepAlive,
-    SubstreamProtocol,
+    ConnectionHandler, ConnectionHandlerEvent, KeepAlive, StreamUpgradeError, SubstreamProtocol,
 };
 use std::collections::VecDeque;
 use std::fmt;
@@ -82,11 +81,11 @@ pub enum Event {
         remote_addr: Multiaddr,
     },
     InboundNegotiationFailed {
-        error: ConnectionHandlerUpgrErr<void::Void>,
+        error: StreamUpgradeError<void::Void>,
     },
     InboundConnectNegotiated(Vec<Multiaddr>),
     OutboundNegotiationFailed {
-        error: ConnectionHandlerUpgrErr<void::Void>,
+        error: StreamUpgradeError<void::Void>,
     },
     OutboundConnectNegotiated {
         remote_addrs: Vec<Multiaddr>,
@@ -127,7 +126,7 @@ pub struct Handler {
     endpoint: ConnectedPoint,
     /// A pending fatal error that results in the connection being closed.
     pending_error: Option<
-        ConnectionHandlerUpgrErr<
+        StreamUpgradeError<
             Either<protocol::inbound::UpgradeError, protocol::outbound::UpgradeError>,
         >,
     >,
@@ -212,45 +211,10 @@ impl Handler {
             <Self as ConnectionHandler>::InboundProtocol,
         >,
     ) {
-        match error {
-            ConnectionHandlerUpgrErr::Timeout => {
-                self.queued_events.push_back(ConnectionHandlerEvent::Custom(
-                    Event::InboundNegotiationFailed {
-                        error: ConnectionHandlerUpgrErr::Timeout,
-                    },
-                ));
-            }
-            ConnectionHandlerUpgrErr::Timer => {
-                self.queued_events.push_back(ConnectionHandlerEvent::Custom(
-                    Event::InboundNegotiationFailed {
-                        error: ConnectionHandlerUpgrErr::Timer,
-                    },
-                ));
-            }
-            ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
-                // The remote merely doesn't support the DCUtR protocol.
-                // This is no reason to close the connection, which may
-                // successfully communicate with other protocols already.
-                self.keep_alive = KeepAlive::No;
-                self.queued_events.push_back(ConnectionHandlerEvent::Custom(
-                    Event::InboundNegotiationFailed {
-                        error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(
-                            NegotiationError::Failed,
-                        )),
-                    },
-                ));
-            }
-            _ => {
-                // Anything else is considered a fatal error or misbehaviour of
-                // the remote peer and results in closing the connection.
-                self.pending_error = Some(error.map_upgrade_err(|e| {
-                    e.map_err(|e| match e {
-                        Either::Left(e) => Either::Left(e),
-                        Either::Right(v) => void::unreachable(v),
-                    })
-                }));
-            }
-        }
+        self.pending_error = Some(StreamUpgradeError::Apply(match error {
+            Either::Left(e) => Either::Left(e),
+            Either::Right(v) => void::unreachable(v),
+        }));
     }
 
     fn on_dial_upgrade_error(
@@ -263,29 +227,27 @@ impl Handler {
         self.keep_alive = KeepAlive::No;
 
         match error {
-            ConnectionHandlerUpgrErr::Timeout => {
+            StreamUpgradeError::Timeout => {
                 self.queued_events.push_back(ConnectionHandlerEvent::Custom(
                     Event::OutboundNegotiationFailed {
-                        error: ConnectionHandlerUpgrErr::Timeout,
+                        error: StreamUpgradeError::Timeout,
                     },
                 ));
             }
-            ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
+            StreamUpgradeError::NegotiationFailed => {
                 // The remote merely doesn't support the DCUtR protocol.
                 // This is no reason to close the connection, which may
                 // successfully communicate with other protocols already.
                 self.queued_events.push_back(ConnectionHandlerEvent::Custom(
                     Event::OutboundNegotiationFailed {
-                        error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(
-                            NegotiationError::Failed,
-                        )),
+                        error: StreamUpgradeError::NegotiationFailed,
                     },
                 ));
             }
             _ => {
                 // Anything else is considered a fatal error or misbehaviour of
                 // the remote peer and results in closing the connection.
-                self.pending_error = Some(error.map_upgrade_err(|e| e.map_err(Either::Right)));
+                self.pending_error = Some(error.map_upgrade_err(Either::Right));
             }
         }
     }
@@ -294,7 +256,7 @@ impl Handler {
 impl ConnectionHandler for Handler {
     type InEvent = Command;
     type OutEvent = Event;
-    type Error = ConnectionHandlerUpgrErr<
+    type Error = StreamUpgradeError<
         Either<protocol::inbound::UpgradeError, protocol::outbound::UpgradeError>,
     >;
     type InboundProtocol = Either<protocol::inbound::Upgrade, DeniedUpgrade>;
@@ -385,9 +347,9 @@ impl ConnectionHandler for Handler {
                     ));
                 }
                 Err(e) => {
-                    return Poll::Ready(ConnectionHandlerEvent::Close(
-                        ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Left(e))),
-                    ))
+                    return Poll::Ready(ConnectionHandlerEvent::Close(StreamUpgradeError::Apply(
+                        Either::Left(e),
+                    )))
                 }
             }
         }

protocols/gossipsub/src/handler.rs

@@ -27,10 +27,10 @@ use futures::future::Either;
 use futures::prelude::*;
 use futures::StreamExt;
 use instant::Instant;
-use libp2p_core::upgrade::{DeniedUpgrade, NegotiationError, UpgradeError};
+use libp2p_core::upgrade::DeniedUpgrade;
 use libp2p_swarm::handler::{
-    ConnectionEvent, ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr,
-    DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound, KeepAlive,
+    ConnectionEvent, ConnectionHandler, ConnectionHandlerEvent, DialUpgradeError,
+    FullyNegotiatedInbound, FullyNegotiatedOutbound, KeepAlive, StreamUpgradeError,
     SubstreamProtocol,
 };
 use libp2p_swarm::NegotiatedSubstream;
@@ -526,20 +526,17 @@ impl ConnectionHandler for Handler {
                         handler.on_fully_negotiated_outbound(fully_negotiated_outbound)
                     }
                     ConnectionEvent::DialUpgradeError(DialUpgradeError {
-                        error: ConnectionHandlerUpgrErr::Timeout | ConnectionHandlerUpgrErr::Timer,
+                        error: StreamUpgradeError::Timeout,
                         ..
                     }) => {
                         log::debug!("Dial upgrade error: Protocol negotiation timeout");
                     }
                     ConnectionEvent::DialUpgradeError(DialUpgradeError {
-                        error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)),
+                        error: StreamUpgradeError::Apply(e),
                         ..
                     }) => void::unreachable(e),
                     ConnectionEvent::DialUpgradeError(DialUpgradeError {
-                        error:
-                            ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(
-                                NegotiationError::Failed,
-                            )),
+                        error: StreamUpgradeError::NegotiationFailed,
                         ..
                     }) => {
                         // The protocol is not supported
@@ -551,10 +548,7 @@ impl ConnectionHandler for Handler {
                         });
                     }
                     ConnectionEvent::DialUpgradeError(DialUpgradeError {
-                        error:
-                            ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(
-                                NegotiationError::ProtocolError(e),
-                            )),
+                        error: StreamUpgradeError::Io(e),
                         ..
                     }) => {
                         log::debug!("Protocol negotiation failed: {e}")

protocols/identify/src/behaviour.rs

@@ -25,8 +25,8 @@ use libp2p_identity::PeerId;
 use libp2p_identity::PublicKey;
 use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm};
 use libp2p_swarm::{
-    AddressScore, ConnectionDenied, ConnectionHandlerUpgrErr, DialError, ExternalAddresses,
-    ListenAddresses, NetworkBehaviour, NotifyHandler, PollParameters, StreamProtocol,
+    AddressScore, ConnectionDenied, DialError, ExternalAddresses, ListenAddresses,
+    NetworkBehaviour, NotifyHandler, PollParameters, StreamProtocol, StreamUpgradeError,
     THandlerInEvent, ToSwarm,
 };
 use libp2p_swarm::{ConnectionId, THandler, THandlerOutEvent};
@@ -495,7 +495,7 @@ pub enum Event {
         /// The peer with whom the error originated.
         peer_id: PeerId,
         /// The error that occurred.
-        error: ConnectionHandlerUpgrErr<UpgradeError>,
+        error: StreamUpgradeError<UpgradeError>,
     },
 }
 

protocols/identify/src/handler.rs

@@ -34,8 +34,8 @@ use libp2p_swarm::handler::{
     ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
 };
 use libp2p_swarm::{
-    ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr, KeepAlive,
-    NegotiatedSubstream, StreamProtocol, SubstreamProtocol,
+    ConnectionHandler, ConnectionHandlerEvent, KeepAlive, NegotiatedSubstream, StreamProtocol,
+    StreamUpgradeError, SubstreamProtocol,
 };
 use log::warn;
 use smallvec::SmallVec;
@@ -111,7 +111,7 @@ pub enum Event {
     /// We received a request for identification.
     Identify,
     /// Failed to identify the remote, or to reply to an identification request.
-    IdentificationError(ConnectionHandlerUpgrErr<UpgradeError>),
+    IdentificationError(StreamUpgradeError<UpgradeError>),
 }
 
 impl Handler {
@@ -205,13 +205,7 @@ impl Handler {
             <Self as ConnectionHandler>::OutboundProtocol,
         >,
     ) {
-        use libp2p_core::upgrade::UpgradeError;
-
-        let err = err.map_upgrade_err(|e| match e {
-            UpgradeError::Select(e) => UpgradeError::Select(e),
-            UpgradeError::Apply(Either::Left(ioe)) => UpgradeError::Apply(ioe),
-            UpgradeError::Apply(Either::Right(ioe)) => UpgradeError::Apply(ioe),
-        });
+        let err = err.map_upgrade_err(|e| e.into_inner());
         self.events
             .push(ConnectionHandlerEvent::Custom(Event::IdentificationError(
                 err,
@@ -317,9 +311,7 @@ impl ConnectionHandler for Handler {
                 Event::Identification(peer_id),
             )),
             Poll::Ready(Some(Err(err))) => Poll::Ready(ConnectionHandlerEvent::Custom(
-                Event::IdentificationError(ConnectionHandlerUpgrErr::Upgrade(
-                    libp2p_core::upgrade::UpgradeError::Apply(err),
-                )),
+                Event::IdentificationError(StreamUpgradeError::Apply(err)),
             )),
             Poll::Ready(None) | Poll::Pending => Poll::Pending,
         }

protocols/kad/src/handler_priv.rs

@@ -33,8 +33,8 @@ use libp2p_swarm::handler::{
     ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
 };
 use libp2p_swarm::{
-    ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr, KeepAlive,
-    NegotiatedSubstream, SubstreamProtocol,
+    ConnectionHandler, ConnectionHandlerEvent, KeepAlive, NegotiatedSubstream, StreamUpgradeError,
+    SubstreamProtocol,
 };
 use log::trace;
 use std::collections::VecDeque;
@@ -325,7 +325,7 @@ pub enum KademliaHandlerEvent<TUserData> {
 #[derive(Debug)]
 pub enum KademliaHandlerQueryErr {
     /// Error while trying to perform the query.
-    Upgrade(ConnectionHandlerUpgrErr<io::Error>),
+    Upgrade(StreamUpgradeError<io::Error>),
     /// Received an answer that doesn't correspond to the request.
     UnexpectedMessage,
     /// I/O error in the substream.
@@ -361,8 +361,8 @@ impl error::Error for KademliaHandlerQueryErr {
     }
 }
 
-impl From<ConnectionHandlerUpgrErr<io::Error>> for KademliaHandlerQueryErr {
-    fn from(err: ConnectionHandlerUpgrErr<io::Error>) -> Self {
+impl From<StreamUpgradeError<io::Error>> for KademliaHandlerQueryErr {
+    fn from(err: StreamUpgradeError<io::Error>) -> Self {
         KademliaHandlerQueryErr::Upgrade(err)
     }
 }

protocols/perf/src/client/behaviour.rs

@@ -28,8 +28,8 @@ use std::{
 use libp2p_core::Multiaddr;
 use libp2p_identity::PeerId;
 use libp2p_swarm::{
-    derive_prelude::ConnectionEstablished, ConnectionClosed, ConnectionHandlerUpgrErr,
-    ConnectionId, FromSwarm, NetworkBehaviour, NotifyHandler, PollParameters, THandlerInEvent,
+    derive_prelude::ConnectionEstablished, ConnectionClosed, ConnectionId, FromSwarm,
+    NetworkBehaviour, NotifyHandler, PollParameters, StreamUpgradeError, THandlerInEvent,
     THandlerOutEvent, ToSwarm,
 };
 use void::Void;
@@ -41,7 +41,7 @@ use super::{RunId, RunParams, RunStats};
 #[derive(Debug)]
 pub struct Event {
     pub id: RunId,
-    pub result: Result<RunStats, ConnectionHandlerUpgrErr<Void>>,
+    pub result: Result<RunStats, StreamUpgradeError<Void>>,
 }
 
 #[derive(Default)]