feat: make Connection::remote_id and Connection::alpn infallible (#3556)

## Description

This PR refactors the connection API to make `Connection::remote_id()`
and `Connection::alpn()` infallible. Previously, these methods could
fail if called before the handshake completed or if the handshake data
was unavailable. Now, `Connection` guarantees that it represents a fully
authenticated connection with verified remote identity and ALPN
protocol, since it now can only be constructed after successful
handshake completion and authentication, eliminating the need for
fallible accessors.

### 0-RTT API Improvements

The 0-RTT API has been restructured with clearer types and semantics:

- Use `Incoming::accept` to return an `Accepting`. Use
`Accepting::into_0rtt` to return an `IncomingZeroRttConnection`

- Use `Connecting::into_0rtt` to return a `OutgoingZeroRttConnection`

- **`OutgoingZeroRttConnection`**: Represents client-side 0-RTT
connections created via `Connecting::into_0rtt()`. Allows sending 0-RTT
data before the handshake completes. Call `handshake_completed()` to get
a `ZeroRttStatus` indicating whether the 0-RTT data was accepted or
rejected by the server.

- **`IncomingZeroRttConnection`**: Represents server-side 0-RTT/0.5-RTT
connections created via `Accepting::into_0rtt()`. Allows receiving 0-RTT
data from clients or sending 0.5-RTT data before the handshake
completes. Call `handshake_completed()` to get a fully authenticated
`Connection`.

- **`ZeroRttStatus` enum**: Returned by
`OutgoingZeroRttConnection::handshake_completed()` to indicate whether
the server accepted or rejected the 0-RTT data:
- `ZeroRttStatus::Accepted(Connection)`: 0-RTT data was accepted,
streams opened before handshake remain valid
- `ZeroRttStatus::Rejected(Connection)`: 0-RTT data was rejected,
pre-handshake streams will error and data must be resent

These types replace the previous the previous version of `Connection` &
the `ZeroRttAccepted` type and provide a more explicit API for handling
0-RTT connection states and outcomes.

### Error Handling

- Connection authentication failures now return `AuthenticationError`
during the conversion between a `quinn:Connection` (which can exist
before the handshake completes) and an `iroh::Connection`. This error
may occur if the remote endpoint is not using the expected TLS
cryptography or if the remote endpoint is a plain quinn connection. You
may also see these errors after calling the `handshake_completed` method
on `IncomingZeroRttConnection` or `OutgoingZeroRttConnection`, if the
connection was closed before the handshake completed.
- New `ConnectingError` type for errors during handshake completion

## Breaking Changes

- `iroh`
    - `changed`
        - `ConnectError::Connection` - fields changed
        - `AcceptError::Connection` - fields changed
        - `AcceptError::MissingRemoteEndpointId` - fields changed
        - `AcceptError::NotAllowed` - fields changed
        - `AcceptError::User` - fields changed
- `Connecting::into_0rtt` -> returns `Result<OutgoingZeroRttConnection,
Connecting>`
    - `removed`
- `ProtocolHandler::on_connecting()` removed - implement
`on_accepting()` instead, which takes `Accepting` rather than
`Connecting`
- `DynProtocolHandler::on_connecting()` removed - implement
`on_accepting()` instead
        - `iroh::endpoint::IncomingFuture` - use `Accepting` instead
- `iroh::endpoint::ZeroRttAccepted` - replaced by explicit 0-RTT
connection types

---------

Co-authored-by: “ramfox” <“[email protected]”>
Co-authored-by: Philipp Krüger <[email protected]>
Co-authored-by: Frando <[email protected]>
Co-authored-by: Friedel Ziegelmayer <[email protected]>

Author: 5 people

iroh/bench/src/iroh.rs

@@ -249,16 +249,16 @@ pub async fn server(endpoint: Endpoint, opt: Opt) -> Result<()> {
     // Handle only the expected amount of clients
     for _ in 0..opt.clients {
         let incoming = endpoint.accept().await.unwrap();
-        let connecting = match incoming.accept() {
-            Ok(connecting) => connecting,
+        let accepting = match incoming.accept() {
+            Ok(accepting) => accepting,
             Err(err) => {
                 warn!("incoming connection failed: {err:#}");
                 // we can carry on in these cases:
                 // this can be caused by retransmitted datagrams
                 continue;
             }
         };
-        let connection = connecting.await.std_context("handshake failed")?;
+        let connection = accepting.await.context("handshake failed")?;
 
         server_tasks.push(tokio::spawn(async move {
             loop {

iroh/bench/src/quinn.rs

@@ -220,16 +220,16 @@ pub async fn server(endpoint: Endpoint, opt: Opt) -> Result<()> {
     // Handle only the expected amount of clients
     for _ in 0..opt.clients {
         let incoming = endpoint.accept().await.unwrap();
-        let connecting = match incoming.accept() {
-            Ok(connecting) => connecting,
+        let accepting = match incoming.accept() {
+            Ok(accepting) => accepting,
             Err(err) => {
                 warn!("incoming connection failed: {err:#}");
                 // we can carry on in these cases:
                 // this can be caused by retransmitted datagrams
                 continue;
             }
         };
-        let connection = connecting.await.std_context("handshake failed")?;
+        let connection = accepting.await.std_context("handshake failed")?;
 
         server_tasks.push(tokio::spawn(async move {
             loop {

iroh/examples/0rtt.rs

@@ -1,14 +1,11 @@
-use std::{env, future::Future, str::FromStr, time::Instant};
+use std::{env, str::FromStr, time::Instant};
 
 use clap::Parser;
 use data_encoding::HEXLOWER;
-use iroh::{
-    EndpointId, SecretKey,
-    endpoint::{Connecting, Connection},
-};
-use n0_error::{Result, StackResultExt, StdResultExt, bail_any};
-use n0_future::{StreamExt, future};
-use n0_watcher::Watcher;
+use iroh::{EndpointId, SecretKey, discovery::Discovery, endpoint::ZeroRttStatus};
+use n0_error::{Result, StackResultExt, StdResultExt};
+use n0_future::StreamExt;
+use quinn::{RecvStream, SendStream};
 use tracing::{info, trace};
 
 const PINGPONG_ALPN: &[u8] = b"0rtt-pingpong";
@@ -44,59 +41,39 @@ pub fn get_or_generate_secret_key() -> Result<SecretKey> {
 }
 
 /// Do a simple ping-pong with the given connection.
-///
-/// We send the data on the connection. If `proceed` resolves to true,
-/// read the response immediately. Otherwise, the stream pair is bad and we need
-/// to open a new stream pair.
-async fn pingpong(
-    connection: &Connection,
-    proceed: impl Future<Output = bool>,
-    x: u64,
-) -> Result<()> {
-    let (mut send, recv) = connection.open_bi().await.anyerr()?;
+async fn pingpong(send: SendStream, recv: RecvStream, x: u64) -> Result<()> {
+    ping(send, x).await?;
+    pong(recv, x).await
+}
+
+async fn ping(mut send: SendStream, x: u64) -> Result<()> {
     let data = x.to_be_bytes();
     send.write_all(&data).await.anyerr()?;
-    send.finish().anyerr()?;
-    let mut recv = if proceed.await {
-        // use recv directly if we can proceed
-        recv
-    } else {
-        // proceed returned false, so we have learned that the 0-RTT send was rejected.
-        // at this point we have a fully handshaked connection, so we try again.
-        let (mut send, recv) = connection.open_bi().await.anyerr()?;
-        send.write_all(&data).await.anyerr()?;
-        send.finish().anyerr()?;
-        recv
-    };
+    send.finish().anyerr()
+}
+
+async fn pong(mut recv: RecvStream, x: u64) -> Result<()> {
+    let data = x.to_be_bytes();
     let echo = recv.read_to_end(8).await.anyerr()?;
     assert!(echo == data);
     Ok(())
 }
 
-async fn pingpong_0rtt(connecting: Connecting, i: u64) -> Result<Connection> {
-    let connection = match connecting.into_0rtt() {
-        Ok((connection, accepted)) => {
-            trace!("0-RTT possible from our side");
-            pingpong(&connection, accepted, i).await?;
-            connection
-        }
-        Err(connecting) => {
-            trace!("0-RTT not possible from our side");
-            let connection = connecting.await.anyerr()?;
-            pingpong(&connection, future::ready(true), i).await?;
-            connection
-        }
-    };
-    Ok(connection)
-}
-
 async fn connect(args: Args) -> Result<()> {
-    let remote_id = args.endpoint_id.context("Missing endpoint id")?;
+    let remote_id = args.endpoint_id.unwrap();
     let endpoint = iroh::Endpoint::builder()
         .relay_mode(iroh::RelayMode::Disabled)
         .keylog(true)
         .bind()
         .await?;
+    // ensure we have resolved the remote_id before connecting
+    // so we get a more accurate connection timing
+    let mut discovery_stream = endpoint
+        .discovery()
+        .resolve(remote_id)
+        .expect("discovery to be enabled");
+    let _ = discovery_stream.next().await;
+
     let t0 = Instant::now();
     for i in 0..args.rounds {
         let t0 = Instant::now();
@@ -106,18 +83,40 @@ async fn connect(args: Args) -> Result<()> {
         let connection = if args.disable_0rtt {
             let connection = connecting.await.anyerr()?;
             trace!("connecting without 0-RTT");
-            pingpong(&connection, future::ready(true), i).await?;
+            let (send, recv) = connection.open_bi().await.anyerr()?;
+            pingpong(send, recv, i).await?;
             connection
         } else {
-            pingpong_0rtt(connecting, i).await?
+            match connecting.into_0rtt() {
+                Ok(zrtt_connection) => {
+                    trace!("0-RTT possible from our side");
+                    let (send, recv) = zrtt_connection.open_bi().await.anyerr()?;
+                    // before we get the full handshake, attempt to send 0-RTT data
+                    let zrtt_task = tokio::spawn(ping(send, i));
+                    match zrtt_connection.handshake_completed().await? {
+                        ZeroRttStatus::Accepted(conn) => {
+                            let _ = zrtt_task.await.anyerr()?;
+                            pong(recv, i).await?;
+                            conn
+                        }
+                        ZeroRttStatus::Rejected(conn) => {
+                            zrtt_task.abort();
+                            let (send, recv) = conn.open_bi().await.anyerr()?;
+                            pingpong(send, recv, i).await?;
+                            conn
+                        }
+                    }
+                }
+                Err(connecting) => {
+                    trace!("0-RTT not possible from our side");
+                    let conn = connecting.await.anyerr()?;
+                    let (send, recv) = conn.open_bi().await.anyerr()?;
+                    pingpong(send, recv, i).await?;
+                    conn
+                }
+            }
         };
-        tokio::spawn(async move {
-            // wait for some time for the handshake to complete and the server
-            // to send a NewSessionTicket. This is less than ideal, but we
-            // don't have a better way to wait for the handshake to complete.
-            tokio::time::sleep(connection.rtt() * 2).await;
-            connection.close(0u8.into(), b"");
-        });
+        connection.close(0u8.into(), b"");
         let elapsed = t0.elapsed();
         println!("round {i}: {} us", elapsed.as_micros());
     }
@@ -138,24 +137,13 @@ async fn accept(_args: Args) -> Result<()> {
         .relay_mode(iroh::RelayMode::Disabled)
         .bind()
         .await?;
-    let mut addrs = endpoint.watch_addr().stream();
-    let addr = loop {
-        let Some(addr) = addrs.next().await else {
-            bail_any!("Address stream closed");
-        };
-        if !addr.ip_addrs().count() == 0 {
-            break addr;
-        }
-    };
-    println!("Listening on: {addr:?}");
+    println!("endpoint id: {}", endpoint.id());
 
     let accept = async move {
         while let Some(incoming) = endpoint.accept().await {
             tokio::spawn(async move {
-                let connecting = incoming.accept().anyerr()?;
-                let (connection, _zero_rtt_accepted) = connecting
-                    .into_0rtt()
-                    .expect("accept into 0.5 RTT always succeeds");
+                let accepting = incoming.accept().anyerr()?;
+                let connection = accepting.into_0rtt();
                 let (mut send, mut recv) = connection.accept_bi().await.anyerr()?;
                 trace!("recv.is_0rtt: {}", recv.is_0rtt());
                 let data = recv.read_to_end(8).await.anyerr()?;

iroh/examples/dht_discovery.rs

@@ -78,8 +78,8 @@ async fn chat_server(args: Args) -> Result<()> {
     println!("pkarr z32: {zid}");
     println!("see https://app.pkarr.org/?pk={zid}");
     while let Some(incoming) = endpoint.accept().await {
-        let connecting = match incoming.accept() {
-            Ok(connecting) => connecting,
+        let accepting = match incoming.accept() {
+            Ok(accepting) => accepting,
             Err(err) => {
                 warn!("incoming connection failed: {err:#}");
                 // we can carry on in these cases:
@@ -88,8 +88,8 @@ async fn chat_server(args: Args) -> Result<()> {
             }
         };
         tokio::spawn(async move {
-            let connection = connecting.await.anyerr()?;
-            let remote_endpoint_id = connection.remote_id()?;
+            let connection = accepting.await?;
+            let remote_endpoint_id = connection.remote_id();
             println!("got connection from {remote_endpoint_id}");
             // just leave the tasks hanging. this is just an example.
             let (mut writer, mut reader) = connection.accept_bi().await.anyerr()?;

iroh/examples/echo-no-router.rs

@@ -89,7 +89,7 @@ async fn start_accept_side() -> Result<Endpoint> {
                     let connection = incoming.await.anyerr()?;
 
                     // We can get the remote's endpoint id from the connection.
-                    let endpoint_id = connection.remote_id()?;
+                    let endpoint_id = connection.remote_id();
                     println!("accepted connection from {endpoint_id}");
 
                     // Our protocol is a simple request-response protocol, so we expect the

iroh/examples/echo.rs

@@ -86,7 +86,7 @@ impl ProtocolHandler for Echo {
     /// the connection lasts.
     async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
         // We can get the remote's endpoint id from the connection.
-        let endpoint_id = connection.remote_id()?;
+        let endpoint_id = connection.remote_id();
         println!("accepted connection from {endpoint_id}");
 
         // Our protocol is a simple request-response protocol, so we expect the

iroh/examples/listen-unreliable.rs

@@ -62,18 +62,18 @@ async fn main() -> Result<()> {
     // accept incoming connections, returns a normal QUIC connection
 
     while let Some(incoming) = endpoint.accept().await {
-        let mut connecting = match incoming.accept() {
-            Ok(connecting) => connecting,
+        let mut accepting = match incoming.accept() {
+            Ok(accepting) => accepting,
             Err(err) => {
                 warn!("incoming connection failed: {err:#}");
                 // we can carry on in these cases:
                 // this can be caused by retransmitted datagrams
                 continue;
             }
         };
-        let alpn = connecting.alpn().await?;
-        let conn = connecting.await.anyerr()?;
-        let endpoint_id = conn.remote_id()?;
+        let alpn = accepting.alpn().await?;
+        let conn = accepting.await?;
+        let endpoint_id = conn.remote_id();
         info!(
             "new (unreliable) connection from {endpoint_id} with ALPN {}",
             String::from_utf8_lossy(&alpn),

iroh/examples/listen.rs

@@ -60,18 +60,18 @@ async fn main() -> Result<()> {
     );
     // accept incoming connections, returns a normal QUIC connection
     while let Some(incoming) = endpoint.accept().await {
-        let mut connecting = match incoming.accept() {
-            Ok(connecting) => connecting,
+        let mut accepting = match incoming.accept() {
+            Ok(accepting) => accepting,
             Err(err) => {
                 warn!("incoming connection failed: {err:#}");
                 // we can carry on in these cases:
                 // this can be caused by retransmitted datagrams
                 continue;
             }
         };
-        let alpn = connecting.alpn().await?;
-        let conn = connecting.await.anyerr()?;
-        let endpoint_id = conn.remote_id()?;
+        let alpn = accepting.alpn().await?;
+        let conn = accepting.await?;
+        let endpoint_id = conn.remote_id();
         info!(
             "new connection from {endpoint_id} with ALPN {}",
             String::from_utf8_lossy(&alpn),

iroh/examples/screening-connection.rs

@@ -14,7 +14,7 @@ use std::sync::{
 
 use iroh::{
     Endpoint, EndpointAddr,
-    endpoint::{Connecting, Connection},
+    endpoint::{Accepting, Connection},
     protocol::{AcceptError, ProtocolHandler, Router},
 };
 use n0_error::{Result, StdResultExt, e};
@@ -100,10 +100,10 @@ struct ScreenedEcho {
 }
 
 impl ProtocolHandler for ScreenedEcho {
-    /// `on_connecting` allows us to intercept a connection as it's being formed,
+    /// `on_accepting` allows us to intercept a connection as it's being formed,
     /// which is the right place to cut off a connection as early as possible.
     /// This is an optional method on the ProtocolHandler trait.
-    async fn on_connecting(&self, connecting: Connecting) -> Result<Connection, AcceptError> {
+    async fn on_accepting(&self, accepting: Accepting) -> Result<Connection, AcceptError> {
         self.conn_attempt_count.fetch_add(1, Ordering::Relaxed);
         let count = self.conn_attempt_count.load(Ordering::Relaxed);
 
@@ -113,8 +113,8 @@ impl ProtocolHandler for ScreenedEcho {
             return Err(e!(AcceptError::NotAllowed));
         }
 
-        // To allow normal connection construction, await the connecting future & return
-        let conn = connecting.await?;
+        // To allow normal connection construction, await the accepting future & return
+        let conn = accepting.await?;
         Ok(conn)
     }
 
@@ -125,7 +125,7 @@ impl ProtocolHandler for ScreenedEcho {
     /// the connection lasts.
     async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
         // We can get the remote's endpoint id from the connection.
-        let endpoint_id = connection.remote_id()?;
+        let endpoint_id = connection.remote_id();
         println!("accepted connection from {endpoint_id}");
 
         // Our protocol is a simple request-response protocol, so we expect the

iroh/examples/search.rs

@@ -128,7 +128,7 @@ impl ProtocolHandler for BlobSearch {
     /// the connection lasts.
     async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
         // We can get the remote's endpoint id from the connection.
-        let endpoint_id = connection.remote_id()?;
+        let endpoint_id = connection.remote_id();
         println!("accepted connection from {endpoint_id}");
 
         // Our protocol is a simple request-response protocol, so we expect the