diff --git a/src/client.rs b/src/client.rs index 9d0ff31d..dce27aec 100644 --- a/src/client.rs +++ b/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`. diff --git a/src/proto/mod.rs b/src/proto/mod.rs index 5e3b4f40..de566b48 100644 --- a/src/proto/mod.rs +++ b/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. diff --git a/src/proto/streams/counts.rs b/src/proto/streams/counts.rs index f74aa14a..f60fd39d 100644 --- a/src/proto/streams/counts.rs +++ b/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()); diff --git a/src/server.rs b/src/server.rs index 931661a1..91d6bd41 100644 --- a/src/server.rs +++ b/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`. diff --git a/tests/h2-tests/tests/stream_states.rs b/tests/h2-tests/tests/stream_states.rs index e302f232..1ec03974 100644 --- a/tests/h2-tests/tests/stream_states.rs +++ b/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!();