protocols/mdns/: Generate peer expiry and fix IPv6 support (#2359)

Co-authored-by: Victor Ermolaev <victorermolaev@gmail.com>
Co-authored-by: Max Inden <mail@max-inden.de>

protocols/mdns/CHANGELOG.md

@@ -11,10 +11,14 @@
 
 - Migrate to Rust edition 2021 (see [PR 2339]).
 
+- Fix generation of peer expiration event and listen on specified IP version (see [PR 2359]).
+
 [PR 2339]: https://github.com/libp2p/rust-libp2p/pull/2339
 
 [PR 2311]: https://github.com/libp2p/rust-libp2p/pull/2311/
 
+[PR 2359]: https://github.com/libp2p/rust-libp2p/pull/2359
+
 # 0.33.0 [2021-11-16]
 
 - Update dependencies.

protocols/mdns/Cargo.toml

@@ -28,4 +28,4 @@ void = "1.0.2"
 [dev-dependencies]
 async-std = { version = "1.9.0", features = ["attributes"] }
 libp2p = { path = "../.." }
-tokio = { version = "1.2.0", default-features = false, features = ["macros", "rt", "rt-multi-thread"] }
+tokio = { version = "1.2.0", default-features = false, features = ["macros", "rt", "rt-multi-thread", "time"] }

protocols/mdns/src/behaviour.rs

@@ -144,12 +144,12 @@ impl Mdns {
             }
         };
         let send_socket = {
-            let addrs = [
+            let addr = match config.multicast_addr {
-                SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
+                IpAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
-                SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0),
+                IpAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0),
-            ];
+            };
 
-            let socket = std::net::UdpSocket::bind(&addrs[..])?;
+            let socket = std::net::UdpSocket::bind(addr)?;
             Async::new(socket)?
         };
         let if_watch = if_watch::IfWatcher::new().await?;
@@ -406,7 +406,7 @@ impl NetworkBehaviour for Mdns {
                 while let Some(pos) = self
                     .discovered_nodes
                     .iter()
-                    .position(|(_, _, exp)| *exp < now)
+                    .position(|(_, _, exp)| *exp <= now)
                 {
                     let (peer_id, addr, _) = self.discovered_nodes.remove(pos);
                     expired.push((peer_id, addr));

protocols/mdns/tests/smoke.rs

@@ -26,6 +26,7 @@ use libp2p::{
     PeerId,
 };
 use std::error::Error;
+use std::time::Duration;
 
 async fn create_swarm(config: MdnsConfig) -> Result<Swarm<Mdns>, Box<dyn Error>> {
     let id_keys = identity::Keypair::generate_ed25519();
@@ -37,7 +38,7 @@ async fn create_swarm(config: MdnsConfig) -> Result<Swarm<Mdns>, Box<dyn Error>>
     Ok(swarm)
 }
 
-async fn run_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
+async fn run_discovery_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
     let mut a = create_swarm(config.clone()).await?;
     let mut b = create_swarm(config).await?;
     let mut discovered_a = false;
@@ -78,24 +79,109 @@ async fn run_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
 
 #[async_std::test]
 async fn test_discovery_async_std_ipv4() -> Result<(), Box<dyn Error>> {
-    run_test(MdnsConfig::default()).await
+    run_discovery_test(MdnsConfig::default()).await
 }
 
 #[async_std::test]
 async fn test_discovery_async_std_ipv6() -> Result<(), Box<dyn Error>> {
     let mut config = MdnsConfig::default();
     config.multicast_addr = *IPV6_MDNS_MULTICAST_ADDRESS;
-    run_test(MdnsConfig::default()).await
+    run_discovery_test(config).await
 }
 
 #[tokio::test]
 async fn test_discovery_tokio_ipv4() -> Result<(), Box<dyn Error>> {
-    run_test(MdnsConfig::default()).await
+    run_discovery_test(MdnsConfig::default()).await
 }
 
 #[tokio::test]
 async fn test_discovery_tokio_ipv6() -> Result<(), Box<dyn Error>> {
     let mut config = MdnsConfig::default();
     config.multicast_addr = *IPV6_MDNS_MULTICAST_ADDRESS;
-    run_test(MdnsConfig::default()).await
+    run_discovery_test(config).await
+}
+
+async fn run_peer_expiration_test(config: MdnsConfig) -> Result<(), Box<dyn Error>> {
+    let mut a = create_swarm(config.clone()).await?;
+    let mut b = create_swarm(config).await?;
+
+    loop {
+        futures::select! {
+            ev = a.select_next_some() => match ev {
+                SwarmEvent::Behaviour(MdnsEvent::Expired(peers)) => {
+                    for (peer, _addr) in peers {
+                        if peer == *b.local_peer_id() {
+                            return Ok(());
+                        }
+                    }
+                }
+                _ => {}
+            },
+            ev = b.select_next_some() => match ev {
+                SwarmEvent::Behaviour(MdnsEvent::Expired(peers)) => {
+                    for (peer, _addr) in peers {
+                        if peer == *a.local_peer_id() {
+                            return Ok(());
+                        }
+                    }
+                }
+                _ => {}
+            }
+
+        }
+    }
+}
+
+#[async_std::test]
+async fn test_expired_async_std_ipv4() -> Result<(), Box<dyn Error>> {
+    let config = MdnsConfig {
+        ttl: Duration::from_millis(500),
+        query_interval: Duration::from_secs(1),
+        ..Default::default()
+    };
+
+    async_std::future::timeout(Duration::from_secs(6), run_peer_expiration_test(config))
+        .await
+        .map(|_| ())
+        .map_err(|e| Box::new(e) as Box<dyn Error>)
+}
+
+#[async_std::test]
+async fn test_expired_async_std_ipv6() -> Result<(), Box<dyn Error>> {
+    let config = MdnsConfig {
+        ttl: Duration::from_millis(500),
+        query_interval: Duration::from_secs(1),
+        multicast_addr: *IPV6_MDNS_MULTICAST_ADDRESS,
+    };
+
+    async_std::future::timeout(Duration::from_secs(6), run_peer_expiration_test(config))
+        .await
+        .map(|_| ())
+        .map_err(|e| Box::new(e) as Box<dyn Error>)
+}
+
+#[tokio::test]
+async fn test_expired_tokio_ipv4() -> Result<(), Box<dyn Error>> {
+    let config = MdnsConfig {
+        ttl: Duration::from_millis(500),
+        query_interval: Duration::from_secs(1),
+        ..Default::default()
+    };
+
+    tokio::time::timeout(Duration::from_secs(6), run_peer_expiration_test(config))
+        .await
+        .unwrap()
+}
+
+#[tokio::test]
+async fn test_expired_tokio_ipv6() -> Result<(), Box<dyn Error>> {
+    let config = MdnsConfig {
+        ttl: Duration::from_millis(500),
+        query_interval: Duration::from_secs(1),
+        multicast_addr: *IPV6_MDNS_MULTICAST_ADDRESS,
+    };
+
+    tokio::time::timeout(Duration::from_secs(6), run_peer_expiration_test(config))
+        .await
+        .unwrap()
 }