refactor(swarm)!: remove handler from NetworkBehaviourAction::Dial

.github/workflows/ci.yml

@@ -298,11 +298,12 @@ jobs:
           require_scope: false
 
       - name: Check PR title length
+        env:
+          TITLE: ${{ github.event.pull_request.title }}
         run: |
-          title="${{ github.event.pull_request.title }}"
-          title_length=${#title}
+          title_length=${#TITLE}
           if [ $title_length -gt 72 ]
           then
             echo "PR title is too long (greater than 72 characters)"
             exit 1
-          fi
+          fi

core/CHANGELOG.md

@@ -1,3 +1,9 @@
+# 0.39.0 [unreleased]
+
+- Deprecate `ConnectionId::new` in favor of `ConnectionId::next`. See [PR 3327].
+
+[PR 3327]: https://github.com/libp2p/rust-libp2p/pull/3327
+
 # 0.38.0
 
 - Remove deprecated functions `StreamMuxerExt::next_{inbound,outbound}`. See [PR 3031].

core/src/connection.rs

@@ -19,6 +19,9 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::multiaddr::{Multiaddr, Protocol};
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+static NEXT_CONNECTION_ID: AtomicUsize = AtomicUsize::new(0);
 
 /// Connection identifier.
 #[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
@@ -30,16 +33,14 @@ impl ConnectionId {
     /// This is primarily useful for creating connection IDs
     /// in test environments. There is in general no guarantee
     /// that all connection IDs are based on non-negative integers.
+    #[deprecated(since = "0.39.0", note = "Use `ConnectionId::next` instead.")]
     pub fn new(id: usize) -> Self {
         Self(id)
     }
-}
-
-impl std::ops::Add<usize> for ConnectionId {
-    type Output = Self;
 
-    fn add(self, other: usize) -> Self {
-        Self(self.0 + other)
+    /// Returns the next available [`ConnectionId`].
+    pub fn next() -> Self {
+        Self(NEXT_CONNECTION_ID.fetch_add(1, Ordering::SeqCst))
     }
 }
 

protocols/autonat/src/behaviour.rs

@@ -40,7 +40,7 @@ use libp2p_swarm::{
         ExpiredListenAddr, FromSwarm,
     },
     ConnectionHandler, ExternalAddresses, IntoConnectionHandler, ListenAddresses, NetworkBehaviour,
-    NetworkBehaviourAction, PollParameters,
+    NetworkBehaviourAction, PollParameters, THandlerInEvent,
 };
 use std::{
     collections::{HashMap, VecDeque},
@@ -210,10 +210,7 @@ pub struct Behaviour {
     last_probe: Option<Instant>,
 
     pending_actions: VecDeque<
-        NetworkBehaviourAction<
-            <Self as NetworkBehaviour>::OutEvent,
-            <Self as NetworkBehaviour>::ConnectionHandler,
-        >,
+        NetworkBehaviourAction<<Self as NetworkBehaviour>::OutEvent, THandlerInEvent<Self>>,
     >,
 
     probe_id: ProbeId,
@@ -391,14 +388,14 @@ impl Behaviour {
         &mut self,
         DialFailure {
             peer_id,
-            handler,
+            connection_id,
             error,
-        }: DialFailure<<Self as NetworkBehaviour>::ConnectionHandler>,
+        }: DialFailure,
     ) {
         self.inner
             .on_swarm_event(FromSwarm::DialFailure(DialFailure {
                 peer_id,
-                handler,
+                connection_id,
                 error,
             }));
         if let Some(event) = self.as_server().on_outbound_dial_error(peer_id, error) {
@@ -563,10 +560,8 @@ impl NetworkBehaviour for Behaviour {
     }
 }
 
-type Action = NetworkBehaviourAction<
-    <Behaviour as NetworkBehaviour>::OutEvent,
-    <Behaviour as NetworkBehaviour>::ConnectionHandler,
->;
+type Action =
+    NetworkBehaviourAction<<Behaviour as NetworkBehaviour>::OutEvent, THandlerInEvent<Behaviour>>;
 
 // Trait implemented for `AsClient` and `AsServer` to handle events from the inner [`request_response::Behaviour`] Protocol.
 trait HandleInnerEvent {

protocols/autonat/src/behaviour/as_server.rs

@@ -29,7 +29,7 @@ use libp2p_request_response::{
 };
 use libp2p_swarm::{
     dial_opts::{DialOpts, PeerCondition},
-    DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
+    DialError, NetworkBehaviourAction, PollParameters,
 };
 use std::{
     collections::{HashMap, HashSet, VecDeque},
@@ -138,7 +138,7 @@ impl<'a> HandleInnerEvent for AsServer<'a> {
                                     )
                                     .addresses(addrs)
                                     .build(),
-                                handler: self.inner.new_handler(),
+                                connection_id: ConnectionId::next(),
                             },
                         ])
                     }

protocols/dcutr/src/behaviour_impl.rs

@@ -29,11 +29,12 @@ use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailu
 use libp2p_swarm::dial_opts::{self, DialOpts};
 use libp2p_swarm::{
     ConnectionHandler, ConnectionHandlerUpgrErr, ExternalAddresses, IntoConnectionHandler,
-    NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
+    NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters, THandlerInEvent,
 };
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::task::{Context, Poll};
 use thiserror::Error;
+use void::Void;
 
 const MAX_NUMBER_OF_UPGRADE_ATTEMPTS: u8 = 3;
 
@@ -67,14 +68,18 @@ pub enum Error {
 
 pub struct Behaviour {
     /// Queue of actions to return when polled.
-    queued_events: VecDeque<NetworkBehaviourAction<Event, handler::Prototype>>,
+    queued_events: VecDeque<NetworkBehaviourAction<Event, Either<handler::relayed::Command, Void>>>,
 
     /// All direct (non-relayed) connections.
     direct_connections: HashMap<PeerId, HashSet<ConnectionId>>,
 
     external_addresses: ExternalAddresses,
 
     local_peer_id: PeerId,
+
+    direct_to_relayed_connections: HashMap<ConnectionId, ConnectionId>,
+
+    outgoing_direct_connection_attempts: HashMap<(ConnectionId, PeerId), u8>,
 }
 
 impl Behaviour {
@@ -84,6 +89,8 @@ impl Behaviour {
             direct_connections: Default::default(),
             external_addresses: Default::default(),
             local_peer_id,
+            direct_to_relayed_connections: Default::default(),
+            outgoing_direct_connection_attempts: Default::default(),
         }
     }
 
@@ -147,40 +154,57 @@ impl Behaviour {
     fn on_dial_failure(
         &mut self,
         DialFailure {
-            peer_id, handler, ..
-        }: DialFailure<<Self as NetworkBehaviour>::ConnectionHandler>,
+            peer_id,
+            connection_id: failed_direct_connection,
+            ..
+        }: DialFailure,
     ) {
-        if let handler::Prototype::DirectConnection {
-            relayed_connection_id,
-            role: handler::Role::Initiator { attempt },
-        } = handler
+        let peer_id = if let Some(peer_id) = peer_id {
+            peer_id
+        } else {
+            return;
+        };
+
+        let relayed_connection_id = if let Some(relayed_connection_id) = self
+            .direct_to_relayed_connections
+            .get(&failed_direct_connection)
         {
-            let peer_id = peer_id.expect("Peer of `Prototype::DirectConnection` is always known.");
-            if attempt < MAX_NUMBER_OF_UPGRADE_ATTEMPTS {
-                self.queued_events
-                    .push_back(NetworkBehaviourAction::NotifyHandler {
-                        handler: NotifyHandler::One(relayed_connection_id),
-                        peer_id,
-                        event: Either::Left(handler::relayed::Command::Connect {
-                            attempt: attempt + 1,
-                            obs_addrs: self.observed_addreses(),
-                        }),
-                    })
-            } else {
-                self.queued_events.extend([
-                    NetworkBehaviourAction::NotifyHandler {
-                        peer_id,
-                        handler: NotifyHandler::One(relayed_connection_id),
-                        event: Either::Left(
-                            handler::relayed::Command::UpgradeFinishedDontKeepAlive,
-                        ),
-                    },
-                    NetworkBehaviourAction::GenerateEvent(Event::DirectConnectionUpgradeFailed {
-                        remote_peer_id: peer_id,
-                        error: Error::Dial,
+            *relayed_connection_id
+        } else {
+            return;
+        };
+
+        let attempt = if let Some(attempt) = self
+            .outgoing_direct_connection_attempts
+            .get(&(relayed_connection_id, peer_id))
+        {
+            *attempt
+        } else {
+            return;
+        };
+
+        if attempt < MAX_NUMBER_OF_UPGRADE_ATTEMPTS {
+            self.queued_events
+                .push_back(NetworkBehaviourAction::NotifyHandler {
+                    handler: NotifyHandler::One(relayed_connection_id),
+                    peer_id,
+                    event: Either::Left(handler::relayed::Command::Connect {
+                        attempt: attempt + 1,
+                        obs_addrs: self.observed_addreses(),
                     }),
-                ]);
-            }
+                })
+        } else {
+            self.queued_events.extend([
+                NetworkBehaviourAction::NotifyHandler {
+                    peer_id,
+                    handler: NotifyHandler::One(relayed_connection_id),
+                    event: Either::Left(handler::relayed::Command::UpgradeFinishedDontKeepAlive),
+                },
+                NetworkBehaviourAction::GenerateEvent(Event::DirectConnectionUpgradeFailed {
+                    remote_peer_id: peer_id,
+                    error: Error::Dial,
+                }),
+            ]);
         }
     }
 
@@ -214,24 +238,34 @@ impl NetworkBehaviour for Behaviour {
     type OutEvent = Event;
 
     fn new_handler(&mut self) -> Self::ConnectionHandler {
-        handler::Prototype::UnknownConnection
+        handler::Prototype
     }
 
     fn on_connection_handler_event(
         &mut self,
         event_source: PeerId,
-        connection: ConnectionId,
-        handler_event: <<Self::ConnectionHandler as IntoConnectionHandler>::Handler as
-            ConnectionHandler>::OutEvent,
+        connection_id: ConnectionId,
+        handler_event: <<Self::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent,
     ) {
+        let relayed_connection_id = match handler_event.as_ref() {
+            Either::Left(_) => connection_id,
+            Either::Right(_) => match self.direct_to_relayed_connections.get(&connection_id) {
+                None => {
+                    // If the connection ID is unknown to us, it means we didn't create it so ignore any event coming from it.
+                    return;
+                }
+                Some(relayed_connection_id) => *relayed_connection_id,
+            },
+        };
+
         match handler_event {
             Either::Left(handler::relayed::Event::InboundConnectRequest {
                 inbound_connect,
                 remote_addr,
             }) => {
                 self.queued_events.extend([
                     NetworkBehaviourAction::NotifyHandler {
-                        handler: NotifyHandler::One(connection),
+                        handler: NotifyHandler::One(relayed_connection_id),
                         peer_id: event_source,
                         event: Either::Left(handler::relayed::Command::AcceptInboundConnect {
                             inbound_connect,
@@ -256,15 +290,17 @@ impl NetworkBehaviour for Behaviour {
                     ));
             }
             Either::Left(handler::relayed::Event::InboundConnectNegotiated(remote_addrs)) => {
+                let maybe_direct_connection_id = ConnectionId::next();
+
+                self.direct_to_relayed_connections
+                    .insert(maybe_direct_connection_id, relayed_connection_id);
+
                 self.queued_events.push_back(NetworkBehaviourAction::Dial {
                     opts: DialOpts::peer_id(event_source)
                         .addresses(remote_addrs)
                         .condition(dial_opts::PeerCondition::Always)
                         .build(),
-                    handler: handler::Prototype::DirectConnection {
-                        relayed_connection_id: connection,
-                        role: handler::Role::Listener,
-                    },
+                    connection_id: maybe_direct_connection_id,
                 });
             }
             Either::Left(handler::relayed::Event::OutboundNegotiationFailed { error }) => {
@@ -276,27 +312,26 @@ impl NetworkBehaviour for Behaviour {
                         },
                     ));
             }
-            Either::Left(handler::relayed::Event::OutboundConnectNegotiated {
-                remote_addrs,
-                attempt,
-            }) => {
+            Either::Left(handler::relayed::Event::OutboundConnectNegotiated { remote_addrs }) => {
+                let maybe_direct_connection_id = ConnectionId::next();
+
+                self.direct_to_relayed_connections
+                    .insert(maybe_direct_connection_id, relayed_connection_id);
+                *self
+                    .outgoing_direct_connection_attempts
+                    .entry((maybe_direct_connection_id, event_source))
+                    .or_default() += 1;
+
                 self.queued_events.push_back(NetworkBehaviourAction::Dial {
                     opts: DialOpts::peer_id(event_source)
                         .condition(dial_opts::PeerCondition::Always)
                         .addresses(remote_addrs)
                         .override_role()
                         .build(),
-                    handler: handler::Prototype::DirectConnection {
-                        relayed_connection_id: connection,
-                        role: handler::Role::Initiator { attempt },
-                    },
+                    connection_id: maybe_direct_connection_id,
                 });
             }
-            Either::Right(Either::Left(
-                handler::direct::Event::DirectConnectionUpgradeSucceeded {
-                    relayed_connection_id,
-                },
-            )) => {
+            Either::Right(handler::direct::Event::DirectConnectionEstablished) => {
                 self.queued_events.extend([
                     NetworkBehaviourAction::NotifyHandler {
                         peer_id: event_source,
@@ -312,15 +347,14 @@ impl NetworkBehaviour for Behaviour {
                     ),
                 ]);
             }
-            Either::Right(Either::Right(event)) => void::unreachable(event),
         };
     }
 
     fn poll(
         &mut self,
         _cx: &mut Context<'_>,
         _: &mut impl PollParameters,
-    ) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
+    ) -> Poll<NetworkBehaviourAction<Self::OutEvent, THandlerInEvent<Self>>> {
         if let Some(event) = self.queued_events.pop_front() {
             return Poll::Ready(event);
         }