add support for public libp2p infrastructure

4 files changed, 143 insertions, 20 deletions

diff --git a/src/config/mod.rs b/src/config/mod.rs
index 08a3742..bd45390 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -278,6 +278,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "With the libp2p transport: find nodes on the LAN (multicast DNS, IPv4). Those that speak continuum are seeds while joining, so nodes on one LAN find each other without configured seeds; a swarm key keeps a new node from joining another swarm on the same LAN.",
     ),
     option(
+        "libp2p-public",
+        Kind::Flag,
+        "BOOL",
+        Some("false"),
+        "With the libp2p transport: use public libp2p infrastructure: join the IPFS DHT (through the IPFS bootstrap nodes) to find peers' addresses and public relays, picked after members that relay. Public relays only keep circuits for minutes and some KiB: enough to punch holes through NATs. Makes this node's peer ID and addresses known to the DHT.",
+    ),
+    option(
         "libp2p-key-file",
         Kind::Text,
         "PATH",
@@ -408,6 +415,9 @@ pub(crate) struct Settings {
     #[serde(default)]
     #[cfg_attr(not(feature = "libp2p"), allow(dead_code))]
     pub(crate) libp2p_mdns: bool,
+    #[serde(default)]
+    #[cfg_attr(not(feature = "libp2p"), allow(dead_code))]
+    pub(crate) libp2p_public: bool,
     #[cfg_attr(not(feature = "libp2p"), allow(dead_code))]
     pub(crate) libp2p_key_file: Option<PathBuf>,
     pub(crate) mycelium_rest_url: Option<String>,
diff --git a/src/main.rs b/src/main.rs
index b677021..f2494d8 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -216,6 +216,7 @@ async fn main() -> Result<()> {
                             relays,
                             relay: args.libp2p_relay,
                             mdns: args.libp2p_mdns,
+                            public: args.libp2p_public,
                             memory: false,
                         },
                     )?),
diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index c16c626..ed35389 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -198,6 +198,7 @@ async fn nodes_over_libp2p() {
                 relays: Vec::new(),
                 relay: false,
                 mdns: false,
+                public: false,
                 memory: true,
             })
             .unwrap();
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();