feat(iroh-net)!: Improve initial connection latency

iroh-bytes/Cargo.toml

@@ -34,7 +34,7 @@ iroh-net = { version = "0.15.0", path = "../iroh-net", optional = true }
 num_cpus = "1.15.0"
 parking_lot = { version = "0.12.1", optional = true }
 postcard = { version = "1", default-features = false, features = ["alloc", "use-std", "experimental-derive"] }
-quinn = "0.10"
+quinn = { package = "iroh-quinn", version = "0.10" }
 rand = "0.8"
 range-collections = "0.4.0"
 redb = { version = "2.0.0", optional = true }

iroh-bytes/examples/provide-bytes.rs

@@ -11,6 +11,7 @@
 //! To provide a collection (multiple blobs)
 use anyhow::Result;
 use tokio_util::task::LocalPoolHandle;
+use tracing::warn;
 use tracing_subscriber::{prelude::*, EnvFilter};
 
 use iroh_bytes::{format::collection::Collection, Hash};
@@ -84,14 +85,22 @@ async fn main() -> Result<()> {
     let lp = LocalPoolHandle::new(1);
 
     let accept_task = tokio::spawn(async move {
-        while let Some(conn) = endpoint.accept().await {
+        while let Some(incoming) = endpoint.accept().await {
             println!("connection incoming");
 
             let db = db.clone();
             let lp = lp.clone();
 
             // spawn a task to handle the connection
             tokio::spawn(async move {
+                let remote_addr = incoming.remote_address();
+                let conn = match incoming.await {
+                    Ok(conn) => conn,
+                    Err(err) => {
+                        warn!(%remote_addr, "Error connecting: {err:#}");
+                        return;
+                    }
+                };
                 iroh_bytes::provider::handle_connection(conn, db, MockEventSender, lp).await
             });
         }

iroh-bytes/src/provider.rs

@@ -280,19 +280,12 @@ pub trait EventSender: Clone + Sync + Send + 'static {
 
 /// Handle a single connection.
 pub async fn handle_connection<D: Map, E: EventSender>(
-    connecting: quinn::Connecting,
+    connection: quinn::Connection,
     db: D,
     events: E,
     rt: LocalPoolHandle,
 ) {
-    let remote_addr = connecting.remote_address();
-    let connection = match connecting.await {
-        Ok(conn) => conn,
-        Err(err) => {
-            warn!(%remote_addr, "Error connecting: {err:#}");
-            return;
-        }
-    };
+    let remote_addr = connection.remote_address();
     let connection_id = connection.stable_id() as u64;
     let span = debug_span!("connection", connection_id, %remote_addr);
     async move {

iroh-cli/Cargo.toml

@@ -44,8 +44,8 @@ parking_lot = "0.12.1"
 pkarr = { version = "1.1.5", default-features = false }
 portable-atomic = "1"
 postcard = "1.0.8"
-quic-rpc = { version = "0.8.0", features = ["flume-transport", "quinn-transport"] }
-quinn = "0.10.2"
+quic-rpc = { version = "0.9.0", features = ["flume-transport", "quinn-transport"] }
+quinn = { package = "iroh-quinn", version = "0.10.2"}
 rand = "0.8.5"
 rustyline = "12.0.0"
 serde = { version = "1.0.197", features = ["derive"] }

iroh-gossip/Cargo.toml

@@ -33,7 +33,7 @@ iroh-base = { version = "0.15.0", path = "../iroh-base" }
 # net dependencies (optional)
 futures-lite = { version = "2.3", optional = true }
 iroh-net = { path = "../iroh-net", version = "0.15.0", optional = true, default-features = false }
-quinn = { version = "0.10", optional = true }
+quinn = { package = "iroh-quinn", version = "0.10", optional = true }
 tokio = { version = "1", optional = true, features = ["io-util", "sync", "rt", "macros", "net", "fs"] }
 tokio-util = { version = "0.7.8", optional = true, features = ["codec"] }
 genawaiter = { version = "0.99.1", default-features = false, features = ["futures03"] }

iroh-gossip/examples/chat.rs

@@ -11,7 +11,6 @@ use iroh_gossip::{
 };
 use iroh_net::{
     key::{PublicKey, SecretKey},
-    magic_endpoint::accept_conn,
     relay::{RelayMap, RelayMode, RelayUrl},
     MagicEndpoint, NodeAddr,
 };
@@ -200,8 +199,13 @@ async fn endpoint_loop(endpoint: MagicEndpoint, gossip: Gossip) {
         });
     }
 }
-async fn handle_connection(conn: quinn::Connecting, gossip: Gossip) -> anyhow::Result<()> {
-    let (peer_id, alpn, conn) = accept_conn(conn).await?;
+async fn handle_connection(
+    mut conn: iroh_net::magic_endpoint::Connecting,
+    gossip: Gossip,
+) -> anyhow::Result<()> {
+    let alpn = conn.alpn().await?;
+    let conn = conn.await?;
+    let peer_id = iroh_net::magic_endpoint::get_remote_node_id(&conn)?;
     match alpn.as_bytes() {
         GOSSIP_ALPN => gossip
             .handle_connection(conn)

iroh-net/Cargo.toml

@@ -44,11 +44,12 @@ libc = "0.2.139"
 num_enum = "0.7"
 once_cell = "1.18.0"
 parking_lot = "0.12.1"
+pin-project = "1"
 pkarr = { version = "1.1.4", default-features = false, features = ["async", "relay"] }
 postcard = { version = "1", default-features = false, features = ["alloc", "use-std", "experimental-derive"] }
-quinn = "0.10"
-quinn-proto = "0.10.5"
-quinn-udp = "0.4"
+quinn = { package = "iroh-quinn", version = "0.10.4" }
+quinn-proto = { package = "iroh-quinn-proto", version = "0.10.7" }
+quinn-udp = { package = "iroh-quinn-udp", version = "0.4" }
 rand = "0.8"
 rand_core = "0.6.4"
 rcgen = "0.11"
@@ -85,6 +86,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"], optional = tr
 # metrics
 iroh-metrics = { version = "0.15.0", path = "../iroh-metrics", default-features = false }
 strum = { version = "0.26.2", features = ["derive"] }
+futures-concurrency = "7.6.0"
 
 [target.'cfg(any(target_os = "linux", target_os = "android"))'.dependencies]
 netlink-packet-core = "0.7.0"

iroh-net/bench/Cargo.toml

@@ -10,7 +10,7 @@ anyhow = "1.0.22"
 bytes = "1"
 hdrhistogram = { version = "7.2", default-features = false }
 iroh-net = { path = ".." }
-quinn = "0.10"
+quinn = { package = "iroh-quinn", version = "0.10"}
 clap = { version = "4", features = ["derive"] }
 tokio = { version = "1.0.1", features = ["rt", "sync"] }
 tracing = "0.1"

iroh-net/examples/listen-unreliable.rs

@@ -63,9 +63,10 @@ async fn main() -> anyhow::Result<()> {
     );
     // accept incoming connections, returns a normal QUIC connection
 
-    while let Some(conn) = endpoint.accept().await {
-        // accept the connection and extract the `node_id` and ALPN
-        let (node_id, alpn, conn) = iroh_net::magic_endpoint::accept_conn(conn).await?;
+    while let Some(mut conn) = endpoint.accept().await {
+        let alpn = conn.alpn().await?;
+        let conn = conn.await?;
+        let node_id = iroh_net::magic_endpoint::get_remote_node_id(&conn)?;
         info!(
             "new (unreliable) connection from {node_id} with ALPN {alpn} (coming from {})",
             conn.remote_address()

iroh-net/examples/listen.rs

@@ -62,9 +62,10 @@ async fn main() -> anyhow::Result<()> {
         "\tcargo run --example connect -- --node-id {me} --addrs \"{local_addrs}\" --relay-url {relay_url}\n"
     );
     // accept incoming connections, returns a normal QUIC connection
-    while let Some(conn) = endpoint.accept().await {
-        // accept the connection and extract the `node_id` and ALPN
-        let (node_id, alpn, conn) = iroh_net::magic_endpoint::accept_conn(conn).await?;
+    while let Some(mut conn) = endpoint.accept().await {
+        let alpn = conn.alpn().await?;
+        let conn = conn.await?;
+        let node_id = iroh_net::magic_endpoint::get_remote_node_id(&conn)?;
         info!(
             "new connection from {node_id} with ALPN {alpn} (coming from {})",
             conn.remote_address()

iroh-net/src/disco.rs

@@ -129,8 +129,10 @@ pub struct Ping {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Pong {
     pub tx_id: stun::TransactionId,
+    /// The observed address off the ping sender.
+    ///
     /// 18 bytes (16+2) on the wire; v4-mapped ipv6 for IPv4.
-    pub src: SendAddr,
+    pub ping_observed_addr: SendAddr,
 }
 
 /// Addresses to which we can send. This is either a UDP or a relay address.
@@ -280,15 +282,18 @@ impl Pong {
         let tx_id = stun::TransactionId::from(tx_id);
         let src = send_addr_from_bytes(&p[TX_LEN..])?;
 
-        Ok(Pong { tx_id, src })
+        Ok(Pong {
+            tx_id,
+            ping_observed_addr: src,
+        })
     }
 
     fn as_bytes(&self) -> Vec<u8> {
         let header = msg_header(MessageType::Pong, V0);
         let mut out = header.to_vec();
         out.extend_from_slice(&self.tx_id);
 
-        let src_bytes = send_addr_to_vec(&self.src);
+        let src_bytes = send_addr_to_vec(&self.ping_observed_addr);
         out.extend(src_bytes);
         out
     }
@@ -412,15 +417,15 @@ mod tests {
                 name: "pong",
                 m: Message::Pong(Pong{
                     tx_id: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12].into(),
-                    src:  SendAddr::Udp("2.3.4.5:1234".parse().unwrap()),
+                    ping_observed_addr:  SendAddr::Udp("2.3.4.5:1234".parse().unwrap()),
                 }),
                 want: "02 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 00 00 00 00 00 00 00 00 00 00 00 ff ff 02 03 04 05 d2 04",
             },
             Test {
                 name: "pongv6",
                 m: Message::Pong(Pong {
                     tx_id: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12].into(),
-                    src: SendAddr::Udp("[fed0::12]:6666".parse().unwrap()),
+                    ping_observed_addr: SendAddr::Udp("[fed0::12]:6666".parse().unwrap()),
                 }),
                 want: "02 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 00 fe d0 00 00 00 00 00 00 00 00 00 00 00 00 00 12 0a 1a",
             },