configured libp2p relays are candidates

1 files changed, 101 insertions, 103 deletions

diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index 12d3da5..63f6335 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -16,9 +16,9 @@
 //! public address forwarded to it), or else those it listens on, loopback left out.
 //!
 //! A node nobody can dial (behind NAT, or a closed port) can still be reached through a
-//! relay: a node of the swarm that relays (`--libp2p-relay`), on which it reserves a slot
-//! (`--libp2p-relays`); its address through the relay (`<relay>/p2p-circuit/p2p/<it>`)
-//! is published with the others. Over a relayed connection, both sides then try to
+//! relay: a node of the swarm that relays (`--libp2p-relay`), or one configured
+//! (`--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`,
 //! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send).
@@ -29,11 +29,11 @@
 //!
 //! 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
-//! member list. A node that needs a relay and has none configured picks up to
-//! `AUTO_RELAYS` of those (`relay_check`). It needs one when it has no address others
-//! could dial, or when AutoNAT says it can't be dialled: AutoNAT asks connected peers
-//! (nodes of the swarm) to dial it back. Once AutoNAT says it can be dialled, it lets
-//! the relays it picked go.
+//! member list. A node that needs a relay picks up to `AUTO_RELAYS` relays
+//! (`relay_check`): those configured first, then members that relay. It needs one when
+//! it has no address others could dial, or when AutoNAT says it can't be dialled:
+//! AutoNAT asks connected peers (nodes of the swarm) to dial it back. Once AutoNAT says
+//! it can be dialled, it lets the relays it picked go.
 //!
 //! With `--libp2p-mdns`, a node finds libp2p peers on its LAN (multicast DNS). Their
 //! LAN addresses are noted, and those that turn out to speak continuum (identify says
@@ -105,8 +105,8 @@ const RELAY_CIRCUIT_BYTES: u64 = 1 << 30;
 /// Circuits through this node at once, and per peer.
 const RELAY_CIRCUITS: usize = 64;
 const RELAY_CIRCUITS_PER_PEER: usize = 8;
-/// A reservation on a relay that ended (the relay restarted, say) is made again after
-/// this.
+/// A configured relay that couldn't be reached to find out its peer ID is tried again
+/// after this.
 const RELAY_RETRY: Duration = Duration::from_secs(30);
 /// Relays picked among the members, at most, and how often a node checks whether it
 /// needs them.
@@ -188,15 +188,10 @@ enum Command {
         address: Multiaddr,
         done: oneshot::Sender<Result<PeerId, String>>,
     },
-    /// See whether relays among the members are needed now (one was just learned of).
+    /// See whether relays are needed now (one was just learned of).
     CheckRelays,
-    /// Reserve a slot on the relay at `relay` (again). `configured`: the address as
-    /// configured, for later attempts (without a peer ID, the relay may since have
-    /// another).
-    Reserve {
-        relay: Multiaddr,
-        configured: Multiaddr,
-    },
+    /// Find out which peer relays at `address`, configured without the relay's peer ID.
+    FindRelay { address: Multiaddr },
 }
 
 /// Where peers are, and where we listen.
@@ -208,6 +203,12 @@ struct Book {
     announced: Vec<Multiaddr>,
     /// `Libp2pConfig::relay`: this node relays, and says so.
     serving: bool,
+    /// Relays configured (`Libp2pConfig::relays`), and where (with the relay's peer ID):
+    /// picked first.
+    configured_relays: HashMap<PeerId, Multiaddr>,
+    /// Of those, the ones configured without the relay's peer ID, as configured: found
+    /// out again when the relay doesn't grant a slot (it may have a new peer ID).
+    unnamed_relays: HashMap<PeerId, Multiaddr>,
     /// Members that say they relay, and where (with the relay's peer ID).
     relays: HashMap<PeerId, Multiaddr>,
     /// With `Libp2pConfig::public`: other peers that relay (identify says so), and
@@ -586,19 +587,25 @@ impl Libp2pTransport {
         let incoming = control
             .accept(PROTOCOL)
             .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?;
-        let configured_relays = !config.relays.is_empty();
+        let (commands, receiver) = mpsc::unbounded_channel();
+        let mut configured_relays = HashMap::new();
+        for relay in config.relays {
+            match split_peer(&relay) {
+                Some(peer) if peer != peer_id => {
+                    configured_relays.insert(peer, relay);
+                }
+                Some(_) => {} // This node.
+                None => {
+                    let _ = commands.send(Command::FindRelay { address: relay });
+                }
+            }
+        }
         let book = Arc::new(Mutex::new(Book {
             announced: config.announce,
             serving: config.relay,
+            configured_relays,
             ..Book::default()
         }));
-        let (commands, receiver) = mpsc::unbounded_channel();
-        for relay in config.relays {
-            let _ = commands.send(Command::Reserve {
-                configured: relay.clone(),
-                relay,
-            });
-        }
         let (stopping, stopped) = watch::channel(false);
         tokio::spawn(run(
             swarm,
@@ -607,7 +614,6 @@ impl Libp2pTransport {
             Arc::clone(&book),
             Relaying {
                 serving: config.relay,
-                configured: configured_relays,
                 public: config.public,
                 local: peer_id,
             },
@@ -824,14 +830,10 @@ fn memory_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> {
     ))
 }
 
-/// Drives the swarm: dials what `commands` ask for and notes where we listen, until the
-/// transport is dropped.
 /// How a node relates to relays.
 struct Relaying {
     /// It relays for others.
     serving: bool,
-    /// Relays to use are configured: it doesn't pick any.
-    configured: bool,
     /// Public relays may be picked, after members.
     public: bool,
     local: PeerId,
@@ -873,8 +875,9 @@ fn relay_check(
     candidates
 }
 
-/// Reserves slots on the relays `relay_check` picks among the members, and lets those
-/// picked go once AutoNAT says this node can be dialled.
+/// Reserves slots on the relays `relay_check` picks (configured ones first, then
+/// members, then public ones), and lets those picked go once AutoNAT says this node can
+/// be dialled.
 fn pick_relays(
     swarm: &mut Swarm<Behaviour>,
     book: &Mutex<Book>,
@@ -891,7 +894,7 @@ fn pick_relays(
         }
         return;
     }
-    let (direct, offers, public) = {
+    let (direct, configured, members, public) = {
         let mut book = lock(book);
         book.failed_relays
             .retain(|_, at| at.elapsed() < RELAY_PAUSE);
@@ -908,21 +911,30 @@ fn pick_relays(
         } else {
             HashMap::new()
         };
-        (!book.direct().is_empty(), usable(&book.relays), public)
+        (
+            !book.direct().is_empty(),
+            usable(&book.configured_relays),
+            usable(&book.relays),
+            public,
+        )
     };
     let mut using: Vec<PeerId> = picked.values().copied().collect();
-    // Members first, then public ones for what's left.
-    let mut wanted = relay_check(relaying.local, direct, nat, &offers, &using);
-    using.extend(wanted.iter().map(|(relay, _)| *relay));
-    wanted.extend(relay_check(relaying.local, direct, nat, &public, &using));
-    for (relay, address) in wanted {
-        match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) {
-            Ok(listener) => {
-                info!(relay = %Addressed(SCHEME, &relay), address = %address, "Picked a member as relay");
-                picked.insert(listener, relay);
-            }
-            Err(error) => {
-                debug!(relay = %Addressed(SCHEME, &relay), error = %error, "Couldn't use a member as relay");
+    // Each kind for what the ones before left.
+    for (kind, offers) in [
+        ("configured", configured),
+        ("member", members),
+        ("public", public),
+    ] {
+        for (relay, address) in relay_check(relaying.local, direct, nat, &offers, &using) {
+            using.push(relay);
+            match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) {
+                Ok(listener) => {
+                    info!(relay = %Addressed(SCHEME, &relay), address = %address, kind, "Picked a relay");
+                    picked.insert(listener, relay);
+                }
+                Err(error) => {
+                    debug!(relay = %Addressed(SCHEME, &relay), kind, error = %error, "Couldn't use a relay");
+                }
             }
         }
     }
@@ -935,7 +947,8 @@ fn find_on_dht(swarm: &mut Swarm<Behaviour>, peer: PeerId) {
     }
 }
 
-/// `again`: to schedule commands.
+/// Drives the swarm: dials what `commands` ask for and notes where we listen, until the
+/// transport is dropped. `again`: to schedule commands.
 async fn run(
     mut swarm: Swarm<Behaviour>,
     mut commands: mpsc::UnboundedReceiver<Command>,
@@ -946,25 +959,12 @@ async fn run(
 ) {
     let serving = relaying.serving;
     let mut nat = ::libp2p::autonat::NatStatus::Unknown;
-    // Relays picked among the members, by listener.
+    // Relays picked, by listener.
     let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new();
     let mut check = tokio::time::interval(RELAY_CHECK);
     check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
     let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP);
     dht_bootstrap.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-    // Relays we listen through, by listener: as configured, and their peer.
-    let mut reservations: HashMap<::libp2p::core::transport::ListenerId, (Multiaddr, PeerId)> =
-        HashMap::new();
-    let reserve_later = |configured: Multiaddr| {
-        let again = again.clone();
-        tokio::spawn(async move {
-            tokio::time::sleep(RELAY_RETRY).await;
-            let _ = again.send(Command::Reserve {
-                relay: configured.clone(),
-                configured,
-            });
-        });
-    };
     let mut connecting: HashMap<PeerId, Vec<oneshot::Sender<Result<(), String>>>> = HashMap::new();
     let mut identifying: HashMap<
         ConnectionId,
@@ -1012,15 +1012,13 @@ async fn run(
                     warn!(error = %error, "Couldn't join the IPFS DHT");
                 }
             }
-            _ = check.tick(), if !relaying.configured => {
+            _ = check.tick() => {
                 pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
             }
             command = commands.recv() => match command {
                 None => break,
                 Some(Command::CheckRelays) => {
-                    if !relaying.configured {
-                        pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
-                    }
+                    pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
                 }
                 Some(Command::Connect { peer, addresses, done }) => {
                     if swarm.is_connected(&peer) {
@@ -1047,46 +1045,39 @@ async fn run(
                         find_on_dht(&mut swarm, peer);
                     }
                 }
-                Some(Command::Reserve { relay, configured }) if split_peer(&relay).is_none() => {
-                    // Which peer relays there first.
+                Some(Command::FindRelay { address }) => {
                     let again = again.clone();
+                    let local = relaying.local;
+                    let book = Arc::clone(&book);
                     tokio::spawn(async move {
                         let (done, identified) = oneshot::channel();
                         if again
-                            .send(Command::Identify { address: relay.clone(), done })
+                            .send(Command::Identify { address: address.clone(), done })
                             .is_err()
                         {
                             return;
                         }
                         match tokio::time::timeout(CONNECT_TIMEOUT, identified).await {
                             Ok(Ok(Ok(peer))) => {
-                                if let Ok(relay) = relay.with_p2p(peer) {
-                                    let _ = again.send(Command::Reserve { relay, configured });
+                                if peer == local {
+                                    return; // This node.
                                 }
+                                if let Ok(relay) = address.clone().with_p2p(peer) {
+                                    debug!(relay = %Addressed(SCHEME, &relay), "Found a configured relay");
+                                    let mut known = lock(&book);
+                                    known.configured_relays.insert(peer, relay);
+                                    known.unnamed_relays.insert(peer, address);
+                                }
+                                let _ = again.send(Command::CheckRelays);
                             }
                             _ => {
-                                warn!(relay = %Addressed(SCHEME, &relay), "Couldn't reach a relay; trying again later");
+                                warn!(relay = %Addressed(SCHEME, &address), "Couldn't reach a configured relay; trying again later");
                                 tokio::time::sleep(RELAY_RETRY).await;
-                                let _ = again.send(Command::Reserve { relay, configured });
+                                let _ = again.send(Command::FindRelay { address });
                             }
                         }
                     });
                 }
-                Some(Command::Reserve { relay, configured }) => {
-                    let Some(relay_peer) = split_peer(&relay) else {
-                        continue; // The arm above takes those.
-                    };
-                    match swarm.listen_on(relay.clone().with(Protocol::P2pCircuit)) {
-                        Ok(listener) => {
-                            debug!(relay = %Addressed(SCHEME, &relay), "Reserving a slot on a relay");
-                            reservations.insert(listener, (configured, relay_peer));
-                        }
-                        Err(error) => {
-                            warn!(relay = %Addressed(SCHEME, &relay), error = %error, "Couldn't use a relay; trying again later");
-                            reserve_later(configured);
-                        }
-                    }
-                }
                 Some(Command::Identify { address, done }) => {
                     let opts = DialOpts::unknown_peer_id().address(address.clone()).build();
                     let id = opts.connection_id();
@@ -1185,20 +1176,8 @@ async fn run(
                     lock(&book)
                         .listening
                         .retain(|known| !addresses.contains(known));
-                    if let Some((relay, relay_peer)) = reservations.remove(&listener_id) {
-                        {
-                            let mut book = lock(&book);
-                            if book.slots.remove(&relay_peer) {
-                                book.counts.slots_lost += 1;
-                            } else {
-                                book.counts.slots_not_granted += 1;
-                            }
-                        }
-                        warn!(relay = %Addressed(SCHEME, &relay), reason = ?reason, "Lost the slot on a relay; reserving again later");
-                        reserve_later(relay);
-                    }
-                    // A relay picked among the members: the next check picks again, another
-                    // one if this one never granted the slot.
+                    // A relay picked: the next check picks again, another one if this one
+                    // never granted the slot.
                     if let Some(relay) = picked.remove(&listener_id) {
                         let lost = {
                             let mut book = lock(&book);
@@ -1208,11 +1187,16 @@ async fn run(
                             } else {
                                 book.counts.slots_not_granted += 1;
                                 book.failed_relays.insert(relay, tokio::time::Instant::now());
+                                // Configured without its peer ID: maybe another one now.
+                                if let Some(address) = book.unnamed_relays.remove(&relay) {
+                                    book.configured_relays.remove(&relay);
+                                    let _ = again.send(Command::FindRelay { address });
+                                }
                             }
                             lost
                         };
                         if lost {
-                            debug!(relay = %Addressed(SCHEME, &relay), reason = ?reason, "Lost the slot on a member relaying");
+                            debug!(relay = %Addressed(SCHEME, &relay), reason = ?reason, "Lost the slot on a relay");
                         } else {
                             info!(
                                 relay = %Addressed(SCHEME, &relay),
@@ -1866,6 +1850,20 @@ mod tests {
         assert!(a.connection(b.local_id()).is_some());
     }
 
+    /// A relay configured without its peer ID is found out, and used.
+    #[tokio::test]
+    async fn a_relay_configured_without_its_peer_id() {
+        let (_r, r_address) = memory_node_with(true, Vec::new(), true);
+        let mut bare = r_address.clone();
+        bare.pop();
+        let (b, _) = memory_node_with(false, vec![bare], false);
+        assert!(
+            eventually(|| !b.local_addresses().is_empty()).await,
+            "b got no slot on the relay"
+        );
+        assert!(b.local_addresses()[0].starts_with(&format!("{r_address}/p2p-circuit/")));
+    }
+
     fn peer() -> PeerId {
         Keypair::generate_ed25519().public().to_peer_id()
     }