libp2p: retry punching

1 files changed, 126 insertions, 15 deletions

diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index 5419f46..a5541f6 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -20,7 +20,8 @@
 //! (`--libp2p-relays`), on which it reserves a slot; its address through the relay
 //! (`<relay>/p2p-circuit/p2p/<it>`) is published with the others. Over a relayed connection, both sides then try to
 //! connect directly (libp2p's DCUtR, hole punching); the relay carries the traffic until
-//! that works, or for good if it doesn't, within its limits (`RELAY_CIRCUIT_DURATION`,
+//! that works, or for good if it doesn't (punched again now and then, `PUNCH_RETRIES`),
+//! within its limits (`RELAY_CIRCUIT_DURATION`,
 //! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send). Hole
 //! punching needs both sides to listen (`HolePunching`: at their LAN addresses too);
 //! where only one does, the other dials it directly once it knows where (`upgrade`).
@@ -156,6 +157,19 @@ const PRUNE_EVERY: Duration = Duration::from_secs(60);
 /// 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);
+/// The same while a peer is connected only through relays: a new relayed connection is
+/// 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);
+/// 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
+/// connection through the relay, on which the peer starts the punch (`retry_punches`).
+const PUNCH_RETRIES: [Duration; 3] = [
+    Duration::from_secs(60),
+    Duration::from_secs(300),
+    Duration::from_secs(1800),
+];
 const DEDUPE_EVERY: Duration = Duration::from_secs(5);
 
 /// How long a message of `bytes` may take on a stream.
@@ -296,6 +310,9 @@ struct Book {
     upgrades: HashMap<PeerId, (tokio::time::Instant, Duration)>,
     /// The shortest round trip measured on each open connection (libp2p's ping).
     rtts: HashMap<ConnectionId, Duration>,
+    /// Peers whose hole punching failed: how often in a row, and when to try again
+    /// (`PUNCH_RETRIES`).
+    punching: HashMap<PeerId, (usize, tokio::time::Instant)>,
     /// 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.
@@ -454,16 +471,7 @@ fn count_dial_failure(failures: &mut DialFailures, error: &::libp2p::swarm::Dial
 impl Book {
     /// How we are connected to `peer`: directly if any connection is direct.
     fn connection(&self, peer: &PeerId) -> Option<Connection> {
-        let mut found = None;
-        for (connected, relayed, _) in self.connections.values() {
-            if connected == peer {
-                if !relayed {
-                    return Some(Connection::Direct);
-                }
-                found = Some(Connection::Relayed);
-            }
-        }
-        found
+        book_connection(&self.connections, peer)
     }
 
     /// The connections to `peer` to close, of several: all but one, the oldest direct
@@ -567,6 +575,7 @@ impl Book {
             self.lan.remove(peer);
             self.failed_relays.remove(peer);
             self.upgrades.remove(peer);
+            self.punching.remove(peer);
             self.seen.remove(peer);
         }
         stale.len()
@@ -972,6 +981,24 @@ impl ::libp2p::swarm::NetworkBehaviour for Ignoring {
     }
 }
 
+/// How `peer` is connected, of `connections` (`Book::connections`): directly if one is
+/// direct.
+fn book_connection(
+    connections: &HashMap<ConnectionId, (PeerId, bool, Multiaddr)>,
+    peer: &PeerId,
+) -> Option<Connection> {
+    let mut found = None;
+    for (connected, relayed, _) in connections.values() {
+        if connected == peer {
+            if !relayed {
+                return Some(Connection::Direct);
+            }
+            found = Some(Connection::Relayed);
+        }
+    }
+    found
+}
+
 fn dialable(known: &[Multiaddr], local: PeerId) -> Vec<Multiaddr> {
     known
         .iter()
@@ -1612,6 +1639,67 @@ fn upgrade(swarm: &mut Swarm<Behaviour>, book: &Mutex<Book>, peer: PeerId, local
     );
 }
 
+/// Tries hole punching again with the peers it failed with that are due (`punching`).
+/// libp2p punches as a relayed connection opens, started by the side that was dialled
+/// through the relay; that side is where a punch's failure is told, so it is this one.
+/// A new relayed connection starts a new punch: we dial the peer through a relay if it
+/// has a slot on one; else we close our relayed connections to it, and its next message
+/// (a probe, within a minute) dials us through ours again. Peers connected directly by
+/// now are dropped; those not connected now are kept, failures and all.
+fn retry_punches(swarm: &mut Swarm<Behaviour>, book: &Mutex<Book>, local: PeerId) {
+    let now = tokio::time::Instant::now();
+    let mut close: Vec<ConnectionId> = Vec::new();
+    let due: Vec<(PeerId, usize, Vec<Multiaddr>)> = {
+        let mut book = lock(book);
+        let book = &mut *book;
+        let mut due = Vec::new();
+        book.punching.retain(|peer, (failed, next)| {
+            match book_connection(&book.connections, peer) {
+                Some(Connection::Direct) => return false,
+                // Kept, failures and all: one we closed to have it dial anew comes back.
+                None => return true,
+                Some(Connection::Relayed) if now < *next => return true,
+                Some(Connection::Relayed) => {}
+            }
+            let through: Vec<Multiaddr> = book
+                .addresses
+                .get(peer)
+                .map(|known| dialable(known, local))
+                .unwrap_or_default()
+                .into_iter()
+                .filter(is_relayed)
+                .collect();
+            // Until its punch fails again (or the peer tries from its side).
+            *next = now + PUNCH_RETRIES[(*failed).min(PUNCH_RETRIES.len() - 1)];
+            if through.is_empty() {
+                close.extend(
+                    book.connections
+                        .iter()
+                        .filter(|(_, (connected, relayed, _))| connected == peer && *relayed)
+                        .map(|(id, _)| *id),
+                );
+            } else {
+                due.push((*peer, *failed, through));
+            }
+            true
+        });
+        due
+    };
+    for connection in close {
+        debug!(connection = %connection, "Hole punching again: closing a relayed connection, for the peer to dial anew");
+        swarm.close_connection(connection);
+    }
+    for (peer, failed, through) in due {
+        debug!(peer = %Addressed(SCHEME, &peer), failed, "Hole punching again, through a new relayed connection");
+        let _ = swarm.dial(
+            DialOpts::peer_id(peer)
+                .addresses(through)
+                .condition(PeerCondition::Always)
+                .build(),
+        );
+    }
+}
+
 /// Asks the DHT (with `--libp2p-public`) where `peer` is, for the next attempt.
 fn find_on_dht(swarm: &mut Swarm<Behaviour>, peer: PeerId) {
     if let Some(dht) = swarm.behaviour_mut().kad.as_mut() {
@@ -1687,6 +1775,7 @@ async fn run(
             }
             _ = check.tick() => {
                 pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
+                retry_punches(&mut swarm, &book, relaying.local);
             }
             command = commands.recv() => match command {
                 None => break,
@@ -1779,11 +1868,17 @@ async fn run(
                         );
                         if !relayed {
                             book.upgrades.remove(&peer_id);
+                            book.punching.remove(&peer_id);
                         }
                         if relaying.local < peer_id
                             && book.connections.values().filter(|(connected, _, _)| *connected == peer_id).count() > 1
                         {
-                            dedupe_due.insert(peer_id, tokio::time::Instant::now() + DUPLICATE_GRACE);
+                            // Only through relays: maybe a punch under way.
+                            let grace = match book.connection(&peer_id) {
+                                Some(Connection::Relayed) => PUNCH_GRACE,
+                                _ => DUPLICATE_GRACE,
+                            };
+                            dedupe_due.insert(peer_id, tokio::time::Instant::now() + grace);
                         }
                         debug!(
                             peer = %Addressed(SCHEME, &peer_id),
@@ -2005,12 +2100,28 @@ async fn run(
                 },
                 SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
                     Ok(_) => {
-                        lock(&book).counts.hole_punches.succeeded += 1;
+                        let mut book = lock(&book);
+                        book.counts.hole_punches.succeeded += 1;
+                        book.punching.remove(&event.remote_peer_id);
+                        drop(book);
                         info!(peer = %Addressed(SCHEME, &event.remote_peer_id), "Connected directly to a peer behind a relay");
                     }
                     Err(error) => {
-                        lock(&book).counts.hole_punches.failed += 1;
-                        debug!(peer = %Addressed(SCHEME, &event.remote_peer_id), error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on");
+                        let retry = {
+                            let mut book = lock(&book);
+                            book.counts.hole_punches.failed += 1;
+                            let failed = book
+                                .punching
+                                .get(&event.remote_peer_id)
+                                .map_or(0, |(failed, _)| *failed);
+                            let retry = PUNCH_RETRIES[failed.min(PUNCH_RETRIES.len() - 1)];
+                            book.punching.insert(
+                                event.remote_peer_id,
+                                (failed + 1, tokio::time::Instant::now() + retry),
+                            );
+                            retry
+                        };
+                        debug!(peer = %Addressed(SCHEME, &event.remote_peer_id), error = %error, retry = ?retry, "Couldn't connect directly to a peer behind a relay; the relay carries on");
                     }
                 },
                 SwarmEvent::Behaviour(BehaviourEvent::Identify(