diff --git a/Cargo.lock b/Cargo.lock index 05798f65f62..080ea1bc61f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3461,9 +3461,14 @@ dependencies = [ "futures-bounded", "futures_ringbuf", "libp2p-core", + "libp2p-identify", "libp2p-identity", + "libp2p-plaintext", + "libp2p-relay", "libp2p-swarm", "libp2p-swarm-test", + "libp2p-tcp", + "libp2p-yamux", "rand 0.8.6", "serde", "serde_json", diff --git a/core/CHANGELOG.md b/core/CHANGELOG.md index be980183c3c..afdae54666f 100644 --- a/core/CHANGELOG.md +++ b/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). diff --git a/core/src/connection.rs b/core/src/connection.rs index 8779f76c03c..fec86de7334 100644 --- a/core/src/connection.rs +++ b/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. diff --git a/core/src/lib.rs b/core/src/lib.rs index 61c4bd1303c..c49d38578fc 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -46,6 +46,7 @@ pub type Negotiated = multistream_select::Negotiated; 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; diff --git a/core/src/multiaddr_ext.rs b/core/src/multiaddr_ext.rs new file mode 100644 index 00000000000..1076912181c --- /dev/null +++ b/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()); + } +} diff --git a/protocols/dcutr/src/behaviour.rs b/protocols/dcutr/src/behaviour.rs index 355cec68571..13138e80f67 100644 --- a/protocols/dcutr/src/behaviour.rs +++ b/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, 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, 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) -} diff --git a/protocols/dcutr/src/protocol/inbound.rs b/protocols/dcutr/src/protocol/inbound.rs index e81782aac40..db2e16bb6c9 100644 --- a/protocols/dcutr/src/protocol/inbound.rs +++ b/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 { diff --git a/protocols/dcutr/src/protocol/outbound.rs b/protocols/dcutr/src/protocol/outbound.rs index d184b011d82..be6440bd985 100644 --- a/protocols/dcutr/src/protocol/outbound.rs +++ b/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 { diff --git a/protocols/relay/src/behaviour.rs b/protocols/relay/src/behaviour.rs index dc129dcaa1a..c8fdf345c21 100644 --- a/protocols/relay/src/behaviour.rs +++ b/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}, }; diff --git a/protocols/relay/src/lib.rs b/protocols/relay/src/lib.rs index 1c32cc8d8d0..8499f5547db 100644 --- a/protocols/relay/src/lib.rs +++ b/protocols/relay/src/lib.rs @@ -25,7 +25,6 @@ mod behaviour; mod copy_future; -mod multiaddr_ext; mod priv_client; mod protocol; diff --git a/protocols/relay/src/multiaddr_ext.rs b/protocols/relay/src/multiaddr_ext.rs deleted file mode 100644 index f9a1f71b4dc..00000000000 --- a/protocols/relay/src/multiaddr_ext.rs +++ /dev/null @@ -1,11 +0,0 @@ -use libp2p_core::{Multiaddr, multiaddr::Protocol}; - -pub(crate) trait MultiaddrExt { - fn is_relayed(&self) -> bool; -} - -impl MultiaddrExt for Multiaddr { - fn is_relayed(&self) -> bool { - self.iter().any(|p| p == Protocol::P2pCircuit) - } -} diff --git a/protocols/relay/src/priv_client.rs b/protocols/relay/src/priv_client.rs index bd3d2c08eee..b8676e545a0 100644 --- a/protocols/relay/src/priv_client.rs +++ b/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}, }; diff --git a/protocols/relay/src/priv_client/transport.rs b/protocols/relay/src/priv_client/transport.rs index 4a09f9afcf2..6907dd239cb 100644 --- a/protocols/relay/src/priv_client/transport.rs +++ b/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, diff --git a/protocols/request-response/CHANGELOG.md b/protocols/request-response/CHANGELOG.md index 4d5f00d1b5a..222b7cca15d 100644 --- a/protocols/request-response/CHANGELOG.md +++ b/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). diff --git a/protocols/request-response/Cargo.toml b/protocols/request-response/Cargo.toml index 5f6f9150e79..ca991ba3d71 100644 --- a/protocols/request-response/Cargo.toml +++ b/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"] } diff --git a/protocols/request-response/src/lib.rs b/protocols/request-response/src/lib.rs index 86ad6f3472e..870f86da745 100644 --- a/protocols/request-response/src/lib.rs +++ b/protocols/request-response/src/lib.rs @@ -63,6 +63,31 @@ //! family can be configured in this way. Such protocols will not be //! advertised during inbound respectively outbound protocol negotiation //! on the substreams. +//! +//! ## Relayed connections +//! +//! By default, outbound requests are sent over any available connection to the peer, +//! including relayed ([`p2p-circuit`](libp2p_core::multiaddr::Protocol::P2pCircuit)) +//! connections. Set [`Config::with_relay_for_requests`] to `false` to restrict requests to +//! direct connections and conserve relay bandwidth. +//! +//! When relayed connections are disabled and the only connection to a peer is relayed, the request +//! is *not* sent over the relay. Instead it stays queued so that a direct connection established +//! shortly after — most notably via DCUtR hole-punching, which this option is meant to be combined +//! with — can carry it. If the peer fully disconnects before a direct connection becomes available, +//! the request fails with [`OutboundFailure::NoDirectConnection`]. This bounds how long a request +//! can stay queued: a relayed connection that nothing keeps alive is closed by the swarm after its +//! `idle_connection_timeout`, and DCUtR keeps it alive only while it is still attempting to +//! upgrade. +//! +//! Two limitations apply: +//! +//! - The relay status of *inbound* connections is not tracked: their address is not retained by the +//! [`Behaviour`], so they are always treated as direct. The policy therefore only governs +//! connections this node dialed. +//! - The queued-request bound is the connection's `idle_connection_timeout`, not the relay +//! reservation. If a relayed connection is kept alive indefinitely by other means while no direct +//! connection is established, requests to that peer remain queued until it closes. #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))] @@ -85,7 +110,7 @@ pub use codec::Codec; use futures::channel::oneshot; use handler::Handler; pub use handler::ProtocolSupport; -use libp2p_core::{ConnectedPoint, Endpoint, Multiaddr, transport::PortUse}; +use libp2p_core::{ConnectedPoint, Endpoint, Multiaddr, MultiaddrExt, transport::PortUse}; use libp2p_identity::PeerId; use libp2p_swarm::{ ConnectionDenied, ConnectionHandler, ConnectionId, DialError, NetworkBehaviour, NotifyHandler, @@ -192,6 +217,9 @@ pub enum OutboundFailure { UnsupportedProtocols, /// An IO failure happened on an outbound stream. Io(io::Error), + /// The request could not be sent because the only connection(s) to the peer are + /// relayed and [`Config::with_relay_for_requests`] is disabled. + NoDirectConnection, } impl fmt::Display for OutboundFailure { @@ -206,6 +234,12 @@ impl fmt::Display for OutboundFailure { write!(f, "The remote supports none of the requested protocols") } OutboundFailure::Io(e) => write!(f, "IO error on outbound stream: {e}"), + OutboundFailure::NoDirectConnection => { + write!( + f, + "No direct connection available and relayed connections are disabled" + ) + } } } } @@ -310,6 +344,10 @@ impl fmt::Display for OutboundRequestId { pub struct Config { request_timeout: Duration, max_concurrent_streams: usize, + /// When true (default), outbound requests can be sent over relay connections. + /// When false, only direct connections are used for requests, which conserves + /// relay bandwidth but may fail if no direct connection exists. + allow_relay_for_requests: bool, } impl Default for Config { @@ -317,6 +355,7 @@ impl Default for Config { Self { request_timeout: Duration::from_secs(10), max_concurrent_streams: 100, + allow_relay_for_requests: true, } } } @@ -340,6 +379,21 @@ impl Config { self.max_concurrent_streams = num_streams; self } + + /// Configures whether outbound requests may be sent over relayed connections. + /// + /// Enabled by default, in which case requests use any connection to the peer. Set to `false` to + /// restrict requests to direct connections, which conserves relay bandwidth. 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 only fails with + /// [`OutboundFailure::NoDirectConnection`] once the peer fully disconnects. + /// + /// See the [crate-level documentation](crate#relayed-connections) for the queuing bound and the + /// inbound-connection caveat. + pub fn with_relay_for_requests(mut self, enabled: bool) -> Self { + self.allow_relay_for_requests = enabled; + self + } } /// A request/response protocol for some message codec. @@ -569,11 +623,25 @@ where request: OutboundMessage, ) -> Option> { if let Some(connections) = self.connected.get_mut(peer) { - if connections.is_empty() { + // Only relayed connections are excluded when `allow_relay_for_requests` is disabled. + let eligible_indices: Vec = connections + .iter() + .enumerate() + .filter(|(_, c)| self.config.allow_relay_for_requests || !c.is_relayed()) + .map(|(idx, _)| idx) + .collect(); + + if eligible_indices.is_empty() { + // The peer is connected only over relayed connection(s) while relay use for + // requests is disabled. Keep the request queued so a later direct connection + // (e.g. established via DCUtR) can carry it; it is failed in + // `on_connection_closed` if the peer fully disconnects beforehand. return Some(request); } - let ix = (request.request_id.0 as usize) % connections.len(); - let conn = &mut connections[ix]; + + // Distribute requests round-robin across the eligible connections. + let pick = eligible_indices[request.request_id.0 as usize % eligible_indices.len()]; + let conn = &mut connections[pick]; conn.pending_outbound_responses.insert(request.request_id); self.pending_events.push_back(ToSwarm::NotifyHandler { peer_id: *peer, @@ -678,6 +746,23 @@ where debug_assert_eq!(connections.is_empty(), remaining_established == 0); if connections.is_empty() { self.connected.remove(&peer_id); + + // The peer is now fully disconnected. Any requests still queued can no longer be + // delivered. In normal operation the queue is drained when a connection is + // established (see `preload_new_handler`); requests only remain here when relay use + // for requests is disabled and the peer was reachable solely over relayed + // connection(s) that have now closed without a direct connection becoming available. + if let Some(pending) = self.pending_outbound_requests.remove(&peer_id) { + for request in pending { + self.pending_events + .push_back(ToSwarm::GenerateEvent(Event::OutboundFailure { + peer: peer_id, + connection_id, + request_id: request.request_id, + error: OutboundFailure::NoDirectConnection, + })); + } + } } for request_id in connection.pending_inbound_responses { @@ -745,7 +830,13 @@ where ) { let mut connection = Connection::new(connection_id, remote_address); - if let Some(pending_requests) = self.pending_outbound_requests.remove(&peer) { + // Only drain queued requests onto connections that are eligible to carry them. A relayed + // connection is skipped when relay use for requests is disabled, leaving the requests + // queued so a later direct connection (e.g. established via DCUtR) can carry them. If the + // peer disconnects first, they are failed in `on_connection_closed`. + if (self.config.allow_relay_for_requests || !connection.is_relayed()) + && let Some(pending_requests) = self.pending_outbound_requests.remove(&peer) + { for request in pending_requests { connection .pending_outbound_responses @@ -1060,4 +1151,64 @@ impl Connection { pending_inbound_responses: Default::default(), } } + + /// Whether this connection is established over a relayed (`p2p-circuit`) address. + /// + /// Inbound connections do not carry a remote address and are therefore treated as + /// direct (see the crate-level documentation on relayed connections). + fn is_relayed(&self) -> bool { + self.remote_address + .as_ref() + .is_some_and(|addr| addr.is_relayed()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const RELAYED_ADDR: &str = "/ip4/127.0.0.1/tcp/1234/p2p/12D3KooWDpJ7As7BWAwRMfu1VU2WCqNjvq387JEYKDBj4kx6nXTN/p2p-circuit/p2p/12D3KooWGzBRs8WVcMaEXGWwsvFSfJKYphwM6ojrPanuJfup1xFz"; + + #[test] + fn connection_over_direct_address_is_not_relayed() { + let conn = Connection::new( + ConnectionId::new_unchecked(0), + Some("/ip4/127.0.0.1/tcp/1234".parse().unwrap()), + ); + assert!(!conn.is_relayed()); + } + + #[test] + fn connection_over_circuit_address_is_relayed() { + let conn = Connection::new( + ConnectionId::new_unchecked(0), + Some(RELAYED_ADDR.parse().unwrap()), + ); + assert!(conn.is_relayed()); + } + + #[test] + fn connection_without_address_is_treated_as_direct() { + let conn = Connection::new(ConnectionId::new_unchecked(0), None); + assert!(!conn.is_relayed()); + } + + #[test] + fn relay_for_requests_is_enabled_by_default() { + assert!(Config::default().allow_relay_for_requests); + } + + #[test] + fn with_relay_for_requests_toggles_the_flag() { + assert!( + !Config::default() + .with_relay_for_requests(false) + .allow_relay_for_requests + ); + assert!( + Config::default() + .with_relay_for_requests(true) + .allow_relay_for_requests + ); + } } diff --git a/protocols/request-response/tests/relay.rs b/protocols/request-response/tests/relay.rs new file mode 100644 index 00000000000..906cab14bf0 --- /dev/null +++ b/protocols/request-response/tests/relay.rs @@ -0,0 +1,342 @@ +//! Integration tests for the [`Config::with_relay_for_requests`] policy. +//! +//! The setup mirrors the relay/DCUtR test harness: a relay server, a `server` that holds a +//! reservation on the relay (reachable via a relayed address) and a `client` that talks to it. + +#![cfg(feature = "cbor")] + +use std::time::Duration; + +use futures::StreamExt; +use libp2p_core::{ + Multiaddr, + multiaddr::Protocol, + transport::{MemoryTransport, Transport, upgrade::Version}, +}; +use libp2p_identify as identify; +use libp2p_identity as identity; +use libp2p_identity::PeerId; +use libp2p_relay as relay; +use libp2p_request_response::{self as request_response, ProtocolSupport}; +use libp2p_swarm::{Config as SwarmConfig, NetworkBehaviour, StreamProtocol, Swarm, SwarmEvent}; +use libp2p_swarm_test::SwarmExt as _; +use serde::{Deserialize, Serialize}; +use tracing_subscriber::EnvFilter; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct Ping(Vec); +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct Pong(Vec); + +#[derive(NetworkBehaviour)] +#[behaviour(prelude = "libp2p_swarm::derive_prelude")] +struct Relay { + relay: relay::Behaviour, + identify: identify::Behaviour, +} + +#[derive(NetworkBehaviour)] +#[behaviour(prelude = "libp2p_swarm::derive_prelude")] +struct Client { + relay: relay::client::Behaviour, + request_response: request_response::cbor::Behaviour, + identify: identify::Behaviour, +} + +/// A relayed-only request succeeds with the default configuration. +#[tokio::test] +async fn request_over_relay_succeeds_by_default() { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); + + let (relay_peer_id, relay_tcp_addr) = spawn_relay().await; + + let server = build_client(request_response::Config::default(), LONG_IDLE); + let server_peer_id = *server.local_peer_id(); + let server_relayed_addr = relayed_addr(&relay_tcp_addr, relay_peer_id, server_peer_id); + reserve_and_spawn_server(server, server_relayed_addr.clone(), relay_peer_id).await; + + let mut client = build_client(request_response::Config::default(), LONG_IDLE); + client.dial_and_wait(server_relayed_addr).await; + client + .behaviour_mut() + .request_response + .send_request(&server_peer_id, Ping(b"ping".to_vec())); + + let pong = client + .wait(|event| match event { + SwarmEvent::Behaviour(ClientEvent::RequestResponse( + request_response::Event::Message { + message: request_response::Message::Response { response, .. }, + .. + }, + )) => Some(response), + SwarmEvent::Behaviour(ClientEvent::RequestResponse( + request_response::Event::OutboundFailure { error, .. }, + )) => panic!("unexpected outbound failure: {error}"), + _ => None, + }) + .await; + + assert_eq!(pong, Pong(b"ping".to_vec())); +} + +/// With relayed requests disabled and only a relayed connection available, the request is kept +/// queued (not sent over the relay, not failed) and only fails with +/// [`request_response::OutboundFailure::NoDirectConnection`] once the peer fully disconnects. +#[tokio::test] +async fn request_over_relay_is_queued_then_fails_on_disconnect() { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); + + let (relay_peer_id, relay_tcp_addr) = spawn_relay().await; + + let server = build_client(request_response::Config::default(), LONG_IDLE); + let server_peer_id = *server.local_peer_id(); + let server_relayed_addr = relayed_addr(&relay_tcp_addr, relay_peer_id, server_peer_id); + reserve_and_spawn_server(server, server_relayed_addr.clone(), relay_peer_id).await; + + // Short idle timeout so the unused relayed connection is closed by the swarm shortly after the + // request is queued (this is the mechanism that bounds how long a request can stay queued). + let mut client = build_client( + request_response::Config::default().with_relay_for_requests(false), + Duration::from_millis(500), + ); + client.dial_and_wait(server_relayed_addr).await; + let request_id = client + .behaviour_mut() + .request_response + .send_request(&server_peer_id, Ping(b"ping".to_vec())); + + // The request is neither sent over the relay nor failed yet: it stays queued. + assert!( + client + .behaviour() + .request_response + .is_pending_outbound(&server_peer_id, &request_id), + "request should be queued while only a relayed connection is available" + ); + + // Once the idle relayed connection is closed, the still-queued request fails. + let error = client + .wait(|event| match event { + SwarmEvent::Behaviour(ClientEvent::RequestResponse( + request_response::Event::OutboundFailure { error, .. }, + )) => Some(error), + SwarmEvent::Behaviour(ClientEvent::RequestResponse( + request_response::Event::Message { .. }, + )) => panic!("request was unexpectedly delivered over the relay"), + _ => None, + }) + .await; + + assert!( + matches!(error, request_response::OutboundFailure::NoDirectConnection), + "expected NoDirectConnection, got {error:?}" + ); +} + +/// With relayed requests disabled, a request sent while only a relayed connection exists stays +/// queued and is delivered once a direct connection is established (modelling a DCUtR upgrade). The +/// server asserts that the request did not arrive over a relay. +#[tokio::test] +async fn queued_request_is_delivered_after_direct_upgrade() { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); + + let (relay_peer_id, relay_tcp_addr) = spawn_relay().await; + + let mut server = build_client(request_response::Config::default(), LONG_IDLE); + let server_peer_id = *server.local_peer_id(); + let server_relayed_addr = relayed_addr(&relay_tcp_addr, relay_peer_id, server_peer_id); + let (_, server_direct_addr) = server.listen().await; + let server_direct_addr = server_direct_addr.with_p2p(server_peer_id).unwrap(); + + reserve(&mut server, server_relayed_addr.clone(), relay_peer_id).await; + tokio::spawn(drive_server(server, true)); + + let mut client = build_client( + request_response::Config::default().with_relay_for_requests(false), + LONG_IDLE, + ); + + // Only a relayed connection exists when the request is sent: it must stay queued. + client.dial_and_wait(server_relayed_addr).await; + let request_id = client + .behaviour_mut() + .request_response + .send_request(&server_peer_id, Ping(b"ping".to_vec())); + assert!( + client + .behaviour() + .request_response + .is_pending_outbound(&server_peer_id, &request_id), + ); + + // Establishing a direct connection (the DCUtR upgrade) drains the queue over the direct link. + client.dial(server_direct_addr).unwrap(); + + let pong = client + .wait(|event| match event { + SwarmEvent::Behaviour(ClientEvent::RequestResponse( + request_response::Event::Message { + message: request_response::Message::Response { response, .. }, + .. + }, + )) => Some(response), + SwarmEvent::Behaviour(ClientEvent::RequestResponse( + request_response::Event::OutboundFailure { error, .. }, + )) => panic!("unexpected outbound failure: {error}"), + _ => None, + }) + .await; + + assert_eq!(pong, Pong(b"ping".to_vec())); +} + +/// A generous idle timeout that keeps relayed connections open for the duration of a test. +const LONG_IDLE: Duration = Duration::from_secs(30); + +fn build_client(config: request_response::Config, idle_timeout: Duration) -> Swarm { + let local_key = identity::Keypair::generate_ed25519(); + let local_peer_id = local_key.public().to_peer_id(); + + let (relay_transport, relay_behaviour) = relay::client::new(local_peer_id); + + let transport = relay_transport + .or_transport(MemoryTransport::default()) + .or_transport(libp2p_tcp::tokio::Transport::default()) + .upgrade(Version::V1) + .authenticate(libp2p_plaintext::Config::new(&local_key)) + .multiplex(libp2p_yamux::Config::default()) + .boxed(); + + let protocols = std::iter::once((StreamProtocol::new("/test-rr/1"), ProtocolSupport::Full)); + + Swarm::new( + transport, + Client { + relay: relay_behaviour, + request_response: request_response::cbor::Behaviour::new(protocols, config), + identify: identify::Behaviour::new(identify::Config::new( + "/client".to_owned(), + local_key.public(), + )), + }, + local_peer_id, + SwarmConfig::with_tokio_executor().with_idle_connection_timeout(idle_timeout), + ) +} + +async fn spawn_relay() -> (PeerId, Multiaddr) { + let mut relay = Swarm::new_ephemeral_tokio(|identity| Relay { + relay: relay::Behaviour::new(identity.public().to_peer_id(), relay::Config::default()), + identify: identify::Behaviour::new(identify::Config::new( + "/relay".to_owned(), + identity.public(), + )), + }); + + let (_, relay_tcp_addr) = relay.listen().with_tcp_addr_external().await; + let relay_peer_id = *relay.local_peer_id(); + tokio::spawn(relay.loop_on_next()); + + (relay_peer_id, relay_tcp_addr) +} + +fn relayed_addr(relay_tcp_addr: &Multiaddr, relay_peer_id: PeerId, peer_id: PeerId) -> Multiaddr { + relay_tcp_addr + .clone() + .with(Protocol::P2p(relay_peer_id)) + .with(Protocol::P2pCircuit) + .with(Protocol::P2p(peer_id)) +} + +/// Reserves a slot on the relay and then spawns the server to keep responding/renewing. +async fn reserve_and_spawn_server( + mut server: Swarm, + relayed_addr: Multiaddr, + relay_peer_id: PeerId, +) { + reserve(&mut server, relayed_addr, relay_peer_id).await; + tokio::spawn(drive_server(server, false)); +} + +/// Drives the server's reservation handshake until its relayed address is usable. +async fn reserve(server: &mut Swarm, relayed_addr: Multiaddr, relay_peer_id: PeerId) { + server.listen_on(relayed_addr.clone()).unwrap(); + + let mut listening = false; + let mut reservation_accepted = false; + let mut address_observed = false; + + while !(listening && reservation_accepted && address_observed) { + match server.next_swarm_event().await { + SwarmEvent::NewListenAddr { address, .. } if address == relayed_addr => { + listening = true; + } + SwarmEvent::Behaviour(ClientEvent::Relay( + relay::client::Event::ReservationReqAccepted { + relay_peer_id: peer_id, + .. + }, + )) if peer_id == relay_peer_id => { + reservation_accepted = true; + } + SwarmEvent::Behaviour(ClientEvent::Identify(identify::Event::Received { .. })) => { + address_observed = true; + } + _ => {} + } + } +} + +/// Responds to every inbound request with a `Pong` echoing the payload. When `require_direct` is +/// set, asserts that requests are never delivered over a relayed connection (tracked via the +/// established connections' endpoints). +async fn drive_server(mut server: Swarm, require_direct: bool) { + use std::collections::HashMap; + + let mut relayed_connections: HashMap<_, bool> = HashMap::new(); + + loop { + match server.select_next_some().await { + SwarmEvent::ConnectionEstablished { + connection_id, + endpoint, + .. + } => { + relayed_connections.insert(connection_id, endpoint.is_relayed()); + } + SwarmEvent::ConnectionClosed { connection_id, .. } => { + relayed_connections.remove(&connection_id); + } + SwarmEvent::Behaviour(ClientEvent::RequestResponse( + request_response::Event::Message { + connection_id, + message: + request_response::Message::Request { + request, channel, .. + }, + .. + }, + )) => { + if require_direct { + assert_eq!( + relayed_connections.get(&connection_id), + Some(&false), + "request was delivered over a relayed connection" + ); + } + let _ = server + .behaviour_mut() + .request_response + .send_response(channel, Pong(request.0)); + } + _ => {} + } + } +}