diff options
Diffstat (limited to 'src/transport/p2p.rs')
| -rw-r--r-- | src/transport/p2p.rs | 272 |
1 files changed, 257 insertions, 15 deletions
diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs index 3464a68..675ff13 100644 --- a/src/transport/p2p.rs +++ b/src/transport/p2p.rs @@ -21,9 +21,17 @@ //! that works, or for good if it doesn't, within its limits (`RELAY_CIRCUIT_DURATION`, //! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send). //! +//! A relay says so where it publishes its addresses: next to each, the same address +//! ending in `/p2p-circuit` (where to reserve a slot), which other nodes find in the +//! member list. A node that needs a relay and has none configured picks up to +//! `AUTO_RELAYS` of those (`relay_check`). It needs one when it has no address others +//! could dial, or when AutoNAT says it can't be dialled: AutoNAT asks connected peers +//! (nodes of the swarm) to dial it back. Once AutoNAT says it can be dialled, it lets +//! the relays it picked go. +//! //! What this transport doesn't do yet (`docs/future-work.md`): find peers on the LAN -//! (mDNS), or pick relays among the members by itself. It never uses public libp2p -//! infrastructure: relays are only those configured. +//! (mDNS). It never uses public libp2p infrastructure: relays are those configured, or +//! members that say they relay. use std::collections::HashMap; use std::path::Path; @@ -75,6 +83,10 @@ const RELAY_CIRCUITS_PER_PEER: usize = 8; /// A reservation on a relay that ended (the relay restarted, say) is made again after /// this. const RELAY_RETRY: Duration = Duration::from_secs(30); +/// Relays picked among the members, at most, and how often a node checks whether it +/// needs them. +const AUTO_RELAYS: usize = 2; +const RELAY_CHECK: Duration = Duration::from_secs(30); /// How long a message of `bytes` may take on a stream. fn stream_timeout(bytes: usize) -> Duration { @@ -132,6 +144,8 @@ enum Command { address: Multiaddr, done: oneshot::Sender<Result<PeerId, String>>, }, + /// See whether relays among the members are needed now (one was just learned of). + CheckRelays, /// Reserve a slot on the relay at `relay` (again). `configured`: the address as /// configured, for later attempts (without a peer ID, the relay may since have /// another). @@ -148,6 +162,27 @@ struct Book { listening: Vec<Multiaddr>, /// `Libp2pConfig::announce`. announced: Vec<Multiaddr>, + /// `Libp2pConfig::relay`: this node relays, and says so. + serving: bool, + /// Members that say they relay, and where (with the relay's peer ID). + relays: HashMap<PeerId, Multiaddr>, +} + +impl Book { + /// Where others could dial us directly: the announced addresses, or else those we + /// listen on (`published`), without the peer ID. + fn direct(&self) -> Vec<Multiaddr> { + if !self.announced.is_empty() { + return self.announced.clone(); + } + let direct: Vec<Multiaddr> = self + .listening + .iter() + .filter(|address| !is_relayed(address)) + .cloned() + .collect(); + published(&direct) + } } /// Whether `address` only reaches this host (or this link): no use to other hosts, and @@ -244,8 +279,10 @@ impl Libp2pTransport { let incoming = control .accept(PROTOCOL) .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?; + let configured_relays = !config.relays.is_empty(); let book = Arc::new(Mutex::new(Book { announced: config.announce, + serving: config.relay, ..Book::default() })); let (commands, receiver) = mpsc::unbounded_channel(); @@ -261,7 +298,11 @@ impl Libp2pTransport { receiver, commands.clone(), Arc::clone(&book), - config.relay, + Relaying { + serving: config.relay, + configured: configured_relays, + local: peer_id, + }, stopping, )); info!(peer_id = %peer_id, "libp2p transport started"); @@ -311,6 +352,7 @@ struct Behaviour { /// Only with `Libp2pConfig::relay`. relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>, dcutr: ::libp2p::dcutr::Behaviour, + autonat: ::libp2p::autonat::Behaviour, } impl Behaviour { @@ -336,6 +378,14 @@ impl Behaviour { relay_client, relay: relay.into(), dcutr: ::libp2p::dcutr::Behaviour::new(peer_id), + autonat: ::libp2p::autonat::Behaviour::new( + peer_id, + ::libp2p::autonat::Config { + // Swarms on private networks (a LAN, WireGuard) count too. + only_global_ips: false, + ..Default::default() + }, + ), } } } @@ -383,15 +433,100 @@ fn memory_swarm(key: Keypair, relay: bool) -> Result<Swarm<Behaviour>> { /// Drives the swarm: dials what `commands` ask for and notes where we listen, until the /// transport is dropped. -/// `again`: to schedule commands; `serving`: whether this node relays for others. +/// How a node relates to relays. +struct Relaying { + /// It relays for others. + serving: bool, + /// Relays to use are configured: it doesn't pick any. + configured: bool, + local: PeerId, +} + +/// The relays to pick among `offers`, if this node needs relays, keeping `using`: a +/// node needs them when it has no address others could dial (`direct`), or AutoNAT says +/// it can't be dialled; up to `AUTO_RELAYS`, in an order of its own (the lowest hash of +/// both peer IDs), so that not every node picks the same. +fn relay_check( + local: PeerId, + direct: bool, + nat: &::libp2p::autonat::NatStatus, + offers: &HashMap<PeerId, Multiaddr>, + using: &[PeerId], +) -> Vec<(PeerId, Multiaddr)> { + use ::libp2p::autonat::NatStatus; + let needed = match nat { + NatStatus::Private => true, + NatStatus::Public(_) => false, + NatStatus::Unknown => !direct, + }; + if !needed || using.len() >= AUTO_RELAYS { + return Vec::new(); + } + let score = |relay: &PeerId| { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + (local, relay).hash(&mut hasher); + hasher.finish() + }; + let mut candidates: Vec<(PeerId, Multiaddr)> = offers + .iter() + .filter(|(relay, _)| **relay != local && !using.contains(relay)) + .map(|(relay, address)| (*relay, address.clone())) + .collect(); + candidates.sort_by_key(|(relay, _)| score(relay)); + candidates.truncate(AUTO_RELAYS - using.len()); + candidates +} + +/// Reserves slots on the relays `relay_check` picks among the members, and lets those +/// picked go once AutoNAT says this node can be dialled. +fn pick_relays( + swarm: &mut Swarm<Behaviour>, + book: &Mutex<Book>, + relaying: &Relaying, + nat: &::libp2p::autonat::NatStatus, + picked: &mut HashMap<::libp2p::core::transport::ListenerId, PeerId>, +) { + if nat.is_public() && !picked.is_empty() { + info!("Reachable directly (AutoNAT); letting the relays picked go"); + for listener in std::mem::take(picked).into_keys() { + swarm.remove_listener(listener); + } + return; + } + let (direct, offers) = { + let book = lock(book); + (!book.direct().is_empty(), book.relays.clone()) + }; + let using: Vec<PeerId> = picked.values().copied().collect(); + for (relay, address) in relay_check(relaying.local, direct, nat, &offers, &using) { + match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) { + Ok(listener) => { + info!(relay = %relay, address = %address, "Picked a member as libp2p relay"); + picked.insert(listener, relay); + } + Err(error) => { + debug!(relay = %relay, error = %error, "Couldn't use a member as libp2p relay"); + } + } + } +} + +/// `again`: to schedule commands. async fn run( mut swarm: Swarm<Behaviour>, mut commands: mpsc::UnboundedReceiver<Command>, again: mpsc::UnboundedSender<Command>, book: Arc<Mutex<Book>>, - serving: bool, + relaying: Relaying, stopping: watch::Sender<bool>, ) { + let serving = relaying.serving; + let mut nat = ::libp2p::autonat::NatStatus::Unknown; + // Relays picked among the members, by listener. + let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new(); + let mut check = tokio::time::interval(RELAY_CHECK); + check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); // Relays we listen through, by listener, as configured. let mut reservations: HashMap<::libp2p::core::transport::ListenerId, Multiaddr> = HashMap::new(); @@ -412,8 +547,16 @@ async fn run( > = HashMap::new(); loop { tokio::select! { + _ = check.tick(), if !relaying.configured => { + pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked); + } command = commands.recv() => match command { None => break, + Some(Command::CheckRelays) => { + if !relaying.configured { + pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked); + } + } Some(Command::Connect { peer, addresses, done }) => { if swarm.is_connected(&peer) { let _ = done.send(Ok(())); @@ -532,6 +675,10 @@ async fn run( warn!(relay = %relay, reason = ?reason, "Lost the slot on a libp2p relay; reserving again later"); reserve_later(relay); } + // A relay picked among the members: the next check picks again. + if let Some(relay) = picked.remove(&listener_id) { + debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying"); + } } SwarmEvent::ListenerError { error, .. } => { warn!(error = %error, "libp2p listener failed"); @@ -543,6 +690,12 @@ async fn run( info!(relay = %relay_peer_id, "Reachable through a libp2p relay"); } } + SwarmEvent::Behaviour(BehaviourEvent::Autonat( + ::libp2p::autonat::Event::StatusChanged { old, new }, + )) => { + info!(old = ?old, new = ?new, "libp2p reachability (AutoNAT)"); + nat = new; + } SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => { debug!(event = ?event, "libp2p relay"); } @@ -611,20 +764,32 @@ impl Transport for Libp2pTransport { } /// Where we can be reached, each with our peer ID: the announced addresses, or else - /// those we listen on (`published`), and those through relays. + /// those we listen on (`published`), and those through relays. A relay adds each + /// direct one again, ending in `/p2p-circuit`: where to reserve a slot. fn local_addresses(&self) -> Vec<String> { let book = lock(&self.book); - let (relayed, direct): (Vec<Multiaddr>, Vec<Multiaddr>) = - book.listening.iter().cloned().partition(is_relayed); - let mut addresses = if book.announced.is_empty() { - published(&direct) + let direct: Vec<Multiaddr> = book + .direct() + .into_iter() + .filter_map(|address| address.with_p2p(self.peer_id).ok()) + .collect(); + let relayed = book + .listening + .iter() + .filter(|address| is_relayed(address)) + .filter_map(|address| address.clone().with_p2p(self.peer_id).ok()); + let offers: Vec<Multiaddr> = if book.serving { + direct + .iter() + .map(|address| address.clone().with(Protocol::P2pCircuit)) + .collect() } else { - book.announced.clone() + Vec::new() }; - addresses.extend(relayed); - addresses + direct .into_iter() - .filter_map(|address| address.with_p2p(self.peer_id).ok()) + .chain(relayed) + .chain(offers) .map(|address| address.to_string()) .collect() } @@ -634,7 +799,27 @@ impl Transport for Libp2pTransport { if let Ok(peer) = PeerId::from_str(target) { return Some(peer.to_string()); } - let address = Multiaddr::from_str(target).ok()?; + let mut address = Multiaddr::from_str(target).ok()?; + // A relay saying where to reserve a slot: `<address>/p2p/<relay>/p2p-circuit`. + if let Some(Protocol::P2pCircuit) = address.iter().last() { + address.pop(); + let relay = split_peer(&address)?; + if relay != self.peer_id { + let new = { + let mut book = lock(&self.book); + let new = book.relays.get(&relay) != Some(&address); + if new && (book.relays.len() < MAX_PEERS || book.relays.contains_key(&relay)) { + book.relays.insert(relay, address.clone()); + } + book.remember(relay, address); + new + }; + if new { + let _ = self.commands.send(Command::CheckRelays); + } + } + return Some(relay.to_string()); + } let peer = split_peer(&address)?; if peer != self.peer_id { lock(&self.book).remember(peer, address); @@ -987,6 +1172,63 @@ mod tests { assert_eq!(from, a.local_id()); } + fn peer() -> PeerId { + Keypair::generate_ed25519().public().to_peer_id() + } + + #[test] + fn relays_are_picked_when_needed() { + use ::libp2p::autonat::NatStatus; + let local = peer(); + let offers: HashMap<PeerId, Multiaddr> = (0..4) + .map(|port| (peer(), format!("/memory/{port}").parse().unwrap())) + .collect(); + let check = |direct, nat: &NatStatus, using: &[PeerId]| { + relay_check(local, direct, nat, &offers, using) + }; + // Dialable, or not known yet with an address to dial: none. + assert!(check(true, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty()); + assert!(check(false, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty()); + assert!(check(true, &NatStatus::Unknown, &[]).is_empty()); + // No address to dial, or AutoNAT says it can't be dialled: some. + assert_eq!(check(false, &NatStatus::Unknown, &[]).len(), AUTO_RELAYS); + let picked = check(true, &NatStatus::Private, &[]); + assert_eq!(picked.len(), AUTO_RELAYS); + // One in use: the next in the same order. + let more = check(true, &NatStatus::Private, &[picked[0].0]); + assert_eq!(more.len(), AUTO_RELAYS - 1); + assert_eq!(more[0].0, picked[1].0); + let all: Vec<PeerId> = picked.iter().map(|(relay, _)| *relay).collect(); + assert!(check(true, &NatStatus::Private, &all).is_empty()); + // Never itself. + let only_us = HashMap::from([(local, Multiaddr::empty())]); + assert!(relay_check(local, false, &NatStatus::Private, &only_us, &[]).is_empty()); + } + + /// A relay says so in its addresses; b, which nobody can dial, learns of it (as from + /// the member list) and becomes reachable through it. + #[tokio::test] + async fn a_member_that_relays_is_picked() { + let (r, _) = memory_node_with(true, Vec::new(), true); + assert!(eventually(|| !r.local_addresses().is_empty()).await); + let offer = r + .local_addresses() + .into_iter() + .find(|address| address.ends_with("/p2p-circuit")) + .expect("the relay offers itself"); + let (b, _) = memory_node_with(false, Vec::new(), false); + assert_eq!(b.locate(&offer).as_deref(), Some(r.local_id())); + assert!( + eventually(|| b + .local_addresses() + .iter() + .any(|address| address.contains("/p2p-circuit/"))) + .await, + "b isn't reachable through r: {:?}", + b.local_addresses() + ); + } + #[tokio::test] async fn unknown_peers_are_refused() { let (b, _) = memory_node(); |