refactor(*): Transport redesign

Cargo.toml

@@ -76,45 +76,44 @@ futures-bounded = { version = "0.2.3" }
 futures-rustls = { version = "0.26.0", default-features = false }
 libp2p = { version = "0.54.0", path = "libp2p" }
 libp2p-allow-block-list = { version = "0.3.0", path = "misc/allow-block-list" }
-libp2p-autonat = { version = "0.12.0", path = "protocols/autonat" }
+libp2p-autonat = { version = "0.13.0", path = "protocols/autonat" }
 libp2p-connection-limits = { version = "0.3.1", path = "misc/connection-limits" }
-libp2p-core = { version = "0.41.2", path = "core" }
-libp2p-dcutr = { version = "0.11.0", path = "protocols/dcutr" }
-libp2p-dns = { version = "0.41.1", path = "transports/dns" }
-libp2p-floodsub = { version = "0.44.0", path = "protocols/floodsub" }
-libp2p-gossipsub = { version = "0.46.1", path = "protocols/gossipsub" }
-libp2p-identify = { version = "0.44.2", path = "protocols/identify" }
+libp2p-core = { version = "0.42.0", path = "core" }
+libp2p-dcutr = { version = "0.12.0", path = "protocols/dcutr" }
+libp2p-dns = { version = "0.42.0", path = "transports/dns" }
+libp2p-floodsub = { version = "0.45.0", path = "protocols/floodsub" }
+libp2p-gossipsub = { version = "0.47.0", path = "protocols/gossipsub" }
+libp2p-identify = { version = "0.45.0", path = "protocols/identify" }
 libp2p-identity = { version = "0.2.8" }
 libp2p-kad = { version = "0.46.0", path = "protocols/kad" }
-libp2p-mdns = { version = "0.45.1", path = "protocols/mdns" }
+libp2p-mdns = { version = "0.46.0", path = "protocols/mdns" }
 libp2p-memory-connection-limits = { version = "0.2.0", path = "misc/memory-connection-limits" }
 libp2p-metrics = { version = "0.14.1", path = "misc/metrics" }
 libp2p-mplex = { version = "0.41.0", path = "muxers/mplex" }
-libp2p-muxer-test-harness = { path = "muxers/test-harness" }
 libp2p-noise = { version = "0.44.0", path = "transports/noise" }
-libp2p-perf = { version = "0.3.0", path = "protocols/perf" }
-libp2p-ping = { version = "0.44.1", path = "protocols/ping" }
+libp2p-perf = { version = "0.4.0", path = "protocols/perf" }
+libp2p-ping = { version = "0.45.0", path = "protocols/ping" }
 libp2p-plaintext = { version = "0.41.0", path = "transports/plaintext" }
 libp2p-pnet = { version = "0.24.0", path = "transports/pnet" }
-libp2p-quic = { version = "0.10.3", path = "transports/quic" }
-libp2p-relay = { version = "0.17.2", path = "protocols/relay" }
-libp2p-rendezvous = { version = "0.14.0", path = "protocols/rendezvous" }
-libp2p-request-response = { version = "0.26.3", path = "protocols/request-response" }
+libp2p-quic = { version = "0.11.0", path = "transports/quic" }
+libp2p-relay = { version = "0.18.0", path = "protocols/relay" }
+libp2p-rendezvous = { version = "0.15.0", path = "protocols/rendezvous" }
+libp2p-request-response = { version = "0.27.0", path = "protocols/request-response" }
 libp2p-server = { version = "0.12.7", path = "misc/server" }
-libp2p-stream = { version = "0.1.0-alpha.1", path = "protocols/stream" }
-libp2p-swarm = { version = "0.44.2", path = "swarm" }
+libp2p-stream = { version = "0.2.0-alpha", path = "protocols/stream" }
+libp2p-swarm = { version = "0.45.0", path = "swarm" }
 libp2p-swarm-derive = { version = "=0.34.2", path = "swarm-derive" } # `libp2p-swarm-derive` may not be compatible with different `libp2p-swarm` non-breaking releases. E.g. `libp2p-swarm` might introduce a new enum variant `FromSwarm` (which is `#[non-exhaustive]`) in a non-breaking release. Older versions of `libp2p-swarm-derive` would not forward this enum variant within the `NetworkBehaviour` hierarchy. Thus the version pinning is required.
 libp2p-swarm-test = { version = "0.3.0", path = "swarm-test" }
-libp2p-tcp = { version = "0.41.1", path = "transports/tcp" }
+libp2p-tcp = { version = "0.42.0", path = "transports/tcp" }
 libp2p-tls = { version = "0.4.0", path = "transports/tls" }
-libp2p-uds = { version = "0.40.0", path = "transports/uds" }
-libp2p-upnp = { version = "0.2.2", path = "protocols/upnp" }
-libp2p-webrtc = { version = "0.7.1-alpha", path = "transports/webrtc" }
+libp2p-uds = { version = "0.41.0", path = "transports/uds" }
+libp2p-upnp = { version = "0.3.0", path = "protocols/upnp" }
+libp2p-webrtc = { version = "0.8.0-alpha", path = "transports/webrtc" }
 libp2p-webrtc-utils = { version = "0.2.0", path = "misc/webrtc-utils" }
-libp2p-webrtc-websys = { version = "0.3.0-alpha", path = "transports/webrtc-websys" }
-libp2p-websocket = { version = "0.43.0", path = "transports/websocket" }
-libp2p-websocket-websys = { version = "0.3.2", path = "transports/websocket-websys" }
-libp2p-webtransport-websys = { version = "0.3.0", path = "transports/webtransport-websys" }
+libp2p-webrtc-websys = { version = "0.4.0-alpha", path = "transports/webrtc-websys" }
+libp2p-websocket = { version = "0.44.0", path = "transports/websocket" }
+libp2p-websocket-websys = { version = "0.4.0", path = "transports/websocket-websys" }
+libp2p-webtransport-websys = { version = "0.4.0", path = "transports/webtransport-websys" }
 libp2p-yamux = { version = "0.45.1", path = "muxers/yamux" }
 multiaddr = "0.18.1"
 multihash = "0.19.1"

core/CHANGELOG.md

@@ -1,3 +1,29 @@
+## 0.42.0 -- unreleased
+
+- Refactor `Transport` to allow the behaviour to tell the implementor of `Transport` to reuse an
+existing port or a new one.
+  See [PR 4568](https://https://github.com/libp2p/rust-libp2p/pull/4568)
+    - Add new enum `PortUse`, defaulting on `PortUse::Reuse`.
+    - Add new struct `DialOpts`, containing `PortUse` and `Endpoint`.
+    - Add field `port_use` to `ConnectedPoint`.
+    - Add new parameter to `Transport::dial`: `DialOpts`.
+    - Remove `Transport::dial_as_listener` and move into `Transport::dial`. To dial as listener, one
+      has to set the `endpoint` field in `DialOpts` to `Endpoint::Listener`.
+    - Adjust utility transports:
+        - `AndThen`
+        - `Boxed`
+        - `Choice`
+        - `Dummy`
+        - `GlobalOnly`
+        - `Map`
+        - `MapErr`
+        - `Memory`
+        - `Optional`
+        - `Timeout`
+        - `Upgrade`
+- Remove `Transport::address_translation` and move this functionality into `libp2p_swarm`.
+  See [umgefahren/libp2p PR 4](https://github.com/umgefahren/rust-libp2p/pull/4)
+
 ## 0.41.2
 
 - Implement `std::fmt::Display` on `ListenerId`.

core/Cargo.toml

@@ -3,7 +3,7 @@ name = "libp2p-core"
 edition = "2021"
 rust-version = { workspace = true }
 description = "Core traits and structs of libp2p"
-version = "0.41.2"
+version = "0.42.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

core/src/connection.rs

@@ -18,7 +18,10 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use crate::multiaddr::{Multiaddr, Protocol};
+use crate::{
+    multiaddr::{Multiaddr, Protocol},
+    transport::PortUse,
+};
 
 /// The endpoint roles associated with a peer-to-peer communication channel.
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
@@ -80,6 +83,13 @@ pub enum ConnectedPoint {
         ///   connection as a dialer and one peer dial the other and upgrade the
         ///   connection _as a listener_ overriding its role.
         role_override: Endpoint,
+        /// Whether the port for the outgoing connection was reused from a listener
+        /// or a new port was allocated. This is useful for address translation.
+        ///
+        /// The port use is implemented on a best-effort basis. It is not guaranteed
+        /// that [`PortUse::Reuse`] actually reused a port. A good example is the case
+        /// where there is no listener available to reuse a port from.
+        port_use: PortUse,
     },
     /// We received the node.
     Listener {
@@ -133,6 +143,7 @@ impl ConnectedPoint {
             ConnectedPoint::Dialer {
                 address,
                 role_override: _,
+                port_use: _,
             } => address,
             ConnectedPoint::Listener { local_addr, .. } => local_addr,
         }

core/src/either.rs

@@ -19,6 +19,7 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::muxing::StreamMuxerEvent;
+use crate::transport::DialOpts;
 use crate::{
     muxing::StreamMuxer,
     transport::{ListenerId, Transport, TransportError, TransportEvent},
@@ -172,48 +173,23 @@ where
         }
     }
 
-    fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
-        use TransportError::*;
-        match self {
-            Either::Left(a) => match a.dial(addr) {
-                Ok(connec) => Ok(EitherFuture::First(connec)),
-                Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
-                Err(Other(err)) => Err(Other(Either::Left(err))),
-            },
-            Either::Right(b) => match b.dial(addr) {
-                Ok(connec) => Ok(EitherFuture::Second(connec)),
-                Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
-                Err(Other(err)) => Err(Other(Either::Right(err))),
-            },
-        }
-    }
-
-    fn dial_as_listener(
+    fn dial(
         &mut self,
         addr: Multiaddr,
-    ) -> Result<Self::Dial, TransportError<Self::Error>>
-    where
-        Self: Sized,
-    {
+        opts: DialOpts,
+    ) -> Result<Self::Dial, TransportError<Self::Error>> {
         use TransportError::*;
         match self {
-            Either::Left(a) => match a.dial_as_listener(addr) {
+            Either::Left(a) => match a.dial(addr, opts) {
                 Ok(connec) => Ok(EitherFuture::First(connec)),
                 Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
                 Err(Other(err)) => Err(Other(Either::Left(err))),
             },
-            Either::Right(b) => match b.dial_as_listener(addr) {
+            Either::Right(b) => match b.dial(addr, opts) {
                 Ok(connec) => Ok(EitherFuture::Second(connec)),
                 Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
                 Err(Other(err)) => Err(Other(Either::Right(err))),
             },
         }
     }
-
-    fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
-        match self {
-            Either::Left(a) => a.address_translation(server, observed),
-            Either::Right(b) => b.address_translation(server, observed),
-        }
-    }
 }

core/src/lib.rs

@@ -45,8 +45,6 @@ mod proto {
 pub use multiaddr;
 pub type Negotiated<T> = multistream_select::Negotiated<T>;
 
-mod translation;
-
 pub mod connection;
 pub mod either;
 pub mod muxing;
@@ -61,7 +59,6 @@ pub use multihash;
 pub use muxing::StreamMuxer;
 pub use peer_record::PeerRecord;
 pub use signed_envelope::SignedEnvelope;
-pub use translation::address_translation;
 pub use transport::Transport;
 pub use upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo};
 

core/src/transport.rs

@@ -48,7 +48,7 @@ pub mod upgrade;
 mod boxed;
 mod optional;
 
-use crate::ConnectedPoint;
+use crate::{ConnectedPoint, Endpoint};
 
 pub use self::boxed::Boxed;
 pub use self::choice::OrTransport;
@@ -58,6 +58,30 @@ pub use self::upgrade::Upgrade;
 
 static NEXT_LISTENER_ID: AtomicUsize = AtomicUsize::new(1);
 
+/// The port use policy for a new connection.
+#[derive(Debug, Copy, Clone, Default, PartialEq, Eq, Hash)]
+pub enum PortUse {
+    /// Always allocate a new port for the dial.
+    New,
+    /// Best effor reusing of an existing port.
+    ///
+    /// If there is no listener present that can be used to dial, a new port is allocated.
+    #[default]
+    Reuse,
+}
+
+/// Options to customize the behaviour during dialing.
+#[derive(Debug, Copy, Clone)]
+pub struct DialOpts {
+    /// The endpoint establishing a new connection.
+    ///
+    /// When attempting a hole-punch, both parties simultaneously "dial" each other but one party has to be the "listener" on the final connection.
+    /// This option specifies the role of this node in the final connection.
+    pub role: Endpoint,
+    /// The port use policy for a new connection.
+    pub port_use: PortUse,
+}
+
 /// A transport provides connection-oriented communication between two peers
 /// through ordered streams of data (i.e. connections).
 ///
@@ -129,16 +153,10 @@ pub trait Transport {
     ///
     /// If [`TransportError::MultiaddrNotSupported`] is returned, it may be desirable to
     /// try an alternative [`Transport`], if available.
-    fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>>;
-
-    /// As [`Transport::dial`] but has the local node act as a listener on the outgoing connection.
-    ///
-    /// This option is needed for NAT and firewall hole punching.
-    ///
-    /// See [`ConnectedPoint::Dialer`] for related option.
-    fn dial_as_listener(
+    fn dial(
         &mut self,
         addr: Multiaddr,
+        opts: DialOpts,
     ) -> Result<Self::Dial, TransportError<Self::Error>>;
 
     /// Poll for [`TransportEvent`]s.
@@ -157,24 +175,6 @@ pub trait Transport {
         cx: &mut Context<'_>,
     ) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>>;
 
-    /// Performs a transport-specific mapping of an address `observed` by a remote onto a
-    /// local `listen` address to yield an address for the local node that may be reachable
-    /// for other peers.
-    ///
-    /// This is relevant for transports where Network Address Translation (NAT) can occur
-    /// so that e.g. the peer is observed at a different IP than the IP of the local
-    /// listening address. See also [`address_translation`][crate::address_translation].
-    ///
-    /// Within [`libp2p::Swarm`](<https://docs.rs/libp2p/latest/libp2p/struct.Swarm.html>) this is
-    /// used when extending the listening addresses of the local peer with external addresses
-    /// observed by remote peers.
-    /// On transports where this is not relevant (i.e. no NATs are present) `None` should be
-    /// returned for the sake of de-duplication.
-    ///
-    /// Note: if the listen or observed address is not a valid address of this transport,
-    /// `None` should be returned as well.
-    fn address_translation(&self, listen: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr>;
-
     /// Boxes the transport, including custom transport errors.
     fn boxed(self) -> boxed::Boxed<Self::Output>
     where

core/src/transport/and_then.rs

@@ -19,8 +19,8 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::{
-    connection::{ConnectedPoint, Endpoint},
-    transport::{ListenerId, Transport, TransportError, TransportEvent},
+    connection::ConnectedPoint,
+    transport::{DialOpts, ListenerId, Transport, TransportError, TransportEvent},
 };
 use either::Either;
 use futures::prelude::*;
@@ -68,51 +68,30 @@ where
         self.transport.remove_listener(id)
     }
 
-    fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
-        let dialed_fut = self
-            .transport
-            .dial(addr.clone())
-            .map_err(|err| err.map(Either::Left))?;
-        let future = AndThenFuture {
-            inner: Either::Left(Box::pin(dialed_fut)),
-            args: Some((
-                self.fun.clone(),
-                ConnectedPoint::Dialer {
-                    address: addr,
-                    role_override: Endpoint::Dialer,
-                },
-            )),
-            _marker: PhantomPinned,
-        };
-        Ok(future)
-    }
-
-    fn dial_as_listener(
+    fn dial(
         &mut self,
         addr: Multiaddr,
+        opts: DialOpts,
     ) -> Result<Self::Dial, TransportError<Self::Error>> {
         let dialed_fut = self
             .transport
-            .dial_as_listener(addr.clone())
+            .dial(addr.clone(), opts)
             .map_err(|err| err.map(Either::Left))?;
         let future = AndThenFuture {
             inner: Either::Left(Box::pin(dialed_fut)),
             args: Some((
                 self.fun.clone(),
                 ConnectedPoint::Dialer {
                     address: addr,
-                    role_override: Endpoint::Listener,
+                    role_override: opts.role,
+                    port_use: opts.port_use,
                 },
             )),
             _marker: PhantomPinned,
         };
         Ok(future)
     }
 
-    fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
-        self.transport.address_translation(server, observed)
-    }
-
     fn poll(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,