register relays in member list & use AutoNAT

2 files changed, 259 insertions, 17 deletions

diff --git a/src/config/mod.rs b/src/config/mod.rs
index 8ab634c..2646e73 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -261,14 +261,14 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         Kind::Flag,
         "BOOL",
         Some("false"),
-        "With the libp2p transport: relay for other nodes of the swarm that can't be dialled (bounded: 64 circuits, 8 per peer, each up to an hour and 1 GiB). Needs a listen address others can dial.",
+        "With the libp2p transport: relay for other nodes of the swarm that can't be dialled (bounded: 64 circuits, 8 per peer, each up to an hour and 1 GiB), and say so in the member list, where nodes that need a relay find it. Needs a listen address others can dial.",
     ),
     option(
         "libp2p-relays",
         Kind::List,
         "MULTIADDR,[MULTIADDR...]",
         None,
-        "With the libp2p transport: relays to be reachable through (multiaddrs, with or without the relay's peer ID), e.g. members started with libp2p-relay. Only these are used; never public ones.",
+        "With the libp2p transport: relays to be reachable through always (multiaddrs, with or without the relay's peer ID). Without, a node that can't be dialled picks members that relay (libp2p-relay) by itself. Never public ones.",
     ),
     option(
         "libp2p-key-file",
diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs
index 3464a68..675ff13 100644
--- a/src/transport/p2p.rs
+++ b/src/transport/p2p.rs
@@ -21,9 +21,17 @@
 //! 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).
 //!
+//! 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.
+//!
 //! What this transport doesn't do yet (`docs/future-work.md`): find peers on the LAN
-//! (mDNS), or pick relays among the members by itself. It never uses public libp2p
-//! infrastructure: relays are only those configured.
+//! (mDNS). It never uses public libp2p infrastructure: relays are those configured, or
+//! members that say they relay.
 
 use std::collections::HashMap;
 use std::path::Path;
@@ -75,6 +83,10 @@ const RELAY_CIRCUITS_PER_PEER: usize = 8;
 /// A reservation on a relay that ended (the relay restarted, say) is made 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.
+const AUTO_RELAYS: usize = 2;
+const RELAY_CHECK: Duration = Duration::from_secs(30);
 
 /// How long a message of `bytes` may take on a stream.
 fn stream_timeout(bytes: usize) -> Duration {
@@ -132,6 +144,8 @@ enum Command {
         address: Multiaddr,
         done: oneshot::Sender<Result<PeerId, String>>,
     },
+    /// See whether relays among the members 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).
@@ -148,6 +162,27 @@ struct Book {
     listening: Vec<Multiaddr>,
     /// `Libp2pConfig::announce`.
     announced: Vec<Multiaddr>,
+    /// `Libp2pConfig::relay`: this node relays, and says so.
+    serving: bool,
+    /// Members that say they relay, and where (with the relay's peer ID).
+    relays: HashMap<PeerId, Multiaddr>,
+}
+
+impl Book {
+    /// Where others could dial us directly: the announced addresses, or else those we
+    /// listen on (`published`), without the peer ID.
+    fn direct(&self) -> Vec<Multiaddr> {
+        if !self.announced.is_empty() {
+            return self.announced.clone();
+        }
+        let direct: Vec<Multiaddr> = self
+            .listening
+            .iter()
+            .filter(|address| !is_relayed(address))
+            .cloned()
+            .collect();
+        published(&direct)
+    }
 }
 
 /// Whether `address` only reaches this host (or this link): no use to other hosts, and
@@ -244,8 +279,10 @@ impl Libp2pTransport {
         let incoming = control
             .accept(PROTOCOL)
             .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?;
+        let configured_relays = !config.relays.is_empty();
         let book = Arc::new(Mutex::new(Book {
             announced: config.announce,
+            serving: config.relay,
             ..Book::default()
         }));
         let (commands, receiver) = mpsc::unbounded_channel();
@@ -261,7 +298,11 @@ impl Libp2pTransport {
             receiver,
             commands.clone(),
             Arc::clone(&book),
-            config.relay,
+            Relaying {
+                serving: config.relay,
+                configured: configured_relays,
+                local: peer_id,
+            },
             stopping,
         ));
         info!(peer_id = %peer_id, "libp2p transport started");
@@ -311,6 +352,7 @@ struct Behaviour {
     /// Only with `Libp2pConfig::relay`.
     relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>,
     dcutr: ::libp2p::dcutr::Behaviour,
+    autonat: ::libp2p::autonat::Behaviour,
 }
 
 impl Behaviour {
@@ -336,6 +378,14 @@ impl Behaviour {
             relay_client,
             relay: relay.into(),
             dcutr: ::libp2p::dcutr::Behaviour::new(peer_id),
+            autonat: ::libp2p::autonat::Behaviour::new(
+                peer_id,
+                ::libp2p::autonat::Config {
+                    // Swarms on private networks (a LAN, WireGuard) count too.
+                    only_global_ips: false,
+                    ..Default::default()
+                },
+            ),
         }
     }
 }
@@ -383,15 +433,100 @@ fn memory_swarm(key: Keypair, relay: bool) -> Result<Swarm<Behaviour>> {
 
 /// Drives the swarm: dials what `commands` ask for and notes where we listen, until the
 /// transport is dropped.
-/// `again`: to schedule commands; `serving`: whether this node relays for others.
+/// 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,
+    local: PeerId,
+}
+
+/// The relays to pick among `offers`, if this node needs relays, keeping `using`: a
+/// node needs them when it has no address others could dial (`direct`), or AutoNAT says
+/// it can't be dialled; up to `AUTO_RELAYS`, in an order of its own (the lowest hash of
+/// both peer IDs), so that not every node picks the same.
+fn relay_check(
+    local: PeerId,
+    direct: bool,
+    nat: &::libp2p::autonat::NatStatus,
+    offers: &HashMap<PeerId, Multiaddr>,
+    using: &[PeerId],
+) -> Vec<(PeerId, Multiaddr)> {
+    use ::libp2p::autonat::NatStatus;
+    let needed = match nat {
+        NatStatus::Private => true,
+        NatStatus::Public(_) => false,
+        NatStatus::Unknown => !direct,
+    };
+    if !needed || using.len() >= AUTO_RELAYS {
+        return Vec::new();
+    }
+    let score = |relay: &PeerId| {
+        use std::hash::{Hash, Hasher};
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        (local, relay).hash(&mut hasher);
+        hasher.finish()
+    };
+    let mut candidates: Vec<(PeerId, Multiaddr)> = offers
+        .iter()
+        .filter(|(relay, _)| **relay != local && !using.contains(relay))
+        .map(|(relay, address)| (*relay, address.clone()))
+        .collect();
+    candidates.sort_by_key(|(relay, _)| score(relay));
+    candidates.truncate(AUTO_RELAYS - using.len());
+    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.
+fn pick_relays(
+    swarm: &mut Swarm<Behaviour>,
+    book: &Mutex<Book>,
+    relaying: &Relaying,
+    nat: &::libp2p::autonat::NatStatus,
+    picked: &mut HashMap<::libp2p::core::transport::ListenerId, PeerId>,
+) {
+    if nat.is_public() && !picked.is_empty() {
+        info!("Reachable directly (AutoNAT); letting the relays picked go");
+        for listener in std::mem::take(picked).into_keys() {
+            swarm.remove_listener(listener);
+        }
+        return;
+    }
+    let (direct, offers) = {
+        let book = lock(book);
+        (!book.direct().is_empty(), book.relays.clone())
+    };
+    let using: Vec<PeerId> = picked.values().copied().collect();
+    for (relay, address) in relay_check(relaying.local, direct, nat, &offers, &using) {
+        match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) {
+            Ok(listener) => {
+                info!(relay = %relay, address = %address, "Picked a member as libp2p relay");
+                picked.insert(listener, relay);
+            }
+            Err(error) => {
+                debug!(relay = %relay, error = %error, "Couldn't use a member as libp2p relay");
+            }
+        }
+    }
+}
+
+/// `again`: to schedule commands.
 async fn run(
     mut swarm: Swarm<Behaviour>,
     mut commands: mpsc::UnboundedReceiver<Command>,
     again: mpsc::UnboundedSender<Command>,
     book: Arc<Mutex<Book>>,
-    serving: bool,
+    relaying: Relaying,
     stopping: watch::Sender<bool>,
 ) {
+    let serving = relaying.serving;
+    let mut nat = ::libp2p::autonat::NatStatus::Unknown;
+    // Relays picked among the members, 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);
     // Relays we listen through, by listener, as configured.
     let mut reservations: HashMap<::libp2p::core::transport::ListenerId, Multiaddr> =
         HashMap::new();
@@ -412,8 +547,16 @@ async fn run(
     > = HashMap::new();
     loop {
         tokio::select! {
+            _ = check.tick(), if !relaying.configured => {
+                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);
+                    }
+                }
                 Some(Command::Connect { peer, addresses, done }) => {
                     if swarm.is_connected(&peer) {
                         let _ = done.send(Ok(()));
@@ -532,6 +675,10 @@ async fn run(
                         warn!(relay = %relay, reason = ?reason, "Lost the slot on a libp2p relay; reserving again later");
                         reserve_later(relay);
                     }
+                    // A relay picked among the members: the next check picks again.
+                    if let Some(relay) = picked.remove(&listener_id) {
+                        debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying");
+                    }
                 }
                 SwarmEvent::ListenerError { error, .. } => {
                     warn!(error = %error, "libp2p listener failed");
@@ -543,6 +690,12 @@ async fn run(
                         info!(relay = %relay_peer_id, "Reachable through a libp2p relay");
                     }
                 }
+                SwarmEvent::Behaviour(BehaviourEvent::Autonat(
+                    ::libp2p::autonat::Event::StatusChanged { old, new },
+                )) => {
+                    info!(old = ?old, new = ?new, "libp2p reachability (AutoNAT)");
+                    nat = new;
+                }
                 SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => {
                     debug!(event = ?event, "libp2p relay");
                 }
@@ -611,20 +764,32 @@ impl Transport for Libp2pTransport {
     }
 
     /// Where we can be reached, each with our peer ID: the announced addresses, or else
-    /// those we listen on (`published`), and those through relays.
+    /// those we listen on (`published`), and those through relays. A relay adds each
+    /// direct one again, ending in `/p2p-circuit`: where to reserve a slot.
     fn local_addresses(&self) -> Vec<String> {
         let book = lock(&self.book);
-        let (relayed, direct): (Vec<Multiaddr>, Vec<Multiaddr>) =
-            book.listening.iter().cloned().partition(is_relayed);
-        let mut addresses = if book.announced.is_empty() {
-            published(&direct)
+        let direct: Vec<Multiaddr> = book
+            .direct()
+            .into_iter()
+            .filter_map(|address| address.with_p2p(self.peer_id).ok())
+            .collect();
+        let relayed = book
+            .listening
+            .iter()
+            .filter(|address| is_relayed(address))
+            .filter_map(|address| address.clone().with_p2p(self.peer_id).ok());
+        let offers: Vec<Multiaddr> = if book.serving {
+            direct
+                .iter()
+                .map(|address| address.clone().with(Protocol::P2pCircuit))
+                .collect()
         } else {
-            book.announced.clone()
+            Vec::new()
         };
-        addresses.extend(relayed);
-        addresses
+        direct
             .into_iter()
-            .filter_map(|address| address.with_p2p(self.peer_id).ok())
+            .chain(relayed)
+            .chain(offers)
             .map(|address| address.to_string())
             .collect()
     }
@@ -634,7 +799,27 @@ impl Transport for Libp2pTransport {
         if let Ok(peer) = PeerId::from_str(target) {
             return Some(peer.to_string());
         }
-        let address = Multiaddr::from_str(target).ok()?;
+        let mut address = Multiaddr::from_str(target).ok()?;
+        // A relay saying where to reserve a slot: `<address>/p2p/<relay>/p2p-circuit`.
+        if let Some(Protocol::P2pCircuit) = address.iter().last() {
+            address.pop();
+            let relay = split_peer(&address)?;
+            if relay != self.peer_id {
+                let new = {
+                    let mut book = lock(&self.book);
+                    let new = book.relays.get(&relay) != Some(&address);
+                    if new && (book.relays.len() < MAX_PEERS || book.relays.contains_key(&relay)) {
+                        book.relays.insert(relay, address.clone());
+                    }
+                    book.remember(relay, address);
+                    new
+                };
+                if new {
+                    let _ = self.commands.send(Command::CheckRelays);
+                }
+            }
+            return Some(relay.to_string());
+        }
         let peer = split_peer(&address)?;
         if peer != self.peer_id {
             lock(&self.book).remember(peer, address);
@@ -987,6 +1172,63 @@ mod tests {
         assert_eq!(from, a.local_id());
     }
 
+    fn peer() -> PeerId {
+        Keypair::generate_ed25519().public().to_peer_id()
+    }
+
+    #[test]
+    fn relays_are_picked_when_needed() {
+        use ::libp2p::autonat::NatStatus;
+        let local = peer();
+        let offers: HashMap<PeerId, Multiaddr> = (0..4)
+            .map(|port| (peer(), format!("/memory/{port}").parse().unwrap()))
+            .collect();
+        let check = |direct, nat: &NatStatus, using: &[PeerId]| {
+            relay_check(local, direct, nat, &offers, using)
+        };
+        // Dialable, or not known yet with an address to dial: none.
+        assert!(check(true, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty());
+        assert!(check(false, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty());
+        assert!(check(true, &NatStatus::Unknown, &[]).is_empty());
+        // No address to dial, or AutoNAT says it can't be dialled: some.
+        assert_eq!(check(false, &NatStatus::Unknown, &[]).len(), AUTO_RELAYS);
+        let picked = check(true, &NatStatus::Private, &[]);
+        assert_eq!(picked.len(), AUTO_RELAYS);
+        // One in use: the next in the same order.
+        let more = check(true, &NatStatus::Private, &[picked[0].0]);
+        assert_eq!(more.len(), AUTO_RELAYS - 1);
+        assert_eq!(more[0].0, picked[1].0);
+        let all: Vec<PeerId> = picked.iter().map(|(relay, _)| *relay).collect();
+        assert!(check(true, &NatStatus::Private, &all).is_empty());
+        // Never itself.
+        let only_us = HashMap::from([(local, Multiaddr::empty())]);
+        assert!(relay_check(local, false, &NatStatus::Private, &only_us, &[]).is_empty());
+    }
+
+    /// A relay says so in its addresses; b, which nobody can dial, learns of it (as from
+    /// the member list) and becomes reachable through it.
+    #[tokio::test]
+    async fn a_member_that_relays_is_picked() {
+        let (r, _) = memory_node_with(true, Vec::new(), true);
+        assert!(eventually(|| !r.local_addresses().is_empty()).await);
+        let offer = r
+            .local_addresses()
+            .into_iter()
+            .find(|address| address.ends_with("/p2p-circuit"))
+            .expect("the relay offers itself");
+        let (b, _) = memory_node_with(false, Vec::new(), false);
+        assert_eq!(b.locate(&offer).as_deref(), Some(r.local_id()));
+        assert!(
+            eventually(|| b
+                .local_addresses()
+                .iter()
+                .any(|address| address.contains("/p2p-circuit/")))
+            .await,
+            "b isn't reachable through r: {:?}",
+            b.local_addresses()
+        );
+    }
+
     #[tokio::test]
     async fn unknown_peers_are_refused() {
         let (b, _) = memory_node();