diff --git a/quic/s2n-quic-transport/src/stream/incoming_connection_flow_controller.rs b/quic/s2n-quic-transport/src/stream/incoming_connection_flow_controller.rs index 1e4ffecc62..c865385762 100644 --- a/quic/s2n-quic-transport/src/stream/incoming_connection_flow_controller.rs +++ b/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() diff --git a/quic/s2n-quic-transport/src/stream/receive_stream.rs b/quic/s2n-quic-transport/src/stream/receive_stream.rs index e0b644dd74..198b1c9faa 100644 --- a/quic/s2n-quic-transport/src/stream/receive_stream.rs +++ b/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); } diff --git a/quic/s2n-quic-transport/src/stream/receive_stream/tests.rs b/quic/s2n-quic-transport/src/stream/receive_stream/tests.rs index fd78e521a5..4c9b4f2c96 100644 --- a/quic/s2n-quic-transport/src/stream/receive_stream/tests.rs +++ b/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::::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::::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::::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::::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::::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::::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::::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" ); } +