fix: replenish empty DATA frame limit with payload

src/client.rs

@@ -1161,8 +1161,9 @@ impl Builder {
     ///
     /// Small DATA frames consume this budget. The budget is restored when
     /// buffered frames are consumed by the application, while sufficiently
-    /// large frames may also restore budget. Empty DATA frames are limited
-    /// separately and do not consume this budget.
+    /// large frames may also restore budget. Empty DATA frames do not consume
+    /// this budget: they have a separate, fixed allowance that sufficiently
+    /// large frames also restore.
     ///
     /// When this budget is exhausted, the connection is closed with
     /// `ENHANCE_YOUR_CALM`.

src/proto/mod.rs

@@ -37,6 +37,9 @@ pub const DEFAULT_LOCAL_RESET_COUNT_MAX: usize = 1024;
 // smaller than this consume more internal bookkeeping than useful data.
 pub const DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD: usize = 256;
 pub const DEFAULT_DATA_FRAME_BUDGET: usize = DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD * 100;
+// Number of empty, non-final DATA frames accepted in a burst. Each non-final
+// DATA frame earns back one more for every DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD
+// bytes of payload beyond that threshold.
 pub const MAX_RECV_EMPTY_DATA_FRAMES: usize = 100;
 // RFC 9113 suggests allowing at minimum 100 streams, it seems reasonable to
 // by default allow a portion of that to be remembered as reset for some time.

src/proto/streams/counts.rs

@@ -83,9 +83,11 @@ pub(super) struct Counts {
     /// Whether the DATA frame budget follows the target connection window.
     data_frame_budget_is_auto: bool,
 
-    /// Number of empty, non-final DATA frames received over the lifetime of
-    /// the connection.
-    num_recv_empty_data_frames: usize,
+    /// connection-level budget for empty, non-final DATA frames.
+    ///
+    /// Closes the connections upon a burst of empty, non-final DATA frames.
+    /// Replenishes the budget when non-empty DATA frames are received.
+    empty_data_frame_budget: Budget,
 }
 
 impl Counts {
@@ -105,7 +107,9 @@ impl Counts {
             num_local_error_reset_streams: 0,
             data_frame_budget: Budget::new(config.data_frame_budget.resolve(None)),
             data_frame_budget_is_auto: config.data_frame_budget.is_auto(),
-            num_recv_empty_data_frames: 0,
+            empty_data_frame_budget: Budget::new(
+                MAX_RECV_EMPTY_DATA_FRAMES * DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD,
+            ),
         }
     }
 
@@ -119,20 +123,15 @@ impl Counts {
     /// Records the framing overhead of a DATA frame.
     pub fn record_data_frame(&mut self, payload_len: usize) -> Result<(), BudgetExhausted> {
         if payload_len == 0 {
-            self.num_recv_empty_data_frames = self
-                .num_recv_empty_data_frames
-                .checked_add(1)
-                .ok_or(BudgetExhausted)?;
-            if self.num_recv_empty_data_frames > MAX_RECV_EMPTY_DATA_FRAMES {
-                return Err(BudgetExhausted);
-            }
-            Ok(())
+            self.empty_data_frame_budget
+                .consume(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD)
         } else if payload_len < DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD {
             self.data_frame_budget
                 .consume(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD - payload_len)
         } else {
-            self.data_frame_budget
-                .replenish(payload_len - DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD);
+            let surplus = payload_len - DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD;
+            self.data_frame_budget.replenish(surplus);
+            self.empty_data_frame_budget.replenish(surplus);
             Ok(())
         }
     }
@@ -442,6 +441,10 @@ mod tests {
         );
         auto.set_target_connection_window_size(1024 * 1024);
         assert_eq!(auto.data_frame_budget.max, 512 * 1024);
+        assert_eq!(
+            auto.empty_data_frame_budget.max,
+            MAX_RECV_EMPTY_DATA_FRAMES * DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD
+        );
         for _ in 0..1_000 {
             auto.record_data_frame(171).unwrap();
         }
@@ -494,13 +497,33 @@ mod tests {
     }
 
     #[test]
-    fn large_data_frames_do_not_replenish_empty_data_frame_limit() {
+    fn large_data_frames_replenish_empty_data_frame_budget() {
+        let mut counts = counts();
+
+        // Each frame earns back twice what one empty frame costs, so an
+        // uncapped refill would build up extra allowance.
+        for _ in 0..MAX_RECV_EMPTY_DATA_FRAMES * 10 {
+            counts.record_data_frame(0).unwrap();
+            counts
+                .record_data_frame(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD * 3)
+                .unwrap();
+        }
+
+        // Replenishment does not accumulate beyond the initial allowance.
+        for _ in 0..MAX_RECV_EMPTY_DATA_FRAMES {
+            counts.record_data_frame(0).unwrap();
+        }
+        assert!(counts.record_data_frame(0).is_err());
+    }
+
+    #[test]
+    fn small_data_frames_do_not_replenish_empty_data_frame_budget() {
         let mut counts = counts();
 
         for _ in 0..MAX_RECV_EMPTY_DATA_FRAMES {
             counts.record_data_frame(0).unwrap();
             counts
-                .record_data_frame(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD * 2)
+                .record_data_frame(DEFAULT_DATA_FRAME_OVERHEAD_THRESHOLD - 1)
                 .unwrap();
         }
         assert!(counts.record_data_frame(0).is_err());

src/server.rs

@@ -1055,8 +1055,9 @@ impl Builder {
     ///
     /// Small DATA frames consume this budget. The budget is restored when
     /// buffered frames are consumed by the application, while sufficiently
-    /// large frames may also restore budget. Empty DATA frames are limited
-    /// separately and do not consume this budget.
+    /// large frames may also restore budget. Empty DATA frames do not consume
+    /// this budget: they have a separate, fixed allowance that sufficiently
+    /// large frames also restore.
     ///
     /// When this budget is exhausted, the connection is closed with
     /// `ENHANCE_YOUR_CALM`.

tests/h2-tests/tests/stream_states.rs

@@ -232,6 +232,71 @@ async fn too_many_padded_empty_data_frames_sends_goaway() {
     join(mock, h2).await;
 }
 
+#[tokio::test]
+async fn padded_empty_data_frames_between_large_data_frames_do_not_exhaust_budget() {
+    h2_support::trace_init!();
+
+    // More streams than the burst of empty DATA frames allowed, with one empty
+    // frame per stream.
+    const NUM_STREAMS: u32 = 150;
+    const BODY_LEN: usize = 1024;
+    // Large enough that the capacity released automatically never reaches the
+    // WINDOW_UPDATE threshold, so the client sends no further WINDOW_UPDATE
+    // frames.
+    const CONNECTION_WINDOW: u32 = 1 << 20;
+
+    let (io, mut srv) = mock::new();
+
+    let mock = async move {
+        let settings = srv.assert_client_handshake().await;
+        assert_default_settings!(settings);
+        srv.recv_frame(frames::window_update(0, CONNECTION_WINDOW - 65_535))
+            .await;
+
+        for i in 0..NUM_STREAMS {
+            let stream_id = 1 + i * 2;
+            srv.recv_frame(
+                frames::headers(stream_id)
+                    .request("GET", "https://http2.akamai.com/")
+                    .eos(),
+            )
+            .await;
+            srv.send_frame(frames::headers(stream_id).response(200))
+                .await;
+            srv.send_frame(frames::data(stream_id, vec![0; BODY_LEN]))
+                .await;
+            // A zero padding length byte with no payload, as some servers
+            // emit between the DATA frames of a body.
+            srv.send_frame(frames::data(stream_id, [0]).padded()).await;
+            srv.send_frame(frames::data(stream_id, "").eos()).await;
+        }
+    };
+
+    let h2 = async move {
+        let (mut client, mut h2) = client::Builder::new()
+            .initial_connection_window_size(CONNECTION_WINDOW)
+            .handshake::<_, Bytes>(io)
+            .await
+            .unwrap();
+
+        for _ in 0..NUM_STREAMS {
+            let request = Request::builder()
+                .uri("https://http2.akamai.com/")
+                .body(())
+                .unwrap();
+            let response = client.send_request(request, true).unwrap().0;
+            let response = h2.drive(response).await.unwrap();
+            let body = h2.drive(util::concat(response.into_body())).await;
+            assert_eq!(body.unwrap().len(), BODY_LEN);
+        }
+
+        drop(client);
+        h2.await.unwrap();
+    };
+
+    join(mock, h2).await;
+}
+
 #[tokio::test]
 async fn too_many_small_data_frames_sends_goaway() {
     h2_support::trace_init!();