Make the SwarmEvent report everything (#1515)

* Improve the SwarmEvent to report everything

* Address review

* Update swarm/src/lib.rs

Co-Authored-By: Roman Borschel <romanb@users.noreply.github.com>

Co-authored-by: Roman Borschel <romanb@users.noreply.github.com>

core/src/connection/pool.rs

@@ -44,7 +44,7 @@ use either::Either;
 use fnv::FnvHashMap;
 use futures::prelude::*;
 use smallvec::SmallVec;
-use std::{error, fmt, hash::Hash, task::Context, task::Poll};
+use std::{convert::TryFrom as _, error, fmt, hash::Hash, num::NonZeroU32, task::Context, task::Poll};
 
 /// A connection `Pool` manages a set of connections for each peer.
 pub struct Pool<TInEvent, TOutEvent, THandler, TTransErr, THandlerErr, TConnInfo = PeerId, TPeerId = PeerId> {
@@ -86,7 +86,7 @@ pub enum PoolEvent<'a, TInEvent, TOutEvent, THandler, TTransErr, THandlerErr, TC
     /// A new connection has been established.
     ConnectionEstablished {
         connection: EstablishedConnection<'a, TInEvent, TConnInfo, TPeerId>,
-        num_established: usize,
+        num_established: NonZeroU32,
     },
 
     /// An established connection has encountered an error.
@@ -99,7 +99,7 @@ pub enum PoolEvent<'a, TInEvent, TOutEvent, THandler, TTransErr, THandlerErr, TC
         /// A reference to the pool that used to manage the connection.
         pool: &'a mut Pool<TInEvent, TOutEvent, THandler, TTransErr, THandlerErr, TConnInfo, TPeerId>,
         /// The remaining number of established connections to the same peer.
-        num_established: usize,
+        num_established: u32,
     },
 
     /// A connection attempt failed.
@@ -580,7 +580,7 @@ where
                     let num_established =
                         if let Some(conns) = self.established.get_mut(connected.peer_id()) {
                             conns.remove(&id);
-                            conns.len()
+                            u32::try_from(conns.len()).unwrap()
                         } else {
                             0
                         };
@@ -600,7 +600,7 @@ where
                                             .map_or(0, |conns| conns.len());
                         if let Err(e) = self.limits.check_established(current) {
                             let connected = entry.close();
-                            let num_established = e.current;
+                            let num_established = u32::try_from(e.current).unwrap();
                             return Poll::Ready(PoolEvent::ConnectionError {
                                 id,
                                 connected,
@@ -623,7 +623,8 @@ where
                         // Add the connection to the pool.
                         let peer = entry.connected().peer_id().clone();
                         let conns = self.established.entry(peer).or_default();
-                        let num_established = conns.len() + 1;
+                        let num_established = NonZeroU32::new(u32::try_from(conns.len() + 1).unwrap())
+                            .expect("n + 1 is always non-zero; qed");
                         conns.insert(id, endpoint);
                         match self.get(id) {
                             Some(PoolConnection::Established(connection)) =>

core/src/network.rs

@@ -52,6 +52,7 @@ use fnv::{FnvHashMap};
 use futures::{prelude::*, future};
 use std::{
     collections::hash_map,
+    convert::TryFrom as _,
     error,
     fmt,
     hash::Hash,
@@ -517,7 +518,7 @@ where
         // A pending outgoing connection to a known peer failed.
         let mut attempt = dialing.remove(&peer_id).expect("by (1)");
 
-        let num_remain = attempt.next.len();
+        let num_remain = u32::try_from(attempt.next.len()).unwrap();
         let failed_addr = attempt.current.clone();
 
         let opts =

core/src/network/event.rs

@@ -42,7 +42,7 @@ use crate::{
     transport::{Transport, TransportError},
 };
 use futures::prelude::*;
-use std::{error, fmt, hash::Hash};
+use std::{error, fmt, hash::Hash, num::NonZeroU32};
 
 /// Event that can happen on the `Network`.
 pub enum NetworkEvent<'a, TTrans, TInEvent, TOutEvent, THandler, TConnInfo, TPeerId>
@@ -88,7 +88,7 @@ where
     /// A new connection arrived on a listener.
     IncomingConnection(IncomingConnectionEvent<'a, TTrans, TInEvent, TOutEvent, THandler, TConnInfo, TPeerId>),
 
-    /// A new connection was arriving on a listener, but an error happened when negotiating it.
+    /// An error happened on a connection during its initial handshake.
     ///
     /// This can include, for example, an error during the handshake of the encryption layer, or
     /// the connection unexpectedly closed.
@@ -105,8 +105,9 @@ where
     ConnectionEstablished {
         /// The newly established connection.
         connection: EstablishedConnection<'a, TInEvent, TConnInfo, TPeerId>,
-        /// The total number of established connections to the same peer.
+        /// The total number of established connections to the same peer, including the one that
-        num_established: usize,
+        /// has just been opened.
+        num_established: NonZeroU32,
     },
 
     /// An established connection to a peer has encountered an error.
@@ -118,13 +119,13 @@ where
         /// The error that occurred.
         error: ConnectionError<<THandler::Handler as ConnectionHandler>::Error>,
         /// The remaining number of established connections to the same peer.
-        num_established: usize,
+        num_established: u32,
     },
 
     /// A dialing attempt to an address of a peer failed.
     DialError {
         /// The number of remaining dialing attempts.
-        attempts_remaining: usize,
+        attempts_remaining: u32,
 
         /// Id of the peer we were trying to dial.
         peer_id: TPeerId,

core/tests/network_dial_error.rs

@@ -258,7 +258,7 @@ fn multiple_addresses_err() {
                         assert_eq!(attempts_remaining, 0);
                         return Poll::Ready(Ok(()));
                     } else {
-                        assert_eq!(attempts_remaining, addresses.len());
+                        assert_eq!(attempts_remaining, addresses.len() as u32);
                     }
                 },
                 Poll::Ready(_) => unreachable!(),

swarm/src/lib.rs

@@ -79,24 +79,30 @@ pub use protocols_handler::{
     SubstreamProtocol
 };
 
-use protocols_handler::NodeHandlerWrapperBuilder;
+use protocols_handler::{
+    NodeHandlerWrapperBuilder,
+    NodeHandlerWrapperError,
+};
 use futures::{
     prelude::*,
     executor::{ThreadPool, ThreadPoolBuilder},
     stream::FusedStream,
 };
 use libp2p_core::{
+    ConnectedPoint,
     Executor,
     Transport,
     Multiaddr,
     Negotiated,
     PeerId,
     connection::{
+        ConnectionError,
         ConnectionId,
         ConnectionInfo,
         EstablishedConnection,
         IntoConnectionHandler,
         ListenerId,
+        PendingConnectionError,
         Substream
     },
     transport::{TransportError, boxed::Boxed as BoxTransport},
@@ -116,6 +122,7 @@ use registry::{Addresses, AddressIntoIter};
 use smallvec::SmallVec;
 use std::{error, fmt, hash::Hash, io, ops::{Deref, DerefMut}, pin::Pin, task::{Context, Poll}};
 use std::collections::HashSet;
+use std::num::NonZeroU32;
 use upgrade::UpgradeInfoSend as _;
 
 /// Contains the state of the network, plus the way it should behave.
@@ -135,29 +142,110 @@ pub type NegotiatedSubstream = Negotiated<Substream<StreamMuxerBox>>;
 
 /// Event generated by the `Swarm`.
 #[derive(Debug)]
-pub enum SwarmEvent<TBvEv> {
+pub enum SwarmEvent<TBvEv, THandleErr> {
     /// Event generated by the `NetworkBehaviour`.
     Behaviour(TBvEv),
-    /// We are now connected to the given peer.
+    /// A connection to the given peer has been opened.
-    Connected(PeerId),
+    ConnectionEstablished {
-    /// We are now disconnected from the given peer.
+        /// Identity of the peer that we have connected to.
-    Disconnected(PeerId),
+        peer_id: PeerId,
+        /// Endpoint of the connection that has been opened.
+        endpoint: ConnectedPoint,
+        /// Number of established connections to this peer, including the one that has just been
+        /// opened.
+        num_established: NonZeroU32,
+    },
+    /// A connection with the given peer has been closed.
+    ConnectionClosed {
+        /// Identity of the peer that we have connected to.
+        peer_id: PeerId,
+        /// Endpoint of the connection that has been closed.
+        endpoint: ConnectedPoint,
+        /// Number of other remaining connections to this same peer.
+        num_established: u32,
+        /// Reason for the disconnection.
+        cause: ConnectionError<NodeHandlerWrapperError<THandleErr>>,
+    },
+    /// A new connection arrived on a listener and is in the process of protocol negotiation.
+    ///
+    /// A corresponding [`ConnectionEstablished`](SwarmEvent::ConnectionEstablished),
+    /// [`BannedPeer`](SwarmEvent::BannedPeer), or
+    /// [`IncomingConnectionError`](SwarmEvent::IncomingConnectionError) event will later be
+    /// generated for this connection.
+    IncomingConnection {
+        /// Local connection address.
+        /// This address has been earlier reported with a [`NewListenAddr`](SwarmEvent::NewListenAddr)
+        /// event.
+        local_addr: Multiaddr,
+        /// Address used to send back data to the remote.
+        send_back_addr: Multiaddr,
+    },
+    /// An error happened on a connection during its initial handshake.
+    ///
+    /// This can include, for example, an error during the handshake of the encryption layer, or
+    /// the connection unexpectedly closed.
+    IncomingConnectionError {
+        /// Local connection address.
+        /// This address has been earlier reported with a [`NewListenAddr`](SwarmEvent::NewListenAddr)
+        /// event.
+        local_addr: Multiaddr,
+        /// Address used to send back data to the remote.
+        send_back_addr: Multiaddr,
+        /// The error that happened.
+        error: PendingConnectionError<io::Error>,
+    },
+    /// We connected to a peer, but we immediately closed the connection because that peer is banned.
+    BannedPeer {
+        /// Identity of the banned peer.
+        peer_id: PeerId,
+        /// Endpoint of the connection that has been closed.
+        endpoint: ConnectedPoint,
+    },
+    /// Starting to try to reach the given peer.
+    ///
+    /// We are trying to connect to this peer until a [`ConnectionEstablished`](SwarmEvent::ConnectionEstablished)
+    /// event is reported, or a [`UnreachableAddr`](SwarmEvent::UnreachableAddr) event is reported
+    /// with `attempts_remaining` equal to 0.
+    Dialing(PeerId),
+    /// Tried to dial an address but it ended up being unreachaable.
+    UnreachableAddr {
+        /// `PeerId` that we were trying to reach.
+        peer_id: PeerId,
+        /// Address that we failed to reach.
+        address: Multiaddr,
+        /// Error that has been encountered.
+        error: PendingConnectionError<io::Error>,
+        /// Number of remaining connection attempts that are being tried for this peer.
+        attempts_remaining: u32,
+    },
+    /// Tried to dial an address but it ended up being unreachaable.
+    /// Contrary to `UnreachableAddr`, we don't know the identity of the peer that we were trying
+    /// to reach.
+    UnknownPeerUnreachableAddr {
+        /// Address that we failed to reach.
+        address: Multiaddr,
+        /// Error that has been encountered.
+        error: PendingConnectionError<io::Error>,
+    },
     /// One of our listeners has reported a new local listening address.
     NewListenAddr(Multiaddr),
     /// One of our listeners has reported the expiration of a listening address.
     ExpiredListenAddr(Multiaddr),
-    /// Tried to dial an address but it ended up being unreachaable.
+    /// One of the listeners gracefully closed.
-    UnreachableAddr {
+    ListenerClosed {
-        /// `PeerId` that we were trying to reach. `None` if we don't know in advance which peer
+        /// The addresses that the listener was listening on. These addresses are now considered
-        /// we were trying to reach.
+        /// expired, similar to if a [`ExpiredListenAddr`](SwarmEvent::ExpiredListenAddr) event
-        peer_id: Option<PeerId>,
+        /// has been generated for each of them.
-        /// Address that we failed to reach.
+        addresses: Vec<Multiaddr>,
-        address: Multiaddr,
+        /// Reason for the closure. Contains `Ok(())` if the stream produced `None`, or `Err`
-        /// Error that has been encountered.
+        /// if the stream produced an error.
-        error: Box<dyn error::Error + Send>,
+        reason: Result<(), io::Error>,
+    },
+    /// One of the listeners reported a non-fatal error.
+    ListenerError {
+        /// The listener error.
+        error: io::Error,
     },
-    /// Startng to try to reach the given peer.
-    StartConnect(PeerId),
 }
 
 /// Contains the state of the network, plus the way it should behave.
@@ -230,14 +318,15 @@ where
 {
 }
 
-impl<TBehaviour, TInEvent, TOutEvent, THandler, TConnInfo>
+impl<TBehaviour, TInEvent, TOutEvent, THandler, TConnInfo, THandleErr>
     ExpandedSwarm<TBehaviour, TInEvent, TOutEvent, THandler, TConnInfo>
 where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
       TInEvent: Clone + Send + 'static,
       TOutEvent: Send + 'static,
       TConnInfo: ConnectionInfo<PeerId = PeerId> + fmt::Debug + Clone + Send + 'static,
       THandler: IntoProtocolsHandler + Send + 'static,
-      THandler::Handler: ProtocolsHandler<InEvent = TInEvent, OutEvent = TOutEvent>,
+      THandler::Handler: ProtocolsHandler<InEvent = TInEvent, OutEvent = TOutEvent, Error = THandleErr>,
+      THandleErr: error::Error + Send + 'static,
 {
     /// Builds a new `Swarm`.
     pub fn new<TTransport, TMuxer>(transport: TTransport, behaviour: TBehaviour, local_peer_id: PeerId) -> Self
@@ -360,7 +449,7 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
     /// Returns the next event that happens in the `Swarm`.
     ///
     /// Includes events from the `NetworkBehaviour` but also events about the connections status.
-    pub async fn next_event(&mut self) -> SwarmEvent<TBehaviour::OutEvent> {
+    pub async fn next_event(&mut self) -> SwarmEvent<TBehaviour::OutEvent, THandleErr> {
         future::poll_fn(move |cx| ExpandedSwarm::poll_next_event(Pin::new(self), cx)).await
     }
 
@@ -380,7 +469,7 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
     ///
     /// Polls the `Swarm` for the next event.
     fn poll_next_event(mut self: Pin<&mut Self>, cx: &mut Context)
-        -> Poll<SwarmEvent<TBehaviour::OutEvent>>
+        -> Poll<SwarmEvent<TBehaviour::OutEvent, THandleErr>>
     {
         // We use a `this` variable because the compiler can't mutably borrow multiple times
         // across a `Deref`.
@@ -398,38 +487,62 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
                     this.behaviour.inject_event(peer, connection, event);
                 },
                 Poll::Ready(NetworkEvent::ConnectionEstablished { connection, num_established }) => {
-                    let peer = connection.peer_id().clone();
+                    let peer_id = connection.peer_id().clone();
-                    if this.banned_peers.contains(&peer) {
+                    let endpoint = connection.endpoint().clone();
-                        this.network.peer(peer)
+                    if this.banned_peers.contains(&peer_id) {
+                        this.network.peer(peer_id.clone())
                             .into_connected()
                             .expect("the Network just notified us that we were connected; QED")
                             .disconnect();
-                    } else if num_established == 1 {
+                        return Poll::Ready(SwarmEvent::BannedPeer {
-                        let endpoint = connection.endpoint().clone();
+                            peer_id,
-                        this.behaviour.inject_connected(peer.clone(), endpoint);
+                            endpoint,
-                        return Poll::Ready(SwarmEvent::Connected(peer));
+                        });
+                    } else if num_established.get() == 1 {
+                        this.behaviour.inject_connected(peer_id.clone(), endpoint.clone());
+                        return Poll::Ready(SwarmEvent::ConnectionEstablished {
+                            peer_id,
+                            endpoint,
+                            num_established,
+                        });
                     } else {
                         // For now, secondary connections are not explicitly reported to
                         // the behaviour. A behaviour only gets awareness of the
                         // connections via the events emitted from the connection handlers.
                         log::trace!("Secondary connection established: {:?}; Total (peer): {}.",
                             connection.connected(), num_established);
+                        return Poll::Ready(SwarmEvent::ConnectionEstablished {
+                            peer_id,
+                            endpoint,
+                            num_established,
+                        });
                     }
                 },
                 Poll::Ready(NetworkEvent::ConnectionError { connected, error, num_established }) => {
                     log::debug!("Connection {:?} closed by {:?}", connected, error);
+                    let peer_id = connected.peer_id().clone();
+                    let endpoint = connected.endpoint;
                     if num_established == 0 {
-                        let peer = connected.peer_id().clone();
+                        this.behaviour.inject_disconnected(&peer_id, endpoint.clone());
-                        let endpoint = connected.endpoint;
-                        this.behaviour.inject_disconnected(&peer, endpoint);
-                        return Poll::Ready(SwarmEvent::Disconnected(peer));
                     }
+                    return Poll::Ready(SwarmEvent::ConnectionClosed {
+                        peer_id,
+                        endpoint,
+                        cause: error,
+                        num_established,
+                    });
                 },
                 Poll::Ready(NetworkEvent::IncomingConnection(incoming)) => {
                     let handler = this.behaviour.new_handler();
+                    let local_addr = incoming.local_addr().clone();
+                    let send_back_addr = incoming.send_back_addr().clone();
                     if let Err(e) = incoming.accept(handler.into_node_handler_builder()) {
                         log::warn!("Incoming connection rejected: {:?}", e);
                     }
+                    return Poll::Ready(SwarmEvent::IncomingConnection {
+                        local_addr,
+                        send_back_addr,
+                    });
                 },
                 Poll::Ready(NetworkEvent::NewListenerAddress { listener_id, listen_addr }) => {
                     log::debug!("Listener {:?}; New address: {:?}", listener_id, listen_addr);
@@ -451,11 +564,24 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
                         this.behaviour.inject_expired_listen_addr(addr);
                     }
                     this.behaviour.inject_listener_closed(listener_id);
+                    return Poll::Ready(SwarmEvent::ListenerClosed {
+                        addresses,
+                        reason,
+                    });
                 }
-                Poll::Ready(NetworkEvent::ListenerError { listener_id, error }) =>
+                Poll::Ready(NetworkEvent::ListenerError { listener_id, error }) => {
-                    this.behaviour.inject_listener_error(listener_id, &error),
+                    this.behaviour.inject_listener_error(listener_id, &error);
-                Poll::Ready(NetworkEvent::IncomingConnectionError { error, .. }) => {
+                    return Poll::Ready(SwarmEvent::ListenerError {
+                        error,
+                    });
+                },
+                Poll::Ready(NetworkEvent::IncomingConnectionError { local_addr, send_back_addr, error }) => {
                     log::debug!("Incoming connection failed: {:?}", error);
+                    return Poll::Ready(SwarmEvent::IncomingConnectionError {
+                        local_addr,
+                        send_back_addr,
+                        error,
+                    });
                 },
                 Poll::Ready(NetworkEvent::DialError { peer_id, multiaddr, error, attempts_remaining }) => {
                     log::debug!(
@@ -466,19 +592,19 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
                         this.behaviour.inject_dial_failure(&peer_id);
                     }
                     return Poll::Ready(SwarmEvent::UnreachableAddr {
-                        peer_id: Some(peer_id.clone()),
+                        peer_id,
                         address: multiaddr,
-                        error: Box::new(error),
+                        error,
+                        attempts_remaining,
                     });
                 },
                 Poll::Ready(NetworkEvent::UnknownPeerDialError { multiaddr, error, .. }) => {
                     log::debug!("Connection attempt to address {:?} of unknown peer failed with {:?}",
                         multiaddr, error);
                     this.behaviour.inject_addr_reach_failure(None, &multiaddr, &error);
-                    return Poll::Ready(SwarmEvent::UnreachableAddr {
+                    return Poll::Ready(SwarmEvent::UnknownPeerUnreachableAddr {
-                        peer_id: None,
                         address: multiaddr,
-                        error: Box::new(error),
+                        error,
                     });
                 },
             }
@@ -542,7 +668,7 @@ where TBehaviour: NetworkBehaviour<ProtocolsHandler = THandler>,
                         this.behaviour.inject_dial_failure(&peer_id);
                     } else {
                         ExpandedSwarm::dial(&mut *this, peer_id.clone());
-                        return Poll::Ready(SwarmEvent::StartConnect(peer_id))
+                        return Poll::Ready(SwarmEvent::Dialing(peer_id))
                     }
                 },
                 Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }) => {