diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/libp2p/book.rs | 22 | ||||
| -rw-r--r-- | src/transport/libp2p/driver.rs | 28 | ||||
| -rw-r--r-- | src/transport/libp2p/mod.rs | 14 | ||||
| -rw-r--r-- | src/transport/libp2p/tests.rs | 14 |
4 files changed, 52 insertions, 26 deletions
diff --git a/src/transport/libp2p/book.rs b/src/transport/libp2p/book.rs index 7de4be7..f466aef 100644 --- a/src/transport/libp2p/book.rs +++ b/src/transport/libp2p/book.rs @@ -70,6 +70,9 @@ pub(super) struct Book { pub(super) failed_relays: HashMap<PeerId, tokio::time::Instant>, /// Relays that granted us a slot we hold. pub(super) slots: std::collections::HashSet<PeerId>, + /// Relays we hold a slot on whose duplicate connections we closed, and when + /// (`duplicates`): a slot lost then moves to the connection kept, it isn't lost. + pub(super) rehoming: HashMap<PeerId, tokio::time::Instant>, /// Peers holding a slot on this node, as a relay. pub(super) slots_here: std::collections::HashSet<PeerId>, /// Peers connected only through a relay, dialled directly (`upgrade`): when to again @@ -117,17 +120,16 @@ impl Book { book_connection(&self.connections, peer) } - /// The connections to `peer` to close, of several: all but one, the oldest direct - /// one (the oldest relayed one if none is direct). + /// The connections to `peer` to close, of several: all but the one to keep + /// (`ranked`). /// - /// None between a relay and a node with a slot on it, or reserving one (`reserving`): - /// a slot lives on one connection, which libp2p doesn't tell, and closing it would - /// end the slot, and with it every circuit to the node (the relay then answers - /// `NoReservation`). - pub(super) fn duplicates(&self, peer: &PeerId, reserving: bool) -> Vec<ConnectionId> { - if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) { - return Vec::new(); - } + /// Between a relay and a node with a slot on it too: messages go over any of the + /// connections (libp2p-stream picks one at random for each), so a slower one kept + /// would carry half of them. A slot lives on one connection, which libp2p doesn't + /// tell; if that one closes, the slot ends, and the node reserves it again at once, + /// over the connection kept (`Book::rehoming`, not counted as lost); circuits + /// through it end, and are made again with the next message. + pub(super) fn duplicates(&self, peer: &PeerId) -> Vec<ConnectionId> { self.ranked(peer).into_iter().skip(1).collect() } diff --git a/src/transport/libp2p/driver.rs b/src/transport/libp2p/driver.rs index 9b4259f..a0434a6 100644 --- a/src/transport/libp2p/driver.rs +++ b/src/transport/libp2p/driver.rs @@ -68,7 +68,14 @@ pub(super) async fn run( for peer in due { dedupe_due.remove(&peer); let reserving = picked.values().any(|relay| *relay == peer); - let duplicates = lock(&book).duplicates(&peer, reserving); + let duplicates = { + let mut book = lock(&book); + let duplicates = book.duplicates(&peer); + if !duplicates.is_empty() && (reserving || book.slots.contains(&peer)) { + book.rehoming.insert(peer, now); + } + duplicates + }; for connection in duplicates { if swarm.close_connection(connection) { closing_duplicates.insert(connection); @@ -305,16 +312,25 @@ pub(super) async fn run( Ok(()) => false, Err(error) => relay_refused(&format!("{error:?}"), &innermost_causes(error)), }; - let (lost, pause) = { + let (lost, pause, rehomed) = { let mut book = lock(&book); let lost = book.slots.remove(&relay); + // Its connection closed as a duplicate (`duplicates`): again + // at once, over the one kept. + let rehomed = lost + && book.rehoming.remove(&relay).is_some_and(|at| { + tokio::time::Instant::now().saturating_duration_since(at) + < REHOMING_WITHIN + }); let pause = match (refused, lost) { (true, _) => Some(RELAY_PAUSE), (false, true) => None, (false, false) => Some(RELAY_RETRY), }; if lost { - book.counts.slots_lost += 1; + if !rehomed { + book.counts.slots_lost += 1; + } } else { book.counts.slots_not_granted += 1; // Named without its peer ID: maybe another one now. @@ -325,11 +341,15 @@ pub(super) async fn run( if let Some(pause) = pause { book.failed_relays.insert(relay, tokio::time::Instant::now() + pause); } - (lost, pause) + (lost, pause, rehomed) }; // Again at once, or others meanwhile. let _ = again.send(Command::CheckRelays); let reason = relay_failure(&reason); + if rehomed { + debug!(relay = %Addressed(SCHEME, &relay), "A duplicate connection to a relay closed with our slot on it; reserving again over the one kept"); + continue; + } match (lost, pause) { (true, None) => { info!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay; reserving again"); diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs index a4fefc3..9d1dcc7 100644 --- a/src/transport/libp2p/mod.rs +++ b/src/transport/libp2p/mod.rs @@ -32,11 +32,12 @@ //! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`), //! by the side with the smaller peer ID: it keeps the direct one with the shortest round //! trip (libp2p's ping, on each connection), so a path over the LAN wins over one -//! through a VPN, whichever opened first. A peer connected directly is dialled again -//! where it might be reached better (`explore`): at addresses it tells of later (a new -//! interface: identify pushes them at once), and at all of them when this node gains -//! an interface. Otherwise a connection made over the WLAN would stay there once -//! ethernet comes. +//! through a VPN, whichever opened first; between a relay and a node with a slot on it +//! too, where a slot on a connection closed is reserved again over the one kept. A +//! peer connected directly is dialled again where it might be reached better +//! (`explore`): at addresses it tells of later (a new interface: identify pushes them +//! at once), and at all of them when this node gains an interface. Otherwise a +//! connection made over the WLAN would stay there once ethernet comes. //! //! 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 @@ -187,6 +188,9 @@ const DUPLICATE_GRACE: Duration = Duration::from_secs(10); /// a hole punch under way (`PUNCH_RETRIES`), which takes its tries (up to a QUIC /// handshake's 5 s each, three times) on that connection. const PUNCH_GRACE: Duration = Duration::from_secs(60); +/// A slot lost this soon after its relay's duplicate connections were closed moved to +/// the one kept (`Book::rehoming`). +const REHOMING_WITHIN: Duration = Duration::from_secs(10); /// After libp2p's hole punching gave up on a peer (three tries, as the relayed /// connection opened), it tries again after these, the last one over and over: what /// failed may have been passing (a NAT's mapping, a busy link). Each time through a new diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs index ed23b34..7311ebf 100644 --- a/src/transport/libp2p/tests.rs +++ b/src/transport/libp2p/tests.rs @@ -898,20 +898,20 @@ fn a_direct_connection_wins_over_a_relayed_one() { // Of several, the oldest direct one stays; a relayed one only without a direct. connect(&mut book, both, false); assert_eq!( - book.duplicates(&both, false), + book.duplicates(&both), [ ConnectionId::new_unchecked(4), ConnectionId::new_unchecked(2) ] ); - assert!(book.duplicates(&relayed_only, false).is_empty()); - // Not while one of them carries a slot, ours or the peer's, or we reserve one. - assert!(book.duplicates(&both, true).is_empty()); + assert!(book.duplicates(&relayed_only).is_empty()); + // Also where one of them carries a slot, ours or the peer's: messages would go over + // either. book.slots.insert(both); - assert!(book.duplicates(&both, false).is_empty()); + assert_eq!(book.duplicates(&both).len(), 2); book.slots.clear(); book.slots_here.insert(both); - assert!(book.duplicates(&both, false).is_empty()); + assert_eq!(book.duplicates(&both).len(), 2); book.slots_here.clear(); // Once measured, the fastest direct one stays, however young. book.rtts @@ -919,7 +919,7 @@ fn a_direct_connection_wins_over_a_relayed_one() { book.rtts .insert(ConnectionId::new_unchecked(4), Duration::from_millis(2)); assert_eq!( - book.duplicates(&both, false), + book.duplicates(&both), [ ConnectionId::new_unchecked(3), ConnectionId::new_unchecked(2) |