libp2p-relay: keep slot connections

1 files changed, 103 insertions, 6 deletions

diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index 7351b16..d621c8e 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -252,6 +252,8 @@ struct Book {
     failed_relays: HashMap<PeerId, tokio::time::Instant>,
     /// Relays that granted us a slot we hold.
     slots: std::collections::HashSet<PeerId>,
+    /// Peers holding a slot on this node, as a relay.
+    slots_here: std::collections::HashSet<PeerId>,
     /// 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.
@@ -305,6 +307,46 @@ fn describe_dial_error(error: &::libp2p::swarm::DialError) -> String {
     described.join("; ")
 }
 
+/// Why a slot on a relay ended, or was never granted, shortly: libp2p nests the
+/// relay's answer several types deep (`Left(Left(Left(Reservation(...))))`), and the
+/// common cases deserve a sentence.
+fn relay_failure(reason: &std::io::Result<()>) -> String {
+    let Err(error) = reason else {
+        return "closed".to_owned();
+    };
+    relay_failure_text(&format!("{error:?}"), &innermost_causes(error))
+}
+
+/// `relay_failure` on the error's debug form, and its innermost cause as text.
+fn relay_failure_text(debug: &str, innermost: &str) -> String {
+    let known = [
+        (
+            "NoAddressesInReservation",
+            "the relay gave no address to be reached through it",
+        ),
+        ("ResourceLimitExceeded", "the relay has no slot left"),
+        ("PermissionDenied", "the relay doesn't let us reserve"),
+        ("NoReservation", "the relay holds no slot for us"),
+        (
+            "connection gated",
+            "the relay refused the connection (connection gated)",
+        ),
+        (
+            "UnsupportedProtocol",
+            "the relay doesn't relay (no circuit relay protocol)",
+        ),
+        ("Timeout", "the relay didn't answer in time"),
+    ];
+    if let Some((_, said)) = known.iter().find(|(pattern, _)| debug.contains(pattern)) {
+        return (*said).to_owned();
+    }
+    // The protocol refused, but only after its first data (`swarm_config`).
+    if innermost == "Failed" {
+        return "the relay refused the circuit relay protocol (is it a relay?)".to_owned();
+    }
+    innermost.to_owned()
+}
+
 /// The innermost of `error`'s causes, each distinct one once: those libp2p lists in a
 /// "Multiple dial errors occurred:" text (one per line, ` - <cause>: <its cause>...`),
 /// or the last of the source chain.
@@ -365,7 +407,15 @@ impl Book {
 
     /// The connections to `peer` to close, of several: all but one, the oldest direct
     /// one (the oldest relayed one if none is direct).
-    fn duplicates(&self, peer: &PeerId) -> Vec<ConnectionId> {
+    ///
+    /// 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`).
+    fn duplicates(&self, peer: &PeerId, reserving: bool) -> Vec<ConnectionId> {
+        if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) {
+            return Vec::new();
+        }
         let mut connections: Vec<(bool, ConnectionId)> = self
             .connections
             .iter()
@@ -1127,7 +1177,8 @@ async fn run(
                     .collect();
                 for peer in due {
                     dedupe_due.remove(&peer);
-                    let duplicates = lock(&book).duplicates(&peer);
+                    let reserving = picked.values().any(|relay| *relay == peer);
+                    let duplicates = lock(&book).duplicates(&peer, reserving);
                     for connection in duplicates {
                         if swarm.close_connection(connection) {
                             closing_duplicates.insert(connection);
@@ -1327,13 +1378,15 @@ async fn run(
                             }
                             lost
                         };
+                        let reason = relay_failure(&reason);
                         if lost {
-                            debug!(relay = %Addressed(SCHEME, &relay), reason = ?reason, "Lost the slot on a relay");
+                            debug!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay");
                         } else {
                             info!(
                                 relay = %Addressed(SCHEME, &relay),
                                 pause = ?RELAY_PAUSE,
-                                "A relay didn't grant a slot (unreachable, or another peer at its address); picking others for a while"
+                                reason = %reason,
+                                "A relay didn't grant a slot; picking others for a while"
                             );
                         }
                     }
@@ -1374,6 +1427,20 @@ async fn run(
                     nat = new;
                 }
                 SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => {
+                    {
+                        use ::libp2p::relay::Event;
+                        let mut book = lock(&book);
+                        match &event {
+                            Event::ReservationReqAccepted { src_peer_id, .. } => {
+                                book.slots_here.insert(*src_peer_id);
+                            }
+                            Event::ReservationClosed { src_peer_id }
+                            | Event::ReservationTimedOut { src_peer_id } => {
+                                book.slots_here.remove(src_peer_id);
+                            }
+                            _ => {}
+                        }
+                    }
                     debug!(event = ?event, "Relay event");
                 }
                 SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
@@ -2218,6 +2285,29 @@ mod tests {
         assert_eq!(book.prune(later, |_| false), 1); // `connected`, once it isn't.
     }
 
+    #[test]
+    fn relay_failures_are_short() {
+        assert_eq!(
+            relay_failure_text(
+                "Custom { kind: Other, error: Other(Left(Left(Left(Reservation(Protocol(NoAddressesInReservation)))))) }",
+                "x"
+            ),
+            "the relay gave no address to be reached through it"
+        );
+        assert_eq!(
+            relay_failure_text(
+                "Codec(Error(Custom { kind: Other, error: Failed }))",
+                "Failed"
+            ),
+            "the relay refused the circuit relay protocol (is it a relay?)"
+        );
+        assert_eq!(
+            relay_failure_text("Io(…)", "connection reset"),
+            "connection reset"
+        );
+        assert_eq!(relay_failure(&Ok(())), "closed");
+    }
+
     /// libp2p's nested dial errors come out as each address's cause, once.
     #[test]
     fn dial_errors_are_short() {
@@ -2300,12 +2390,19 @@ mod tests {
         // Of several, the oldest direct one stays; a relayed one only without a direct.
         connect(&mut book, both, false);
         assert_eq!(
-            book.duplicates(&both),
+            book.duplicates(&both, false),
             [
                 ConnectionId::new_unchecked(4),
                 ConnectionId::new_unchecked(2)
             ]
         );
-        assert!(book.duplicates(&relayed_only).is_empty());
+        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());
+        book.slots.insert(both);
+        assert!(book.duplicates(&both, false).is_empty());
+        book.slots.clear();
+        book.slots_here.insert(both);
+        assert!(book.duplicates(&both, false).is_empty());
     }
 }