use libp2p mDNS

9 files changed, 164 insertions, 12 deletions

diff --git a/src/config/mod.rs b/src/config/mod.rs
index 2646e73..08a3742 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -271,6 +271,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "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-mdns",
+        Kind::Flag,
+        "BOOL",
+        Some("false"),
+        "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-key-file",
         Kind::Text,
         "PATH",
@@ -398,6 +405,9 @@ pub(crate) struct Settings {
     #[serde(default)]
     #[cfg_attr(not(feature = "libp2p"), allow(dead_code))]
     pub(crate) libp2p_relays: Vec<String>,
+    #[serde(default)]
+    #[cfg_attr(not(feature = "libp2p"), allow(dead_code))]
+    pub(crate) libp2p_mdns: 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 c933497..b677021 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -215,6 +215,7 @@ async fn main() -> Result<()> {
                             announce,
                             relays,
                             relay: args.libp2p_relay,
+                            mdns: args.libp2p_mdns,
                             memory: false,
                         },
                     )?),
diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index 21b5f42..c16c626 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -197,6 +197,7 @@ async fn nodes_over_libp2p() {
                 announce: Vec::new(),
                 relays: Vec::new(),
                 relay: false,
+                mdns: false,
                 memory: true,
             })
             .unwrap();
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index ad1de61..7c4c444 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -124,6 +124,17 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
         }
 
         let now = Instant::now();
+        // Nodes the transport found by itself (libp2p on the LAN): seeds too.
+        let found = state.transport.discovered_seeds();
+        if !found.is_empty() {
+            for seed in found {
+                if !unresolved.contains(&seed) {
+                    info!(seed = %seed, "Found a node nearby; joining through it");
+                    unresolved.push(seed);
+                }
+            }
+            next_resolve = now;
+        }
         if !unresolved.is_empty() && now >= next_resolve {
             resolve_seeds(state, &mut unresolved, &mut failing).await;
             if !unresolved.is_empty() {
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs
index 926acdd..e6816d6 100644
--- a/src/transport/firewall.rs
+++ b/src/transport/firewall.rs
@@ -88,6 +88,10 @@ impl Transport for FirewallTransport {
         self.inner.can_reach(peer)
     }
 
+    fn discovered_seeds(&self) -> Vec<String> {
+        self.inner.discovered_seeds()
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index bfd9432..04375f6 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -504,6 +504,20 @@ impl Transport for IdentityTransport {
         self.shared.destination(peer).is_some()
     }
 
+    /// Each transport's, with its scheme.
+    fn discovered_seeds(&self) -> Vec<String> {
+        self.shared
+            .links
+            .iter()
+            .flat_map(|link| {
+                link.inner
+                    .discovered_seeds()
+                    .into_iter()
+                    .map(|seed| link.address(&seed))
+            })
+            .collect()
+    }
+
     /// A node ID resolves to itself if this node knows where it is; anything else is
     /// an address (`<scheme>:...`) or what one of the transports resolves to one (the
     /// first that does), asked WHO.
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 41dac71..6ae3814 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -555,6 +555,12 @@ pub(crate) trait Transport: Send + Sync {
         vec![peer.to_owned()]
     }
 
+    /// Seeds the transport found by itself since last asked (libp2p: nodes on the LAN),
+    /// for a node that is joining. By default none.
+    fn discovered_seeds(&self) -> Vec<String> {
+        Vec::new()
+    }
+
     /// Whether this node has a way to reach `peer` at all: an address of it on one of
     /// its transports. Members it can't reach aren't its neighbours (`swarm::topology`).
     /// By default yes, as for `local_addresses`.
diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs
index 675ff13..e6abb0a 100644
--- a/src/transport/p2p.rs
+++ b/src/transport/p2p.rs
@@ -29,9 +29,13 @@
 //! (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). It never uses public libp2p infrastructure: relays are those configured, or
-//! members that say they relay.
+//! 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
+//! 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.
 
 use std::collections::HashMap;
 use std::path::Path;
@@ -105,6 +109,8 @@ pub(crate) struct Libp2pConfig {
     pub(crate) relays: Vec<Multiaddr>,
     /// Relay for other nodes.
     pub(crate) relay: bool,
+    /// Find peers on the LAN (mDNS).
+    pub(crate) mdns: bool,
     /// libp2p's in-memory transport instead of TCP and QUIC: for tests.
     pub(crate) memory: bool,
 }
@@ -166,9 +172,29 @@ struct Book {
     serving: bool,
     /// Members that say they relay, and where (with the relay's peer ID).
     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).
+    continuum: std::collections::HashSet<PeerId>,
+    /// Peers found on the LAN that speak continuum, with their address: seeds not
+    /// handed out yet (`discovered_seeds`).
+    discovered: Vec<Multiaddr>,
 }
 
 impl Book {
+    /// `peer`, found on the LAN, speaks continuum: a seed, at the address we know first.
+    fn discover(&mut self, peer: PeerId) {
+        let Some(address) = self.addresses.get(&peer).and_then(|known| known.first()) else {
+            return;
+        };
+        if let Ok(address) = address.clone().with_p2p(peer)
+            && !self.discovered.contains(&address)
+            && self.discovered.len() < MAX_PEERS
+        {
+            self.discovered.push(address);
+        }
+    }
+
     /// 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> {
@@ -262,9 +288,9 @@ impl Libp2pTransport {
     pub(crate) fn start(config: Libp2pConfig) -> Result<Self> {
         let peer_id = config.key.public().to_peer_id();
         let mut swarm = if config.memory {
-            memory_swarm(config.key, config.relay)?
+            memory_swarm(config.key, config.relay, config.mdns)?
         } else {
-            network_swarm(config.key, config.relay)?
+            network_swarm(config.key, config.relay, config.mdns)?
         };
         for address in config.listen {
             swarm
@@ -353,10 +379,17 @@ struct Behaviour {
     relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>,
     dcutr: ::libp2p::dcutr::Behaviour,
     autonat: ::libp2p::autonat::Behaviour,
+    /// Only with `Libp2pConfig::mdns`.
+    mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>,
 }
 
 impl Behaviour {
-    fn new(key: &Keypair, relay_client: ::libp2p::relay::client::Behaviour, relay: bool) -> Self {
+    fn new(
+        key: &Keypair,
+        relay_client: ::libp2p::relay::client::Behaviour,
+        relay: bool,
+        mdns: bool,
+    ) -> std::io::Result<Self> {
         let peer_id = key.public().to_peer_id();
         let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public())
             .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION")));
@@ -372,7 +405,14 @@ impl Behaviour {
                 },
             )
         });
-        Self {
+        let mdns = match mdns {
+            true => Some(::libp2p::mdns::tokio::Behaviour::new(
+                ::libp2p::mdns::Config::default(),
+                peer_id,
+            )?),
+            false => None,
+        };
+        Ok(Self {
             stream: libp2p_stream::Behaviour::new(),
             identify: ::libp2p::identify::Behaviour::new(identify),
             relay_client,
@@ -386,11 +426,12 @@ impl Behaviour {
                     ..Default::default()
                 },
             ),
-        }
+            mdns: mdns.into(),
+        })
     }
 }
 
-fn network_swarm(key: Keypair, relay: bool) -> Result<Swarm<Behaviour>> {
+fn network_swarm(key: Keypair, relay: bool, mdns: bool) -> Result<Swarm<Behaviour>> {
     Ok(::libp2p::SwarmBuilder::with_existing_identity(key)
         .with_tokio()
         .with_tcp(
@@ -404,12 +445,15 @@ fn network_swarm(key: Keypair, relay: bool) -> Result<Swarm<Behaviour>> {
             ::libp2p::noise::Config::new,
             ::libp2p::yamux::Config::default,
         )?
-        .with_behaviour(|key, relay_client| Behaviour::new(key, relay_client, relay))?
+        .with_behaviour(|key, relay_client| {
+            Behaviour::new(key, relay_client, relay, mdns)
+                .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) -> Result<Swarm<Behaviour>> {
+fn memory_swarm(key: Keypair, relay: bool, mdns: bool) -> Result<Swarm<Behaviour>> {
     use ::libp2p::Transport as _;
     use ::libp2p::core::transport::{MemoryTransport, OrTransport};
     use ::libp2p::core::upgrade::Version;
@@ -425,7 +469,7 @@ fn memory_swarm(key: Keypair, relay: bool) -> Result<Swarm<Behaviour>> {
         .boxed();
     Ok(Swarm::new(
         transport,
-        Behaviour::new(&key, relay_client, relay),
+        Behaviour::new(&key, relay_client, relay, mdns)?,
         peer_id,
         ::libp2p::swarm::Config::with_tokio_executor().with_idle_connection_timeout(IDLE_TIMEOUT),
     ))
@@ -718,6 +762,39 @@ async fn run(
                     for address in published(&info.listen_addrs).into_iter().rev() {
                         book.remember(peer_id, address);
                     }
+                    if info.protocols.contains(&PROTOCOL) {
+                        book.continuum.insert(peer_id);
+                        if book.lan.remove(&peer_id) {
+                            book.discover(peer_id);
+                        }
+                    }
+                }
+                SwarmEvent::Behaviour(BehaviourEvent::Mdns(
+                    ::libp2p::mdns::Event::Discovered(found),
+                )) => {
+                    for (peer, address) in found {
+                        if peer == relaying.local {
+                            continue;
+                        }
+                        debug!(peer = %peer, address = %address, "libp2p peer on the LAN");
+                        let continuum = {
+                            let mut book = lock(&book);
+                            book.remember(peer, address.clone());
+                            book.continuum.contains(&peer)
+                        };
+                        if continuum {
+                            lock(&book).discover(peer);
+                        } else if lock(&book).lan.len() < MAX_PEERS {
+                            // Connected, identify says whether it speaks continuum.
+                            lock(&book).lan.insert(peer);
+                            let _ = swarm.dial(
+                                DialOpts::peer_id(peer)
+                                    .addresses(vec![address])
+                                    .condition(PeerCondition::DisconnectedAndNotDialing)
+                                    .build(),
+                            );
+                        }
+                    }
                 }
                 _ => {}
             },
@@ -827,6 +904,14 @@ impl Transport for Libp2pTransport {
         Some(peer.to_string())
     }
 
+    /// Nodes found on the LAN (mDNS) that speak continuum.
+    fn discovered_seeds(&self) -> Vec<String> {
+        std::mem::take(&mut lock(&self.book).discovered)
+            .iter()
+            .map(Multiaddr::to_string)
+            .collect()
+    }
+
     /// Where `target` is, each address with its peer ID.
     fn locations(&self, target: &str) -> Vec<String> {
         let Ok(peer) = PeerId::from_str(target) else {
@@ -980,6 +1065,7 @@ mod tests {
             announce: Vec::new(),
             relays,
             relay,
+            mdns: false,
             memory: true,
         })
         .unwrap();
@@ -1117,6 +1203,7 @@ mod tests {
             announce: vec![announced.clone()],
             relays: Vec::new(),
             relay: false,
+            mdns: false,
             memory: true,
         })
         .unwrap();
@@ -1229,6 +1316,20 @@ mod tests {
         );
     }
 
+    /// A peer on the LAN that speaks continuum becomes a seed once, at its address.
+    #[test]
+    fn lan_peers_become_seeds_once() {
+        let mut book = Book::default();
+        let lan = peer();
+        let address: Multiaddr = "/ip4/192.168.1.5/tcp/9661".parse().unwrap();
+        book.discover(lan); // No address known: nothing.
+        assert!(book.discovered.is_empty());
+        book.remember(lan, address.clone());
+        book.discover(lan);
+        book.discover(lan);
+        assert_eq!(book.discovered, [address.with_p2p(lan).unwrap()]);
+    }
+
     #[tokio::test]
     async fn unknown_peers_are_refused() {
         let (b, _) = memory_node();
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 7e4dac7..83b0cff 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -165,6 +165,10 @@ impl Transport for CountingTransport {
         self.inner.can_reach(peer)
     }
 
+    fn discovered_seeds(&self) -> Vec<String> {
+        self.inner.discovered_seeds()
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }