libp2p: keep fastest connection

1 files changed, 67 insertions, 19 deletions

diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index b5cee5f..9f83295 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -27,7 +27,9 @@
 //!
 //! 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`),
-//! by the side with the smaller peer ID.
+//! 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 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
@@ -136,8 +138,9 @@ const UPGRADE_RETRY: Duration = Duration::from_secs(30);
 const STALE_AFTER: Duration = Duration::from_secs(600);
 const PRUNE_EVERY: Duration = Duration::from_secs(60);
 /// Several connections to one peer (both dialled at once, or AutoNAT dialled back to
-/// see whether it can) are cut to one (`Book::duplicates`) this long after the last one
-/// opened: long enough for what it was dialled for. Looked for every `DEDUPE_EVERY`.
+/// see whether it can) are cut to one, the fastest (`Book::duplicates`), this long after
+/// the last one opened: long enough for what it was dialled for, and for each to be
+/// pinged (at once, as it opens). Looked for every `DEDUPE_EVERY`.
 const DUPLICATE_GRACE: Duration = Duration::from_secs(10);
 const DEDUPE_EVERY: Duration = Duration::from_secs(5);
 
@@ -269,6 +272,8 @@ struct Book {
     /// 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)>,
+    /// The shortest round trip measured on each open connection (libp2p's ping).
+    rtts: HashMap<ConnectionId, 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.
@@ -450,29 +455,30 @@ impl Book {
         if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) {
             return Vec::new();
         }
-        let mut connections: Vec<(bool, ConnectionId)> = self
+        self.ranked(peer).into_iter().skip(1).collect()
+    }
+
+    /// The connections to `peer`, the one to keep first: direct ones before relayed
+    /// ones, measured ones by their round trip (`rtts`: over the LAN before a VPN's hub,
+    /// whichever opened first), then the oldest.
+    fn ranked(&self, peer: &PeerId) -> Vec<ConnectionId> {
+        let mut connections: Vec<(bool, Duration, ConnectionId)> = self
             .connections
             .iter()
             .filter(|(_, (connected, _, _))| connected == peer)
-            .map(|(id, (_, relayed, _))| (*relayed, *id))
+            .map(|(id, (_, relayed, _))| {
+                let rtt = self.rtts.get(id).copied().unwrap_or(Duration::MAX);
+                (*relayed, rtt, *id)
+            })
             .collect();
-        // Direct (false) first, then the oldest.
         connections.sort();
-        connections.into_iter().skip(1).map(|(_, id)| id).collect()
+        connections.into_iter().map(|(_, _, id)| id).collect()
     }
 
-    /// The remote address of a connection to `peer`: a direct one if there is one.
+    /// The remote address of the connection to `peer` that is kept (`ranked`).
     fn location(&self, peer: &PeerId) -> Option<&Multiaddr> {
-        let mut found = None;
-        for (connected, relayed, address) in self.connections.values() {
-            if connected == peer {
-                if !relayed {
-                    return Some(address);
-                }
-                found = Some(address);
-            }
-        }
-        found
+        let kept = *self.ranked(peer).first()?;
+        self.connections.get(&kept).map(|(_, _, address)| address)
     }
 
     fn stats(&self) -> Libp2pStats {
@@ -920,6 +926,9 @@ struct Behaviour {
     /// Only with `Libp2pConfig::relay`.
     relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>,
     dcutr: ::libp2p::dcutr::Behaviour,
+    /// Each connection's round trip (`Book::rtts`): of several to one peer, the fastest
+    /// is kept.
+    ping: ::libp2p::ping::Behaviour,
     autonat: ::libp2p::autonat::Behaviour,
     /// Only with `Libp2pConfig::mdns`.
     mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>,
@@ -1013,6 +1022,7 @@ impl Behaviour {
             relay_client,
             relay: relay.into(),
             dcutr: ::libp2p::dcutr::Behaviour::new(peer_id),
+            ping: ::libp2p::ping::Behaviour::new(::libp2p::ping::Config::new()),
             autonat: ::libp2p::autonat::Behaviour::new(
                 peer_id,
                 ::libp2p::autonat::Config {
@@ -1428,7 +1438,11 @@ async fn run(
                     }
                 }
                 SwarmEvent::ConnectionClosed { peer_id, connection_id, cause, .. } => {
-                    lock(&book).connections.remove(&connection_id);
+                    {
+                        let mut book = lock(&book);
+                        book.connections.remove(&connection_id);
+                        book.rtts.remove(&connection_id);
+                    }
                     let cause = match cause {
                         _ if closing_duplicates.remove(&connection_id) => "a duplicate".to_owned(),
                         // By us, or idle: libp2p doesn't tell which.
@@ -1582,6 +1596,23 @@ async fn run(
                     }
                     debug!(event = ?event, "Relay event");
                 }
+                SwarmEvent::Behaviour(BehaviourEvent::Ping(::libp2p::ping::Event {
+                    peer,
+                    connection,
+                    result,
+                })) => match result {
+                    Ok(rtt) => {
+                        let mut book = lock(&book);
+                        if book.connections.contains_key(&connection) {
+                            let shortest = book.rtts.entry(connection).or_insert(rtt);
+                            *shortest = (*shortest).min(rtt);
+                        }
+                        trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, rtt = ?rtt, "Connection's round trip");
+                    }
+                    Err(error) => {
+                        trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, error = %error, "Connection's ping failed");
+                    }
+                },
                 SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
                     Ok(_) => {
                         lock(&book).counts.hole_punches.succeeded += 1;
@@ -2698,5 +2729,22 @@ mod tests {
         book.slots.clear();
         book.slots_here.insert(both);
         assert!(book.duplicates(&both, false).is_empty());
+        book.slots_here.clear();
+        // Once measured, the fastest direct one stays, however young.
+        book.rtts
+            .insert(ConnectionId::new_unchecked(3), Duration::from_millis(40));
+        book.rtts
+            .insert(ConnectionId::new_unchecked(4), Duration::from_millis(2));
+        assert_eq!(
+            book.duplicates(&both, false),
+            [
+                ConnectionId::new_unchecked(3),
+                ConnectionId::new_unchecked(2)
+            ]
+        );
+        assert_eq!(
+            book.location(&both).map(|address| address.to_string()),
+            Some("/ip4/10.0.0.4/tcp/1".to_owned())
+        );
     }
 }