diff options
Diffstat (limited to 'src/transport/p2p.rs')
| -rw-r--r-- | src/transport/p2p.rs | 252 |
1 files changed, 232 insertions, 20 deletions
diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs index 3023fb2..3464a68 100644 --- a/src/transport/p2p.rs +++ b/src/transport/p2p.rs @@ -13,9 +13,17 @@ //! reached (`local_addresses`): the addresses it announces (`--libp2p-announce`, e.g. a //! public address forwarded to it), or else those it listens on, loopback left out. //! +//! A node nobody can dial (behind NAT, or a closed port) can still be reached through a +//! relay: a node of the swarm that relays (`--libp2p-relay`), on which it reserves a slot +//! (`--libp2p-relays`); its address through the relay (`<relay>/p2p-circuit/p2p/<it>`) +//! is published with the others. Over a relayed connection, both sides then try to +//! connect directly (libp2p's DCUtR, hole punching); the relay carries the traffic until +//! 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). +//! //! What this transport doesn't do yet (`docs/future-work.md`): find peers on the LAN -//! (mDNS), or reach nodes behind NAT (relays, hole punching). It never uses public -//! libp2p infrastructure. +//! (mDNS), or pick relays among the members by itself. It never uses public libp2p +//! infrastructure: relays are only those configured. use std::collections::HashMap; use std::path::Path; @@ -55,6 +63,19 @@ const READS: usize = 32; const ADDRESSES_PER_PEER: usize = 8; const MAX_PEERS: usize = 4096; +/// A relayed connection (circuit) through this node lasts at most this long and carries +/// at most this much: generous, since the swarm's own relays may carry sync traffic for +/// good where hole punching fails (libp2p's defaults, 2 minutes and 128 KiB, are for +/// the punching only). +const RELAY_CIRCUIT_DURATION: Duration = Duration::from_secs(3600); +const RELAY_CIRCUIT_BYTES: u64 = 1 << 30; +/// Circuits through this node at once, and per peer. +const RELAY_CIRCUITS: usize = 64; +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); + /// How long a message of `bytes` may take on a stream. fn stream_timeout(bytes: usize) -> Duration { STREAM_TIMEOUT + Duration::from_millis(bytes as u64 * 1000 / MIN_RATE) @@ -67,6 +88,11 @@ pub(crate) struct Libp2pConfig { pub(crate) listen: Vec<Multiaddr>, /// Multiaddrs to publish instead of the listen addresses, if any. pub(crate) announce: Vec<Multiaddr>, + /// Relays to reserve a slot on (multiaddrs; without the relay's peer ID, a connection + /// finds it out). + pub(crate) relays: Vec<Multiaddr>, + /// Relay for other nodes. + pub(crate) relay: bool, /// libp2p's in-memory transport instead of TCP and QUIC: for tests. pub(crate) memory: bool, } @@ -106,6 +132,13 @@ enum Command { address: Multiaddr, done: oneshot::Sender<Result<PeerId, String>>, }, + /// 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). + Reserve { + relay: Multiaddr, + configured: Multiaddr, + }, } /// Where peers are, and where we listen. @@ -160,6 +193,13 @@ impl Book { } } +/// Whether `address` goes through a relay. +fn is_relayed(address: &Multiaddr) -> bool { + address + .iter() + .any(|protocol| matches!(protocol, Protocol::P2pCircuit)) +} + /// `address` with the peer ID it ends in, if it ends in one. fn split_peer(address: &Multiaddr) -> Option<PeerId> { match address.iter().last() { @@ -187,9 +227,9 @@ impl Libp2pTransport { pub(crate) fn start(config: Libp2pConfig) -> Result<Self> { let peer_id = config.key.public().to_peer_id(); let mut swarm = if config.memory { - memory_swarm(config.key)? + memory_swarm(config.key, config.relay)? } else { - network_swarm(config.key)? + network_swarm(config.key, config.relay)? }; for address in config.listen { swarm @@ -209,8 +249,21 @@ impl Libp2pTransport { ..Book::default() })); let (commands, receiver) = mpsc::unbounded_channel(); + for relay in config.relays { + let _ = commands.send(Command::Reserve { + configured: relay.clone(), + relay, + }); + } let (stopping, stopped) = watch::channel(false); - tokio::spawn(run(swarm, receiver, Arc::clone(&book), stopping)); + tokio::spawn(run( + swarm, + receiver, + commands.clone(), + Arc::clone(&book), + config.relay, + stopping, + )); info!(peer_id = %peer_id, "libp2p transport started"); Ok(Self { local_id: peer_id.to_string(), @@ -254,20 +307,40 @@ const IDENTIFY_VERSION: &str = "/continuum/1"; struct Behaviour { stream: libp2p_stream::Behaviour, identify: ::libp2p::identify::Behaviour, + relay_client: ::libp2p::relay::client::Behaviour, + /// Only with `Libp2pConfig::relay`. + relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>, + dcutr: ::libp2p::dcutr::Behaviour, } impl Behaviour { - fn new(key: &Keypair) -> Self { - let config = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public()) + fn new(key: &Keypair, relay_client: ::libp2p::relay::client::Behaviour, relay: bool) -> Self { + let peer_id = key.public().to_peer_id(); + let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public()) .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION"))); + let relay = relay.then(|| { + ::libp2p::relay::Behaviour::new( + peer_id, + ::libp2p::relay::Config { + max_circuits: RELAY_CIRCUITS, + max_circuits_per_peer: RELAY_CIRCUITS_PER_PEER, + max_circuit_duration: RELAY_CIRCUIT_DURATION, + max_circuit_bytes: RELAY_CIRCUIT_BYTES, + ..Default::default() + }, + ) + }); Self { stream: libp2p_stream::Behaviour::new(), - identify: ::libp2p::identify::Behaviour::new(config), + identify: ::libp2p::identify::Behaviour::new(identify), + relay_client, + relay: relay.into(), + dcutr: ::libp2p::dcutr::Behaviour::new(peer_id), } } } -fn network_swarm(key: Keypair) -> Result<Swarm<Behaviour>> { +fn network_swarm(key: Keypair, relay: bool) -> Result<Swarm<Behaviour>> { Ok(::libp2p::SwarmBuilder::with_existing_identity(key) .with_tokio() .with_tcp( @@ -277,18 +350,24 @@ fn network_swarm(key: Keypair) -> Result<Swarm<Behaviour>> { )? .with_quic() .with_dns()? - .with_behaviour(Behaviour::new)? + .with_relay_client( + ::libp2p::noise::Config::new, + ::libp2p::yamux::Config::default, + )? + .with_behaviour(|key, relay_client| Behaviour::new(key, relay_client, relay))? .with_swarm_config(|config| config.with_idle_connection_timeout(IDLE_TIMEOUT)) .build()) } -fn memory_swarm(key: Keypair) -> Result<Swarm<Behaviour>> { +fn memory_swarm(key: Keypair, relay: bool) -> Result<Swarm<Behaviour>> { use ::libp2p::Transport as _; - use ::libp2p::core::transport::MemoryTransport; + use ::libp2p::core::transport::{MemoryTransport, OrTransport}; use ::libp2p::core::upgrade::Version; let peer_id = key.public().to_peer_id(); - let transport = MemoryTransport::default() + let (relay_transport, relay_client) = ::libp2p::relay::client::new(peer_id); + let transport = OrTransport::new(relay_transport, MemoryTransport::default()) + .boxed() .upgrade(Version::V1) .authenticate(::libp2p::noise::Config::new(&key)?) .multiplex(::libp2p::yamux::Config::default()) @@ -296,7 +375,7 @@ fn memory_swarm(key: Keypair) -> Result<Swarm<Behaviour>> { .boxed(); Ok(Swarm::new( transport, - Behaviour::new(&key), + Behaviour::new(&key, relay_client, relay), peer_id, ::libp2p::swarm::Config::with_tokio_executor().with_idle_connection_timeout(IDLE_TIMEOUT), )) @@ -304,12 +383,28 @@ fn memory_swarm(key: Keypair) -> 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. async fn run( mut swarm: Swarm<Behaviour>, mut commands: mpsc::UnboundedReceiver<Command>, + again: mpsc::UnboundedSender<Command>, book: Arc<Mutex<Book>>, + serving: bool, stopping: watch::Sender<bool>, ) { + // Relays we listen through, by listener, as configured. + let mut reservations: HashMap<::libp2p::core::transport::ListenerId, Multiaddr> = + HashMap::new(); + let reserve_later = |configured: Multiaddr| { + let again = again.clone(); + tokio::spawn(async move { + tokio::time::sleep(RELAY_RETRY).await; + let _ = again.send(Command::Reserve { + relay: configured.clone(), + configured, + }); + }); + }; let mut connecting: HashMap<PeerId, Vec<oneshot::Sender<Result<(), String>>>> = HashMap::new(); let mut identifying: HashMap< ConnectionId, @@ -339,6 +434,43 @@ async fn run( } } } + Some(Command::Reserve { relay, configured }) if split_peer(&relay).is_none() => { + // Which peer relays there first. + let again = again.clone(); + tokio::spawn(async move { + let (done, identified) = oneshot::channel(); + if again + .send(Command::Identify { address: relay.clone(), done }) + .is_err() + { + return; + } + match tokio::time::timeout(CONNECT_TIMEOUT, identified).await { + Ok(Ok(Ok(peer))) => { + if let Ok(relay) = relay.with_p2p(peer) { + let _ = again.send(Command::Reserve { relay, configured }); + } + } + _ => { + warn!(relay = %relay, "Couldn't reach a libp2p relay; trying again later"); + tokio::time::sleep(RELAY_RETRY).await; + let _ = again.send(Command::Reserve { relay, configured }); + } + } + }); + } + Some(Command::Reserve { relay, configured }) => { + match swarm.listen_on(relay.clone().with(Protocol::P2pCircuit)) { + Ok(listener) => { + debug!(relay = %relay, "Reserving a slot on a libp2p relay"); + reservations.insert(listener, configured); + } + Err(error) => { + warn!(relay = %relay, error = %error, "Couldn't use a libp2p relay; trying again later"); + reserve_later(configured); + } + } + } Some(Command::Identify { address, done }) => { let opts = DialOpts::unknown_peer_id().address(address.clone()).build(); let id = opts.connection_id(); @@ -376,6 +508,10 @@ async fn run( } SwarmEvent::NewListenAddr { address, .. } => { info!(address = %address, "libp2p listening"); + // A relay tells those reserving where it is: its external addresses. + if serving && !is_local(&address) && !is_relayed(&address) { + swarm.add_external_address(address.clone()); + } let mut book = lock(&book); if !book.listening.contains(&address) { book.listening.push(address); @@ -383,11 +519,37 @@ async fn run( } SwarmEvent::ExpiredListenAddr { address, .. } => { info!(address = %address, "libp2p no longer listening"); + if serving { + swarm.remove_external_address(&address); + } lock(&book).listening.retain(|known| *known != address); } + SwarmEvent::ListenerClosed { listener_id, addresses, reason } => { + lock(&book) + .listening + .retain(|known| !addresses.contains(known)); + if let Some(relay) = reservations.remove(&listener_id) { + warn!(relay = %relay, reason = ?reason, "Lost the slot on a libp2p relay; reserving again later"); + reserve_later(relay); + } + } SwarmEvent::ListenerError { error, .. } => { warn!(error = %error, "libp2p listener failed"); } + SwarmEvent::Behaviour(BehaviourEvent::RelayClient( + ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. }, + )) => { + if !renewal { + info!(relay = %relay_peer_id, "Reachable through a libp2p relay"); + } + } + SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => { + debug!(event = ?event, "libp2p relay"); + } + SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result { + Ok(_) => info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay"), + Err(error) => debug!(peer = %event.remote_peer_id, error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on"), + }, SwarmEvent::Behaviour(BehaviourEvent::Identify( ::libp2p::identify::Event::Received { peer_id, info, .. }, )) => { @@ -449,14 +611,17 @@ impl Transport for Libp2pTransport { } /// Where we can be reached, each with our peer ID: the announced addresses, or else - /// those we listen on (`published`). + /// those we listen on (`published`), and those through relays. fn local_addresses(&self) -> Vec<String> { let book = lock(&self.book); - let addresses = if book.announced.is_empty() { - published(&book.listening) + let (relayed, direct): (Vec<Multiaddr>, Vec<Multiaddr>) = + book.listening.iter().cloned().partition(is_relayed); + let mut addresses = if book.announced.is_empty() { + published(&direct) } else { book.announced.clone() }; + addresses.extend(relayed); addresses .into_iter() .filter_map(|address| address.with_p2p(self.peer_id).ok()) @@ -609,19 +774,40 @@ mod tests { /// Memory ports are process-wide. static NEXT_PORT: AtomicU64 = AtomicU64::new(40_000); - fn memory_node() -> (Libp2pTransport, Multiaddr) { + /// A node on the in-memory transport, listening on a port of its own (or nowhere), + /// and its address with its peer ID (or just its peer ID). + fn memory_node_with( + listen: bool, + relays: Vec<Multiaddr>, + relay: bool, + ) -> (Libp2pTransport, Multiaddr) { let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed); let key = Keypair::generate_ed25519(); let address: Multiaddr = format!("/memory/{port}").parse().unwrap(); let peer = key.public().to_peer_id(); let transport = Libp2pTransport::start(Libp2pConfig { key, - listen: vec![address.clone()], + listen: if listen { + vec![address.clone()] + } else { + Vec::new() + }, announce: Vec::new(), + relays, + relay, memory: true, }) .unwrap(); - (transport, address.with_p2p(peer).unwrap()) + let address = if listen { + address.with_p2p(peer).unwrap() + } else { + Multiaddr::empty().with(Protocol::P2p(peer)) + }; + (transport, address) + } + + fn memory_node() -> (Libp2pTransport, Multiaddr) { + memory_node_with(true, Vec::new(), false) } fn ping() -> ParsedMessage { @@ -744,6 +930,8 @@ mod tests { key, listen: vec![format!("/memory/{port}").parse().unwrap()], announce: vec![announced.clone()], + relays: Vec::new(), + relay: false, memory: true, }) .unwrap(); @@ -775,6 +963,30 @@ mod tests { ); } + /// b listens nowhere but through the relay r; a reaches it there. + #[tokio::test] + async fn a_node_nobody_can_dial_is_reached_through_a_relay() { + let (_r, r_address) = memory_node_with(true, Vec::new(), true); + let (b, _) = memory_node_with(false, vec![r_address.clone()], false); + let b = Arc::new(b); + let mut at_b = received(Arc::clone(&b)); + assert!( + eventually(|| !b.local_addresses().is_empty()).await, + "b got no slot on the relay" + ); + let relayed = b.local_addresses()[0].clone(); + assert!( + relayed.starts_with(&format!("{r_address}/p2p-circuit/")), + "{relayed}" + ); + + let (a, _) = memory_node(); + assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id()); + a.send_message(b.local_id(), &ping()).await.unwrap(); + let (from, _) = at_b.recv().await.unwrap(); + assert_eq!(from, a.local_id()); + } + #[tokio::test] async fn unknown_peers_are_refused() { let (b, _) = memory_node(); |