feat(request-response): configurable use of relayed connections

core/CHANGELOG.md

@@ -1,5 +1,9 @@
 ## 0.44.0
 
+- Add `MultiaddrExt::is_relayed()` for checking whether a `Multiaddr` is relayed
+  (contains a `p2p-circuit`).
+  See [PR 6501](https://github.com/libp2p/rust-libp2p/pull/6501).
+
 - Raise MSRV to 1.88.0.
   See [PR 6273](https://github.com/libp2p/rust-libp2p/pull/6273).
 

core/src/connection.rs

@@ -18,10 +18,7 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use crate::{
-    multiaddr::{Multiaddr, Protocol},
-    transport::PortUse,
-};
+use crate::{multiaddr::Multiaddr, multiaddr_ext::MultiaddrExt, transport::PortUse};
 
 /// The endpoint roles associated with a peer-to-peer communication channel.
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
@@ -135,8 +132,7 @@ impl ConnectedPoint {
             ConnectedPoint::Dialer { address, .. } => address,
             ConnectedPoint::Listener { local_addr, .. } => local_addr,
         }
-        .iter()
-        .any(|p| p == Protocol::P2pCircuit)
+        .is_relayed()
     }
 
     /// Returns the address of the remote stored in this struct.

core/src/lib.rs

@@ -46,6 +46,7 @@ pub type Negotiated<T> = multistream_select::Negotiated<T>;
 
 pub mod connection;
 pub mod either;
+pub mod multiaddr_ext;
 pub mod muxing;
 pub mod peer_record;
 pub mod signed_envelope;
@@ -55,6 +56,7 @@ pub mod upgrade;
 pub use connection::{ConnectedPoint, Endpoint};
 pub use libp2p_identity::PeerId;
 pub use multiaddr::Multiaddr;
+pub use multiaddr_ext::MultiaddrExt;
 pub use multihash;
 pub use muxing::StreamMuxer;
 pub use peer_record::PeerRecord;

core/src/multiaddr_ext.rs

@@ -0,0 +1,31 @@
+use crate::multiaddr::{Multiaddr, Protocol};
+
+/// Extension trait providing convenience helpers for [`Multiaddr`].
+pub trait MultiaddrExt {
+    /// Returns `true` if the address is relayed, i.e. it contains a
+    /// [`p2p-circuit`](Protocol::P2pCircuit) component.
+    fn is_relayed(&self) -> bool;
+}
+
+impl MultiaddrExt for Multiaddr {
+    fn is_relayed(&self) -> bool {
+        self.iter().any(|p| p == Protocol::P2pCircuit)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn direct_address_is_not_relayed() {
+        let addr: Multiaddr = "/ip4/127.0.0.1/tcp/1234".parse().unwrap();
+        assert!(!addr.is_relayed());
+    }
+
+    #[test]
+    fn circuit_address_is_relayed() {
+        let addr: Multiaddr = "/ip4/127.0.0.1/tcp/1234/p2p/12D3KooWDpJ7As7BWAwRMfu1VU2WCqNjvq387JEYKDBj4kx6nXTN/p2p-circuit/p2p/12D3KooWGzBRs8WVcMaEXGWwsvFSfJKYphwM6ojrPanuJfup1xFz".parse().unwrap();
+        assert!(addr.is_relayed());
+    }
+}

protocols/dcutr/src/behaviour.rs

@@ -29,7 +29,8 @@ use std::{
 use either::Either;
 use hashlink::LruCache;
 use libp2p_core::{
-    Endpoint, Multiaddr, connection::ConnectedPoint, multiaddr::Protocol, transport::PortUse,
+    Endpoint, Multiaddr, MultiaddrExt, connection::ConnectedPoint, multiaddr::Protocol,
+    transport::PortUse,
 };
 use libp2p_identity::PeerId;
 use libp2p_swarm::{
@@ -177,7 +178,7 @@ impl NetworkBehaviour for Behaviour {
         local_addr: &Multiaddr,
         remote_addr: &Multiaddr,
     ) -> Result<THandler<Self>, ConnectionDenied> {
-        if is_relayed(local_addr) {
+        if local_addr.is_relayed() {
             let connected_point = ConnectedPoint::Listener {
                 local_addr: local_addr.clone(),
                 send_back_addr: remote_addr.clone(),
@@ -212,7 +213,7 @@ impl NetworkBehaviour for Behaviour {
         role_override: Endpoint,
         port_use: PortUse,
     ) -> Result<THandler<Self>, ConnectionDenied> {
-        if is_relayed(addr) {
+        if addr.is_relayed() {
             return Ok(Either::Left(handler::relayed::Handler::new(
                 ConnectedPoint::Dialer {
                     address: addr.clone(),
@@ -367,7 +368,7 @@ impl Candidates {
     }
 
     fn add(&mut self, mut address: Multiaddr) {
-        if is_relayed(&address) {
+        if address.is_relayed() {
             return;
         }
 
@@ -382,7 +383,3 @@ impl Candidates {
         self.inner.iter().map(|(a, _)| a)
     }
 }
-
-fn is_relayed(addr: &Multiaddr) -> bool {
-    addr.iter().any(|p| p == Protocol::P2pCircuit)
-}

protocols/dcutr/src/protocol/inbound.rs

@@ -22,7 +22,7 @@ use std::io;
 
 use asynchronous_codec::Framed;
 use futures::prelude::*;
-use libp2p_core::{Multiaddr, multiaddr::Protocol};
+use libp2p_core::{Multiaddr, MultiaddrExt};
 use libp2p_swarm::Stream;
 use thiserror::Error;
 
@@ -57,7 +57,7 @@ pub(crate) async fn handshake(
         })
         // Filter out relayed addresses.
         .filter(|a| {
-            if a.iter().any(|p| p == Protocol::P2pCircuit) {
+            if a.is_relayed() {
                 tracing::debug!(address=%a, "Dropping relayed address");
                 false
             } else {

protocols/dcutr/src/protocol/outbound.rs

@@ -23,7 +23,7 @@ use std::io;
 use asynchronous_codec::Framed;
 use futures::prelude::*;
 use futures_timer::Delay;
-use libp2p_core::{Multiaddr, multiaddr::Protocol};
+use libp2p_core::{Multiaddr, MultiaddrExt};
 use libp2p_swarm::Stream;
 use thiserror::Error;
 use web_time::Instant;
@@ -74,7 +74,7 @@ pub(crate) async fn handshake(
         })
         // Filter out relayed addresses.
         .filter(|a| {
-            if a.iter().any(|p| p == Protocol::P2pCircuit) {
+            if a.is_relayed() {
                 tracing::debug!(address=%a, "Dropping relayed address");
                 false
             } else {

protocols/relay/src/behaviour.rs

@@ -31,7 +31,9 @@ use std::{
 };
 
 use either::Either;
-use libp2p_core::{ConnectedPoint, Endpoint, Multiaddr, multiaddr::Protocol, transport::PortUse};
+use libp2p_core::{
+    ConnectedPoint, Endpoint, Multiaddr, MultiaddrExt, multiaddr::Protocol, transport::PortUse,
+};
 use libp2p_identity::PeerId;
 use libp2p_swarm::{
     ConnectionClosed, ConnectionDenied, ConnectionId, ExternalAddresses, FromSwarm,
@@ -42,7 +44,6 @@ use web_time::Instant;
 
 use crate::{
     behaviour::handler::Handler,
-    multiaddr_ext::MultiaddrExt,
     proto,
     protocol::{inbound_hop, outbound_stop},
 };

protocols/relay/src/lib.rs

@@ -25,7 +25,6 @@
 
 mod behaviour;
 mod copy_future;
-mod multiaddr_ext;
 mod priv_client;
 mod protocol;
 

protocols/relay/src/priv_client.rs

@@ -41,7 +41,7 @@ use futures::{
     stream::StreamExt,
 };
 use libp2p_core::{
-    Endpoint, Multiaddr,
+    Endpoint, Multiaddr, MultiaddrExt,
     multiaddr::Protocol,
     transport::{ListenerId, PortUse},
 };
@@ -56,7 +56,6 @@ use libp2p_swarm::{
 use transport::Transport;
 
 use crate::{
-    multiaddr_ext::MultiaddrExt,
     priv_client::handler::Handler,
     protocol::{self, inbound_stop},
 };

protocols/relay/src/priv_client/transport.rs

@@ -32,6 +32,7 @@ use futures::{
     stream::{SelectAll, Stream, StreamExt},
 };
 use libp2p_core::{
+    MultiaddrExt,
     multiaddr::{Multiaddr, Protocol},
     transport::{DialOpts, ListenerId, TransportError, TransportEvent},
 };
@@ -40,7 +41,6 @@ use thiserror::Error;
 
 use crate::{
     RequestId,
-    multiaddr_ext::MultiaddrExt,
     priv_client::Connection,
     protocol::{
         outbound_hop,

protocols/request-response/CHANGELOG.md

@@ -1,5 +1,12 @@
 ## 0.30.0
 
+- Add `Config::with_relay_for_requests` to control whether outbound requests may be sent over
+  relayed connections (enabled by default). When disabled and only a relayed connection to the peer
+  is available, the request is kept queued so a direct connection established shortly after (e.g. via
+  DCUtR) can carry it, and fails with the new `OutboundFailure::NoDirectConnection` variant if the
+  peer fully disconnects first.
+  See [PR 6501](https://github.com/libp2p/rust-libp2p/pull/6501).
+
 - Use `futures-timer` instead of tokio's timer for stream timeouts so the bounded `Delay` works on
   `wasm32`; tokio's timer has no driver in the browser and panics at runtime.
   See [PR 6488](https://github.com/libp2p/rust-libp2p/pull/6488).

protocols/request-response/Cargo.toml

@@ -31,7 +31,13 @@ cbor = ["dep:serde", "dep:cbor4ii", "libp2p-swarm/macros"]
 anyhow = "1.0.102"
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
 rand = { workspace = true }
+libp2p-identify = { workspace = true }
+libp2p-plaintext = { workspace = true }
+libp2p-relay = { workspace = true }
+libp2p-swarm = { workspace = true, features = ["macros"] }
 libp2p-swarm-test = { path = "../../swarm-test" }
+libp2p-tcp = { workspace = true, features = ["tokio"] }
+libp2p-yamux = { workspace = true }
 futures_ringbuf = "0.4.0"
 serde = { version = "1.0", features = ["derive"] }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }