fix(s2n-quic-transport): release connection flow control credits on stop_sending

quic/s2n-quic-transport/src/stream/incoming_connection_flow_controller.rs

@@ -177,6 +177,11 @@ impl IncomingConnectionFlowController {
         self.inner.borrow().acquired_window
     }
 
+    #[cfg(test)]
+    pub fn consumed_window(&self) -> VarInt {
+        self.inner.borrow().consumed_window
+    }
+
     #[cfg(test)]
     pub fn remaining_window(&self) -> VarInt {
         self.inner.borrow_mut().remaining_window()

quic/s2n-quic-transport/src/stream/receive_stream.rs

@@ -405,6 +405,30 @@ impl ReceiveStream {
                 ref mut missing_data,
                 ..
             } => {
+                //= https://www.rfc-editor.org/rfc/rfc9000#section-4.5
+                //# The receiver MUST use the final size of the stream to
+                //# account for all bytes sent on the stream in its connection level flow
+                //# controller.
+
+                // If we don't know the final size then try acquiring flow control
+                // for any data that arrived before the peer processed STOP_SENDING.
+                if self.receive_buffer.final_size().is_none() {
+                    let data_end = frame
+                        .offset
+                        .checked_add_usize(frame.data.len())
+                        .ok_or_else(|| {
+                            transport::Error::FLOW_CONTROL_ERROR
+                                .with_reason("data size overflow")
+                                .with_frame_type(frame.tag().into())
+                        })?;
+                    self.flow_controller
+                        .acquire_window_up_to(data_end, frame.tag().into())?;
+                }
+
+                // Release any outstanding credits immediately since the data
+                // will not be consumed by the application on this stream.
+                self.flow_controller.release_outstanding_window();
+
                 if missing_data.on_data(frame).is_ready() {
                     self.stop_sending_sync.stop_sync();
                     self.final_state_observed = true;
@@ -762,14 +786,12 @@ impl ReceiveStream {
                     self.state = ReceiveStreamState::DataRead;
                     self.final_state_observed = true;
                     response.status = ops::Status::Finished;
-                    return Ok(response);
                 }
                 // If we've already buffered everything, transition to the final state
                 ReceiveStreamState::Receiving if self.receive_buffer.is_writing_complete() => {
                     self.state = ReceiveStreamState::DataRead;
                     self.final_state_observed = true;
                     response.status = ops::Status::Finished;
-                    return Ok(response);
                 }
                 //= https://www.rfc-editor.org/rfc/rfc9000#section-3.5
                 //# If the stream is in the "Recv" or "Size Known" states, the transport
@@ -785,6 +807,8 @@ impl ReceiveStream {
                         error,
                         missing_data,
                     };
+
+                    response.status = ops::Status::Reset(error);
                 }
             }
 
@@ -795,9 +819,11 @@ impl ReceiveStream {
             // space which had been allocated but not used
             self.receive_buffer.reset();
 
-            // Mark the stream as reset. Note that the request doesn't have a flush so there's
-            // currently no way to wait for the reset to be acknowledged.
-            response.status = ops::Status::Reset(error);
+            // Stop synchronizing the stream flow control window and release all
+            // outstanding connection flow control credits so that other streams
+            // are not starved.
+            self.flow_controller.stop_sync();
+            self.flow_controller.release_outstanding_window();
 
             return Ok(response);
         }

quic/s2n-quic-transport/src/stream/receive_stream/tests.rs

@@ -1740,6 +1740,92 @@ fn do_not_send_flow_control_update_if_stream_is_reset_or_eof() {
     }
 }
 
+#[test]
+fn stop_sending_releases_outstanding_connection_flow_control_credits() {
+    let test_env_config = conn_flow_control_test_env_config();
+    let mut test_env = setup_stream_test_env_with_config(test_env_config);
+
+    // Feed 2000 bytes of data
+    test_env.feed_data(VarInt::from_u32(0), 2000);
+
+    // Don't consume any data - all 2000 credits are held by the stream
+    assert_eq!(
+        VarInt::new(test_env_config.initial_connection_receive_window_size - 2000).unwrap(),
+        Into::<u64>::into(test_env.rx_connection_flow_controller.remaining_window())
+    );
+
+    // Call stop_sending - this simulates the Drop path.
+    // The receive buffer should be reset AND the outstanding flow control
+    // credits should be released back to the connection.
+    assert!(test_env
+        .stop_sending(ApplicationErrorCode::UNKNOWN)
+        .is_ok());
+
+    // After stop_sending, the full desired connection flow control window
+    // should be available again because all credits were released.
+    assert_eq!(
+        VarInt::from_u32(test_env_config.desired_connection_flow_control_window),
+        Into::<u64>::into(test_env.rx_connection_flow_controller.remaining_window())
+    );
+}
+
+#[test]
+fn stop_sending_releases_credits_for_data_arriving_in_stopping_state() {
+    let test_env_config = conn_flow_control_test_env_config();
+    let mut test_env = setup_stream_test_env_with_config(test_env_config);
+
+    // Feed 2000 bytes of data — stream acquires connection flow control credits
+    test_env.feed_data(VarInt::from_u32(0), 2000);
+    assert_eq!(
+        VarInt::new(
+            test_env_config.initial_connection_receive_window_size - 2000,
+        )
+        .unwrap(),
+        Into::<u64>::into(test_env.rx_connection_flow_controller.remaining_window())
+    );
+
+    // STOP_SENDING releases outstanding credits
+    assert!(test_env
+        .stop_sending(ApplicationErrorCode::UNKNOWN)
+        .is_ok());
+    assert_eq!(
+        VarInt::from_u32(test_env_config.desired_connection_flow_control_window),
+        Into::<u64>::into(test_env.rx_connection_flow_controller.remaining_window())
+    );
+
+    let consumed_before = test_env.rx_connection_flow_controller.consumed_window();
+
+    // Feed additional data while in Stopping state.
+    // This data was sent before the peer received STOP_SENDING.
+    // It must be acquired (RFC 9000 §4.5) and then immediately released
+    // so the connection flow control credits are not stranded.
+    let mut events = StreamEvents::new();
+    assert!(test_env
+        .stream
+        .on_data(
+            &stream_data(
+                test_env.stream.stream_id,
+                VarInt::from_u32(2000),
+                &[0u8; 1000],
+                false,
+            ),
+            &mut events,
+        )
+        .is_ok());
+
+    // Connection flow control window should be fully available again
+    assert_eq!(
+        VarInt::from_u32(test_env_config.desired_connection_flow_control_window),
+        Into::<u64>::into(test_env.rx_connection_flow_controller.remaining_window())
+    );
+
+    // consumed_window must increase to reflect the additional data
+    assert_eq!(
+        consumed_before + VarInt::from_u32(1000),
+        test_env.rx_connection_flow_controller.consumed_window()
+    );
+}
+
 #[test]
 fn stop_sending_will_trigger_a_stop_sending_frame() {
     for available_data in &[0, 1] {
@@ -1977,6 +2063,49 @@ fn stop_sending_is_ignored_if_stream_has_already_received_all_data() {
     test_env.assert_write_frames(0);
 }
 
+#[test]
+fn stop_sending_releases_credits_when_all_data_received_but_not_consumed() {
+    let test_env_config = conn_flow_control_test_env_config();
+    let mut test_env = setup_stream_test_env_with_config(test_env_config);
+
+    // Feed 2000 bytes and FIN so is_writing_complete() becomes true
+    test_env.feed_data(VarInt::from_u32(0), 2000);
+    let mut events = StreamEvents::new();
+    assert!(test_env
+        .stream
+        .on_data(
+            &stream_data(
+                test_env.stream.stream_id,
+                VarInt::from_u32(2000),
+                &[],
+                true
+            ),
+            &mut events,
+        )
+        .is_ok());
+    assert!(
+        test_env.stream.receive_stream.receive_buffer.is_writing_complete(),
+        "all data should be received"
+    );
+
+    // Don't consume any data - credits are held
+    assert_eq!(
+        VarInt::new(test_env_config.initial_connection_receive_window_size - 2000).unwrap(),
+        Into::<u64>::into(test_env.rx_connection_flow_controller.remaining_window())
+    );
+
+    // Call stop_sending - hits the is_writing_complete() branch
+    assert!(test_env
+        .stop_sending(ApplicationErrorCode::UNKNOWN)
+        .is_ok());
+
+    // Credits must be released
+    assert_eq!(
+        VarInt::from_u32(test_env_config.desired_connection_flow_control_window),
+        Into::<u64>::into(test_env.rx_connection_flow_controller.remaining_window())
+    );
+}
+
 #[test]
 fn stop_sending_can_be_sent_if_size_is_known_but_data_is_still_missing() {
     for send_missing_data_before_ack in &[true, false] {
@@ -2548,3 +2677,4 @@ fn receiving_into_non_empty_buffers_returns_an_error() {
         "data should not be lost when returning an error"
     );
 }
+