diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/libp2p.rs | 285 |
1 files changed, 228 insertions, 57 deletions
diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs index d621c8e..9217c79 100644 --- a/src/transport/libp2p.rs +++ b/src/transport/libp2p.rs @@ -21,7 +21,9 @@ //! (`<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). +//! `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`), @@ -103,9 +105,13 @@ const MAX_PEERS: usize = 4096; /// 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; +/// Circuits through this node at once, and per peer (either end: a node behind the +/// relay is an end of every circuit to it). libp2p's rate limits on circuits and slots +/// (a burst, then one every minute or two, per peer and per IP address: every node +/// behind one NAT shares one) are left out: they're for public relays serving +/// strangers, and refused circuits used them up too. +const RELAY_CIRCUITS: usize = 1024; +const RELAY_CIRCUITS_PER_PEER: usize = 64; /// A configured relay that couldn't be reached to find out its peer ID is tried again /// after this. const RELAY_RETRY: Duration = Duration::from_secs(30); @@ -113,10 +119,16 @@ const RELAY_RETRY: Duration = Duration::from_secs(30); /// needs them. const AUTO_RELAYS: usize = 2; const RELAY_CHECK: Duration = Duration::from_secs(30); -/// A relay that lost the slot before granting it (it doesn't answer, or another peer -/// answers at its address: a relay restarted without its libp2p key) isn't picked again -/// for this long. +/// A relay that refused a slot (`relay_refused`: none left, not allowed, not a relay, +/// another peer at its address: a relay restarted without its libp2p key) isn't picked +/// again for this long. One that didn't grant it for another reason (it doesn't answer, +/// 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 @@ -248,12 +260,15 @@ struct Book { /// With `Libp2pConfig::public`: other peers that relay (identify says so), and /// where. public_relays: HashMap<PeerId, Multiaddr>, - /// Relays picked that never granted a slot, and when (`RELAY_PAUSE`). + /// Relays not to be picked, until when (`RELAY_PAUSE`, `RELAY_RETRY`). failed_relays: HashMap<PeerId, tokio::time::Instant>, /// Relays that granted us a slot we hold. slots: std::collections::HashSet<PeerId>, /// Peers holding a slot on this node, as a relay. slots_here: std::collections::HashSet<PeerId>, + /// Peers connected only through a relay, dialled directly (`upgrade`): when to again + /// at the soonest, and how long to wait after that. + upgrades: HashMap<PeerId, (tokio::time::Instant, Duration)>, /// Open connections: to whom, whether through a relay, and the remote address. connections: HashMap<ConnectionId, (PeerId, bool, Multiaddr)>, /// What AutoNAT says, and the address it confirmed while public. @@ -347,6 +362,22 @@ fn relay_failure_text(debug: &str, innermost: &str) -> String { innermost.to_owned() } +/// Whether a slot ended, or wasn't granted, because the relay refused it (its error's +/// debug form and innermost cause, as for `relay_failure_text`), rather than because it +/// didn't answer or the connection closed: only then is it worth not asking again for a +/// while. +fn relay_refused(debug: &str, innermost: &str) -> bool { + let refusals = [ + "NoAddressesInReservation", + "ResourceLimitExceeded", + "PermissionDenied", + "connection gated", + "UnsupportedProtocol", + "WrongPeerId", + ]; + refusals.iter().any(|pattern| debug.contains(pattern)) || innermost == "Failed" +} + /// The innermost of `error`'s causes, each distinct one once: those libp2p lists in a /// "Multiple dial errors occurred:" text (one per line, ` - <cause>: <its cause>...`), /// or the last of the source chain. @@ -459,7 +490,7 @@ impl Book { paused: self .failed_relays .values() - .filter(|at| at.elapsed() < RELAY_PAUSE) + .filter(|until| tokio::time::Instant::now() < **until) .count(), granted: self.counts.slots_granted, lost: self.counts.slots_lost, @@ -503,6 +534,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() @@ -924,6 +956,8 @@ impl Behaviour { max_circuits_per_peer: RELAY_CIRCUITS_PER_PEER, max_circuit_duration: RELAY_CIRCUIT_DURATION, max_circuit_bytes: RELAY_CIRCUIT_BYTES, + reservation_rate_limiters: Vec::new(), + circuit_src_rate_limiters: Vec::new(), ..Default::default() }, ) @@ -1080,8 +1114,8 @@ fn pick_relays( } let (direct, configured, listed, members, public) = { let mut book = lock(book); - book.failed_relays - .retain(|_, at| at.elapsed() < RELAY_PAUSE); + let now = tokio::time::Instant::now(); + book.failed_relays.retain(|_, until| now < *until); let failed = book.failed_relays.clone(); let usable = |offers: &HashMap<PeerId, Multiaddr>| -> HashMap<PeerId, Multiaddr> { offers @@ -1126,6 +1160,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<Behaviour>, book: &Mutex<Book>, peer: PeerId, local: PeerId) { + let direct: Vec<Multiaddr> = { + let mut book = lock(book); + if book.connection(&peer) != Some(Connection::Relayed) { + return; + } + let direct: Vec<Multiaddr> = 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<Behaviour>, peer: PeerId) { if let Some(dht) = swarm.behaviour_mut().kad.as_mut() { @@ -1210,6 +1288,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(); @@ -1285,6 +1364,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 { @@ -1363,31 +1445,50 @@ async fn run( // A relay picked: the next check picks again, another one if this one // never granted the slot. if let Some(relay) = picked.remove(&listener_id) { - let lost = { + let refused = match &reason { + Ok(()) => false, + Err(error) => relay_refused(&format!("{error:?}"), &innermost_causes(error)), + }; + let (lost, pause) = { let mut book = lock(&book); let lost = book.slots.remove(&relay); + let pause = match (refused, lost) { + (true, _) => Some(RELAY_PAUSE), + (false, true) => None, + (false, false) => Some(RELAY_RETRY), + }; if lost { book.counts.slots_lost += 1; } else { book.counts.slots_not_granted += 1; - book.failed_relays.insert(relay, tokio::time::Instant::now()); // Named without its peer ID: maybe another one now. if let Some(address) = book.find_again(relay) { let _ = again.send(Command::FindRelay { address }); } } - lost + if let Some(pause) = pause { + book.failed_relays.insert(relay, tokio::time::Instant::now() + pause); + } + (lost, pause) }; + // Again at once, or others meanwhile. + let _ = again.send(Command::CheckRelays); let reason = relay_failure(&reason); - if lost { - debug!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay"); - } else { - info!( - relay = %Addressed(SCHEME, &relay), - pause = ?RELAY_PAUSE, - reason = %reason, - "A relay didn't grant a slot; picking others for a while" - ); + match (lost, pause) { + (true, None) => { + info!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay; reserving again"); + } + (true, Some(pause)) => { + info!(relay = %Addressed(SCHEME, &relay), pause = ?pause, reason = %reason, "Lost the slot on a relay; picking others for a while"); + } + (false, pause) => { + info!( + relay = %Addressed(SCHEME, &relay), + pause = ?pause.unwrap_or_default(), + reason = %reason, + "A relay didn't grant a slot; picking others for a while" + ); + } } } } @@ -1456,43 +1557,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), @@ -1507,6 +1611,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 { @@ -2066,6 +2171,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() { @@ -2308,6 +2448,37 @@ mod tests { assert_eq!(relay_failure(&Ok(())), "closed"); } + /// Only refusals pause a relay; a slot that ended otherwise is asked for again. + #[test] + fn only_refusals_pause_relays() { + let refused = [ + ("Reservation(Protocol(NoAddressesInReservation))", "x"), + ("Reservation(Status(ResourceLimitExceeded))", "x"), + ("Reservation(Status(PermissionDenied))", "x"), + ("Denied { cause: \"connection gated\" }", "x"), + ("Upgrade(UnsupportedProtocol)", "x"), + ("Dial(WrongPeerId { obtained: … })", "x"), + ( + "Codec(Error(Custom { kind: Other, error: Failed }))", + "Failed", + ), + ]; + for (debug, innermost) in refused { + assert!(relay_refused(debug, innermost), "{debug}"); + } + let not = [ + ("Io(Custom { kind: Other, error: Timeout })", "Timeout"), + ( + "Io(Custom { kind: Other, error: Right(Closed) })", + "connection is closed", + ), + ("Io(…)", "connection reset"), + ]; + for (debug, innermost) in not { + assert!(!relay_refused(debug, innermost), "{debug}"); + } + } + /// libp2p's nested dial errors come out as each address's cause, once. #[test] fn dial_errors_are_short() { |