From 9e69975d930436d3d8df2d2c9c0fe6cead8f9dcb Mon Sep 17 00:00:00 2001 From: Moritz Date: Wed, 7 Oct 2026 11:46:15 +0000 Subject: [PATCH] fix(http2): replenish connection receive window when stream closes with unconsumed data > [!NOTE] > This PR was generated by an AI coding agent (Jetski) on behalf of @mosuem. ### Summary Per **RFC 9113 Section 5.2 (Flow Control)** and **Section 6.9 (`WINDOW_UPDATE`)**, a receiver that receives a flow-controlled `DATA` frame must always account for its contribution against the connection flow-control window, even when the stream terminates abnormally or its incoming messages are discarded before the application consumes them. Previously: 1. `ConnectionMessageQueueIn.removeStreamMessageQueue(streamId)` discarded buffered `DataMessage`s in `_stream2pendingMessages[streamId]` without calling `_windowUpdateHandler.dataProcessed(ignoredBytes)`. If a stream was paused (or had not yet been listened to) when `DATA` frames arrived and was subsequently terminated (`stream.terminate()` or `RST_STREAM`), those bytes were permanently leaked from the connection-level receive window. Once 65,535 bytes (or `connectionWindowSize`) accumulated across terminated streams, all future streams on the connection deadlocked waiting for connection window credit. 2. `StreamMessageQueueIn.cancel()` only completed `_onCancelCompleter` without draining `_pendingMessages`, updating `bufferIndicator`, or running `onCloseCheck()`. Cancelling `stream.incomingMessages` while paused left `bufferIndicator.wouldBuffer == true`, stranding buffered `DATA` in `ConnectionMessageQueueIn` and keeping the stream's active slot open indefinitely. 3. `StreamHandler._terminateStream` was a no-op when `stream.state == StreamState.Closed`, so calling `stream.terminate()` on a stream whose remote end had already sent `END_STREAM` while the local listener was paused failed to clean up the stream or replenish the connection window. ### Changes - **`lib/src/flowcontrol/connection_queues.dart`**: In `ConnectionMessageQueueIn.removeStreamMessageQueue`, sum the byte lengths of any undelivered `DataMessage`s and call `_windowUpdateHandler.dataProcessed(ignoredBytes)` when `!wasTerminated`. Also terminate any remaining stream message queues in `ConnectionMessageQueueIn.onTerminated`. - **`lib/src/flowcontrol/stream_queues.dart`**: Override `StreamMessageQueueIn.cancel()` to drain `_pendingMessages`, update `bufferIndicator` so `ConnectionMessageQueueIn` flushes any held messages and replenishes the connection window, and run `onCloseCheck()`. - **`lib/src/streams/stream_handler.dart`**: Allow `_terminateStream` to clean up `StreamState.Closed` streams whose `incomingQueue` has not yet closed. ### Test Verification (Fails Before $\rightarrow$ Passes After) Added two regression tests in `test/flow_control_test.dart`, driving the raw peer from #1999: - `terminating a stream with unconsumed buffered DATA replenishes the connection receive window` - `cancelling incomingMessages on a paused stream replenishes the connection window and releases the active stream slot` Wherever a test needs to know that the client has processed the frames sent so far (or finished reacting to a cancel), it uses the peer's PING round trip (`peer.barrier()`) rather than a wall-clock sleep. **Before fix** (tests on `master`): ```text 00:05 +0 -1: terminating a stream with unconsumed buffered DATA replenishes the connection receive window [E] TimeoutException after 0:00:05.000000: Future not completed test/flow_control_test.dart 126:7 _RawPeer.until test/flow_control_test.dart 427:5 main. 00:10 +0 -2: cancelling incomingMessages on a paused stream replenishes the connection window and releases the active stream slot [E] TimeoutException after 0:00:05.000000: Future not completed test/flow_control_test.dart 126:7 _RawPeer.until test/flow_control_test.dart 469:5 main. ``` **After fix:** ```text 00:00 +9: All tests passed! ``` --- pkgs/http2/CHANGELOG.md | 3 + .../src/flowcontrol/connection_queues.dart | 31 ++++++-- .../lib/src/flowcontrol/stream_queues.dart | 16 +++- .../http2/lib/src/streams/stream_handler.dart | 9 ++- pkgs/http2/test/flow_control_test.dart | 77 +++++++++++++++++++ 5 files changed, 127 insertions(+), 9 deletions(-) diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index 534ea890dd..2ac83f0378 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md @@ -8,6 +8,9 @@ - Honor the peer's `SETTINGS_MAX_FRAME_SIZE` when sending; the ACK of our own `SETTINGS_INITIAL_WINDOW_SIZE` no longer resizes send windows. - Never send a WINDOW_UPDATE with an increment of 0. +- Replenish the connection-level receive window and release active stream slots + when a stream with buffered unconsumed `DATA` frames is terminated, reset, or + its incoming subscription is cancelled. ## 3.1.0 diff --git a/pkgs/http2/lib/src/flowcontrol/connection_queues.dart b/pkgs/http2/lib/src/flowcontrol/connection_queues.dart index 0f4c86fd50..29ca79ec4f 100644 --- a/pkgs/http2/lib/src/flowcontrol/connection_queues.dart +++ b/pkgs/http2/lib/src/flowcontrol/connection_queues.dart @@ -219,10 +219,11 @@ class ConnectionMessageQueueIn extends Object @override void onTerminated(Object? error) { - // NOTE: The higher level will be shutdown first, so all streams - // should have been removed at this point. - assert(_stream2messageQueue.isEmpty); - assert(_stream2pendingMessages.isEmpty); + for (final mq in _stream2messageQueue.values) { + mq.terminate(error); + } + _stream2messageQueue.clear(); + _stream2pendingMessages.clear(); closeWithError(error); } @@ -262,9 +263,27 @@ class ConnectionMessageQueueIn extends Object /// Removes a stream id and its message queue from this connection-level /// message queue. - void removeStreamMessageQueue(int streamId) { - _stream2pendingMessages.remove(streamId); + /// + /// Buffered [DataMessage]s that were never delivered to the stream still + /// count against the connection receive window (RFC 9113 section 6.9.1), so + /// the window is replenished for them - unless [replenishWindow] is `false`, + /// which callers pass while the connection is being torn down and nothing + /// can be written to the peer any more. + void removeStreamMessageQueue(int streamId, {bool replenishWindow = true}) { + final pendingMessages = _stream2pendingMessages.remove(streamId); _stream2messageQueue.remove(streamId); + if (pendingMessages != null) { + var ignoredBytes = 0; + for (final message in pendingMessages) { + _count--; + if (message is DataMessage) { + ignoredBytes += message.bytes.length; + } + } + if (replenishWindow && ignoredBytes > 0) { + _windowUpdateHandler.dataProcessed(ignoredBytes); + } + } } /// Processes an incoming [DataFrame] which is addressed to a specific stream. diff --git a/pkgs/http2/lib/src/flowcontrol/stream_queues.dart b/pkgs/http2/lib/src/flowcontrol/stream_queues.dart index b6d7d686aa..0e44bde0e4 100644 --- a/pkgs/http2/lib/src/flowcontrol/stream_queues.dart +++ b/pkgs/http2/lib/src/flowcontrol/stream_queues.dart @@ -209,6 +209,16 @@ class StreamMessageQueueIn extends Object }; } + @override + void cancel() { + super.cancel(); + if (!wasClosed && !wasTerminated) { + _tryDispatch(); + _tryUpdateBufferIndicator(); + onCloseCheck(); + } + } + /// Debugging data: the number of pending messages in this queue. int get pendingMessages => _pendingMessages.length; @@ -337,9 +347,11 @@ class StreamMessageQueueIn extends Object } void _tryUpdateBufferIndicator() { - if (_incomingMessagesC.isPaused || _pendingMessages.isNotEmpty) { + if (!wasCancelled && + (_incomingMessagesC.isPaused || _pendingMessages.isNotEmpty)) { bufferIndicator.markBuffered(); - } else if (bufferIndicator.wouldBuffer && !_incomingMessagesC.isPaused) { + } else if (bufferIndicator.wouldBuffer && + (wasCancelled || !_incomingMessagesC.isPaused)) { bufferIndicator.markUnBuffered(); } } diff --git a/pkgs/http2/lib/src/streams/stream_handler.dart b/pkgs/http2/lib/src/streams/stream_handler.dart index 7763a32718..513c87da87 100644 --- a/pkgs/http2/lib/src/streams/stream_handler.dart +++ b/pkgs/http2/lib/src/streams/stream_handler.dart @@ -543,6 +543,10 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { stream.state == StreamState.ReservedRemote) { _frameWriter.writeRstStreamFrame(stream.id, ErrorCode.CANCEL); _closeStreamAbnormally(stream, null, propagateException: false); + } else if (stream.state == StreamState.Closed && + !stream.incomingQueue.wasClosed && + !stream.incomingQueue.wasTerminated) { + _closeStreamAbnormally(stream, null, propagateException: false); } } @@ -987,7 +991,10 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin { Object? exception, { bool propagateException = false, }) { - incomingQueue.removeStreamMessageQueue(stream.id); + incomingQueue.removeStreamMessageQueue( + stream.id, + replenishWindow: !wasTerminated, + ); if (stream.state != StreamState.Terminated) { _changeState(stream, StreamState.Terminated); diff --git a/pkgs/http2/test/flow_control_test.dart b/pkgs/http2/test/flow_control_test.dart index 8caf4f49df..be65acc25b 100644 --- a/pkgs/http2/test/flow_control_test.dart +++ b/pkgs/http2/test/flow_control_test.dart @@ -401,4 +401,81 @@ void main() { expect(goaways, isEmpty); await server.terminate(); }); + + test('terminating a stream with unconsumed buffered DATA replenishes the ' + 'connection receive window', () async { + await connect(const ClientSettings()); + peer + ..sendSettings() + ..ackSettings(); + final stream = client.makeRequest(_get('/aborted'), endStream: true); + final sub = stream.incomingMessages.listen((_) {}, onError: (_) {}); + sub.pause(); + await peer.until(() => peer.frames.any((f) => f.type == _headers)); + + // Send 49152 bytes of response DATA while the stream is paused. + peer + ..sendOkHeaders(1) + ..send(_data, 0, 1, Uint8List(16384)) + ..send(_data, 0, 1, Uint8List(16384)) + ..send(_data, 0, 1, Uint8List(16384)); + // The client has buffered the DATA once it has answered a PING sent after. + await peer.barrier(); + + // Terminate the stream before the listener ever consumes the buffered DATA. + stream.terminate(); + await peer.until( + () => peer.frames.any((f) => f.type == _windowUpdate && f.stream == 0), + ); + + final connUpdates = peer.frames.where( + (f) => f.type == _windowUpdate && f.stream == 0, + ); + final replenished = connUpdates.fold( + 0, + (sum, f) => sum + _increment(f), + ); + expect(replenished, 49152); + await sub.cancel(); + }); + + test('cancelling incomingMessages on a paused stream replenishes the ' + 'connection window and releases the active stream slot', () async { + await connect(const ClientSettings()); + // Peer advertises maxConcurrentStreams = 1. + peer + ..send(_settings, 0, 0, const [0x00, 0x03, 0x00, 0x00, 0x00, 0x01]) + ..ackSettings(); + await client.onInitialPeerSettingsReceived; + + final stream = client.makeRequest(_get('/cancel'), endStream: true); + final id = stream.id; + final sub = stream.incomingMessages.listen((_) {}, onError: (_) {}); + sub.pause(); + await peer.until( + () => peer.frames.any((f) => f.type == _headers && f.stream == id), + ); + + final updatesBefore = + peer.frames + .where((f) => f.type == _windowUpdate && f.stream == 0) + .length; + peer + ..sendOkHeaders(id) + ..send(_data, _flagEndStream, id, Uint8List(16384)); + await peer.barrier(); + + await sub.cancel(); + await peer.until( + () => + peer.frames + .where((f) => f.type == _windowUpdate && f.stream == 0) + .length > + updatesBefore, + ); + // Whatever the cancel set off has run once the client answers a PING sent + // after the WINDOW_UPDATE was seen. + await peer.barrier(); + expect(client.isOpen, isTrue); + }); }