feat(iroh): Enable applications to establish 0-RTT connections (#3163)

## Description

Implements necessary APIs to make use of 0-RTT QUIC connections.

0-RTT allows you to skip a round-trip in case you have connected to a
known endpoint ahead of time, and stored the given TLS session ticket.
With this PR, we by default will cache up to 8 session tickets per
endpoint you connect to, and remember up to 32 endpoints maximum.
This cache only lives in-memory. We might add customization to the
`EndpointBuilder` in the future to allow for customizing this cache
(allowing you to persist it), but that obviously has security
implications, so will need careful consideration.

This PR enables using 0-RTT via the `Endpoint::connect_with_opts`
function, which - unlike `Endpoint::connect` - returns a `Connecting`, a
state prior to a full `Connection`. By calling `Connecting::into_0rtt`
you can attempt to turn this connection into a full 0-RTT connection.
However, security caveats apply. See that function's documentation for
details.

Migration guide:
```rs
let connection = endpoint.connect_with(node_addr, alpn, transport_config).await?;
```
to
```rs
let connection = endpoint.connect_with_opts(
    node_addr,
    alpn,
    ConnectOptions::new().with_transport_config(transport_config),
)
.await?
.await?; // second await for Connecting -> Connection
```

Closes #3146 

## Breaking Changes

- `iroh::Endpoint::connect_with` was removed, and
`iroh::Endpoint::connect_with_opts` was added instead, but returning an
`iroh::endpoint::Connecting` instead of an `iroh::endpoint::Connection`,
allowing use of QUIC's 0-RTT feature.
- `iroh::endpoint::Connection::into_0rtt` now returns
`iroh::endpoint::ZeroRttAccepted` (among other things), instead of
`iroh_quinn::ZeroRttAccepted`. This wrapper is equivalent in
functionality, but makes sure we're not depending on API-breaking
changes in quinn and can keep a discovery task alive for as long as
needed, until a connection is established.

## Change checklist

- [x] Self-review.
- [x] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [x] Tests if relevant.
- [x] All breaking changes documented.

iroh/src/discovery.rs

@@ -398,15 +398,15 @@ impl DiscoveryTask {
             }
         };
         let mut on_first_tx = Some(on_first_tx);
-        debug!("discovery: start");
+        debug!("starting");
         loop {
             match stream.next().await {
                 Some(Ok(r)) => {
                     if r.node_addr.is_empty() {
-                        debug!(provenance = %r.provenance, "discovery: empty address found");
+                        debug!(provenance = %r.provenance, "empty address found");
                         continue;
                     }
-                    debug!(provenance = %r.provenance, addr = ?r.node_addr, "discovery: new address found");
+                    debug!(provenance = %r.provenance, addr = ?r.node_addr, "new address found");
                     ep.add_node_addr_with_source(r.node_addr, r.provenance).ok();
                     if let Some(tx) = on_first_tx.take() {
                         tx.send(Ok(())).ok();
@@ -443,13 +443,14 @@ mod tests {
 
     use anyhow::Context;
     use iroh_base::SecretKey;
+    use quinn::{IdleTimeout, TransportConfig};
     use rand::Rng;
     use testresult::TestResult;
     use tokio_util::task::AbortOnDropHandle;
     use tracing_test::traced_test;
 
     use super::*;
-    use crate::RelayMode;
+    use crate::{endpoint::ConnectOptions, RelayMode};
 
     type InfoStore = HashMap<NodeId, (Option<RelayUrl>, BTreeSet<SocketAddr>, u64)>;
 
@@ -507,15 +508,14 @@ mod tests {
             endpoint: Endpoint,
             node_id: NodeId,
         ) -> Option<BoxStream<Result<DiscoveryItem>>> {
-            let addr_info = match self.resolve_wrong {
-                false => self.shared.nodes.lock().unwrap().get(&node_id).cloned(),
-                true => {
-                    let ts = system_time_now() - 100_000;
-                    let port: u16 = rand::thread_rng().gen_range(10_000..20_000);
-                    // "240.0.0.0/4" is reserved and unreachable
-                    let addr: SocketAddr = format!("240.0.0.1:{port}").parse().unwrap();
-                    Some((None, BTreeSet::from([addr]), ts))
-                }
+            let addr_info = if self.resolve_wrong {
+                let ts = system_time_now() - 100_000;
+                let port: u16 = rand::thread_rng().gen_range(10_000..20_000);
+                // "240.0.0.0/4" is reserved and unreachable
+                let addr: SocketAddr = format!("240.0.0.1:{port}").parse().unwrap();
+                Some((None, BTreeSet::from([addr]), ts))
+            } else {
+                self.shared.nodes.lock().unwrap().get(&node_id).cloned()
             };
             let stream = match addr_info {
                 Some((url, addrs, ts)) => {
@@ -636,10 +636,9 @@ mod tests {
             disco.add(disco3);
             new_endpoint(secret, disco).await
         };
-        let ep1_addr = NodeAddr::new(ep1.node_id());
         // wait for out address to be updated and thus published at least once
         ep1.node_addr().await?;
-        let _conn = ep2.connect(ep1_addr, TEST_ALPN).await?;
+        let _conn = ep2.connect(ep1.node_id(), TEST_ALPN).await?;
         Ok(())
     }
 
@@ -659,10 +658,19 @@ mod tests {
             let disco = ConcurrentDiscovery::from_services(vec![Box::new(disco1)]);
             new_endpoint(secret, disco).await
         };
-        let ep1_addr = NodeAddr::new(ep1.node_id());
         // wait for out address to be updated and thus published at least once
         ep1.node_addr().await?;
-        let res = ep2.connect(ep1_addr, TEST_ALPN).await;
+
+        // 10x faster test via a 3s idle timeout instead of the 30s default
+        let mut config = TransportConfig::default();
+        config.keep_alive_interval(Some(Duration::from_secs(1)));
+        config.max_idle_timeout(Some(IdleTimeout::try_from(Duration::from_secs(3))?));
+        let opts = ConnectOptions::new().with_transport_config(Arc::new(config));
+
+        let res = ep2
+            .connect_with_opts(ep1.node_id(), TEST_ALPN, opts)
+            .await? // -> Connecting works
+            .await; // -> Connection is expected to fail
         assert!(res.is_err());
         Ok(())
     }