[WIP] feat: use quic multipath

Cargo.toml

@@ -40,3 +40,12 @@ unexpected_cfgs = { level = "warn", check-cfg = ["cfg(iroh_docsrs)", "cfg(iroh_l
 
 [workspace.lints.clippy]
 unused-async = "warn"
+
+
+[patch.crates-io]
+netwatch = { git = "https://github.com/n0-computer/net-tools", branch = "feat-multipath" }
+
+[patch."https://github.com/n0-computer/quinn"]
+# iroh-quinn = { path = "../iroh-quinn/quinn" }
+# iroh-quinn-proto = { path = "../iroh-quinn/quinn-proto" }
+# iroh-quinn-udp = { path = "../iroh-quinn/quinn-udp" }

iroh-base/src/endpoint_addr.rs

@@ -36,7 +36,9 @@ use crate::{EndpointId, PublicKey, RelayUrl};
 /// [discovery]: https://docs.rs/iroh/*/iroh/index.html#endpoint-discovery
 /// [home relay]: https://docs.rs/iroh/*/iroh/relay/index.html
 /// [Relay server]: https://docs.rs/iroh/*/iroh/index.html#relay-servers
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[derive(
+    derive_more::Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash,
+)]
 pub struct EndpointAddr {
     /// The endpoint's identifier.
     pub id: EndpointId,
@@ -54,6 +56,18 @@ pub enum TransportAddr {
     Ip(SocketAddr),
 }
 
+impl TransportAddr {
+    /// Whether this is a transport address via a relay server.
+    pub fn is_relay(&self) -> bool {
+        matches!(self, Self::Relay(_))
+    }
+
+    /// Whether this is an IP transport address.
+    pub fn is_ip(&self) -> bool {
+        matches!(self, Self::Ip(_))
+    }
+}
+
 impl EndpointAddr {
     /// Creates a new [`EndpointAddr`] with no network level addresses.
     ///

iroh-relay/Cargo.toml

@@ -42,8 +42,8 @@ postcard = { version = "1", default-features = false, features = [
     "use-std",
     "experimental-derive",
 ] }
-quinn = { package = "iroh-quinn", version = "0.14.0", default-features = false, features = ["rustls-ring"] }
-quinn-proto = { package = "iroh-quinn-proto", version = "0.13.0" }
+quinn = { package = "iroh-quinn", git = "https://github.com/n0-computer/quinn", branch = "multipath-quinn-0.11.x", default-features = false, features = ["rustls-ring"] }
+quinn-proto = { package = "iroh-quinn-proto", git = "https://github.com/n0-computer/quinn", branch = "multipath-quinn-0.11.x" }
 rand = "0.9.2"
 reqwest = { version = "0.12", default-features = false, features = [
     "rustls-tls",

iroh-relay/src/client.rs

@@ -281,7 +281,7 @@ impl ClientBuilder {
         let conn = Conn::new(conn, self.key_cache.clone(), &self.secret_key).await?;
 
         event!(
-            target: "events.net.relay.connected",
+            target: "iroh::_events::net::relay::connected",
             Level::DEBUG,
             url = %self.url,
         );
@@ -337,7 +337,7 @@ impl ClientBuilder {
         let conn = Conn::new(ws_stream, self.key_cache.clone(), &self.secret_key).await?;
 
         event!(
-            target: "events.net.relay.connected",
+            target: "iroh::_events::net::relay::connected",
             Level::DEBUG,
             url = %self.url,
         );

iroh-relay/src/client/conn.rs

@@ -11,7 +11,7 @@ use iroh_base::SecretKey;
 use n0_future::{Sink, Stream};
 use nested_enum_utils::common_fields;
 use snafu::{Backtrace, Snafu};
-use tracing::debug;
+use tracing::{debug, trace};
 
 use super::KeyCache;
 #[cfg(not(wasm_browser))]
@@ -99,9 +99,9 @@ impl Conn {
         let mut conn = WsBytesFramed { io };
 
         // exchange information with the server
-        debug!("server_handshake: started");
+        trace!("server_handshake: started");
         handshake::clientside(&mut conn, secret_key).await?;
-        debug!("server_handshake: done");
+        trace!("server_handshake: done");
 
         Ok(Self { conn, key_cache })
     }

iroh-relay/src/client/tls.rs

@@ -138,7 +138,6 @@ impl MaybeTlsStreamBuilder {
 
     async fn dial_url_direct(&self) -> Result<tokio::net::TcpStream, DialError> {
         use tokio::net::TcpStream;
-        debug!(%self.url, "dial url");
         let dst_ip = self
             .dns_resolver
             .resolve_host(&self.url, self.prefer_ipv6, DNS_TIMEOUT)
@@ -147,7 +146,7 @@ impl MaybeTlsStreamBuilder {
         let port = url_port(&self.url).context(InvalidTargetPortSnafu)?;
         let addr = SocketAddr::new(dst_ip, port);
 
-        debug!("connecting to {}", addr);
+        trace!("connecting to {}", addr);
         let tcp_stream = time::timeout(DIAL_ENDPOINT_TIMEOUT, async move {
             TcpStream::connect(addr).await
         })

iroh/Cargo.toml

@@ -46,15 +46,16 @@ nested_enum_utils = "0.2.1"
 netwatch = { version = "0.10" }
 pin-project = "1"
 pkarr = { version = "5", default-features = false, features = ["relays"] }
-quinn = { package = "iroh-quinn", version = "0.14.0", default-features = false, features = ["rustls-ring"] }
-quinn-proto = { package = "iroh-quinn-proto", version = "0.13.0" }
-quinn-udp = { package = "iroh-quinn-udp", version = "0.5.7" }
+quinn = { package = "iroh-quinn", git = "https://github.com/n0-computer/quinn", branch = "multipath-quinn-0.11.x", default-features = false, features = ["rustls-ring"] }
+quinn-proto = { package = "iroh-quinn-proto", git = "https://github.com/n0-computer/quinn", branch = "multipath-quinn-0.11.x" }
+quinn-udp = { package = "iroh-quinn-udp", git = "https://github.com/n0-computer/quinn", branch = "multipath-quinn-0.11.x" }
 rand = "0.9.2"
 reqwest = { version = "0.12", default-features = false, features = [
     "rustls-tls",
     "stream",
 ] }
 ring = "0.17"
+rustc-hash = "2"
 rustls = { version = "0.23.33", default-features = false, features = ["ring"] }
 serde = { version = "1.0.219", features = ["derive", "rc"] }
 smallvec = "1.11.1"
@@ -97,7 +98,7 @@ hickory-resolver = "0.25.1"
 igd-next = { version = "0.16", features = ["aio_tokio"] }
 netdev = { version = "0.38.1" }
 portmapper = { version = "0.10", default-features = false }
-quinn = { package = "iroh-quinn", version = "0.14.0", default-features = false, features = ["runtime-tokio", "rustls-ring"] }
+quinn = { package = "iroh-quinn", git = "https://github.com/n0-computer/quinn", branch = "multipath-quinn-0.11.x", default-features = false, features = ["runtime-tokio", "rustls-ring"] }
 tokio = { version = "1", features = [
     "io-util",
     "macros",

iroh/bench/Cargo.toml

@@ -12,7 +12,7 @@ iroh = { path = ".." }
 iroh-metrics = "0.36"
 n0-future = "0.3.0"
 n0-snafu = "0.2.0"
-quinn = { package = "iroh-quinn", version = "0.14" }
+quinn = { package = "iroh-quinn", git = "https://github.com/n0-computer/quinn", branch = "multipath-quinn-0.11.x" }
 rand = "0.9.2"
 rcgen = "0.14"
 rustls = { version = "0.23.33", default-features = false, features = ["ring"] }

iroh/src/disco.rs

@@ -24,7 +24,7 @@ use std::{
 };
 
 use data_encoding::HEXLOWER;
-use iroh_base::{PublicKey, RelayUrl};
+use iroh_base::{EndpointId, PublicKey, RelayUrl};
 use nested_enum_utils::common_fields;
 use rand::Rng;
 use serde::{Deserialize, Serialize};
@@ -119,12 +119,50 @@ pub struct Ping {
     /// Random client-generated per-ping transaction ID.
     pub tx_id: TransactionId,
 
-    /// Allegedly the ping sender's wireguard public key.
-    /// It shouldn't be trusted by itself, but can be combined with
-    /// netmap data to reduce the discokey:endpointkey relation from 1:N to 1:1.
+    /// Allegedly the ping sender's public key.
+    ///
+    /// It shouldn't be trusted by itself.
     pub endpoint_key: PublicKey,
 }
 
+impl Ping {
+    /// Creates a ping message to ping `node_id`.
+    ///
+    /// Uses a randomly generated STUN transaction ID.
+    pub(crate) fn new(endpoint_id: EndpointId) -> Self {
+        Self {
+            tx_id: TransactionId::default(),
+            endpoint_key: endpoint_id,
+        }
+    }
+
+    fn from_bytes(p: &[u8]) -> Result<Self, ParseError> {
+        // Deliberately lax on longer-than-expected messages, for future compatibility.
+        ensure!(p.len() >= PING_LEN, TooShortSnafu);
+        let tx_id: [u8; TX_LEN] = p[..TX_LEN].try_into().expect("length checked");
+        let raw_key = &p[TX_LEN..TX_LEN + iroh_base::PublicKey::LENGTH];
+        let endpoint_key =
+            PublicKey::try_from(raw_key).map_err(|_| InvalidEncodingSnafu.build())?;
+        let tx_id = TransactionId::from(tx_id);
+
+        Ok(Ping {
+            tx_id,
+            endpoint_key,
+        })
+    }
+
+    fn as_bytes(&self) -> Vec<u8> {
+        let header = msg_header(MessageType::Ping, V0);
+        let mut out = vec![0u8; PING_LEN + HEADER_LEN];
+
+        out[..HEADER_LEN].copy_from_slice(&header);
+        out[HEADER_LEN..HEADER_LEN + TX_LEN].copy_from_slice(&self.tx_id);
+        out[HEADER_LEN + TX_LEN..].copy_from_slice(self.endpoint_key.as_ref());
+
+        out
+    }
+}
+
 /// A response a Ping.
 ///
 /// It includes the sender's source IP + port, so it's effectively a STUN response.
@@ -146,21 +184,6 @@ pub enum SendAddr {
     Relay(RelayUrl),
 }
 
-impl SendAddr {
-    /// Returns if this is a `relay` addr.
-    pub fn is_relay(&self) -> bool {
-        matches!(self, Self::Relay(_))
-    }
-
-    /// Returns the `Some(Url)` if it is a relay addr.
-    pub fn relay_url(&self) -> Option<RelayUrl> {
-        match self {
-            Self::Relay(url) => Some(url.clone()),
-            Self::Udp(_) => None,
-        }
-    }
-}
-
 impl From<transports::Addr> for SendAddr {
     fn from(addr: transports::Addr) -> Self {
         match addr {
@@ -214,34 +237,6 @@ pub struct CallMeMaybe {
     pub my_numbers: Vec<SocketAddr>,
 }
 
-impl Ping {
-    fn from_bytes(p: &[u8]) -> Result<Self, ParseError> {
-        // Deliberately lax on longer-than-expected messages, for future compatibility.
-        ensure!(p.len() >= PING_LEN, TooShortSnafu);
-        let tx_id: [u8; TX_LEN] = p[..TX_LEN].try_into().expect("length checked");
-        let raw_key = &p[TX_LEN..TX_LEN + iroh_base::PublicKey::LENGTH];
-        let endpoint_key =
-            PublicKey::try_from(raw_key).map_err(|_| InvalidEncodingSnafu.build())?;
-        let tx_id = TransactionId::from(tx_id);
-
-        Ok(Ping {
-            tx_id,
-            endpoint_key,
-        })
-    }
-
-    fn as_bytes(&self) -> Vec<u8> {
-        let header = msg_header(MessageType::Ping, V0);
-        let mut out = vec![0u8; PING_LEN + HEADER_LEN];
-
-        out[..HEADER_LEN].copy_from_slice(&header);
-        out[HEADER_LEN..HEADER_LEN + TX_LEN].copy_from_slice(&self.tx_id);
-        out[HEADER_LEN + TX_LEN..].copy_from_slice(self.endpoint_key.as_ref());
-
-        out
-    }
-}
-
 #[allow(missing_docs)]
 #[common_fields({
     backtrace: Option<snafu::Backtrace>,

iroh/src/discovery.rs

@@ -124,12 +124,11 @@ use snafu::{IntoError, Snafu, ensure};
 use tokio::sync::oneshot;
 use tracing::{Instrument, debug, error_span, warn};
 
-use crate::Endpoint;
 pub use crate::endpoint_info::{EndpointData, EndpointInfo, ParseError, UserData};
+use crate::{Endpoint, magicsock::endpoint_map::Source};
 
 #[cfg(not(wasm_browser))]
 pub mod dns;
-
 #[cfg(feature = "discovery-local-network")]
 pub mod mdns;
 pub mod pkarr;
@@ -492,11 +491,7 @@ impl Discovery for ConcurrentDiscovery {
     }
 }
 
-/// Maximum duration since the last control or data message received from an endpoint to make us
-/// start a discovery task.
-const MAX_AGE: Duration = Duration::from_secs(10);
-
-/// A wrapper around a tokio task which runs an endpoint discovery.
+/// A wrapper around a tokio task which runs a node discovery.
 pub(super) struct DiscoveryTask {
     on_first_rx: oneshot::Receiver<Result<(), DiscoveryError>>,
     _task: AbortOnDropHandle<()>,
@@ -527,30 +522,19 @@ impl DiscoveryTask {
     /// If `delay` is set, the [`DiscoveryTask`] will first wait for `delay` and then check again
     /// if we recently received messages from remote endpoint. If true, the task will abort.
     /// Otherwise, or if no `delay` is set, the discovery will be started.
-    pub(super) fn maybe_start_after_delay(
+    pub(super) fn start_after_delay(
         ep: &Endpoint,
         endpoint_id: EndpointId,
-        delay: Option<Duration>,
+        delay: Duration,
     ) -> Result<Option<Self>, DiscoveryError> {
         // If discovery is not needed, don't even spawn a task.
-        if !ep.needs_discovery(endpoint_id, MAX_AGE) {
-            return Ok(None);
-        }
         ensure!(!ep.discovery().is_empty(), NoServiceConfiguredSnafu);
         let (on_first_tx, on_first_rx) = oneshot::channel();
         let ep = ep.clone();
         let me = ep.id();
         let task = task::spawn(
             async move {
-                // If delay is set, wait and recheck if discovery is needed. If not, early-exit.
-                if let Some(delay) = delay {
-                    time::sleep(delay).await;
-                    if !ep.needs_discovery(endpoint_id, MAX_AGE) {
-                        debug!("no discovery needed, abort");
-                        on_first_tx.send(Ok(())).ok();
-                        return;
-                    }
-                }
+                time::sleep(delay).await;
                 Self::run(ep, endpoint_id, on_first_tx).await
             }
             .instrument(
@@ -606,10 +590,10 @@ impl DiscoveryTask {
                         continue;
                     }
                     debug!(%provenance, addr = ?endpoint_addr, "new address found");
-                    let source = crate::magicsock::Source::Discovery {
+                    let source = Source::Discovery {
                         name: provenance.to_string(),
                     };
-                    ep.add_endpoint_addr(endpoint_addr, source).ok();
+                    ep.add_endpoint_addr(endpoint_addr, source).await.ok();
 
                     if let Some(tx) = on_first_tx.take() {
                         tx.send(Ok(())).ok();