{core/,swarm/}: Dial with handler and return handler on error and closed

core/src/connection/manager/task.rs

@@ -270,10 +270,8 @@ where
                                 continue 'poll;
                             }
                             Poll::Ready(None) => {
-                                todo!("Safe to assume that this should never happen?");
-                                // The manager has dropped the task or disappeared; abort.
-                                // TODO: Should we return the handler in this case?
-                                // return Poll::Ready(());
+                                // The manager has disappeared; abort.
+                                return Poll::Ready(());
                             }
                         }
                     }

core/src/network.rs

@@ -766,9 +766,8 @@ pub enum DialError<THandler> {
         address: Multiaddr,
         handler: THandler,
     },
-    LocalPeerId {
-        handler: THandler,
-    },
+    /// The dialing attempt is rejected because the peer being dialed is the local peer.
+    LocalPeerId { handler: THandler },
 }
 
 impl<THandler> fmt::Debug for DialError<THandler> {

protocols/identify/src/identify.rs

@@ -27,8 +27,8 @@ use libp2p_core::{
     ConnectedPoint, Multiaddr, PeerId, PublicKey,
 };
 use libp2p_swarm::{
-    AddressScore, DialPeerCondition, IntoProtocolsHandler, NegotiatedSubstream, NetworkBehaviour,
-    NetworkBehaviourAction, NotifyHandler, PollParameters, ProtocolsHandler,
+    AddressScore, DialError, DialPeerCondition, IntoProtocolsHandler, NegotiatedSubstream,
+    NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters, ProtocolsHandler,
     ProtocolsHandlerUpgrErr,
 };
 use std::{
@@ -223,7 +223,7 @@ impl NetworkBehaviour for Identify {
         }
     }
 
-    fn inject_dial_failure(&mut self, peer_id: &PeerId, _: Self::ProtocolsHandler) {
+    fn inject_dial_failure(&mut self, peer_id: &PeerId, _: Self::ProtocolsHandler, _: DialError) {
         if !self.connected.contains_key(peer_id) {
             self.pending_push.remove(peer_id);
         }

protocols/kad/src/behaviour.rs

@@ -43,7 +43,8 @@ use libp2p_core::{
     ConnectedPoint, Multiaddr, PeerId,
 };
 use libp2p_swarm::{
-    DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
+    DialError, DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler,
+    PollParameters,
 };
 use log::{debug, info, warn};
 use smallvec::SmallVec;
@@ -1864,9 +1865,32 @@ where
         }
     }
 
-    fn inject_dial_failure(&mut self, peer_id: &PeerId, _: Self::ProtocolsHandler) {
-        for query in self.queries.iter_mut() {
-            query.on_failure(peer_id);
+    fn inject_dial_failure(
+        &mut self,
+        peer_id: &PeerId,
+        _: Self::ProtocolsHandler,
+        error: DialError,
+    ) {
+        match error {
+            DialError::Banned
+            | DialError::ConnectionLimit(_)
+            | DialError::InvalidAddress(_)
+            | DialError::UnreachableAddr(_)
+            | DialError::LocalPeerId
+            | DialError::NoAddresses => {
+                for query in self.queries.iter_mut() {
+                    query.on_failure(peer_id);
+                }
+            }
+            DialError::DialPeerConditionFalse(
+                DialPeerCondition::Disconnected | DialPeerCondition::NotDialing,
+            ) => {
+                // We might (still) be connected, or about to be connected, thus do not report the
+                // failure to the queries.
+            }
+            DialError::DialPeerConditionFalse(DialPeerCondition::Always) => {
+                unreachable!("DialPeerCondition::Always can not trigger DialPeerConditionFalse.");
+            }
         }
     }
 

protocols/relay/src/behaviour.rs

@@ -29,7 +29,7 @@ use libp2p_core::connection::{ConnectedPoint, ConnectionId, ListenerId};
 use libp2p_core::multiaddr::Multiaddr;
 use libp2p_core::PeerId;
 use libp2p_swarm::{
-    DialPeerCondition, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction,
+    DialError, DialPeerCondition, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction,
     NotifyHandler, PollParameters,
 };
 use std::collections::{hash_map::Entry, HashMap, HashSet, VecDeque};
@@ -302,7 +302,20 @@ impl NetworkBehaviour for Relay {
         }
     }
 
-    fn inject_dial_failure(&mut self, peer_id: &PeerId, _: Self::ProtocolsHandler) {
+    fn inject_dial_failure(
+        &mut self,
+        peer_id: &PeerId,
+        _: Self::ProtocolsHandler,
+        error: DialError,
+    ) {
+        if let DialError::DialPeerConditionFalse(
+            DialPeerCondition::Disconnected | DialPeerCondition::NotDialing,
+        ) = error
+        {
+            // Return early. The dial, that this dial was canceled for, might still succeed.
+            return;
+        }
+
         if let Entry::Occupied(o) = self.listeners.entry(*peer_id) {
             if matches!(o.get(), RelayListener::Connecting { .. }) {
                 // By removing the entry, the channel to the listener is dropped and thus the

protocols/request-response/src/lib.rs

@@ -68,7 +68,7 @@ use futures::channel::oneshot;
 use handler::{RequestProtocol, RequestResponseHandler, RequestResponseHandlerEvent};
 use libp2p_core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
 use libp2p_swarm::{
-    DialPeerCondition, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction,
+    DialError, DialPeerCondition, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction,
     NotifyHandler, PollParameters,
 };
 use smallvec::SmallVec;
@@ -686,7 +686,7 @@ where
         self.connected.remove(peer);
     }
 
-    fn inject_dial_failure(&mut self, peer: &PeerId, _: Self::ProtocolsHandler) {
+    fn inject_dial_failure(&mut self, peer: &PeerId, _: Self::ProtocolsHandler, _: DialError) {
         // If there are pending outgoing requests when a dial failure occurs,
         // it is implied that we are not connected to the peer, since pending
         // outgoing requests are drained when a connection is established and

protocols/request-response/src/throttled.rs

@@ -45,7 +45,7 @@ use codec::{Codec, Message, ProtocolWrapper, Type};
 use futures::ready;
 use libp2p_core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
 use libp2p_swarm::{
-    IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
+    DialError, IntoProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
 };
 use lru::LruCache;
 use std::{cmp::max, num::NonZeroU16};
@@ -493,8 +493,13 @@ where
         self.behaviour.inject_disconnected(p)
     }
 
-    fn inject_dial_failure(&mut self, p: &PeerId, handler: Self::ProtocolsHandler) {
-        self.behaviour.inject_dial_failure(p, handler)
+    fn inject_dial_failure(
+        &mut self,
+        p: &PeerId,
+        handler: Self::ProtocolsHandler,
+        error: DialError,
+    ) {
+        self.behaviour.inject_dial_failure(p, handler, error)
     }
 
     fn inject_event(

swarm-derive/src/lib.rs

@@ -57,6 +57,7 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
     let connection_id = quote! {::libp2p::core::connection::ConnectionId};
     let connected_point = quote! {::libp2p::core::ConnectedPoint};
     let listener_id = quote! {::libp2p::core::connection::ListenerId};
+    let dial_error = quote! {::libp2p::swarm::DialError};
 
     let poll_parameters = quote! {::libp2p::swarm::PollParameters};
 
@@ -227,11 +228,13 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
             .fields
             .iter()
             .enumerate()
+            // The outmost handler belongs to the last behaviour.
             .rev()
             .filter(|f| !is_ignored(&f.1))
             .enumerate()
             .map(move |(enum_n, (field_n, field))| {
                 let handler = if field_n == 0 {
+                    // Given that the iterator is reversed, this is the innermost handler only.
                     quote! { let handler = handlers }
                 } else {
                     quote! {
@@ -271,11 +274,13 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
             .fields
             .iter()
             .enumerate()
+            // The outmost handler belongs to the last behaviour.
             .rev()
             .filter(|f| !is_ignored(&f.1))
             .enumerate()
             .map(move |(enum_n, (field_n, field))| {
                 let handler = if field_n == 0 {
+                    // Given that the iterator is reversed, this is the innermost handler only.
                     quote! { let handler = handlers }
                 } else {
                     quote! {
@@ -284,8 +289,42 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
                 };
 
                 let inject = match field.ident {
-                    Some(ref i) => quote! { self.#i.inject_dial_failure(peer_id, handler) },
-                    None => quote! { self.#enum_n.inject_dial_failure(peer_id, handler) },
+                    Some(ref i) => {
+                        quote! { self.#i.inject_dial_failure(peer_id, handler, error.clone()) }
+                    }
+                    None => {
+                        quote! { self.#enum_n.inject_dial_failure(peer_id, handler, error.clone()) }
+                    }
+                };
+
+                quote! {
+                    #handler;
+                    #inject;
+                }
+            })
+    };
+
+    // Build the list of statements to put in the body of `inject_listen_failure()`.
+    let inject_listen_failure_stmts = {
+        data_struct
+            .fields
+            .iter()
+            .enumerate()
+            .rev()
+            .filter(|f| !is_ignored(&f.1))
+            .enumerate()
+            .map(move |(enum_n, (field_n, field))| {
+                let handler = if field_n == 0 {
+                    quote! { let handler = handlers }
+                } else {
+                    quote! {
+                        let (handlers, handler) = handlers.into_inner()
+                    }
+                };
+
+                let inject = match field.ident {
+                    Some(ref i) => quote! { self.#i.inject_listen_failure(local_addr, send_back_addr, handler) },
+                    None => quote! { self.#enum_n.inject_listen_failure(local_addr, send_back_addr, handler) },
                 };
 
                 quote! {
@@ -485,7 +524,7 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
             }
         }
 
-        out_handler.unwrap_or(quote! {()}) // TODO: incorrect
+        out_handler.unwrap_or(quote! {()}) // TODO: See test `empty`.
     };
 
     // The method to use to poll.
@@ -529,7 +568,11 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
             wrapped_event = quote!{ #either_ident::First(#wrapped_event) };
         }
 
-        let wrapped_handler = {
+        // `DialPeer` and `DialAddress` each provide a handler of the specific
+        // behaviour triggering the event. Though in order for the final handler
+        // to be able to handle protocols of all behaviours, the provided
+        // handler needs to be combined with handlers of all other behaviours.
+        let provided_handler_and_new_handlers = {
             let mut out_handler = None;
 
             for (f_n, f) in data_struct.fields.iter().enumerate() {
@@ -543,7 +586,9 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
                 };
 
                 let builder = if field_n == f_n {
-                    quote! { handler }
+                    // The behaviour that triggered the event. Thus, instead of
+                    // creating a new handler, use the provided handler.
+                    quote! { provided_handler }
                 } else {
                     quote! { #f_name.new_handler() }
                 };
@@ -556,7 +601,7 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
                 }
             }
 
-            out_handler.unwrap_or(quote! {()}) // TODO: incorrect
+            out_handler.unwrap_or(quote! {()}) // TODO: See test `empty`.
         };
 
         let generate_event_match_arm = if event_process {
@@ -577,11 +622,11 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
             loop {
                 match #trait_to_impl::poll(&mut #field_name, cx, poll_params) {
                     #generate_event_match_arm
-                    std::task::Poll::Ready(#network_behaviour_action::DialAddress { address, handler }) => {
-                        return std::task::Poll::Ready(#network_behaviour_action::DialAddress { address, handler: #wrapped_handler });
+                    std::task::Poll::Ready(#network_behaviour_action::DialAddress { address, providedhandler }) => {
+                        return std::task::Poll::Ready(#network_behaviour_action::DialAddress { address, handler: #provided_handler_and_new_handlers });
                     }
-                    std::task::Poll::Ready(#network_behaviour_action::DialPeer { peer_id, condition, handler }) => {
-                        return std::task::Poll::Ready(#network_behaviour_action::DialPeer { peer_id, condition, handler: #wrapped_handler });
+                    std::task::Poll::Ready(#network_behaviour_action::DialPeer { peer_id, condition, provided_handler }) => {
+                        return std::task::Poll::Ready(#network_behaviour_action::DialPeer { peer_id, condition, handler: #provided_handler_and_new_handlers });
                     }
                     std::task::Poll::Ready(#network_behaviour_action::NotifyHandler { peer_id, handler, event }) => {
                         return std::task::Poll::Ready(#network_behaviour_action::NotifyHandler {
@@ -645,10 +690,14 @@ fn build_struct(ast: &DeriveInput, data_struct: &DataStruct) -> TokenStream {
                 #(#inject_addr_reach_failure_stmts);*
             }
 
-            fn inject_dial_failure(&mut self, peer_id: &#peer_id, handlers: Self::ProtocolsHandler) {
+            fn inject_dial_failure(&mut self, peer_id: &#peer_id, handlers: Self::ProtocolsHandler, error: #dial_error) {
                 #(#inject_dial_failure_stmts);*
             }
 
+            fn inject_listen_failure(&mut self, local_addr: &#multiaddr, send_back_addr: &#multiaddr, handlers: Self::ProtocolsHandler) {
+                #(#inject_listen_failure_stmts);*
+            }
+
             fn inject_new_listener(&mut self, id: #listener_id) {
                 #(#inject_new_listener_stmts);*
             }

swarm/CHANGELOG.md

@@ -19,6 +19,8 @@
 - Implement `ProtocolsHandler` on `either::Either`representing either of two
   `ProtocolsHandler` implementations (see [PR 2192]).
 
+TODO: Mention `inject_dial_failure` now called on DialPeerCondition not met.
+
 [PR 2150]: https://github.com/libp2p/rust-libp2p/pull/2150
 [PR 2182]: https://github.com/libp2p/rust-libp2p/pull/2182
 [PR 2183]: https://github.com/libp2p/rust-libp2p/pull/2183