diff --git a/Cargo.lock b/Cargo.lock index 0fa48556eed..99c32c583fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3403,6 +3403,7 @@ dependencies = [ "futures-bounded", "futures-timer", "libp2p-core", + "libp2p-identify", "libp2p-identity", "libp2p-ping", "libp2p-plaintext", @@ -5286,6 +5287,18 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "relay-client-example" +version = "0.1.0" +dependencies = [ + "clap", + "futures", + "libp2p", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "relay-server-example" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 092e759eed0..c2a0e1b662f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ members = [ "examples/ipfs-private", "examples/metrics", "examples/ping", + "examples/relay-client", "examples/relay-server", "examples/rendezvous", "examples/stream", diff --git a/examples/relay-client/Cargo.toml b/examples/relay-client/Cargo.toml new file mode 100644 index 00000000000..67bacd17005 --- /dev/null +++ b/examples/relay-client/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "relay-client-example" +version = "0.1.0" +edition.workspace = true +publish = false +license = "MIT" + +[package.metadata.release] +release = false + +[dependencies] +clap = { version = "4.6.1", features = ["derive"] } +futures = { workspace = true } +libp2p = { path = "../../libp2p", features = ["dns", "identify", "macros", "noise", "ping", "quic", "relay", "tcp", "tokio", "yamux"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "signal"] } +tracing = { workspace = true } +tracing-subscriber = { workspace = true, features = ["env-filter"] } + +[lints] +workspace = true diff --git a/examples/relay-client/README.md b/examples/relay-client/README.md new file mode 100644 index 00000000000..74b109ea4ec --- /dev/null +++ b/examples/relay-client/README.md @@ -0,0 +1,17 @@ +## Description + +A small relay client that demonstrates the autorelay. + +## Run the client + +In another terminal, pointing it at the relay: + +``` +cargo run \ + --secret-key-seed 1 \ + --relay /ip4/$RELAY_IP/tcp/$PORT/p2p/$RELAY_PEERID \ + --max-reservations 2 +``` + +Provide `relay` multiple times to point at additional relays; autorelay will +pick among them up to `max-reservations`. \ No newline at end of file diff --git a/examples/relay-client/src/main.rs b/examples/relay-client/src/main.rs new file mode 100644 index 00000000000..b577b92b878 --- /dev/null +++ b/examples/relay-client/src/main.rs @@ -0,0 +1,144 @@ +use std::{error::Error, num::NonZeroU8}; + +use clap::Parser; +use futures::stream::StreamExt; +use libp2p::{ + core::multiaddr::Multiaddr, + identify, + identity::Keypair, + noise, ping, + relay::{self, autorelay}, + swarm::{NetworkBehaviour, SwarmEvent}, + tcp, yamux, +}; +use tracing_subscriber::EnvFilter; + +#[derive(Debug, Parser)] +#[command(name = "libp2p relay client")] +struct Opts { + /// Fixed value used to derive a deterministic peer id. + #[arg(long)] + secret_key_seed: Option, + + /// List of relay addresses + #[arg(long = "relay", required = true)] + relays: Vec, + + /// Maximum number of relay reservations autorelay should maintain. + #[arg(long, default_value_t = 2)] + max_reservations: u8, +} + +#[derive(NetworkBehaviour)] +struct Behaviour { + relay_client: relay::client::Behaviour, + autorelay: autorelay::Behaviour, + identify: identify::Behaviour, + ping: ping::Behaviour, +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); + + let opts = Opts::parse(); + + let max_reservations = NonZeroU8::new(opts.max_reservations) + .ok_or("--max-reservations must be greater than zero")?; + let autorelay_config = autorelay::Config::default().set_max_reservations(max_reservations); + + let mut swarm = + libp2p::SwarmBuilder::with_existing_identity(generate_ed25519(opts.secret_key_seed)) + .with_tokio() + .with_tcp( + tcp::Config::default().nodelay(true), + noise::Config::new, + yamux::Config::default, + )? + .with_quic() + .with_dns()? + .with_relay_client(noise::Config::new, yamux::Config::default)? + .with_behaviour(|keypair, relay_behaviour| Behaviour { + relay_client: relay_behaviour, + autorelay: autorelay::Behaviour::new_with_config(autorelay_config), + identify: identify::Behaviour::new(identify::Config::new( + "/autorelay-example/0.1.0".to_owned(), + keypair.public(), + )), + ping: ping::Behaviour::new(ping::Config::new()), + })? + .build(); + + swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?; + swarm.listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse()?)?; + + let local_peer_id = *swarm.local_peer_id(); + tracing::info!(%local_peer_id, "Local peer id"); + + for addr in &opts.relays { + tracing::info!(%addr, "Dialing relay"); + swarm.dial(addr.clone())?; + } + + loop { + tokio::select! { + event = swarm.select_next_some() => match event { + SwarmEvent::NewListenAddr { address, .. } => { + tracing::info!(%address, "Listening"); + } + SwarmEvent::ConnectionEstablished { + peer_id, endpoint, .. + } => { + tracing::info!(peer=%peer_id, address=%endpoint.get_remote_address(), "Connected"); + } + SwarmEvent::ConnectionClosed { peer_id, .. } => { + tracing::info!(peer=%peer_id, "Disconnected"); + } + SwarmEvent::ExternalAddrConfirmed { address } => { + tracing::info!(%address, "External address confirmed"); + } + SwarmEvent::ExternalAddrExpired { address } => { + tracing::info!(%address, "External address expired"); + } + SwarmEvent::Behaviour(BehaviourEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { + relay_peer_id, renewal, .. + }, + )) => { + if renewal { + tracing::info!(%relay_peer_id, "Reservation renewed"); + } else { + tracing::info!(%relay_peer_id, "Reservation accepted"); + } + } + SwarmEvent::Behaviour(BehaviourEvent::RelayClient(event)) => { + tracing::debug!(?event, "Relay client event"); + } + SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received { + peer_id, info, .. + })) => { + tracing::debug!(peer=%peer_id, protocols=?info.protocols, "Identify received"); + } + SwarmEvent::Behaviour(BehaviourEvent::Identify(_)) => {} + SwarmEvent::Behaviour(BehaviourEvent::Ping(_)) => {} + SwarmEvent::Behaviour(BehaviourEvent::Autorelay(e)) => { + tracing::debug!(?e, "Autorelay event"); + } + _ => {} + }, + } + } +} + +fn generate_ed25519(secret_key_seed: Option) -> Keypair { + match secret_key_seed { + Some(secret_key_seed) => { + let mut bytes = [0u8; 32]; + bytes[0] = secret_key_seed; + Keypair::ed25519_from_bytes(bytes).expect("only errors on wrong length") + } + None => Keypair::generate_ed25519(), + } +} diff --git a/protocols/relay/CHANGELOG.md b/protocols/relay/CHANGELOG.md index b0c24aa0365..107bf5e2c31 100644 --- a/protocols/relay/CHANGELOG.md +++ b/protocols/relay/CHANGELOG.md @@ -21,6 +21,8 @@ See [PR 6285](https://github.com/libp2p/rust-libp2p/pull/6285). - Reset reservation state on listener close. See [PR 6461](https://github.com/libp2p/rust-libp2p/pull/6461). +- Implements autorelay that would make a reservation as soon as a connection reports supporting HOP protocol. + See [PR 6156](https://github.com/libp2p/rust-libp2p/pull/6156) ## 0.21.1 - reduce allocations by replacing `get_or_insert` with `get_or_insert_with` diff --git a/protocols/relay/Cargo.toml b/protocols/relay/Cargo.toml index ffecc209821..8a1daf3126e 100644 --- a/protocols/relay/Cargo.toml +++ b/protocols/relay/Cargo.toml @@ -29,6 +29,7 @@ thiserror = { workspace = true } tracing = { workspace = true } [dev-dependencies] +libp2p-identify = { workspace = true } libp2p-identity = { workspace = true, features = ["rand"] } libp2p-ping = { workspace = true } libp2p-plaintext = { workspace = true } diff --git a/protocols/relay/src/autorelay.rs b/protocols/relay/src/autorelay.rs new file mode 100644 index 00000000000..402efce559a --- /dev/null +++ b/protocols/relay/src/autorelay.rs @@ -0,0 +1,871 @@ +use std::{ + collections::{BTreeMap, HashMap, HashSet, VecDeque}, + num::NonZeroU8, + task::{Context, Poll, Waker}, + time::Duration, +}; + +use either::Either; +use futures::FutureExt; +use futures_timer::Delay; +use libp2p_core::{ + Endpoint, + multiaddr::Protocol, + transport::{ListenerId, PortUse}, +}; +use libp2p_identity::PeerId; +use libp2p_swarm::{ + ExternalAddresses, ListenOpts, NewListenAddr, + derive_prelude::{ + AddressChange, ConnectionClosed, ConnectionDenied, ConnectionEstablished, ConnectionId, + DialFailure, ExpiredListenAddr, FromSwarm, ListenerClosed, ListenerError, Multiaddr, + NetworkBehaviour, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, + }, + dial_opts::DialOpts, + dummy, +}; +use web_time::{Instant, SystemTime}; + +use crate::{ + autorelay::handler::Out, + multiaddr_ext::{MultiaddrExt, relay_peer_id}, +}; + +mod handler; + +#[derive(Debug)] +pub struct Behaviour { + config: Config, + status: Status, + auto_status_change: bool, + external_addresses: ExternalAddresses, + events: VecDeque::ToSwarm, THandlerInEvent>>, + + connections: HashMap<(PeerId, ConnectionId), Connection>, + + reservations: HashMap, + + external_reservations: HashMap, + + static_relays: HashMap>, + + static_dial_cooldowns: HashMap, + + failure_counts: HashMap, + + reservation_cooldowns: HashMap, + + previous_relays: VecDeque<(PeerId, Multiaddr, SystemTime)>, + + relays_available: bool, + + cooldown_wakeup: Option<(Instant, Delay)>, + + waker: Option, +} + +impl Default for Behaviour { + fn default() -> Self { + Self { + config: Config::default(), + status: Status::Enable, + auto_status_change: true, + external_addresses: ExternalAddresses::default(), + events: VecDeque::new(), + connections: HashMap::new(), + reservations: HashMap::new(), + external_reservations: HashMap::new(), + static_relays: HashMap::new(), + static_dial_cooldowns: HashMap::new(), + failure_counts: HashMap::new(), + reservation_cooldowns: HashMap::new(), + previous_relays: VecDeque::new(), + relays_available: false, + cooldown_wakeup: None, + waker: None, + } + } +} + +#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)] +pub enum Status { + #[default] + Enable, + Disable, +} + +#[derive(Debug)] +struct Connection { + address: Multiaddr, + relay_status: RelayStatus, +} + +impl Connection { + /// Mark relayed connection as not supported + pub(crate) fn disqualify_connection_if_relayed(&mut self) { + if self.address.is_relayed() { + self.relay_status = RelayStatus::NotSupported; + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RelayStatus { + Supported { status: ReservationStatus }, + NotSupported, + Pending, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ReservationStatus { + Idle, + Pending { id: ListenerId }, + Active { id: ListenerId }, +} + +#[derive(Debug)] +pub struct Config { + max_reservations: NonZeroU8, + failure_cooldown: Duration, + failure_cooldown_max: Duration, + max_previous_relays: usize, + static_relays: HashMap>, +} + +impl Default for Config { + fn default() -> Self { + Self { + max_reservations: NonZeroU8::new(2).unwrap(), + failure_cooldown: Duration::from_secs(30), + failure_cooldown_max: Duration::from_secs(10 * 60), + max_previous_relays: 16, + static_relays: HashMap::new(), + } + } +} + +impl Config { + pub fn set_max_reservations(mut self, max_reservations: NonZeroU8) -> Self { + self.max_reservations = max_reservations; + self + } + + pub fn set_failure_cooldown(mut self, duration: Duration) -> Self { + self.failure_cooldown = duration; + self + } + + pub fn set_failure_cooldown_max(mut self, duration: Duration) -> Self { + self.failure_cooldown_max = duration; + self + } + + pub fn set_max_previous_relays(mut self, max: usize) -> Self { + self.max_previous_relays = max; + self + } + + pub fn add_static_relay(mut self, peer_id: PeerId, addresses: Vec) -> Self { + let entry = self.static_relays.entry(peer_id).or_default(); + for addr in addresses { + if !entry.contains(&addr) { + entry.push(addr); + } + } + self + } +} + +#[derive(Debug)] +#[non_exhaustive] +pub enum Event { + /// The status of the local node has changed. + StatusChanged { status: Status }, + /// No connected peer supports the HOP protocol. + NoRelaysAvailable, + /// At least one connected peer supports the HOP protocol. + RelaysAvailable, +} + +impl Behaviour { + pub fn new_with_config(mut config: Config) -> Self { + let initial_static_relays = std::mem::take(&mut config.static_relays); + let mut behaviour = Self { + config, + ..Default::default() + }; + for (peer_id, addresses) in initial_static_relays { + for address in addresses { + behaviour.add_static_relay(peer_id, address); + } + } + behaviour + } + + /// Sets the autorelay status. + pub fn set_status(&mut self, status: Option) { + match status { + Some(status) => { + self.auto_status_change = false; + if self.status != status { + self.status = status; + self.events + .push_back(ToSwarm::GenerateEvent(Event::StatusChanged { status })); + if status == Status::Enable { + self.meet_reservation_target(); + } + } + } + None => { + self.auto_status_change = true; + self.determine_status_from_external_addresses(); + } + } + + if let Some(waker) = self.waker.take() { + waker.wake(); + } + } + + /// Register a peer as a static relay. + /// + /// This will dial and establish a connection to the peer if it doesn't already have a direct + /// connection. + /// Note that peers that are through a relay cannot be used as a static peer + pub fn add_static_relay(&mut self, peer_id: PeerId, address: Multiaddr) { + if address.is_relayed() { + tracing::warn!(%peer_id, %address, "static relay address is relayed. ignoring."); + return; + } + + let entry = self.static_relays.entry(peer_id).or_default(); + if entry.contains(&address) { + tracing::warn!(%peer_id, %address, "static relay address already exist"); + } else { + entry.push(address); + } + let combined = entry.clone(); + + if self.is_peer_idle(&peer_id) { + self.evict_for_static_peer(peer_id); + } + + if !self.queue_static_dial(peer_id, combined) { + self.meet_reservation_target(); + } + + if let Some(waker) = self.waker.take() { + waker.wake(); + } + } + + /// Remove peer as a static relay. + /// This will not close any connections or terminate any existing reservation with the relay + pub fn remove_static_relay(&mut self, peer_id: &PeerId) -> bool { + self.static_dial_cooldowns.remove(peer_id); + self.static_relays.remove(peer_id).is_some() + } + + pub fn static_relays(&self) -> impl Iterator { + self.static_relays + .iter() + .map(|(peer, addrs)| (peer, addrs.as_slice())) + } + + pub fn previous_relays(&self) -> impl Iterator { + self.previous_relays + .iter() + .map(|(peer, addr, ts)| (peer, addr, ts)) + } + + fn static_dial_in_cooldown(&self, peer_id: &PeerId) -> bool { + self.static_dial_cooldowns + .get(peer_id) + .is_some_and(|deadline| *deadline > Instant::now()) + } + + fn queue_static_dial(&mut self, peer_id: PeerId, addresses: Vec) -> bool { + if addresses.is_empty() + || self.has_direct_connection(&peer_id) + || self.static_dial_in_cooldown(&peer_id) + { + return false; + } + let opts = DialOpts::peer_id(peer_id).addresses(addresses).build(); + self.events.push_back(ToSwarm::Dial { opts }); + true + } + + fn record_previous_relay(&mut self, peer_id: PeerId, address: Multiaddr) { + let max = self.config.max_previous_relays; + if max == 0 { + return; + } + self.previous_relays.retain(|(p, _, _)| *p != peer_id); + if self.previous_relays.len() >= max { + self.previous_relays.pop_front(); + } + self.previous_relays + .push_back((peer_id, address, SystemTime::now())); + } + + fn forget_previous_relay(&mut self, peer_id: &PeerId) { + self.previous_relays.retain(|(p, _, _)| p != peer_id); + } + + fn record_failure(&mut self, peer_id: PeerId) -> Duration { + let attempts = self.failure_counts.entry(peer_id).or_insert(0); + *attempts = attempts.saturating_add(1); + let exponent = attempts.saturating_sub(1).min(20); + let scale = 1u32 << exponent; + self.config + .failure_cooldown + .saturating_mul(scale) + .min(self.config.failure_cooldown_max) + } + + fn clear_failure(&mut self, peer_id: &PeerId) { + self.failure_counts.remove(peer_id); + self.reservation_cooldowns.remove(peer_id); + } + + fn reservation_in_cooldown(&self, peer_id: &PeerId) -> bool { + self.reservation_cooldowns + .get(peer_id) + .is_some_and(|deadline| *deadline > Instant::now()) + } + + fn poll_reservation_cooldowns(&mut self, cx: &mut Context<'_>) -> bool { + let now = Instant::now(); + + if self + .reservation_cooldowns + .values() + .any(|deadline| *deadline <= now) + { + self.reservation_cooldowns + .retain(|_, deadline| *deadline > now); + self.cooldown_wakeup = None; + self.meet_reservation_target(); + return true; + } + + match self.reservation_cooldowns.values().copied().min() { + Some(deadline) => { + if self.cooldown_wakeup.as_ref().map(|(at, _)| *at) != Some(deadline) { + let delay = Delay::new(deadline.saturating_duration_since(now)); + self.cooldown_wakeup = Some((deadline, delay)); + } + } + None => { + self.cooldown_wakeup = None; + return false; + } + } + + match self.cooldown_wakeup.as_mut() { + Some((_, timer)) => timer.poll_unpin(cx).is_ready(), + None => false, + } + } + + fn determine_status_from_external_addresses(&mut self) { + let has_public_addr = self + .external_addresses + .iter() + .any(|addr| !addr.is_relayed()); + + let new_status = match has_public_addr { + true => Status::Disable, + false => Status::Enable, + }; + if new_status != self.status { + self.status = new_status; + self.events + .push_back(ToSwarm::GenerateEvent(Event::StatusChanged { + status: new_status, + })); + match new_status { + Status::Enable => self.meet_reservation_target(), + Status::Disable => self.remove_all_reservations(), + } + } + } + + fn is_peer_idle(&self, peer_id: &PeerId) -> bool { + self.connections.iter().any(|((pid, _), info)| { + pid == peer_id + && info.relay_status + == RelayStatus::Supported { + status: ReservationStatus::Idle, + } + }) + } + + fn has_direct_connection(&self, peer_id: &PeerId) -> bool { + self.connections + .iter() + .any(|((pid, _), info)| pid == peer_id && !info.address.is_relayed()) + } + + fn evict_for_static_peer(&mut self, new_static: PeerId) { + let covered = self.covered_peers(); + if covered.contains(&new_static) { + tracing::debug!(%new_static, "peer is already covered by a reservation"); + return; + } + let max = self.config.max_reservations.get() as usize; + if covered.len() < max { + tracing::debug!(%new_static, "free reservation slot available. no eviction needed"); + return; + } + + if let Some((peer_id, listener_id)) = self + .reservations + .iter() + .find(|(_, (peer_id, _))| !self.static_relays.contains_key(peer_id)) + .map(|(listener_id, (peer_id, _))| (peer_id, *listener_id)) + { + tracing::debug!(%peer_id, %listener_id, "evicting peer to for static relay"); + self.events + .push_back(ToSwarm::RemoveListener { id: listener_id }); + } + } + + fn select_connection_for_reservation(&mut self, peer_id: PeerId, connection_id: ConnectionId) { + let info = self + .connections + .get_mut(&(peer_id, connection_id)) + .expect("connection is present"); + + if info.relay_status + != (RelayStatus::Supported { + status: ReservationStatus::Idle, + }) + { + return; + } + + let addr_with_peer_id = match info.address.clone().with_p2p(peer_id) { + Ok(addr) => addr, + Err(addr) => { + tracing::warn!(%addr, "address unexpectedly contains a different peer id than the connection; marking relay connection ineligible"); + info.relay_status = RelayStatus::NotSupported; + return; + } + }; + + let opts = ListenOpts::new(addr_with_peer_id.with(Protocol::P2pCircuit)); + let id = opts.listener_id(); + + info.relay_status = RelayStatus::Supported { + status: ReservationStatus::Pending { id }, + }; + self.reservations.insert(id, (peer_id, connection_id)); + self.events.push_back(ToSwarm::ListenOn { opts }); + } + + /// Removes all existing reservations. + fn remove_all_reservations(&mut self) { + let relay_listeners = self + .reservations + .iter() + .map(|(id, (peer_id, conn_id))| (*id, *peer_id, *conn_id)) + .collect::>(); + + for (listener_id, peer_id, connection_id) in relay_listeners { + let Some(connection) = self.connections.get_mut(&(peer_id, connection_id)) else { + continue; + }; + + if !matches!( + connection.relay_status, + RelayStatus::Supported { + status: ReservationStatus::Active { id } | ReservationStatus::Pending { id } + } if id == listener_id + ) { + continue; + } + + connection.relay_status = RelayStatus::Supported { + status: ReservationStatus::Idle, + }; + + self.events + .push_back(ToSwarm::RemoveListener { id: listener_id }); + } + } + + fn disable_reservation(&mut self, id: ListenerId, failed: bool) { + if self.external_reservations.remove(&id).is_some() { + self.meet_reservation_target(); + return; + } + + let Some((peer_id, connection_id)) = self.reservations.remove(&id) else { + return; + }; + + let Some(address) = self + .connections + .get(&(peer_id, connection_id)) + .filter(|info| { + matches!( + info.relay_status, + RelayStatus::Supported { + status: ReservationStatus::Active { .. } + | ReservationStatus::Pending { .. } + } + ) + }) + .map(|info| info.address.clone()) + else { + self.meet_reservation_target(); + return; + }; + + let cooldown_duration = failed.then(|| self.record_failure(peer_id)); + + let connection = self + .connections + .get_mut(&(peer_id, connection_id)) + .expect("connection is tracked"); + connection.relay_status = RelayStatus::Supported { + status: ReservationStatus::Idle, + }; + + if let Some(duration) = cooldown_duration { + self.reservation_cooldowns + .insert(peer_id, Instant::now() + duration); + } + + self.record_previous_relay(peer_id, address); + self.meet_reservation_target(); + } + + fn covered_peers(&self) -> HashSet { + self.reservations + .values() + .map(|(peer_id, _)| *peer_id) + .chain(self.external_reservations.values().copied()) + .collect() + } + + /// Meet the reservation target by selecting connections to establish a reservation. + fn meet_reservation_target(&mut self) { + if self.status == Status::Disable { + return; + } + + let max = self.config.max_reservations.get() as usize; + let covered = self.covered_peers(); + let budget = max.saturating_sub(covered.len()); + if budget == 0 { + return; + } + + let mut static_candidates = BTreeMap::new(); + let mut candidates: BTreeMap<_, ConnectionId> = BTreeMap::new(); + for ((peer_id, connection_id), info) in self.connections.iter() { + if covered.contains(peer_id) { + continue; + } + if self.reservation_in_cooldown(peer_id) { + continue; + } + if info.relay_status + != (RelayStatus::Supported { + status: ReservationStatus::Idle, + }) + { + continue; + } + let bucket = if self.static_relays.contains_key(peer_id) { + &mut static_candidates + } else { + &mut candidates + }; + bucket + .entry(*peer_id) + .and_modify(|existing| *existing = (*existing).min(*connection_id)) + .or_insert(*connection_id); + } + + let selected_candidates: Vec<(PeerId, ConnectionId)> = static_candidates + .into_iter() + .chain(candidates) + .take(budget) + .collect(); + + for (peer_id, connection_id) in selected_candidates { + self.select_connection_for_reservation(peer_id, connection_id); + } + + debug_assert!(self.covered_peers().len() <= max); + } + + fn update_relay_availability(&mut self) { + let has_hop_peer = self + .connections + .values() + .any(|info| matches!(info.relay_status, RelayStatus::Supported { .. })); + + match (has_hop_peer, self.relays_available) { + (true, false) => { + self.relays_available = true; + self.events + .push_back(ToSwarm::GenerateEvent(Event::RelaysAvailable)); + } + (false, true) => { + self.relays_available = false; + self.events + .push_back(ToSwarm::GenerateEvent(Event::NoRelaysAvailable)); + } + _ => {} + } + } +} + +impl NetworkBehaviour for Behaviour { + type ConnectionHandler = Either; + type ToSwarm = Event; + + fn handle_established_inbound_connection( + &mut self, + _connection_id: ConnectionId, + _peer: PeerId, + local_addr: &Multiaddr, + _remote_addr: &Multiaddr, + ) -> Result, ConnectionDenied> { + if local_addr.is_relayed() { + Ok(Either::Right(dummy::ConnectionHandler)) + } else { + Ok(Either::Left(handler::Handler::default())) + } + } + + fn handle_established_outbound_connection( + &mut self, + _connection_id: ConnectionId, + _peer: PeerId, + addr: &Multiaddr, + _role_override: Endpoint, + _port_use: PortUse, + ) -> Result, ConnectionDenied> { + if addr.is_relayed() { + Ok(Either::Right(dummy::ConnectionHandler)) + } else { + Ok(Either::Left(handler::Handler::default())) + } + } + + fn on_swarm_event(&mut self, event: FromSwarm) { + let change = self.external_addresses.on_swarm_event(&event); + + if self.auto_status_change && change { + self.determine_status_from_external_addresses(); + } + + match event { + FromSwarm::ConnectionEstablished(ConnectionEstablished { + peer_id, + endpoint, + connection_id, + .. + }) => { + let remote_addr = endpoint.get_remote_address().clone(); + + let mut connection = Connection { + address: remote_addr, + relay_status: RelayStatus::Pending, + }; + + connection.disqualify_connection_if_relayed(); + + self.connections + .insert((peer_id, connection_id), connection); + + if self.static_relays.contains_key(&peer_id) { + self.static_dial_cooldowns.remove(&peer_id); + } + } + FromSwarm::ConnectionClosed(ConnectionClosed { + peer_id, + connection_id, + .. + }) => { + let Some(connection) = self.connections.remove(&(peer_id, connection_id)) else { + return; + }; + + if !self.connections.keys().any(|(pid, _)| *pid == peer_id) { + self.clear_failure(&peer_id); + } + + let had_reservation = matches!( + connection.relay_status, + RelayStatus::Supported { + status: ReservationStatus::Active { .. } + | ReservationStatus::Pending { .. } + } + ); + + if let RelayStatus::Supported { + status: ReservationStatus::Active { id } | ReservationStatus::Pending { id }, + } = connection.relay_status + { + self.reservations.remove(&id); + self.meet_reservation_target(); + } + + if had_reservation { + self.record_previous_relay(peer_id, connection.address); + } + + if let Some(addresses) = self.static_relays.get(&peer_id).cloned() { + self.queue_static_dial(peer_id, addresses); + } + + self.update_relay_availability(); + } + FromSwarm::AddressChange(AddressChange { + peer_id, + connection_id, + old: _, + new, + }) => { + let Some(connection) = self.connections.get_mut(&(peer_id, connection_id)) else { + return; + }; + + let new_addr = new.get_remote_address(); + + connection.address = new_addr.clone(); + } + FromSwarm::NewListenAddr(NewListenAddr { listener_id, addr }) => { + if !addr.is_relayed() { + return; + } + + if let Some((peer_id, connection_id)) = self.reservations.get(&listener_id).copied() + { + let Some(connection) = self.connections.get_mut(&(peer_id, connection_id)) + else { + return; + }; + + if matches!( + connection.relay_status, + RelayStatus::Supported { + status: ReservationStatus::Pending { id } + } if id == listener_id + ) { + connection.relay_status = RelayStatus::Supported { + status: ReservationStatus::Active { id: listener_id }, + }; + self.forget_previous_relay(&peer_id); + self.clear_failure(&peer_id); + } + return; + } + + if let Some(relay_peer_id) = relay_peer_id(addr) { + self.external_reservations + .insert(listener_id, relay_peer_id); + } + } + FromSwarm::ExpiredListenAddr(ExpiredListenAddr { listener_id, .. }) => { + self.disable_reservation(listener_id, false); + } + FromSwarm::ListenerError(ListenerError { listener_id, .. }) => { + self.disable_reservation(listener_id, true); + } + FromSwarm::ListenerClosed(ListenerClosed { + listener_id, + reason, + .. + }) => { + self.disable_reservation(listener_id, reason.is_err()); + } + FromSwarm::DialFailure(DialFailure { + peer_id: Some(peer_id), + error, + .. + }) if self.static_relays.contains_key(&peer_id) => { + tracing::warn!(%peer_id, %error, "dial to static relay failed"); + self.static_dial_cooldowns + .insert(peer_id, Instant::now() + self.config.failure_cooldown); + } + _ => {} + } + } + + fn on_connection_handler_event( + &mut self, + peer_id: PeerId, + connection_id: ConnectionId, + event: THandlerOutEvent, + ) { + let Either::Left(event) = event; + + let Some(connection) = self.connections.get_mut(&(peer_id, connection_id)) else { + return; + }; + + match event { + Out::Supported => { + if matches!( + connection.relay_status, + RelayStatus::Pending | RelayStatus::NotSupported + ) { + connection.relay_status = RelayStatus::Supported { + status: ReservationStatus::Idle, + }; + if self.static_relays.contains_key(&peer_id) { + self.evict_for_static_peer(peer_id); + } + self.meet_reservation_target(); + self.update_relay_availability(); + } + } + Out::Unsupported => { + let drop_listener = match connection.relay_status { + RelayStatus::Supported { + status: ReservationStatus::Pending { id } | ReservationStatus::Active { id }, + } => Some(id), + _ => None, + }; + let lost_address = drop_listener.map(|_| connection.address.clone()); + connection.relay_status = RelayStatus::NotSupported; + if let Some(id) = drop_listener { + self.reservations.remove(&id); + self.events.push_back(ToSwarm::RemoveListener { id }); + self.meet_reservation_target(); + } + if let Some(address) = lost_address { + self.record_previous_relay(peer_id, address); + } + self.update_relay_availability(); + } + } + } + + fn poll( + &mut self, + cx: &mut Context<'_>, + ) -> Poll>> { + loop { + if let Some(event) = self.events.pop_front() { + return Poll::Ready(event); + } + + if self.poll_reservation_cooldowns(cx) { + continue; + } + + self.waker = Some(cx.waker().clone()); + + return Poll::Pending; + } + } +} diff --git a/protocols/relay/src/autorelay/handler.rs b/protocols/relay/src/autorelay/handler.rs new file mode 100644 index 00000000000..e1ef47dac3a --- /dev/null +++ b/protocols/relay/src/autorelay/handler.rs @@ -0,0 +1,103 @@ +use std::{ + collections::VecDeque, + convert::Infallible, + task::{Context, Poll}, +}; + +use libp2p_core::upgrade::DeniedUpgrade; +use libp2p_swarm::{ + ConnectionHandler, ConnectionHandlerEvent, SubstreamProtocol, SupportedProtocols, + handler::ConnectionEvent, +}; + +use crate::HOP_PROTOCOL_NAME; + +#[derive(Default, Debug)] +pub struct Handler { + events: VecDeque< + ConnectionHandlerEvent< + ::OutboundProtocol, + ::OutboundOpenInfo, + ::ToBehaviour, + >, + >, + + supported: bool, + + supported_protocol: SupportedProtocols, +} + +#[derive(Debug, Copy, Clone)] +pub enum Out { + Supported, + Unsupported, +} + +impl ConnectionHandler for Handler { + type FromBehaviour = Infallible; + type ToBehaviour = Out; + type InboundProtocol = DeniedUpgrade; + type OutboundProtocol = DeniedUpgrade; + type InboundOpenInfo = (); + type OutboundOpenInfo = (); + + fn listen_protocol(&self) -> SubstreamProtocol { + SubstreamProtocol::new(DeniedUpgrade, ()) + } + + fn connection_keep_alive(&self) -> bool { + false + } + + fn on_behaviour_event(&mut self, event: Self::FromBehaviour) { + match event {} + } + + fn on_connection_event( + &mut self, + event: ConnectionEvent< + Self::InboundProtocol, + Self::OutboundProtocol, + Self::InboundOpenInfo, + Self::OutboundOpenInfo, + >, + ) { + if let ConnectionEvent::RemoteProtocolsChange(protocol) = event { + let change = self.supported_protocol.on_protocols_change(protocol); + if change { + let valid = self + .supported_protocol + .iter() + .any(|proto| HOP_PROTOCOL_NAME.eq(proto)); + + match (valid, self.supported) { + (true, false) => { + self.supported = true; + self.events + .push_back(ConnectionHandlerEvent::NotifyBehaviour(Out::Supported)); + } + (false, true) => { + self.supported = false; + self.events + .push_back(ConnectionHandlerEvent::NotifyBehaviour(Out::Unsupported)); + } + (true, true) => {} + _ => {} + } + } + } + } + + fn poll( + &mut self, + _cx: &mut Context<'_>, + ) -> Poll< + ConnectionHandlerEvent, + > { + if let Some(event) = self.events.pop_front() { + return Poll::Ready(event); + } + + Poll::Pending + } +} diff --git a/protocols/relay/src/lib.rs b/protocols/relay/src/lib.rs index 1c32cc8d8d0..65bc83fa2e3 100644 --- a/protocols/relay/src/lib.rs +++ b/protocols/relay/src/lib.rs @@ -23,6 +23,7 @@ #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))] +pub mod autorelay; mod behaviour; mod copy_future; mod multiaddr_ext; diff --git a/protocols/relay/src/multiaddr_ext.rs b/protocols/relay/src/multiaddr_ext.rs index f9a1f71b4dc..d1f66cb032c 100644 --- a/protocols/relay/src/multiaddr_ext.rs +++ b/protocols/relay/src/multiaddr_ext.rs @@ -1,4 +1,5 @@ use libp2p_core::{Multiaddr, multiaddr::Protocol}; +use libp2p_identity::PeerId; pub(crate) trait MultiaddrExt { fn is_relayed(&self) -> bool; @@ -9,3 +10,15 @@ impl MultiaddrExt for Multiaddr { self.iter().any(|p| p == Protocol::P2pCircuit) } } + +pub(crate) fn relay_peer_id(addr: &Multiaddr) -> Option { + let mut last_p2p = None; + for proto in addr.iter() { + match proto { + Protocol::P2p(peer) => last_p2p = Some(peer), + Protocol::P2pCircuit => return last_p2p, + _ => {} + } + } + None +} diff --git a/protocols/relay/tests/autorelay.rs b/protocols/relay/tests/autorelay.rs new file mode 100644 index 00000000000..77becfe7bce --- /dev/null +++ b/protocols/relay/tests/autorelay.rs @@ -0,0 +1,1097 @@ +use std::{ + collections::{HashMap, HashSet}, + future::Future, + num::NonZeroU8, + time::Duration, +}; + +use futures::{ + io::{AsyncRead, AsyncWrite}, + stream::StreamExt, +}; +use libp2p_core::{ + multiaddr::{Multiaddr, Protocol}, + muxing::StreamMuxerBox, + transport::{Boxed, MemoryTransport, Transport, choice::OrTransport}, + upgrade, +}; +use libp2p_identify as identify; +use libp2p_identity as identity; +use libp2p_identity::PeerId; +use libp2p_plaintext as plaintext; +use libp2p_relay::{self as relay, autorelay}; +use libp2p_swarm::{Config, ConnectionId, NetworkBehaviour, Swarm, SwarmEvent}; +use tracing_subscriber::EnvFilter; + +#[tokio::test] +async fn autorelay_respects_max_reservations() { + init_tracing(); + + let (relay_a_peer_id, relay_a_addr) = spawn_relay(); + let (relay_b_peer_id, relay_b_addr) = spawn_relay(); + + let mut client = + build_client(autorelay::Config::default().set_max_reservations(NonZeroU8::new(1).unwrap())); + client.dial(relay_a_addr).unwrap(); + client.dial(relay_b_addr).unwrap(); + + let mut accepted = 0usize; + let mut timeout = futures_timer::Delay::new(Duration::from_secs(20)); + loop { + tokio::select! { + _ = &mut timeout => break, + ev = client.select_next_some() => { + if let SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { relay_peer_id, .. }, + )) = ev + { + assert!(relay_peer_id == relay_a_peer_id || relay_peer_id == relay_b_peer_id); + accepted += 1; + if accepted > 1 { + panic!("autorelay opened more reservations than max_reservations=1"); + } + futures_timer::Delay::new(Duration::from_secs(2)).await; + break; + } + } + } + } + + assert_eq!( + accepted, 1, + "expected exactly one reservation, observed {accepted}" + ); +} + +#[tokio::test] +async fn autorelay_with_two_reservations_among_five_relays() { + init_tracing(); + + let relay_addrs: Vec<(PeerId, Multiaddr)> = (0..5).map(|_| spawn_relay()).collect(); + let relay_peers: HashSet = relay_addrs.iter().map(|(p, _)| *p).collect(); + + let mut client = + build_client(autorelay::Config::default().set_max_reservations(NonZeroU8::new(2).unwrap())); + for (_, addr) in &relay_addrs { + client.dial(addr.clone()).unwrap(); + } + + let mut direct_conns: HashMap = HashMap::new(); + let mut reservations: HashSet = HashSet::new(); + + let mut sleep = futures_timer::Delay::new(Duration::from_secs(30)); + loop { + tokio::select! { + _ = &mut sleep => panic!( + "timeout: got {} reservations, expected 2", + reservations.len() + ), + ev = client.select_next_some() => match ev { + SwarmEvent::ConnectionEstablished { + peer_id, connection_id, endpoint, .. + } if !endpoint.is_relayed() && relay_peers.contains(&peer_id) => { + direct_conns.insert(peer_id, connection_id); + } + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { + relay_peer_id, + renewal: false, + .. + }, + )) => { + reservations.insert(relay_peer_id); + } + _ => {} + } + } + if reservations.len() == 2 { + break; + } + } + + let drop_peer = *reservations.iter().next().expect("two reservations held"); + let keep_peer = reservations + .iter() + .find(|p| **p != drop_peer) + .copied() + .expect("two reservations held"); + let drop_conn = *direct_conns + .get(&drop_peer) + .expect("direct connection observed"); + + assert!( + client.close_connection(drop_conn), + "should close the relay connection holding a reservation" + ); + + let mut sleep = futures_timer::Delay::new(Duration::from_secs(30)); + + loop { + tokio::select! { + _ = &mut sleep => panic!("timeout waiting for replacement reservation"), + ev = client.select_next_some() => { + if let SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { + relay_peer_id, + renewal: false, + .. + }, + )) = ev + && relay_peer_id != keep_peer + && relay_peer_id != drop_peer + { + return; + } + } + } + } +} + +#[tokio::test] +async fn autorelay_drops_reservations_when_public_address_appears() { + init_tracing(); + + let (_, relay_a_addr) = spawn_relay(); + let (_, relay_b_addr) = spawn_relay(); + + let mut client = + build_client(autorelay::Config::default().set_max_reservations(NonZeroU8::new(2).unwrap())); + client.dial(relay_a_addr).unwrap(); + client.dial(relay_b_addr).unwrap(); + + let mut confirmed: HashSet = HashSet::new(); + let mut sleep = futures_timer::Delay::new(Duration::from_secs(30)); + + loop { + tokio::select! { + _ = &mut sleep => panic!( + "timeout: got {} confirmed external addresses, expected 2", + confirmed.len() + ), + ev = client.select_next_some() => { + if let SwarmEvent::ExternalAddrConfirmed { address } = ev + && address.iter().any(|p| p == Protocol::P2pCircuit) + { + confirmed.insert(address); + } + } + } + if confirmed.len() == 2 { + break; + } + } + + let public_addr = Multiaddr::empty().with(Protocol::Memory(rand::random::())); + client.add_external_address(public_addr); + + let mut expired: HashSet = HashSet::new(); + let mut sleep = futures_timer::Delay::new(Duration::from_secs(15)); + + loop { + tokio::select! { + _ = &mut sleep => panic!( + "timeout: only {}/{} relayed addresses expired", + expired.len(), + confirmed.len() + ), + ev = client.select_next_some() => { + if let SwarmEvent::ExternalAddrExpired { address } = ev + && confirmed.contains(&address) + { + expired.insert(address); + } + } + } + if expired == confirmed { + break; + } + } +} + +#[tokio::test] +async fn autorelay_blacklists_failing_relay_and_retries_after_cooldown() { + init_tracing(); + + let (_, relay_addr) = spawn_rejecting_relay(); + + let cooldown = Duration::from_secs(1); + let mut client = build_client( + autorelay::Config::default() + .set_max_reservations(NonZeroU8::new(1).unwrap()) + .set_failure_cooldown(cooldown), + ); + client.dial(relay_addr).unwrap(); + + let first_failure_at = wait_for_listener_failure(&mut client, Duration::from_secs(10)).await; + + let early_retry = with_timeout( + wait_for_listener_failure(&mut client, cooldown * 5), + cooldown / 2, + ) + .await; + assert!( + early_retry.is_none(), + "autorelay retried during the cooldown window" + ); + + let second_failure_at = wait_for_listener_failure(&mut client, cooldown * 5).await; + let elapsed = second_failure_at.duration_since(first_failure_at); + assert!( + elapsed >= cooldown, + "retry should respect cooldown (elapsed {elapsed:?}, cooldown {cooldown:?})" + ); +} + +async fn wait_for_listener_failure( + client: &mut Swarm, + timeout: Duration, +) -> std::time::Instant { + let mut sleep = futures_timer::Delay::new(timeout); + + loop { + tokio::select! { + _ = &mut sleep => panic!("timeout waiting for listener failure"), + ev = client.select_next_some() => { + if let SwarmEvent::ListenerClosed { reason: Err(_), .. } = ev { + return std::time::Instant::now(); + } + } + } + } +} + +#[tokio::test] +async fn autorelay_disabled_does_not_reserve() { + init_tracing(); + + let (_, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client + .behaviour_mut() + .autorelay + .set_status(Some(autorelay::Status::Disable)); + client.dial(relay_addr).unwrap(); + + let observed = with_timeout( + wait_until(&mut client, Duration::from_secs(5), |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { .. } + )) + ) + }), + Duration::from_secs(3), + ) + .await; + + assert!( + observed.is_none(), + "autorelay opened a reservation while disabled" + ); +} + +#[tokio::test] +async fn autorelay_re_enable_triggers_reservation() { + init_tracing(); + + let (_, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client + .behaviour_mut() + .autorelay + .set_status(Some(autorelay::Status::Disable)); + client.dial(relay_addr).unwrap(); + + let mut sleep = futures_timer::Delay::new(Duration::from_secs(3)); + + loop { + tokio::select! { + _ = &mut sleep => break, + ev = client.select_next_some() => { + if matches!( + ev, + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { .. } + )) + ) { + panic!("autorelay reserved while disabled"); + } + } + } + } + + client + .behaviour_mut() + .autorelay + .set_status(Some(autorelay::Status::Enable)); + + wait_until(&mut client, Duration::from_secs(10), |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { .. } + )) + ) + }) + .await; +} + +#[tokio::test] +async fn autorelay_disable_preserves_active_reservation() { + init_tracing(); + + let (_, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client.dial(relay_addr).unwrap(); + + wait_until(&mut client, Duration::from_secs(20), |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { .. } + )) + ) + }) + .await; + + client + .behaviour_mut() + .autorelay + .set_status(Some(autorelay::Status::Disable)); + + let mut sleep = futures_timer::Delay::new(Duration::from_secs(3)); + + loop { + tokio::select! { + _ = &mut sleep => break, + ev = client.select_next_some() => { + if let SwarmEvent::ListenerClosed { reason: Err(_), .. } = ev { + panic!("disabling autorelay dropped an active reservation"); + } + if let SwarmEvent::ExternalAddrExpired { .. } = ev { + panic!("disabling autorelay expired an external address"); + } + } + } + } +} + +#[tokio::test] +async fn autorelay_prefers_static_relay() { + init_tracing(); + + let (opportunistic_peer, opportunistic_addr) = spawn_relay(); + let (static_peer, static_addr) = spawn_relay(); + + let mut client = + build_client(autorelay::Config::default().set_max_reservations(NonZeroU8::new(1).unwrap())); + client + .behaviour_mut() + .autorelay + .set_status(Some(autorelay::Status::Disable)); + + client.dial(opportunistic_addr).unwrap(); + client + .behaviour_mut() + .autorelay + .add_static_relay(static_peer, static_addr); + + // Let both connections establish and identify exchanges complete. + let mut warmup = futures_timer::Delay::new(Duration::from_secs(3)); + loop { + tokio::select! { + _ = &mut warmup => break, + _ = client.select_next_some() => {} + } + } + + client + .behaviour_mut() + .autorelay + .set_status(Some(autorelay::Status::Enable)); + + let accepted_peer = wait_until_some(&mut client, Duration::from_secs(15), |event| { + if let SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { relay_peer_id, .. }, + )) = event + { + Some(*relay_peer_id) + } else { + None + } + }) + .await; + + assert_eq!( + accepted_peer, static_peer, + "autorelay should pick the static relay over the opportunistic one" + ); + assert_ne!(accepted_peer, opportunistic_peer); +} + +#[tokio::test] +async fn remove_static_relay_preserves_active_reservation() { + init_tracing(); + + let (relay_peer, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client + .behaviour_mut() + .autorelay + .add_static_relay(relay_peer, relay_addr); + + wait_until(&mut client, Duration::from_secs(15), |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { .. } + )) + ) + }) + .await; + + assert!( + client + .behaviour_mut() + .autorelay + .remove_static_relay(&relay_peer) + ); + + let mut sleep = futures_timer::Delay::new(Duration::from_secs(3)); + + loop { + tokio::select! { + _ = &mut sleep => break, + ev = client.select_next_some() => { + if let SwarmEvent::ListenerClosed { reason: Err(_), .. } = ev { + panic!("removing static relay dropped an active reservation"); + } + if let SwarmEvent::ExternalAddrExpired { .. } = ev { + panic!("removing static relay expired an external address"); + } + } + } + } +} + +#[tokio::test] +async fn static_relay_redials_after_connection_drop() { + init_tracing(); + + let (relay_peer, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client + .behaviour_mut() + .autorelay + .add_static_relay(relay_peer, relay_addr); + + let conn_id = + wait_for_reservation_with_conn(&mut client, relay_peer, Duration::from_secs(15)).await; + + assert!(client.close_connection(conn_id)); + + wait_until(&mut client, Duration::from_secs(20), { + let mut redialed = false; + let mut reserved_again = false; + move |event| { + match event { + SwarmEvent::ConnectionEstablished { + peer_id, endpoint, .. + } if *peer_id == relay_peer && !endpoint.is_relayed() => { + redialed = true; + } + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { relay_peer_id, .. }, + )) if *relay_peer_id == relay_peer => { + reserved_again = true; + } + _ => {} + } + redialed && reserved_again + } + }) + .await; +} + +async fn wait_until_some(client: &mut Swarm, timeout: Duration, mut extract: F) -> T +where + F: FnMut(&SwarmEvent) -> Option, +{ + let mut sleep = futures_timer::Delay::new(timeout); + + loop { + tokio::select! { + _ = &mut sleep => panic!("timeout waiting on predicate"), + ev = client.select_next_some() => { + if let Some(value) = extract(&ev) { + return value; + } + } + } + } +} + +#[tokio::test] +async fn autorelay_emits_relay_available_after_recovery() { + init_tracing(); + + let (relay_peer, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client.dial(relay_addr.clone()).unwrap(); + + let conn_id = + wait_for_reservation_with_conn(&mut client, relay_peer, Duration::from_secs(15)).await; + + assert!(client.close_connection(conn_id)); + + wait_until(&mut client, Duration::from_secs(10), |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::Autorelay(autorelay::Event::NoRelaysAvailable)) + ) + }) + .await; + + client.dial(relay_addr).unwrap(); + + wait_until(&mut client, Duration::from_secs(15), |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::Autorelay(autorelay::Event::RelaysAvailable)) + ) + }) + .await; +} + +#[tokio::test] +async fn autorelay_no_relays_available_is_edge_triggered() { + init_tracing(); + + let (relay_a_peer, relay_a_addr) = spawn_relay(); + let (relay_b_peer, relay_b_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client.dial(relay_a_addr).unwrap(); + client.dial(relay_b_addr).unwrap(); + + let mut conns: HashMap = HashMap::new(); + let mut reserved: HashSet = HashSet::new(); + let mut sleep = futures_timer::Delay::new(Duration::from_secs(20)); + + loop { + tokio::select! { + _ = &mut sleep => panic!("did not get both reservations in time"), + ev = client.select_next_some() => match ev { + SwarmEvent::ConnectionEstablished { + peer_id, connection_id, endpoint, .. + } if !endpoint.is_relayed() + && (peer_id == relay_a_peer || peer_id == relay_b_peer) => + { + conns.insert(peer_id, connection_id); + } + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { relay_peer_id, .. } + )) if relay_peer_id == relay_a_peer || relay_peer_id == relay_b_peer => { + reserved.insert(relay_peer_id); + } + _ => {} + } + } + if reserved.len() == 2 { + break; + } + } + + let conn_a = *conns.get(&relay_a_peer).unwrap(); + let conn_b = *conns.get(&relay_b_peer).unwrap(); + + assert!(client.close_connection(conn_a)); + assert!(client.close_connection(conn_b)); + + let mut starved_count = 0usize; + let mut sleep = futures_timer::Delay::new(Duration::from_secs(5)); + + loop { + tokio::select! { + _ = &mut sleep => break, + ev = client.select_next_some() => { + if matches!( + ev, + SwarmEvent::Behaviour(ClientEvent::Autorelay( + autorelay::Event::NoRelaysAvailable + )) + ) { + starved_count += 1; + } + } + } + } + + assert_eq!( + starved_count, 1, + "NoRelaysAvailable should fire exactly once across multiple meet_reservation_target invocations" + ); +} + +#[tokio::test] +async fn autorelay_resumes_after_public_address_removed() { + init_tracing(); + + let (relay_peer, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client.dial(relay_addr).unwrap(); + + wait_for_reservation_from(&mut client, relay_peer, Duration::from_secs(15)).await; + + let public_addr = memory_addr(); + client.add_external_address(public_addr.clone()); + + wait_until(&mut client, Duration::from_secs(10), |event| { + matches!(event, SwarmEvent::ExternalAddrExpired { .. }) + }) + .await; + + client.remove_external_address(&public_addr); + + wait_for_reservation_from(&mut client, relay_peer, Duration::from_secs(15)).await; +} + +#[tokio::test] +async fn autorelay_manual_enable_ignores_public_address() { + init_tracing(); + + let (relay_peer, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client + .behaviour_mut() + .autorelay + .set_status(Some(autorelay::Status::Enable)); + client.dial(relay_addr).unwrap(); + + wait_for_reservation_from(&mut client, relay_peer, Duration::from_secs(15)).await; + + client.add_external_address(memory_addr()); + + let mut sleep = futures_timer::Delay::new(Duration::from_secs(3)); + + loop { + tokio::select! { + _ = &mut sleep => break, + ev = client.select_next_some() => { + if let SwarmEvent::ListenerClosed { reason: Err(_), .. } = ev { + panic!("manual-Enable autorelay dropped reservation after public addr appeared"); + } + if let SwarmEvent::ExternalAddrExpired { address } = &ev + && address.iter().any(|p| p == Protocol::P2pCircuit) + { + panic!("manual-Enable autorelay expired the relayed external address"); + } + if let SwarmEvent::Behaviour(ClientEvent::Autorelay( + autorelay::Event::StatusChanged { status: autorelay::Status::Disable }, + )) = ev + { + panic!("manual-Enable autorelay flipped to Disable on public addr"); + } + } + } + } +} + +#[tokio::test] +async fn autorelay_forgets_previous_relay_on_reacquire() { + init_tracing(); + + let (relay_peer, relay_addr) = spawn_relay(); + + let mut client = build_client(autorelay::Config::default()); + client.dial(relay_addr.clone()).unwrap(); + + let conn_id = + wait_for_reservation_with_conn(&mut client, relay_peer, Duration::from_secs(15)).await; + + assert!(client.close_connection(conn_id)); + + wait_until(&mut client, Duration::from_secs(10), |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::Autorelay(autorelay::Event::NoRelaysAvailable)) + ) + }) + .await; + + assert!( + client + .behaviour() + .autorelay + .previous_relays() + .any(|(p, _, _)| *p == relay_peer), + "expected {relay_peer} in previous_relays after loss" + ); + + client.dial(relay_addr).unwrap(); + + wait_until(&mut client, Duration::from_secs(15), |event| { + matches!( + event, + SwarmEvent::NewListenAddr { address, .. } if address.iter().any(|p| p == Protocol::P2pCircuit) + ) + }) + .await; + + let previous: Vec = client + .behaviour() + .autorelay + .previous_relays() + .map(|(p, _, _)| *p) + .collect(); + assert!( + !previous.contains(&relay_peer), + "expected {relay_peer} to be removed from previous_relays after re-acquire, got {previous:?}" + ); +} + +#[tokio::test] +async fn autorelay_previous_relays_is_bounded() { + init_tracing(); + + let peers_and_addrs: Vec<(PeerId, Multiaddr)> = (0..3).map(|_| spawn_relay()).collect(); + + let mut client = build_client( + autorelay::Config::default() + .set_max_reservations(NonZeroU8::new(1).unwrap()) + .set_max_previous_relays(2), + ); + + for (peer, addr) in &peers_and_addrs { + client.dial(addr.clone()).unwrap(); + + let conn_id = + wait_for_reservation_with_conn(&mut client, *peer, Duration::from_secs(15)).await; + + assert!(client.close_connection(conn_id)); + + wait_until(&mut client, Duration::from_secs(10), |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::Autorelay(autorelay::Event::NoRelaysAvailable)) + ) + }) + .await; + } + + let previous: Vec = client + .behaviour() + .autorelay + .previous_relays() + .map(|(p, _, _)| *p) + .collect(); + + assert_eq!( + previous.len(), + 2, + "expected previous_relays to be bounded to 2, got {previous:?}" + ); + assert!( + !previous.contains(&peers_and_addrs[0].0), + "oldest relay should have been evicted: {previous:?}" + ); + assert!(previous.contains(&peers_and_addrs[1].0)); + assert!(previous.contains(&peers_and_addrs[2].0)); +} + +#[tokio::test] +async fn autorelay_static_relay_dial_cooldown_after_failure() { + init_tracing(); + + let cooldown = Duration::from_secs(2); + let mut client = build_client(autorelay::Config::default().set_failure_cooldown(cooldown)); + + let unreachable_peer = PeerId::random(); + let unreachable_addr = memory_addr(); + + client + .behaviour_mut() + .autorelay + .add_static_relay(unreachable_peer, unreachable_addr.clone()); + + wait_until(&mut client, Duration::from_secs(5), |event| { + matches!( + event, + SwarmEvent::OutgoingConnectionError { peer_id: Some(p), .. } if *p == unreachable_peer + ) + }) + .await; + + let first_failure_at = std::time::Instant::now(); + + client + .behaviour_mut() + .autorelay + .add_static_relay(unreachable_peer, unreachable_addr.clone()); + + let mut redialed = false; + let mut watch = futures_timer::Delay::new(cooldown / 2); + loop { + tokio::select! { + _ = &mut watch => break, + ev = client.select_next_some() => { + if matches!( + ev, + SwarmEvent::OutgoingConnectionError { peer_id: Some(p), .. } if p == unreachable_peer + ) { + redialed = true; + break; + } + } + } + } + assert!(!redialed, "autorelay redialed within cooldown"); + + let remaining = cooldown + .checked_sub(first_failure_at.elapsed()) + .unwrap_or_default(); + if !remaining.is_zero() { + futures_timer::Delay::new(remaining + Duration::from_millis(200)).await; + } + + client + .behaviour_mut() + .autorelay + .add_static_relay(unreachable_peer, unreachable_addr); + + wait_until(&mut client, Duration::from_secs(5), |event| { + matches!( + event, + SwarmEvent::OutgoingConnectionError { peer_id: Some(p), .. } if *p == unreachable_peer + ) + }) + .await; +} + +#[tokio::test] +async fn autorelay_evicts_discovered_peers_for_static() { + init_tracing(); + + let (opp_a_peer, opp_a_addr) = spawn_relay(); + let (opp_b_peer, opp_b_addr) = spawn_relay(); + let (static_peer, static_addr) = spawn_relay(); + + let mut client = + build_client(autorelay::Config::default().set_max_reservations(NonZeroU8::new(1).unwrap())); + + client.dial(opp_a_addr).unwrap(); + client.dial(opp_b_addr).unwrap(); + + wait_until_some(&mut client, Duration::from_secs(20), |event| { + if let SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { relay_peer_id, .. }, + )) = event + && (*relay_peer_id == opp_a_peer || *relay_peer_id == opp_b_peer) + { + Some(*relay_peer_id) + } else { + None + } + }) + .await; + + client + .behaviour_mut() + .autorelay + .add_static_relay(static_peer, static_addr); + + wait_for_reservation_from(&mut client, static_peer, Duration::from_secs(20)).await; +} + +async fn wait_until(client: &mut Swarm, timeout: Duration, mut predicate: F) +where + F: FnMut(&SwarmEvent) -> bool, +{ + let mut sleep = futures_timer::Delay::new(timeout); + loop { + tokio::select! { + _ = &mut sleep => panic!("timeout waiting on predicate"), + ev = client.select_next_some() => { + if predicate(&ev) { + return; + } + } + } + } +} + +async fn with_timeout(future: F, timeout: Duration) -> Option { + use futures::future::Either; + let timer = futures_timer::Delay::new(timeout); + futures::pin_mut!(future); + match futures::future::select(future, timer).await { + Either::Left((output, _)) => Some(output), + Either::Right(_) => None, + } +} + +async fn wait_for_reservation_from(client: &mut Swarm, peer: PeerId, timeout: Duration) { + wait_until(client, timeout, |event| { + matches!( + event, + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { relay_peer_id, .. } + )) if *relay_peer_id == peer + ) + }) + .await; +} + +async fn wait_for_reservation_with_conn( + client: &mut Swarm, + peer: PeerId, + timeout: Duration, +) -> ConnectionId { + wait_until_some(client, timeout, { + let mut established: Option = None; + let mut reserved = false; + move |event| { + match event { + SwarmEvent::ConnectionEstablished { + peer_id, + connection_id, + endpoint, + .. + } if *peer_id == peer && !endpoint.is_relayed() => { + established = Some(*connection_id); + } + SwarmEvent::Behaviour(ClientEvent::RelayClient( + relay::client::Event::ReservationReqAccepted { relay_peer_id, .. }, + )) if *relay_peer_id == peer => { + reserved = true; + } + _ => {} + } + if reserved { established } else { None } + } + }) + .await +} + +fn init_tracing() { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); +} + +fn memory_addr() -> Multiaddr { + Multiaddr::empty().with(Protocol::Memory(rand::random::())) +} + +fn spawn_relay() -> (PeerId, Multiaddr) { + spawn_relay_swarm(build_relay()) +} + +fn spawn_rejecting_relay() -> (PeerId, Multiaddr) { + spawn_relay_swarm(build_rejecting_relay()) +} + +fn spawn_relay_swarm(mut relay: Swarm) -> (PeerId, Multiaddr) { + let addr = memory_addr(); + let peer = *relay.local_peer_id(); + relay.listen_on(addr.clone()).unwrap(); + relay.add_external_address(addr.clone()); + tokio::spawn(relay.collect::>()); + (peer, addr) +} + +fn build_relay() -> Swarm { + build_relay_with_config(relay::Config { + reservation_duration: Duration::from_secs(60), + ..Default::default() + }) +} + +fn build_rejecting_relay() -> Swarm { + build_relay_with_config(relay::Config { + max_reservations: 0, + ..Default::default() + }) +} + +fn build_relay_with_config(config: relay::Config) -> Swarm { + let local_key = identity::Keypair::generate_ed25519(); + let local_peer_id = local_key.public().to_peer_id(); + let transport = upgrade_transport(MemoryTransport::default().boxed(), &local_key); + + Swarm::new( + transport, + Relay { + relay: relay::Behaviour::new(local_peer_id, config), + identify: identify::Behaviour::new(identify::Config::new( + "/autorelay-test/1.0.0".to_owned(), + local_key.public(), + )), + }, + local_peer_id, + Config::with_tokio_executor(), + ) +} + +fn build_client(autorelay_config: autorelay::Config) -> Swarm { + let local_key = identity::Keypair::generate_ed25519(); + let local_peer_id = local_key.public().to_peer_id(); + let (relay_transport, relay_client) = relay::client::new(local_peer_id); + + let transport = upgrade_transport( + OrTransport::new(relay_transport, MemoryTransport::default()).boxed(), + &local_key, + ); + + Swarm::new( + transport, + Client { + relay_client, + autorelay: autorelay::Behaviour::new_with_config(autorelay_config), + identify: identify::Behaviour::new(identify::Config::new( + "/autorelay-test/1.0.0".to_owned(), + local_key.public(), + )), + }, + local_peer_id, + Config::with_tokio_executor(), + ) +} + +fn upgrade_transport( + transport: Boxed, + identity: &identity::Keypair, +) -> Boxed<(PeerId, StreamMuxerBox)> +where + StreamSink: AsyncRead + AsyncWrite + Send + Unpin + 'static, +{ + transport + .upgrade(upgrade::Version::V1) + .authenticate(plaintext::Config::new(identity)) + .multiplex(libp2p_yamux::Config::default()) + .boxed() +} + +#[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_client: relay::client::Behaviour, + autorelay: autorelay::Behaviour, + identify: identify::Behaviour, +}