Limit outgoing datagram buffer size

quinn-proto/src/connection.rs

@@ -356,7 +356,6 @@ where
         if ack.largest >= self.space(space).next_packet_number {
             return Err(TransportError::PROTOCOL_VIOLATION("unsent packet acked"));
         }
-        let was_congested = self.congestion_blocked();
         let was_blocked = self.blocked();
         let new_largest = {
             let space = self.space_mut(space);
@@ -446,12 +445,6 @@ where
                 self.events.push_back(Event::StreamWritable { stream });
             }
         }
-        if was_congested
-            && !self.congestion_blocked()
-            && mem::replace(&mut self.datagrams.send_blocked, false)
-        {
-            self.events.push_back(Event::DatagramSendUnblocked);
-        }
         Ok(())
     }
 
@@ -1955,7 +1948,7 @@ where
                     // TODO: Cache, or perhaps forward to user?
                 }
                 Frame::Datagram(datagram) => {
-                    let window = match self.config.datagram_window {
+                    let window = match self.config.datagram_receive_window {
                         None => {
                             return Err(TransportError::PROTOCOL_VIOLATION(
                                 "unexpected DATAGRAM frame",
@@ -2576,6 +2569,10 @@ where
                 self.datagrams.outgoing.push_front(datagram);
                 break;
             }
+            if self.datagrams.outgoing_total >= self.config.datagram_send_window {
+                self.events.push_back(Event::DatagramSendUnblocked);
+            }
+            self.datagrams.outgoing_total -= datagram.data.len();
             datagram.encode(true, buf);
         }
 
@@ -2989,14 +2986,13 @@ where
     /// If `Err(SendDatagramError::Blocked)` is returned, `Event::DatagramSendUnblocked` may be
     /// emitted in the future.
     pub fn send_datagram(&mut self) -> Result<DatagramSender<'_, S>, SendDatagramError> {
-        if self.config.datagram_window.is_none() {
+        if self.config.datagram_receive_window.is_none() {
             return Err(SendDatagramError::Disabled);
         }
         let max = self
             .max_datagram_size()
             .ok_or(SendDatagramError::UnsupportedByPeer)?;
-        if self.congestion_blocked() {
-            self.datagrams.send_blocked = true;
+        if self.datagrams.outgoing_total >= self.config.datagram_send_window {
             return Err(SendDatagramError::Blocked);
         }
         Ok(DatagramSender { max, conn: self })
@@ -3026,7 +3022,7 @@ where
             - 4                 // worst-case packet number size
             - self.space(SpaceId::Data).crypto.as_ref().or(self.zero_rtt_crypto.as_ref()).unwrap().packet.tag_len()
             - Datagram::SIZE_BOUND;
-        self.config.datagram_window?;
+        self.config.datagram_receive_window?;
         let limit = self.params.max_datagram_frame_size?.into_inner();
         Some(limit.min(max_size as u64) as usize)
     }
@@ -3616,6 +3612,7 @@ impl<S: crypto::Session> DatagramSender<'_, S> {
         if data.len() > self.max {
             return Err(DatagramTooLarge);
         }
+        self.conn.datagrams.outgoing_total += data.len();
         self.conn.datagrams.outgoing.push_back(Datagram { data });
         Ok(())
     }
@@ -3625,19 +3622,18 @@ struct DatagramState {
     /// Number of bytes of datagrams that have been received by the local transport but not
     /// delivered to the application
     recv_buffered: usize,
-    /// Whether a `send_datagram` call failed due to congestion
-    send_blocked: bool,
     incoming: VecDeque<Datagram>,
     outgoing: VecDeque<Datagram>,
+    outgoing_total: usize,
 }
 
 impl DatagramState {
     fn new() -> Self {
         Self {
             recv_buffered: 0,
-            send_blocked: false,
             incoming: VecDeque::new(),
             outgoing: VecDeque::new(),
+            outgoing_total: 0,
         }
     }
 }

quinn-proto/src/shared.rs

@@ -118,12 +118,19 @@ pub struct TransportConfig {
     /// This allows passive observers to easily judge the round trip time of a connection, which can
     /// be useful for network administration but sacrifices a small amount of privacy.
     pub allow_spin: bool,
-    /// Maximum number of application datagram bytes to buffer, or None to disable datagrams
+    /// Maximum number of incoming application datagram bytes to buffer, or None to disable
+    /// datagrams
     ///
     /// The peer is forbidden to send single datagrams larger than this size. If the aggregate size
     /// of all datagrams that have been received from the peer but not consumed by the application
     /// exceeds this value, old datagrams are dropped until it is no longer exceeded.
-    pub datagram_window: Option<usize>,
+    pub datagram_receive_window: Option<usize>,
+    /// Maximum number of outgoing application datagram bytes to buffer
+    ///
+    /// While datagrams are sent ASAP, it is possible for an application to generate data faster
+    /// than the link, or even the underlying hardware, can transmit them. This limits the amount of
+    /// memory that may be consumed in that case.
+    pub datagram_send_window: usize,
 }
 
 impl Default for TransportConfig {
@@ -160,7 +167,8 @@ impl Default for TransportConfig {
             keep_alive_interval: 0,
             crypto_buffer_size: 16 * 1024,
             allow_spin: true,
-            datagram_window: Some(STREAM_RWND as usize),
+            datagram_receive_window: Some(STREAM_RWND as usize),
+            datagram_send_window: 1 * 1024 * 1024,
         }
     }
 }

quinn-proto/src/tests/mod.rs

@@ -1175,7 +1175,7 @@ fn datagram_window() {
     const WINDOW: usize = 100;
     let server = ServerConfig {
         transport: Arc::new(TransportConfig {
-            datagram_window: Some(WINDOW),
+            datagram_receive_window: Some(WINDOW),
             ..TransportConfig::default()
         }),
         ..server_config()
@@ -1238,7 +1238,7 @@ fn datagram_window() {
 fn datagram_unsupported() {
     let server = ServerConfig {
         transport: Arc::new(TransportConfig {
-            datagram_window: None,
+            datagram_receive_window: None,
             ..TransportConfig::default()
         }),
         ..server_config()

quinn-proto/src/transport_parameters.rs

@@ -89,7 +89,7 @@ impl TransportParameters {
             disable_active_migration: server_config.map_or(false, |c| !c.migration),
             active_connection_id_limit: REM_CID_COUNT,
             max_datagram_frame_size: config
-                .datagram_window
+                .datagram_receive_window
                 .map(|x| (x.min(u16::max_value().into()) as u16).into()),
             ..Self::default()
         }