libp2p: introduce rehoming

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)