libp2p: introduce rehoming

1 files changed, 24 insertions, 4 deletions

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");