Queue recv window update on shrink (#976)

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 =====

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;
+}