From 26a9ed2302b87bd3b1dc28ed02022b53c8bbd9ef Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 20:29:12 +0100 Subject: libp2p: upgrade relayed connections --- src/transport/libp2p.rs | 167 ++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 132 insertions(+), 35 deletions(-) (limited to 'src/transport/libp2p.rs') diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs index 060f614..008c7ec 100644 --- a/src/transport/libp2p.rs +++ b/src/transport/libp2p.rs @@ -21,7 +21,9 @@ //! (`/p2p-circuit/p2p/`) 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). +//! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send). Hole +//! punching needs both sides to listen; where only one does, the other dials it +//! directly once it knows where (`upgrade`). //! //! Of several connections to one peer (both dialled at once, or AutoNAT dialled back), //! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`), @@ -119,6 +121,10 @@ const RELAY_CHECK: Duration = Duration::from_secs(30); /// the connection closed) isn't for `RELAY_RETRY`; a slot lost is reserved again at /// once. const RELAY_PAUSE: Duration = Duration::from_secs(600); +/// A peer connected only through a relay is dialled directly where it can be reached +/// (`upgrade`): first at once, then after a failure no sooner than this, doubling up to +/// `RELAY_PAUSE`. +const UPGRADE_RETRY: Duration = Duration::from_secs(30); /// A peer nobody mentioned for this long (the member list, identify, mDNS, a /// connection) is forgotten, unless connected: mostly old peer IDs of nodes restarted /// without their libp2p key, whose new ones the member list has since. Members are @@ -256,6 +262,9 @@ struct Book { slots: std::collections::HashSet, /// Peers holding a slot on this node, as a relay. slots_here: std::collections::HashSet, + /// Peers connected only through a relay, dialled directly (`upgrade`): when to again + /// at the soonest, and how long to wait after that. + upgrades: HashMap, /// Open connections: to whom, whether through a relay, and the remote address. connections: HashMap, /// What AutoNAT says, and the address it confirmed while public. @@ -521,6 +530,7 @@ impl Book { self.continuum.remove(peer); self.lan.remove(peer); self.failed_relays.remove(peer); + self.upgrades.remove(peer); self.seen.remove(peer); } stale.len() @@ -1144,6 +1154,50 @@ fn pick_relays( } } +/// Dials `peer` directly if it's connected only through a relay and is known elsewhere +/// too (the member list, identify, mDNS), since the first connection, made through a +/// relay (a relayed seed), would otherwise be kept for good: a peer connected isn't +/// dialled again, and hole punching needs both sides to listen. The direct connection, +/// once open, wins over the relayed one (`Book::duplicates`). +fn upgrade(swarm: &mut Swarm, book: &Mutex, peer: PeerId, local: PeerId) { + let direct: Vec = { + let mut book = lock(book); + if book.connection(&peer) != Some(Connection::Relayed) { + return; + } + let direct: Vec = book + .addresses + .get(&peer) + .map(|known| dialable(known, local)) + .unwrap_or_default() + .into_iter() + .filter(|address| !is_relayed(address)) + .collect(); + if direct.is_empty() { + return; + } + let now = tokio::time::Instant::now(); + let (due, wait) = book + .upgrades + .get(&peer) + .copied() + .unwrap_or((now, UPGRADE_RETRY)); + if now < due { + return; + } + book.upgrades + .insert(peer, (now + wait, (wait * 2).min(RELAY_PAUSE))); + direct + }; + debug!(peer = %Addressed(SCHEME, &peer), addresses = ?direct, "Connected only through a relay; dialling directly"); + let _ = swarm.dial( + DialOpts::peer_id(peer) + .addresses(direct) + .condition(PeerCondition::NotDialing) + .build(), + ); +} + /// Asks the DHT (with `--libp2p-public`) where `peer` is, for the next attempt. fn find_on_dht(swarm: &mut Swarm, peer: PeerId) { if let Some(dht) = swarm.behaviour_mut().kad.as_mut() { @@ -1228,6 +1282,7 @@ async fn run( Some(Command::Connect { peer, addresses, done }) => { if swarm.is_connected(&peer) { let _ = done.send(Ok(())); + upgrade(&mut swarm, &book, peer, relaying.local); continue; } let waiting = connecting.entry(peer).or_default(); @@ -1303,6 +1358,9 @@ async fn run( connection_id, (peer_id, relayed, endpoint.get_remote_address().clone()), ); + if !relayed { + book.upgrades.remove(&peer_id); + } if relaying.local < peer_id && book.connections.values().filter(|(connected, _, _)| *connected == peer_id).count() > 1 { @@ -1493,43 +1551,46 @@ async fn run( SwarmEvent::Behaviour(BehaviourEvent::Identify( ::libp2p::identify::Event::Received { connection_id, peer_id, info }, )) => { - let mut book = lock(&book); - let listen = published(&info.listen_addrs); - // Every new connection identifies again: logged when it says something new. - if book.knows(peer_id, &listen) { - trace!(peer = %Addressed(SCHEME, &peer_id), connection = %connection_id, "Peer identified again"); - } else { - debug!( - peer = %Addressed(SCHEME, &peer_id), - connection = %connection_id, - listen = ?info.listen_addrs, - observed = %info.observed_addr, - agent = %info.agent_version, - "Peer identified" - ); - } - // Last first: `remember` puts each in front. - for address in listen.into_iter().rev() { - book.remember(peer_id, address); - } - if relaying.public - && info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME) - && (book.public_relays.len() < MAX_PEERS - || book.public_relays.contains_key(&peer_id)) - && let Some(address) = book - .addresses - .get(&peer_id) - .and_then(|known| known.first()) - .and_then(|address| address.clone().with_p2p(peer_id).ok()) { - book.public_relays.insert(peer_id, address); - } - if info.protocols.contains(&PROTOCOL) { - book.continuum.insert(peer_id); - if book.lan.remove(&peer_id) { - book.discover(peer_id); + let mut book = lock(&book); + let listen = published(&info.listen_addrs); + // Every new connection identifies again: logged when it says something new. + if book.knows(peer_id, &listen) { + trace!(peer = %Addressed(SCHEME, &peer_id), connection = %connection_id, "Peer identified again"); + } else { + debug!( + peer = %Addressed(SCHEME, &peer_id), + connection = %connection_id, + listen = ?info.listen_addrs, + observed = %info.observed_addr, + agent = %info.agent_version, + "Peer identified" + ); + } + // Last first: `remember` puts each in front. + for address in listen.into_iter().rev() { + book.remember(peer_id, address); + } + if relaying.public + && info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME) + && (book.public_relays.len() < MAX_PEERS + || book.public_relays.contains_key(&peer_id)) + && let Some(address) = book + .addresses + .get(&peer_id) + .and_then(|known| known.first()) + .and_then(|address| address.clone().with_p2p(peer_id).ok()) + { + book.public_relays.insert(peer_id, address); + } + if info.protocols.contains(&PROTOCOL) { + book.continuum.insert(peer_id); + if book.lan.remove(&peer_id) { + book.discover(peer_id); + } } } + upgrade(&mut swarm, &book, peer_id, relaying.local); } SwarmEvent::Behaviour(BehaviourEvent::Mdns( ::libp2p::mdns::Event::Discovered(found), @@ -1544,6 +1605,7 @@ async fn run( book.remember(peer, address.clone()); book.continuum.contains(&peer) }; + upgrade(&mut swarm, &book, peer, relaying.local); if continuum { lock(&book).discover(peer); } else if lock(&book).lan.len() < MAX_PEERS { @@ -2103,6 +2165,41 @@ mod tests { assert!(a.connection(b.local_id()).is_some()); } + /// b listens nowhere; a, which can be dialled, reaches it through the relay r: b + /// then dials a directly (identify told it where), and both keep that. + #[tokio::test] + async fn a_relayed_connection_is_upgraded_where_one_side_can_be_dialled() { + 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 (a, _) = memory_node(); + let relayed = b.local_addresses()[0].clone(); + assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id()); + a.send_message(b.local_id(), &status()).await.unwrap(); + at_b.recv().await.unwrap(); + assert!( + eventually(|| b.connection(a.local_id()) == Some(Connection::Direct)).await, + "b didn't dial a directly" + ); + assert_eq!(a.connection(b.local_id()), Some(Connection::Direct)); + // The relayed one goes (`DUPLICATE_GRACE`); the direct one stays. + let deadline = tokio::time::Instant::now() + DUPLICATE_GRACE * 2; + while a.stats().libp2p.unwrap().connections.relayed > 0 { + assert!( + tokio::time::Instant::now() < deadline, + "the relayed connection stayed" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + b.send_message(a.local_id(), &status()).await.unwrap(); + assert_eq!(a.connection(b.local_id()), Some(Connection::Direct)); + } + /// A relay that is no node (`--relay-only`) relays, and takes no messages itself. #[tokio::test] async fn a_relay_that_is_no_node() { -- cgit v1.3.1