[connection] retransmit CRYPTO to speedup handshake completion

Under some conditions, and endpoint may retransmit CRYPTO data before
the PTO expiry, see:

https://datatracker.ietf.org/doc/html/rfc9002#section-6.2.3

src/aioquic/quic/connection.py

@@ -269,6 +269,7 @@ def __init__(
         self._connect_called = False
         self._cryptos: Dict[tls.Epoch, CryptoPair] = {}
         self._crypto_buffers: Dict[tls.Epoch, Buffer] = {}
+        self._crypto_retransmitted = False
         self._crypto_streams: Dict[tls.Epoch, QuicStream] = {}
         self._events: Deque[events.QuicEvent] = deque()
         self._handshake_complete = False
@@ -362,6 +363,7 @@ def __init__(
             peer_completed_address_validation=not self._is_client,
             quic_logger=self._quic_logger,
             send_probe=self._send_probe,
+            logger=self._logger,
         )
 
         # things to send
@@ -890,6 +892,16 @@ def receive_datagram(self, data: bytes, addr: NetworkAddress, now: float) -> Non
                         event="packet_dropped",
                         data={"trigger": "key_unavailable"},
                     )
+
+                # if a client receives HANDSHAKE or 1-RTT packets before it has handshake keys,
+                # it can assume that the server's INITIAL was lost
+                if (
+                    self._is_client
+                    and epoch in (tls.Epoch.HANDSHAKE, tls.Epoch.ONE_RTT)
+                    and not self._crypto_retransmitted
+                ):
+                    self._loss.reschedule_data(now=now)
+                    self._crypto_retransmitted = True
                 continue
             except CryptoError as exc:
                 self._logger.debug(exc)
@@ -1548,6 +1560,20 @@ def _handle_crypto_frame(
                 self._logger.info(
                     "ALPN negotiated protocol %s", self.tls.alpn_negotiated
                 )
+        else:
+            self._logger.info(
+                "Duplicate CRYPTO data received for epoch %s", context.epoch
+            )
+
+            # if a server receives duplicate CRYPTO in an INITIAL packet,
+            # it can assume the client did not receive the server's CRYPTO
+            if (
+                not self._is_client
+                and context.epoch == tls.Epoch.INITIAL
+                and not self._crypto_retransmitted
+            ):
+                self._loss.reschedule_data(now=context.time)
+                self._crypto_retransmitted = True
 
     def _handle_data_blocked_frame(
         self, context: QuicReceiveContext, frame_type: int, buf: Buffer
@@ -1614,7 +1640,7 @@ def _handle_handshake_done_frame(
                 reason_phrase="Clients must not send HANDSHAKE_DONE frames",
             )
 
-        #  for clients, the handshake is now confirmed
+        # for clients, the handshake is now confirmed
         if not self._handshake_confirmed:
             self._discard_epoch(tls.Epoch.HANDSHAKE)
             self._handshake_confirmed = True

src/aioquic/quic/recovery.py

@@ -1,3 +1,4 @@
+import logging
 import math
 from typing import Any, Callable, Dict, Iterable, List, Optional
 
@@ -154,13 +155,15 @@ def __init__(
         initial_rtt: float,
         peer_completed_address_validation: bool,
         send_probe: Callable[[], None],
+        logger: Optional[logging.LoggerAdapter] = None,
         quic_logger: Optional[QuicLoggerTrace] = None,
     ) -> None:
         self.max_ack_delay = 0.025
         self.peer_completed_address_validation = peer_completed_address_validation
         self.spaces: List[QuicPacketSpace] = []
 
         # callbacks
+        self._logger = logger
         self._quic_logger = quic_logger
         self._send_probe = send_probe
 
@@ -319,20 +322,7 @@ def on_loss_detection_timeout(self, now: float) -> None:
             self._detect_loss(loss_space, now=now)
         else:
             self._pto_count += 1
-
-            # reschedule some data
-            for space in self.spaces:
-                self._on_packets_lost(
-                    tuple(
-                        filter(
-                            lambda i: i.is_crypto_packet, space.sent_packets.values()
-                        )
-                    ),
-                    space=space,
-                    now=now,
-                )
-
-            self._send_probe()
+            self.reschedule_data(now=now)
 
     def on_packet_sent(self, packet: QuicSentPacket, space: QuicPacketSpace) -> None:
         space.sent_packets[packet.packet_number] = packet
@@ -349,6 +339,25 @@ def on_packet_sent(self, packet: QuicSentPacket, space: QuicPacketSpace) -> None
             if self._quic_logger is not None:
                 self._log_metrics_updated()
 
+    def reschedule_data(self, now: float) -> None:
+        """
+        Schedule some data for retransmission.
+        """
+        # if there is any outstanding CRYPTO, retransmit it
+        crypto_scheduled = False
+        for space in self.spaces:
+            packets = tuple(
+                filter(lambda i: i.is_crypto_packet, space.sent_packets.values())
+            )
+            if packets:
+                self._on_packets_lost(packets, space=space, now=now)
+                crypto_scheduled = True
+        if crypto_scheduled and self._logger is not None:
+            self._logger.debug("Scheduled CRYPTO data for retransmission")
+
+        # ensure an ACK-elliciting packet is sent
+        self._send_probe()
+
     def _detect_loss(self, space: QuicPacketSpace, now: float) -> None:
         """
         Check whether any packets should be declared lost.