feat: add event and function to wait for resumption tickets

Author: Frando

Cargo.lock

@@ -63,3 +63,6 @@ debug = true
 
 [profile.release]
 debug = true
+
+[patch.crates-io]
+rustls = { git = "https://github.com/rustls/rustls", branch = "main" }

Cargo.toml

@@ -127,6 +127,10 @@ impl crypto::Session for NoProtectionSession {
     ) -> Result<(), crypto::ExportKeyingMaterialError> {
         self.inner.export_keying_material(output, label, context)
     }
+
+    fn resumption_tickets_received(&self) -> Option<u32> {
+        self.inner.resumption_tickets_received()
+    }
 }
 
 impl crypto::ClientConfig for NoProtectionClientConfig {

perf/src/noprotection.rs

@@ -233,6 +233,8 @@ pub struct Connection {
     stats: ConnectionStats,
     /// QUIC version used for the connection.
     version: u32,
+    /// True if we emitted the event for received resumption tickets
+    resumption_tickets_received: bool,
 }
 
 impl Connection {
@@ -350,6 +352,7 @@ impl Connection {
             rng,
             stats: ConnectionStats::default(),
             version,
+            resumption_tickets_received: false,
         };
         if path_validated {
             this.on_path_validated();
@@ -2035,6 +2038,12 @@ impl Connection {
             }
         }
 
+        if !self.resumption_tickets_received && self.crypto.resumption_tickets_received() > Some(0)
+        {
+            self.resumption_tickets_received = true;
+            self.events.push_back(Event::ResumptionTicketsReceived)
+        }
+
         Ok(())
     }
 
@@ -3920,6 +3929,10 @@ pub enum Event {
     DatagramReceived,
     /// One or more application datagrams have been sent after blocking
     DatagramsUnblocked,
+    /// One or more TLS session resumption tickets have been received
+    ///
+    /// This event is only emitted on the client, and is emitted at most once per connection.
+    ResumptionTicketsReceived,
 }
 
 fn instant_saturating_sub(x: Instant, y: Instant) -> Duration {

quinn-proto/src/connection/mod.rs

@@ -91,6 +91,11 @@ pub trait Session: Send + Sync + 'static {
         label: &[u8],
         context: &[u8],
     ) -> Result<(), ExportKeyingMaterialError>;
+
+    /// Returns the number of TLS1.3 session resumption tickets that were received
+    ///
+    /// Returns `None` on the server side.
+    fn resumption_tickets_received(&self) -> Option<u32>;
 }
 
 /// A pair of keys for bidirectional communication

quinn-proto/src/crypto.rs

@@ -199,6 +199,13 @@ impl crypto::Session for TlsSession {
             .map_err(|_| ExportKeyingMaterialError)?;
         Ok(())
     }
+
+    fn resumption_tickets_received(&self) -> Option<u32> {
+        match &self.inner {
+            Connection::Client(conn) => Some(conn.tls13_tickets_received()),
+            Connection::Server(_) => None,
+        }
+    }
 }
 
 const RETRY_INTEGRITY_KEY_DRAFT: [u8; 16] = [

quinn-proto/src/crypto/rustls.rs

@@ -128,6 +128,10 @@ fn lifecycle() {
     let _guard = subscribe();
     let mut pair = Pair::default();
     let (client_ch, server_ch) = pair.connect();
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(pair.client_conn_mut(client_ch).poll(), None);
     assert!(pair.client_conn_mut(client_ch).using_ecn());
     assert!(pair.server_conn_mut(server_ch).using_ecn());
@@ -161,6 +165,10 @@ fn draft_version_compat() {
     let mut pair = Pair::default();
     let (client_ch, server_ch) = pair.connect_with(client_config);
 
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(pair.client_conn_mut(client_ch).poll(), None);
     assert!(pair.client_conn_mut(client_ch).using_ecn());
     assert!(pair.server_conn_mut(server_ch).using_ecn());
@@ -206,6 +214,10 @@ fn server_stateless_reset() {
     pair.client.connections.get_mut(&client_ch).unwrap().ping();
     info!("resetting");
     pair.drive();
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(
         pair.client_conn_mut(client_ch).poll(),
         Some(Event::ConnectionLost {
@@ -327,6 +339,10 @@ fn finish_stream_simple() {
     pair.client_send(client_ch, s).finish().unwrap();
     pair.drive();
 
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(
         pair.client_conn_mut(client_ch).poll(),
         Some(Event::Stream(StreamEvent::Finished { id })) if id == s
@@ -379,6 +395,10 @@ fn reset_stream() {
     let mut chunks = recv.read(false).unwrap();
     assert_matches!(chunks.next(usize::MAX), Err(ReadError::Reset(ERROR)));
     let _ = chunks.finalize();
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(pair.client_conn_mut(client_ch).poll(), None);
 }
 
@@ -596,7 +616,10 @@ fn zero_rtt_happypath() {
         Ok(Some(chunk)) if chunk.offset == 0 && chunk.bytes == MSG
     );
     let _ = chunks.finalize();
-    assert_eq!(pair.client_conn_mut(client_ch).lost_packets(), 0);
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
 }
 
 #[test]
@@ -905,6 +928,10 @@ fn stream_id_limit() {
     pair.client_send(client_ch, s).write(MSG).unwrap();
     pair.client_send(client_ch, s).finish().unwrap();
     pair.drive();
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(
         pair.client_conn_mut(client_ch).poll(),
         Some(Event::Stream(StreamEvent::Finished { id })) if id == s
@@ -1192,6 +1219,10 @@ fn idle_timeout() {
     }
 
     assert!(pair.time - start < Duration::from_millis(2 * IDLE_TIMEOUT));
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(
         pair.client_conn_mut(client_ch).poll(),
         Some(Event::ConnectionLost {
@@ -1271,6 +1302,10 @@ fn migration() {
     assert_ne!(pair.server_conn_mut(server_ch).total_recvd(), 0);
 
     pair.drive();
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(pair.client_conn_mut(client_ch).poll(), None);
     assert_eq!(
         pair.server_conn_mut(server_ch).remote_address(),
@@ -1657,6 +1692,10 @@ fn finish_stream_flow_control_reordered() {
     pair.server.finish_delay(); // Add flow control packets after
     pair.drive();
 
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(
         pair.client_conn_mut(client_ch).poll(),
         Some(Event::Stream(StreamEvent::Finished { id })) if id == s
@@ -1749,6 +1788,10 @@ fn stop_during_finish() {
     pair.drive_server();
     pair.client_send(client_ch, s).finish().unwrap();
     pair.drive_client();
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(
         pair.client_conn_mut(client_ch).poll(),
         Some(Event::Stream(StreamEvent::Stopped { id, error_code: ERROR })) if id == s
@@ -2036,6 +2079,10 @@ fn finish_acked() {
     // Send FIN, receive data ack
     info!("client receives ACK, sends FIN");
     pair.drive_client();
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     // Check for premature finish from data ack
     assert_matches!(pair.client_conn_mut(client_ch).poll(), None);
     // Process FIN ack
@@ -2074,6 +2121,10 @@ fn finish_retransmit() {
     // Receive FIN ack, but no data ack
     pair.drive_client();
     // Check for premature finish from FIN ack
+    assert_matches!(
+        pair.client_conn_mut(client_ch).poll(),
+        Some(Event::ResumptionTicketsReceived)
+    );
     assert_matches!(pair.client_conn_mut(client_ch).poll(), None);
     // Recover
     pair.drive();

quinn-proto/src/tests/mod.rs

@@ -25,8 +25,8 @@ use crate::{
     udp_transmit,
 };
 use proto::{
-    ConnectionError, ConnectionHandle, ConnectionStats, Dir, EndpointEvent, StreamEvent, StreamId,
-    congestion::Controller,
+    ConnectionError, ConnectionHandle, ConnectionStats, Dir, EndpointEvent, Side, StreamEvent,
+    StreamId, congestion::Controller,
 };
 
 /// In-progress connection attempt future
@@ -636,6 +636,37 @@ impl Connection {
         // May need to send MAX_STREAMS to make progress
         conn.wake();
     }
+
+    /// Waits until the connection received TLS resumption tickets
+    ///
+    /// Yields `true` once resumption tickets were received. Resolves immediately
+    /// if tickets were already received, otherwise it resolves once tickets arrive.
+    /// If the server does not send any tickets, the returned future will remain pending forever.
+    ///
+    /// This should only be used on the client side. On the server side, it will
+    /// always resolve immediately and yield `false`.
+    pub fn resumption_tickets_received(&self) -> impl Future<Output = bool> + Send + 'static {
+        let conn = self.0.clone();
+        async move {
+            let notify;
+            let (mut notified, out) = {
+                let conn = conn.state.lock("resumption_tickets_received");
+                let (notified, out) = match conn.resumption_tickets.as_ref() {
+                    Some(ResumptionTicketState::Received) => (None, true),
+                    Some(ResumptionTicketState::Pending(n)) => {
+                        notify = n.clone();
+                        (Some(notify.notified()), true)
+                    }
+                    None => (None, false),
+                };
+                (notified, out)
+            };
+            if let Some(notified) = notified.take() {
+                notified.await;
+            }
+            out
+        }
+    }
 }
 
 pin_project! {
@@ -870,6 +901,10 @@ impl ConnectionRef {
         socket: Arc<dyn AsyncUdpSocket>,
         runtime: Arc<dyn Runtime>,
     ) -> Self {
+        let resumption_tickets = match conn.side() {
+            Side::Client => Some(ResumptionTicketState::Pending(Default::default())),
+            Side::Server => None,
+        };
         Self(Arc::new(ConnectionInner {
             state: Mutex::new(State {
                 inner: conn,
@@ -892,6 +927,7 @@ impl ConnectionRef {
                 runtime,
                 send_buffer: Vec::new(),
                 buffered_transmit: None,
+                resumption_tickets,
             }),
             shared: Shared::default(),
         }))
@@ -974,6 +1010,8 @@ pub(crate) struct State {
     send_buffer: Vec<u8>,
     /// We buffer a transmit when the underlying I/O would block
     buffered_transmit: Option<proto::Transmit>,
+    /// Whether we received resumption tickets. None on the server side.
+    resumption_tickets: Option<ResumptionTicketState>,
 }
 
 impl State {
@@ -1108,6 +1146,14 @@ impl State {
                         wake_all_notify(&mut self.stopped);
                     }
                 }
+                ResumptionTicketsReceived => {
+                    if let Some(ResumptionTicketState::Pending(notify)) =
+                        self.resumption_tickets.as_mut()
+                    {
+                        notify.notify_waiters();
+                        self.resumption_tickets = Some(ResumptionTicketState::Received);
+                    }
+                }
                 ConnectionLost { reason } => {
                     self.terminate(reason, shared);
                 }
@@ -1278,6 +1324,12 @@ fn wake_all_notify(wakers: &mut FxHashMap<StreamId, Arc<Notify>>) {
         .for_each(|(_, notify)| notify.notify_waiters())
 }
 
+#[derive(Debug)]
+enum ResumptionTicketState {
+    Received,
+    Pending(Arc<Notify>),
+}
+
 /// Errors that can arise when sending a datagram
 #[derive(Debug, Error, Clone, Eq, PartialEq)]
 pub enum SendDatagramError {