diff options
| -rw-r--r-- | src/transport/p2p.rs | 43 |
1 files changed, 37 insertions, 6 deletions
diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs index 3fa0bef..955c0f7 100644 --- a/src/transport/p2p.rs +++ b/src/transport/p2p.rs @@ -95,6 +95,10 @@ 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. +const RELAY_PAUSE: Duration = Duration::from_secs(600); /// How long a message of `bytes` may take on a stream. fn stream_timeout(bytes: usize) -> Duration { @@ -181,6 +185,8 @@ 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`). + failed_relays: HashMap<PeerId, tokio::time::Instant>, /// Peers found on the LAN (mDNS) not known yet to speak continuum. lan: std::collections::HashSet<PeerId>, /// Peers that speak continuum (identify says so). @@ -628,13 +634,23 @@ fn pick_relays( return; } let (direct, offers, public) = { - let book = lock(book); + let mut book = lock(book); + book.failed_relays + .retain(|_, at| at.elapsed() < RELAY_PAUSE); + let failed = book.failed_relays.clone(); + let usable = |offers: &HashMap<PeerId, Multiaddr>| -> HashMap<PeerId, Multiaddr> { + offers + .iter() + .filter(|(relay, _)| !failed.contains_key(relay)) + .map(|(relay, address)| (*relay, address.clone())) + .collect() + }; let public = if relaying.public { - book.public_relays.clone() + usable(&book.public_relays) } else { HashMap::new() }; - (!book.direct().is_empty(), book.relays.clone(), public) + (!book.direct().is_empty(), usable(&book.relays), public) }; let mut using: Vec<PeerId> = picked.values().copied().collect(); // Members first, then public ones for what's left. @@ -672,8 +688,9 @@ async fn run( ) { let serving = relaying.serving; let mut nat = ::libp2p::autonat::NatStatus::Unknown; - // Relays picked among the members, by listener. + // Relays picked among the members, by listener, and those that granted a slot. let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new(); + let mut granted: std::collections::HashSet<PeerId> = std::collections::HashSet::new(); let mut check = tokio::time::interval(RELAY_CHECK); check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP); @@ -840,9 +857,21 @@ 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. + // A relay picked among the members: the next check picks again, another + // one if this one never granted the slot. if let Some(relay) = picked.remove(&listener_id) { - debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying"); + if granted.remove(&relay) { + debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying"); + } else { + info!( + relay = %relay, + pause = ?RELAY_PAUSE, + "A relay didn't grant a slot (unreachable, or another peer at its address); picking others for a while" + ); + lock(&book) + .failed_relays + .insert(relay, tokio::time::Instant::now()); + } } } SwarmEvent::ListenerError { error, .. } => { @@ -851,6 +880,8 @@ async fn run( SwarmEvent::Behaviour(BehaviourEvent::RelayClient( ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. }, )) => { + granted.insert(relay_peer_id); + lock(&book).failed_relays.remove(&relay_peer_id); if !renewal { info!(relay = %relay_peer_id, "Reachable through a libp2p relay"); } |