add support for public libp2p infrastructure

1 files changed, 131 insertions, 20 deletions

diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs
index e6abb0a..5829419 100644
--- a/src/transport/p2p.rs
+++ b/src/transport/p2p.rs
@@ -34,8 +34,12 @@
 //! so) are seeds while the node joins (`discovered_seeds`): nodes on one LAN find each
 //! other without configured seeds.
 //!
-//! It never uses public libp2p infrastructure: relays are those configured, or members
-//! that say they relay.
+//! It uses public libp2p infrastructure only with `--libp2p-public`: then it joins the
+//! IPFS DHT (Kademlia, through the IPFS bootstrap nodes), where it finds addresses of
+//! peers the member list doesn't have, and public peers that relay; those are picked
+//! after members that relay. Public relays keep circuits short (by default 2 minutes and
+//! 128 KiB): enough to punch holes through NATs, not to carry a swarm's traffic. Without
+//! it, relays are those configured, or members that say they relay.
 
 use std::collections::HashMap;
 use std::path::Path;
@@ -111,6 +115,8 @@ pub(crate) struct Libp2pConfig {
     pub(crate) relay: bool,
     /// Find peers on the LAN (mDNS).
     pub(crate) mdns: bool,
+    /// Use public infrastructure: the IPFS DHT, public relays.
+    pub(crate) public: bool,
     /// libp2p's in-memory transport instead of TCP and QUIC: for tests.
     pub(crate) memory: bool,
 }
@@ -172,6 +178,9 @@ struct Book {
     serving: bool,
     /// 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
+    /// where.
+    public_relays: HashMap<PeerId, Multiaddr>,
     /// Peers found on the LAN (mDNS) not known yet to speak continuum.
     lan: std::collections::HashSet<PeerId>,
     /// Peers that speak continuum (identify says so).
@@ -287,10 +296,11 @@ impl Libp2pTransport {
     /// Builds the swarm, listens, and runs it in a task of its own.
     pub(crate) fn start(config: Libp2pConfig) -> Result<Self> {
         let peer_id = config.key.public().to_peer_id();
+        let features = Features::of(&config);
         let mut swarm = if config.memory {
-            memory_swarm(config.key, config.relay, config.mdns)?
+            memory_swarm(config.key, features)?
         } else {
-            network_swarm(config.key, config.relay, config.mdns)?
+            network_swarm(config.key, features)?
         };
         for address in config.listen {
             swarm
@@ -327,6 +337,7 @@ impl Libp2pTransport {
             Relaying {
                 serving: config.relay,
                 configured: configured_relays,
+                public: config.public,
                 local: peer_id,
             },
             stopping,
@@ -381,16 +392,67 @@ struct Behaviour {
     autonat: ::libp2p::autonat::Behaviour,
     /// Only with `Libp2pConfig::mdns`.
     mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>,
+    /// Only with `Libp2pConfig::public`: the IPFS DHT.
+    kad: ::libp2p::swarm::behaviour::toggle::Toggle<
+        ::libp2p::kad::Behaviour<::libp2p::kad::store::MemoryStore>,
+    >,
 }
 
+/// What the behaviour has besides what every node has.
+#[derive(Clone, Copy)]
+struct Features {
+    relay: bool,
+    mdns: bool,
+    public: bool,
+}
+
+impl Features {
+    fn of(config: &Libp2pConfig) -> Self {
+        Self {
+            relay: config.relay,
+            mdns: config.mdns,
+            public: config.public,
+        }
+    }
+}
+
+/// The IPFS DHT's protocol, and its bootstrap nodes (all at `IPFS_BOOTSTRAP_ADDRESS`).
+const IPFS_DHT: StreamProtocol = StreamProtocol::new("/ipfs/kad/1.0.0");
+const IPFS_BOOTSTRAP_ADDRESS: &str = "/dnsaddr/bootstrap.libp2p.io";
+const IPFS_BOOTSTRAP_PEERS: [&str; 4] = [
+    "QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
+    "QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
+    "QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
+    "QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
+];
+/// How often a node on the DHT refreshes its routing table.
+const DHT_BOOTSTRAP: Duration = Duration::from_secs(300);
+
 impl Behaviour {
     fn new(
         key: &Keypair,
         relay_client: ::libp2p::relay::client::Behaviour,
-        relay: bool,
-        mdns: bool,
+        features: Features,
     ) -> std::io::Result<Self> {
+        let Features {
+            relay,
+            mdns,
+            public,
+        } = features;
         let peer_id = key.public().to_peer_id();
+        let kad = public.then(|| {
+            let mut dht = ::libp2p::kad::Behaviour::with_config(
+                peer_id,
+                ::libp2p::kad::store::MemoryStore::new(peer_id),
+                ::libp2p::kad::Config::new(IPFS_DHT),
+            );
+            let bootstrap: Multiaddr = IPFS_BOOTSTRAP_ADDRESS.parse().expect("a multiaddr");
+            for peer in IPFS_BOOTSTRAP_PEERS {
+                let peer: PeerId = peer.parse().expect("a peer ID");
+                let _ = dht.add_address(&peer, bootstrap.clone());
+            }
+            dht
+        });
         let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public())
             .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION")));
         let relay = relay.then(|| {
@@ -427,11 +489,12 @@ impl Behaviour {
                 },
             ),
             mdns: mdns.into(),
+            kad: kad.into(),
         })
     }
 }
 
-fn network_swarm(key: Keypair, relay: bool, mdns: bool) -> Result<Swarm<Behaviour>> {
+fn network_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> {
     Ok(::libp2p::SwarmBuilder::with_existing_identity(key)
         .with_tokio()
         .with_tcp(
@@ -446,14 +509,14 @@ fn network_swarm(key: Keypair, relay: bool, mdns: bool) -> Result<Swarm<Behaviou
             ::libp2p::yamux::Config::default,
         )?
         .with_behaviour(|key, relay_client| {
-            Behaviour::new(key, relay_client, relay, mdns)
+            Behaviour::new(key, relay_client, features)
                 .map_err(|error| Box::new(error) as Box<dyn std::error::Error + Send + Sync>)
         })?
         .with_swarm_config(|config| config.with_idle_connection_timeout(IDLE_TIMEOUT))
         .build())
 }
 
-fn memory_swarm(key: Keypair, relay: bool, mdns: bool) -> Result<Swarm<Behaviour>> {
+fn memory_swarm(key: Keypair, features: Features) -> Result<Swarm<Behaviour>> {
     use ::libp2p::Transport as _;
     use ::libp2p::core::transport::{MemoryTransport, OrTransport};
     use ::libp2p::core::upgrade::Version;
@@ -469,7 +532,7 @@ fn memory_swarm(key: Keypair, relay: bool, mdns: bool) -> Result<Swarm<Behaviour
         .boxed();
     Ok(Swarm::new(
         transport,
-        Behaviour::new(&key, relay_client, relay, mdns)?,
+        Behaviour::new(&key, relay_client, features)?,
         peer_id,
         ::libp2p::swarm::Config::with_tokio_executor().with_idle_connection_timeout(IDLE_TIMEOUT),
     ))
@@ -483,6 +546,8 @@ struct Relaying {
     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,
 }
 
@@ -538,12 +603,21 @@ fn pick_relays(
         }
         return;
     }
-    let (direct, offers) = {
+    let (direct, offers, public) = {
         let book = lock(book);
-        (!book.direct().is_empty(), book.relays.clone())
+        let public = if relaying.public {
+            book.public_relays.clone()
+        } else {
+            HashMap::new()
+        };
+        (!book.direct().is_empty(), book.relays.clone(), public)
     };
-    let using: Vec<PeerId> = picked.values().copied().collect();
-    for (relay, address) in relay_check(relaying.local, direct, nat, &offers, &using) {
+    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 = %relay, address = %address, "Picked a member as libp2p relay");
@@ -556,6 +630,13 @@ fn pick_relays(
     }
 }
 
+/// 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() {
+        dht.get_closest_peers(peer);
+    }
+}
+
 /// `again`: to schedule commands.
 async fn run(
     mut swarm: Swarm<Behaviour>,
@@ -571,6 +652,8 @@ async fn run(
     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.
     let mut reservations: HashMap<::libp2p::core::transport::ListenerId, Multiaddr> =
         HashMap::new();
@@ -591,6 +674,13 @@ async fn run(
     > = HashMap::new();
     loop {
         tokio::select! {
+            _ = dht_bootstrap.tick(), if relaying.public => {
+                if let Some(dht) = swarm.behaviour_mut().kad.as_mut()
+                    && let Err(error) = dht.bootstrap()
+                {
+                    warn!(error = %error, "Couldn't join the IPFS DHT");
+                }
+            }
             _ = check.tick(), if !relaying.configured => {
                 pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
             }
@@ -611,14 +701,18 @@ async fn run(
                     if waiting.len() > 1 {
                         continue; // Already dialing.
                     }
-                    let opts = DialOpts::peer_id(peer)
+                    let mut opts = DialOpts::peer_id(peer)
                         .addresses(addresses)
-                        .condition(PeerCondition::DisconnectedAndNotDialing)
-                        .build();
-                    if let Err(error) = swarm.dial(opts) {
+                        .condition(PeerCondition::DisconnectedAndNotDialing);
+                    if relaying.public {
+                        // Addresses the DHT knows too.
+                        opts = opts.extend_addresses_through_behaviour();
+                    }
+                    if let Err(error) = swarm.dial(opts.build()) {
                         for done in connecting.remove(&peer).into_iter().flatten() {
                             let _ = done.send(Err(error.to_string()));
                         }
+                        find_on_dht(&mut swarm, peer);
                     }
                 }
                 Some(Command::Reserve { relay, configured }) if split_peer(&relay).is_none() => {
@@ -688,8 +782,11 @@ async fn run(
                     if let Some(peer) = peer_id
                         && !swarm.is_connected(&peer)
                     {
-                        for done in connecting.remove(&peer).into_iter().flatten() {
-                            let _ = done.send(Err(error.to_string()));
+                        if let Some(waiting) = connecting.remove(&peer) {
+                            for done in waiting {
+                                let _ = done.send(Err(error.to_string()));
+                            }
+                            find_on_dht(&mut swarm, peer);
                         }
                     }
                 }
@@ -762,6 +859,18 @@ async fn run(
                     for address in published(&info.listen_addrs).into_iter().rev() {
                         book.remember(peer_id, address);
                     }
+                    if relaying.public
+                        && info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME)
+                        && (book.public_relays.len() < MAX_PEERS
+                            || book.public_relays.contains_key(&peer_id))
+                        && let Some(address) = book
+                            .addresses
+                            .get(&peer_id)
+                            .and_then(|known| known.first())
+                            .and_then(|address| address.clone().with_p2p(peer_id).ok())
+                    {
+                        book.public_relays.insert(peer_id, address);
+                    }
                     if info.protocols.contains(&PROTOCOL) {
                         book.continuum.insert(peer_id);
                         if book.lan.remove(&peer_id) {
@@ -1066,6 +1175,7 @@ mod tests {
             relays,
             relay,
             mdns: false,
+            public: false,
             memory: true,
         })
         .unwrap();
@@ -1204,6 +1314,7 @@ mod tests {
             relays: Vec::new(),
             relay: false,
             mdns: false,
+            public: false,
             memory: true,
         })
         .unwrap();