Merge remote-tracking branch 'origin/main' into feat/dns

Author: Frando

Cargo.lock

@@ -40,6 +40,10 @@ pub(crate) struct Cli {
     #[cfg(unix)]
     #[clap(long)]
     pub(crate) log_fd: Option<i32>,
+
+    /// Port to serve metrics on. -1 to disable.
+    #[clap(long)]
+    pub(crate) metrics_port: Option<i16>,
 }
 
 #[derive(Parser, Debug, Clone)]
@@ -127,7 +131,14 @@ impl Cli {
                         path.display()
                     );
                 }
-                let config = NodeConfig::from_env(self.config.as_deref())?;
+                let mut config = NodeConfig::from_env(self.config.as_deref())?;
+                if let Some(metrics_port) = self.metrics_port {
+                    if metrics_port < 0 {
+                        config.metrics_addr = None;
+                    } else {
+                        config.metrics_addr = Some(([127, 0, 0, 1], metrics_port as u16).into())
+                    }
+                }
 
                 let add_command = add.map(|source| blob::BlobCommands::Add {
                     source,

iroh-cli/src/commands.rs

@@ -99,7 +99,7 @@ impl Default for NodeConfig {
             // TODO(ramfox): this should probably just be a relay map
             relay_nodes: [default_na_relay_node(), default_eu_relay_node()].into(),
             gc_policy: GcPolicy::Disabled,
-            metrics_addr: None,
+            metrics_addr: Some(([127, 0, 0, 1], 9090).into()),
         }
     }
 }

iroh-cli/src/config.rs

@@ -36,7 +36,7 @@ http = "1"
 http-body-util = "0.1.0"
 hyper = { version = "1", features = ["server", "client", "http1"] }
 hyper-util = "0.1.1"
-igd = { version = "0.12.1", features = ["aio"] }
+igd-next = { version = "0.14.3", features = ["aio_tokio"] }
 iroh-base = { version = "0.13.0", path = "../iroh-base", features = ["key"] }
 libc = "0.2.139"
 num_enum = "0.7"

iroh-net/Cargo.toml

@@ -210,7 +210,7 @@ impl Client {
 struct Probe {
     /// When was the probe last updated.
     last_probe: Instant,
-    /// The last [`igd::aio::Gateway`] and when was it last seen.
+    /// The last [`upnp::Gateway`] and when was it last seen.
     last_upnp_gateway_addr: Option<(upnp::Gateway, Instant)>,
     /// Last time PCP was seen.
     last_pcp: Option<Instant>,

iroh-net/src/portmapper.rs

@@ -57,7 +57,7 @@ impl Mapping {
     pub(crate) async fn new_upnp(
         local_ip: Ipv4Addr,
         local_port: NonZeroU16,
-        gateway: Option<igd::aio::Gateway>,
+        gateway: Option<upnp::Gateway>,
         external_port: Option<NonZeroU16>,
     ) -> Result<Self> {
         upnp::Mapping::new(local_ip, local_port, gateway, external_port)

iroh-net/src/portmapper/mapping.rs

@@ -4,15 +4,15 @@ use std::{
     time::Duration,
 };
 
-use anyhow::Result;
-use igd::aio as aigd;
+use anyhow::{anyhow, Result};
+use igd_next::aio as aigd;
 
 use iroh_metrics::inc;
 use tracing::debug;
 
 use super::Metrics;
 
-pub use aigd::Gateway;
+pub type Gateway = aigd::Gateway<aigd::tokio::Tokio>;
 
 /// Seconds we ask the router to maintain the port mapping. 0 means infinite.
 const PORT_MAPPING_LEASE_DURATION_SECONDS: u32 = 0;
@@ -31,7 +31,7 @@ const PORT_MAPPING_DESCRIPTION: &str = "iroh-portmap";
 pub struct Mapping {
     /// The internet Gateway device (router) used to create this mapping.
     #[debug("{}", gateway)]
-    gateway: aigd::Gateway,
+    gateway: Gateway,
     /// The external address obtained by this mapping.
     external_ip: Ipv4Addr,
     /// External port obtained by this mapping.
@@ -42,7 +42,7 @@ impl Mapping {
     pub(crate) async fn new(
         local_addr: Ipv4Addr,
         port: NonZeroU16,
-        gateway: Option<aigd::Gateway>,
+        gateway: Option<Gateway>,
         preferred_port: Option<NonZeroU16>,
     ) -> Result<Self> {
         let local_addr = SocketAddrV4::new(local_addr, port.into());
@@ -51,23 +51,25 @@ impl Mapping {
         let gateway = if let Some(known_gateway) = gateway {
             known_gateway
         } else {
-            aigd::search_gateway(igd::SearchOptions {
+            aigd::tokio::search_gateway(igd_next::SearchOptions {
                 timeout: Some(SEARCH_TIMEOUT),
                 ..Default::default()
             })
             .await?
         };
 
-        let external_ip = gateway.get_external_ip().await?;
+        let std::net::IpAddr::V4(external_ip) = gateway.get_external_ip().await? else {
+            return Err(anyhow!("igd device's external ip is ipv6"));
+        };
 
         // if we are trying to get a specific external port, try this first. If this fails, default
         // to try to get any port
         if let Some(external_port) = preferred_port {
             if gateway
                 .add_port(
-                    igd::PortMappingProtocol::UDP,
+                    igd_next::PortMappingProtocol::UDP,
                     external_port.into(),
-                    local_addr,
+                    local_addr.into(),
                     PORT_MAPPING_LEASE_DURATION_SECONDS,
                     PORT_MAPPING_DESCRIPTION,
                 )
@@ -84,8 +86,8 @@ impl Mapping {
 
         let external_port = gateway
             .add_any_port(
-                igd::PortMappingProtocol::UDP,
-                local_addr,
+                igd_next::PortMappingProtocol::UDP,
+                local_addr.into(),
                 PORT_MAPPING_LEASE_DURATION_SECONDS,
                 PORT_MAPPING_DESCRIPTION,
             )
@@ -112,7 +114,7 @@ impl Mapping {
             ..
         } = self;
         gateway
-            .remove_port(igd::PortMappingProtocol::UDP, external_port.into())
+            .remove_port(igd_next::PortMappingProtocol::UDP, external_port.into())
             .await?;
         Ok(())
     }
@@ -126,7 +128,7 @@ impl Mapping {
 /// Searches for UPnP gateways.
 pub async fn probe_available() -> Option<Gateway> {
     inc!(Metrics, upnp_probes);
-    match aigd::search_gateway(igd::SearchOptions {
+    match aigd::tokio::search_gateway(igd_next::SearchOptions {
         timeout: Some(SEARCH_TIMEOUT),
         ..Default::default()
     })

iroh-net/src/portmapper/upnp.rs

@@ -57,7 +57,7 @@ clap = { version = "4", features = ["derive"], optional = true }
 indicatif = { version = "0.17", features = ["tokio"], optional = true }
 
 [features]
-default = ["metrics"]
+default = ["metrics", "fs-store"]
 metrics = ["iroh-metrics", "iroh-bytes/metrics"]
 fs-store = ["iroh-bytes/fs-store"]
 test = []

iroh/Cargo.toml

@@ -277,7 +277,7 @@ impl<D> NodeInner<D> {
     }
 }
 
-#[cfg(all(test, feature = "fs-store"))]
+#[cfg(test)]
 mod tests {
     use std::path::Path;
     use std::time::Duration;
@@ -387,4 +387,24 @@ mod tests {
 
         Ok(())
     }
+
+    #[cfg(feature = "fs-store")]
+    #[tokio::test]
+    async fn test_shutdown() -> Result<()> {
+        let _guard = iroh_test::logging::setup();
+
+        let iroh_root = tempfile::TempDir::new()?;
+        {
+            let iroh = Node::persistent(iroh_root.path()).await?.spawn().await?;
+            let doc = iroh.docs.create().await?;
+            drop(doc);
+            iroh.shutdown();
+            iroh.await?;
+        }
+
+        let iroh = Node::persistent(iroh_root.path()).await?.spawn().await?;
+        let _doc = iroh.docs.create().await?;
+
+        Ok(())
+    }
 }

iroh/src/node.rs

@@ -492,13 +492,15 @@ where
             debug!(me = ?server.node_id(), "gossip initial update: {local_endpoints:?}");
             gossip.update_endpoints(&local_endpoints).ok();
         }
-
         loop {
             tokio::select! {
                 biased;
                 _ = cancel_token.cancelled() => {
                     // clean shutdown of the blobs db to close the write transaction
                     handler.inner.db.shutdown().await;
+                    if let Err(err) = handler.inner.sync.shutdown().await {
+                        warn!("sync shutdown error: {:?}", err);
+                    }
                     break
                 },
                 // handle rpc requests. This will do nothing if rpc is not configured, since