Fix race condition between 0-RTT and Incoming

Closes #1820

The fix:

- Endpoint now maintains a slab with an entry for each pending Incoming
  to buffer received data.
- ConnectionIndex now maps initial DCID to that slab key immediately
  upon construction of Incoming.
- If Incoming is accepted, association is overridden with association
  with ConnectionHandle, and all buffered datagrams are fed to newly
  constructed Connection.
- If Incoming is refused/retried/ignored, or accepting errors,
  association and slab entry are cleaned up to prevent memory leak.

Additional considerations:

- The Incoming::ignore operation can no longer be implemented as just
  dropping it. To help prevent incorrect API usage, proto::Incoming is
  modified to log a warning if it is dropped without being passed to
  Endpoint::accept/refuse/retry/ignore.
- Three things protect against memory exhaustion attacks here:

  1. The MAX_INCOMING_CONNECTIONS limit is moved from quinn to proto,
     limiting the number of concurrent incoming connections for which
     datagrams will be buffered before the application decides what to
     do with them. Also, it is changed from a constant to a field of the
     server config, max_incoming.
  2. Per-incoming buffered data is limited to a new limit stored in the
     server config, incoming_buffer_size, beyond which subsequent
     packets are discarded if received in these conditions.
  3. The sum total of all incoming buffered data is limited to a new
     limit stored in the server config, incoming_buffer_size_total,
     beyond which subsequent packets are discarded if received in these
     conditions.

quinn-proto/src/config.rs

@@ -776,6 +776,10 @@ pub struct ServerConfig {
 
     pub(crate) preferred_address_v4: Option<SocketAddrV4>,
     pub(crate) preferred_address_v6: Option<SocketAddrV6>,
+
+    pub(crate) max_incoming: usize,
+    pub(crate) incoming_buffer_size: u64,
+    pub(crate) incoming_buffer_size_total: u64,
 }
 
 impl ServerConfig {
@@ -795,6 +799,10 @@ impl ServerConfig {
 
             preferred_address_v4: None,
             preferred_address_v6: None,
+
+            max_incoming: 1 << 16,
+            incoming_buffer_size: 10 << 20,
+            incoming_buffer_size_total: 100 << 20,
         }
     }
 
@@ -838,6 +846,54 @@ impl ServerConfig {
         self.preferred_address_v6 = address;
         self
     }
+
+    /// Maximum number of [`Incoming`][crate::Incoming] to allow to exist at a time
+    ///
+    /// An [`Incoming`][crate::Incoming] comes into existence when an incoming connection attempt
+    /// is received and stops existing when the application either accepts it or otherwise disposes
+    /// of it. While this limit is reached, new incoming connection attempts are immediately
+    /// refused. Larger values have greater worst-case memory consumption, but accommodate greater
+    /// application latency in handling incoming connection attempts.
+    ///
+    /// The default value is set to 65536. Assuming each [`Incoming`][crate::Incoming] accounts for
+    /// little over 1200 bytes of memory at most, this should limit memory consumption from this to
+    /// under 100 MiB--a generous amount that still prevents memory exhaustion in most contexts.
+    pub fn max_incoming(&mut self, max_incoming: usize) -> &mut Self {
+        self.max_incoming = max_incoming;
+        self
+    }
+
+    /// Maximum number of received bytes to buffer for each [`Incoming`][crate::Incoming]
+    ///
+    /// An [`Incoming`][crate::Incoming] comes into existence when an incoming connection attempt
+    /// is received and stops existing when the application either accepts it or otherwise disposes
+    /// of it. This limit governs only packets received within that period, and does not include
+    /// the first packet. Packets received in excess of this limit are dropped, which may cause
+    /// 0-RTT data to have to be retransmitted.
+    ///
+    /// The default value is set to 10 MiB--an amount such that in most situations a client would
+    /// not transmit that much 0-RTT data faster than the server handles the corresponding
+    /// [`Incoming`][crate::Incoming].
+    pub fn incoming_buffer_size(&mut self, incoming_buffer_size: u64) -> &mut Self {
+        self.incoming_buffer_size = incoming_buffer_size;
+        self
+    }
+
+    /// Maximum number of received bytes to buffer for all [`Incoming`][crate::Incoming]
+    /// collectively
+    /// 
+    /// An [`Incoming`][crate::Incoming] comes into existence when an incoming connection attempt
+    /// is received and stops existing when the application either accepts it or otherwise disposes
+    /// of it. This limit governs only packets received within that period, and does not include
+    /// the first packet. Packets received in excess of this limit are dropped, which may cause
+    /// 0-RTT data to have to be retransmitted.
+    ///
+    /// The default value is set to 100 MiB--a generous amount that still prevents memory
+    /// exhaustion in most contexts.
+    pub fn incoming_buffer_size_total(&mut self, incoming_buffer_size_total: u64) -> &mut Self {
+        self.incoming_buffer_size_total = incoming_buffer_size_total;
+        self
+    }
 }
 
 #[cfg(feature = "rustls")]
@@ -880,6 +936,12 @@ impl fmt::Debug for ServerConfig {
             .field("migration", &self.migration)
             .field("preferred_address_v4", &self.preferred_address_v4)
             .field("preferred_address_v6", &self.preferred_address_v6)
+            .field("max_incoming", &self.max_incoming)
+            .field("incoming_buffer_size", &self.incoming_buffer_size)
+            .field(
+                "incoming_buffer_size_total",
+                &self.incoming_buffer_size_total,
+            )
             .finish()
     }
 }

quinn-proto/src/connection/mod.rs

@@ -25,8 +25,8 @@ use crate::{
     packet::{Header, InitialHeader, InitialPacket, LongType, Packet, PartialDecode, SpaceId},
     range_set::ArrayRangeSet,
     shared::{
-        ConnectionEvent, ConnectionEventInner, ConnectionId, EcnCodepoint, EndpointEvent,
-        EndpointEventInner,
+        ConnectionEvent, ConnectionEventInner, ConnectionId, DatagramConnectionEvent, EcnCodepoint,
+        EndpointEvent, EndpointEventInner,
     },
     token::ResetToken,
     transport_parameters::TransportParameters,
@@ -989,13 +989,13 @@ impl Connection {
     pub fn handle_event(&mut self, event: ConnectionEvent) {
         use self::ConnectionEventInner::*;
         match event.0 {
-            Datagram {
+            Datagram(DatagramConnectionEvent {
                 now,
                 remote,
                 ecn,
                 first_decode,
                 remaining,
-            } => {
+            }) => {
                 // If this packet could initiate a migration and we're a client or a server that
                 // forbids migration, drop the datagram. This could be relaxed to heuristically
                 // permit NAT-rebinding-like migration.
@@ -3346,11 +3346,11 @@ impl Connection {
     #[cfg(test)]
     pub(crate) fn decode_packet(&self, event: &ConnectionEvent) -> Option<Vec<u8>> {
         let (first_decode, remaining) = match &event.0 {
-            ConnectionEventInner::Datagram {
+            ConnectionEventInner::Datagram(DatagramConnectionEvent {
                 first_decode,
                 remaining,
                 ..
-            } => (first_decode, remaining),
+            }) => (first_decode, remaining),
             _ => return None,
         };