From c4fb923d89a497b9768bdf3f68d0c96857687158 Mon Sep 17 00:00:00 2001 From: Jack Callahan <346653883+BlizTechDev@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:06:51 +0200 Subject: [PATCH] Queue recv window update on shrink (#976) Problem When a delayed SETTINGS ACK reduces SETTINGS_INITIAL_WINDOW_SIZE, Recv::apply_local_settings decrements each open stream's receive window via dec_recv_window(dec) without re-checking whether previously released capacity has now crossed the WINDOW_UPDATE batching threshold. If the application already consumed and released all received DATA while a sub-threshold remainder stayed batched, the stream is never added to pending_window_updates, leaving the peer stalled on a non-positive send window with no further incoming DATA to trigger progress. Solution After decrementing stream.recv_flow in the Ordering::Less branch of Recv::apply_local_settings, check stream.recv_flow.unclaimed_capacity() and push the stream onto self.pending_window_updates when eligible. Validation - Added unit tests in src/proto/streams/recv.rs covering released credit becoming eligible on shrink, shrinking with no released credit, and deduplicating an already queued stream. - Added an end-to-end wire test in tests/h2-tests/tests/flow_control.rs verifying that a delayed SETTINGS ACK flushes the batched stream WINDOW_UPDATE and completes the request body transfer. - Ran cargo fmt, cargo clippy --lib -- -D warnings, cargo check --features unstable, cargo test, and cargo test -p h2-tests. --- src/proto/streams/recv.rs | 179 +++++++++++++++++++++++++++ tests/h2-tests/tests/flow_control.rs | 102 +++++++++++++++ 2 files changed, 281 insertions(+) diff --git a/src/proto/streams/recv.rs b/src/proto/streams/recv.rs index bf4d53dfe..b1e0d8c66 100644 --- a/src/proto/streams/recv.rs +++ b/src/proto/streams/recv.rs @@ -612,6 +612,11 @@ impl Recv { .recv_flow .dec_recv_window(dec) .map_err(proto::Error::library_go_away)?; + + if stream.recv_flow.unclaimed_capacity().is_some() { + self.pending_window_updates.push(&mut stream); + } + Ok::<_, proto::Error>(()) })?; } @@ -1319,6 +1324,180 @@ mod tests { assert_eq!(stream.in_flight_recv_data, 0); assert_eq!(recv.in_flight_data, 0); } + + fn test_config() -> Config { + Config { + initial_max_send_streams: 100, + local_max_buffer_size: 1024, + local_next_stream_id: StreamId::from(2), + local_push_enabled: false, + extended_connect_protocol_enabled: false, + local_reset_duration: Duration::from_secs(30), + local_reset_max: 100, + remote_reset_max: 100, + remote_init_window_sz: DEFAULT_INITIAL_WINDOW_SIZE, + remote_max_initiated: Some(100), + local_max_error_reset_streams: None, + data_frame_budget: DataFrameBudget::Configured(DEFAULT_DATA_FRAME_BUDGET), + } + } + + #[test] + fn apply_local_settings_shrink_queues_released_capacity() { + let config = test_config(); + let mut recv = Recv::new(peer::Dyn::Server, &config); + let mut store = Store::new(); + let id = StreamId::from(1); + + { + let mut stream = store.insert( + id, + Stream::new(id, DEFAULT_INITIAL_WINDOW_SIZE, DEFAULT_INITIAL_WINDOW_SIZE), + ); + stream + .recv_flow + .send_data(DEFAULT_INITIAL_WINDOW_SIZE) + .unwrap(); + stream.in_flight_recv_data = DEFAULT_INITIAL_WINDOW_SIZE; + } + recv.flow.send_data(DEFAULT_INITIAL_WINDOW_SIZE).unwrap(); + recv.in_flight_data = DEFAULT_INITIAL_WINDOW_SIZE; + + let mut task = None; + for bytes in [16_384, 16_384, 16_384, 16_383] { + { + let mut stream = store.find_mut(&id).unwrap(); + recv.release_capacity(bytes, &mut stream, &mut task) + .unwrap(); + } + if let Some(mut stream) = recv.pending_window_updates.pop(&mut store) { + let amount = stream.recv_flow.unclaimed_capacity().unwrap(); + stream.recv_flow.inc_window(amount).unwrap(); + } + } + + { + let stream = store.find_mut(&id).unwrap(); + assert_eq!(stream.recv_flow.window_size(), 49_152); + assert_eq!(stream.recv_flow.available().as_size(), 65_535); + assert_eq!(stream.in_flight_recv_data, 0); + } + assert!(recv.pending_window_updates.pop(&mut store).is_none()); + + let mut settings = frame::Settings::default(); + settings.set_initial_window_size(Some(1024)); + recv.apply_local_settings(&settings, &mut store).unwrap(); + + let mut stream = recv + .pending_window_updates + .pop(&mut store) + .expect("shrinking window must queue previously released capacity"); + assert_eq!(stream.id, id); + assert_eq!(stream.recv_flow.unclaimed_capacity(), Some(16_383)); + assert_eq!(stream.recv_flow.available().as_size(), 1024); + stream.recv_flow.inc_window(16_383).unwrap(); + assert_eq!(stream.recv_flow.window_size(), 1024); + assert_eq!(stream.recv_flow.unclaimed_capacity(), None); + drop(stream); + + assert!(recv.pending_window_updates.pop(&mut store).is_none()); + } + + #[test] + fn apply_local_settings_shrink_without_released_capacity_does_not_queue() { + let config = test_config(); + let mut recv = Recv::new(peer::Dyn::Server, &config); + let mut store = Store::new(); + let id = StreamId::from(1); + + { + let mut stream = store.insert( + id, + Stream::new(id, DEFAULT_INITIAL_WINDOW_SIZE, DEFAULT_INITIAL_WINDOW_SIZE), + ); + stream + .recv_flow + .send_data(DEFAULT_INITIAL_WINDOW_SIZE) + .unwrap(); + stream.in_flight_recv_data = DEFAULT_INITIAL_WINDOW_SIZE; + } + + let mut settings = frame::Settings::default(); + settings.set_initial_window_size(Some(1024)); + recv.apply_local_settings(&settings, &mut store).unwrap(); + + { + let stream = store.find_mut(&id).unwrap(); + assert_eq!(stream.recv_flow.unclaimed_capacity(), None); + } + assert!(recv.pending_window_updates.pop(&mut store).is_none()); + } + + #[test] + fn apply_local_settings_shrink_deduplicates_already_queued_stream() { + let config = test_config(); + let mut recv = Recv::new(peer::Dyn::Server, &config); + let mut store = Store::new(); + let id1 = StreamId::from(1); + let id3 = StreamId::from(3); + + { + let mut s1 = store.insert( + id1, + Stream::new( + id1, + DEFAULT_INITIAL_WINDOW_SIZE, + DEFAULT_INITIAL_WINDOW_SIZE, + ), + ); + s1.recv_flow.send_data(DEFAULT_INITIAL_WINDOW_SIZE).unwrap(); + s1.in_flight_recv_data = DEFAULT_INITIAL_WINDOW_SIZE; + } + { + let mut s3 = store.insert( + id3, + Stream::new( + id3, + DEFAULT_INITIAL_WINDOW_SIZE, + DEFAULT_INITIAL_WINDOW_SIZE, + ), + ); + s3.recv_flow.send_data(DEFAULT_INITIAL_WINDOW_SIZE).unwrap(); + s3.in_flight_recv_data = DEFAULT_INITIAL_WINDOW_SIZE; + } + recv.flow.send_data(DEFAULT_INITIAL_WINDOW_SIZE).unwrap(); + recv.in_flight_data = DEFAULT_INITIAL_WINDOW_SIZE; + + // Leave s1 already queued in pending_window_updates. + { + let mut task = None; + let mut s1 = store.find_mut(&id1).unwrap(); + recv.release_capacity(32_768, &mut s1, &mut task).unwrap(); + } + + // Put s3 into the sub-threshold batched state (16,383 unclaimed, not queued). + recv.flow.inc_window(32_768).unwrap(); + recv.flow.send_data(32_768).unwrap(); + recv.in_flight_data += 32_768; + { + let mut s3 = store.find_mut(&id3).unwrap(); + s3.recv_flow.inc_window(49_152).unwrap(); + s3.recv_flow.assign_capacity(49_152).unwrap(); + s3.in_flight_recv_data = 16_383; + let mut task = None; + recv.release_capacity(16_383, &mut s3, &mut task).unwrap(); + assert!(!s3.is_pending_window_update); + } + + let mut settings = frame::Settings::default(); + settings.set_initial_window_size(Some(1024)); + recv.apply_local_settings(&settings, &mut store).unwrap(); + + let first = recv.pending_window_updates.pop(&mut store).unwrap().id; + let second = recv.pending_window_updates.pop(&mut store).unwrap().id; + assert_eq!((first, second), (id1, id3)); + assert!(recv.pending_window_updates.pop(&mut store).is_none()); + } } // ===== impl Open ===== diff --git a/tests/h2-tests/tests/flow_control.rs b/tests/h2-tests/tests/flow_control.rs index 11d1207e1..8397737e6 100644 --- a/tests/h2-tests/tests/flow_control.rs +++ b/tests/h2-tests/tests/flow_control.rs @@ -3072,3 +3072,105 @@ async fn poll_capacity_window_update_settings_race() { join(srv, h2).await; } +#[tokio::test] +async fn late_settings_ack_shrinks_window_and_sends_batched_window_update() { + h2_support::trace_init!(); + let (io, mut client) = mock::new(); + let (fourth_released_tx, fourth_released_rx) = oneshot::channel(); + + let peer = async move { + client.write_preface().await; + client + .send(frame::Settings::default().into()) + .await + .unwrap(); + + // Read server's initial SETTINGS (initial_window_size = 1024) and its ACK of ours, + // intentionally delaying our SETTINGS ACK so stream 1 starts at 65,535. + let frame = client.next().await.expect("unexpected EOF").unwrap(); + let settings = assert_settings!(frame); + assert_eq!(settings.initial_window_size(), Some(1024)); + + let frame = client.next().await.expect("unexpected EOF").unwrap(); + let ack = assert_settings!(frame); + assert!(ack.is_ack()); + + client + .send_frame(frames::headers(1).request("POST", "https://example.com/")) + .await; + + // Exhaust the initial 65,535-byte stream window across 4 DATA frames. + for len in [16_384, 16_384, 16_384, 16_383] { + client.send_frame(frames::data(1, vec![0u8; len])).await; + } + + // First three 16,384-byte releases each cross the 50% threshold; the 4th + // (16,383 < 49,152 / 2) remains batched below threshold. + for _ in 0..3 { + client.recv_frame(frames::window_update(0, 16_384)).await; + client.recv_frame(frames::window_update(1, 16_384)).await; + } + + fourth_released_rx.await.unwrap(); + yield_once().await; + + // Now ACK the server's initial SETTINGS (shrinking stream window to 1024). + // The server must advertise the batched 16,383-byte stream credit so our window + // moves from -15,359 to +1,024 without waiting for another DATA frame. + client.send_frame(frames::settings_ack()).await; + tokio::time::timeout( + Duration::from_secs(1), + client.recv_frame(frames::window_update(1, 16_383)), + ) + .await + .expect("timed out waiting for stream WINDOW_UPDATE after delayed SETTINGS ACK"); + + client + .send_frame(frames::data(1, vec![0u8; 1024]).eos()) + .await; + client + .recv_frame(frames::headers(1).response(200).eos()) + .await; + }; + + let mut builder = server::Builder::new(); + builder.initial_window_size(1024); + + let srv = async move { + let mut h2 = builder.handshake::<_, Bytes>(io).await.expect("handshake"); + let (req, mut respond) = h2.accept().await.unwrap().unwrap(); + let mut body = req.into_body(); + + let drive_conn = async move { + poll_fn(|cx| h2.poll_closed(cx)).await.unwrap(); + }; + + let handle_stream = async move { + // Drain all 4 initial chunks (65,535 bytes) before releasing chunk-by-chunk + // so the stream window starts at 0 and reproduces the exact batching steps. + let mut lengths = Vec::with_capacity(4); + for _ in 0..4 { + let chunk = body.data().await.unwrap().unwrap(); + lengths.push(chunk.len()); + } + assert_eq!(lengths, vec![16_384, 16_384, 16_384, 16_383]); + + for len in lengths { + body.flow_control().release_capacity(len).unwrap(); + yield_once().await; + } + fourth_released_tx.send(()).unwrap(); + + let tail = body.data().await.unwrap().unwrap(); + assert_eq!(tail.len(), 1024); + assert!(body.data().await.is_none()); + + let rsp = http::Response::builder().status(200).body(()).unwrap(); + respond.send_response(rsp, true).unwrap(); + }; + + join(drive_conn, handle_stream).await; + }; + + join(peer, srv).await; +}