fix(http2): do not head-of-line block HEADERS, RST_STREAM and GOAWAY behind window-exhausted DATA (#2013) > [!NOTE] > This PR was generated by an AI coding agent (Jetski) on behalf of @mosuem. `ConnectionMessageQueueOut` is a strict FIFO. Once the connection-level send window is exhausted, a `DataMessage` at the head of the queue blocks *everything* behind it, including frames that are not subject to flow control (RFC 9113 Section 6.9: *"Flow control only applies to frames that are identified as being subject to flow control. Of the frame types defined in this document, this includes only DATA frames."*). In practice: while one stream was waiting for `WINDOW_UPDATE`, no new request could be opened (`HEADERS`), no stream could be cancelled (`RST_STREAM`), and the connection could not be shut down gracefully (`GOAWAY`) until the peer granted credit. - **`lib/src/flowcontrol/connection_queues.dart`**: - `_peekSendableMessage()` replaces the "first message or nothing" rule. When the window is exhausted and the head of the queue is a `DataMessage`, it returns the first `GOAWAY`, or the first `HEADERS` / `RST_STREAM` whose stream has no earlier queued message. Skipping a message of a stream blocks every later message of that stream (and, for a `PUSH_PROMISE`, of the promised stream), so per-stream frame order is preserved: trailers never overtake their own `DATA`. - The scan is linear in the queue length, and `StreamMessageQueueOut` keeps forwarding `DATA` into this queue while the connection window is exhausted (it only checks the stream window). To avoid rescanning on every `sendData()` call, `enqueueMessage` skips `_trySendMessages()` when a `DataMessage` is enqueued while `_connectionWindow.positiveWindow.wouldBuffer` is `true` (since the new `DataMessage` cannot be sent and cannot unblock any earlier queued message). Enqueueing a non-DATA message while blocked still costs one pass over the queue. Added to `test/client_test.dart`: - `headers-and-rst-not-blocked-or-reordered-on-exhausted-window`: stream 1 exhausts the connection window; stream 3's `HEADERS` and stream 5's `HEADERS` + `RST_STREAM` must still go out, in that order, before the window is replenished. - `trailers-stay-behind-blocked-data-while-other-headers-bypass`: stream 1 has blocked `DATA` *and* queued trailers; stream 3's `HEADERS` bypass them, stream 1's trailers are only written after its `DATA`. **Before fix:** ```text 00:05 +0 -1: client-tests client-errors headers-and-rst-not-blocked-or-reordered-on-exhausted-window [E] TimeoutException after 0:00:05.000000: Test timed out after 5 seconds. 00:10 +0 -2: client-tests client-errors trailers-stay-behind-blocked-data-while-other-headers-bypass [E] TimeoutException after 0:00:05.000000: Test timed out after 5 seconds. ``` (stream 3's `HEADERS` never leave the client while stream 1 waits for credit) **After fix:** ```text 00:00 +2: All tests passed! ```
diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index a5ca6db..fc2d454 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md
@@ -25,6 +25,8 @@ stream, so a slow reader stops the server at the flow-control window instead of buffering the whole response in memory, and uses a 4 MiB stream / 16 MiB connection receive window instead of the protocol's 64 KiB defaults. +- Stop head-of-line blocking `HEADERS`, `RST_STREAM`, and `GOAWAY` behind + `DATA` that is waiting for connection-level flow-control credit. - Bound peer-initiated concurrent streams and reject peers that ignore the advertised `SETTINGS_MAX_CONCURRENT_STREAMS` value. Reserved peer streams are also bounded locally. The default limit is 100.
diff --git a/pkgs/http2/lib/src/flowcontrol/connection_queues.dart b/pkgs/http2/lib/src/flowcontrol/connection_queues.dart index 77bb52c..60158ee 100644 --- a/pkgs/http2/lib/src/flowcontrol/connection_queues.dart +++ b/pkgs/http2/lib/src/flowcontrol/connection_queues.dart
@@ -58,6 +58,14 @@ ensureNotClosingSync(() { if (!wasTerminated) { _messages.addLast(message); + // When the connection send window is exhausted, a newly queued + // DataMessage cannot be sent and cannot unblock any earlier message. + // Skipping _trySendMessages() avoids rescanning _messages on every + // sendData() call (which would otherwise be quadratic). + if (message is DataMessage && + _connectionWindow.positiveWindow.wouldBuffer) { + return; + } _trySendMessages(); } }); @@ -90,27 +98,60 @@ } } + /// The next message which can be written now, or `null` if none can. + /// + /// Normally that is the first queued message. When the connection-level send + /// window is exhausted and the first message is a [DataMessage], frames which + /// are not subject to flow control (RFC 9113 section 6.9: only DATA frames + /// are) may still be written: `GOAWAY`, and `HEADERS` / `RST_STREAM` of + /// streams that have no earlier queued message. Skipping a message of a + /// stream blocks all later messages of that stream, so the per-stream frame + /// order is preserved. + Message? _peekSendableMessage() { + if (_messages.isEmpty || _frameWriter.bufferIndicator.wouldBuffer) { + return null; + } + final first = _messages.first; + if (first is! DataMessage || + !_connectionWindow.positiveWindow.wouldBuffer) { + return first; + } + + final blockedStreams = <int>{}; + for (final message in _messages) { + if (message is GoawayMessage) return message; + if (message is DataMessage || blockedStreams.contains(message.streamId)) { + blockedStreams.add(message.streamId); + } else if (message is PushPromiseMessage) { + // A PUSH_PROMISE must precede any frame of the promised stream and + // the END_STREAM of its associated stream (RFC 9113 section 8.4.1); + // keep it, and everything after it on both streams, in order. + blockedStreams + ..add(message.streamId) + ..add(message.promisedStreamId); + } else { + return message; + } + } + return null; + } + void _trySendMessages() { if (!wasTerminated) { // We can make progress if // * there is at least one message to send // * the underlying frame writer / sink / socket doesn't block // * either one - // * the next message is a non-flow control message (e.g. headers) + // * the next sendable message is a non-flow control message // * the connection window is positive - if (_messages.isNotEmpty && - !_frameWriter.bufferIndicator.wouldBuffer && - (!_connectionWindow.positiveWindow.wouldBuffer || - _messages.first is! DataMessage)) { - _trySendMessage(); + final message = _peekSendableMessage(); + if (message != null) { + _trySendMessage(message); // If we have more messages and we can send them, we'll run them // using `Timer.run()` to let other things get in-between. - if (_messages.isNotEmpty && - !_frameWriter.bufferIndicator.wouldBuffer && - (!_connectionWindow.positiveWindow.wouldBuffer || - _messages.first is! DataMessage)) { + if (_peekSendableMessage() != null) { // TODO: If all the frame writer methods would return the // number of bytes written, we could just say, we loop here until 10kb // and after words, we'll make `Timer.run()`. @@ -122,25 +163,28 @@ } } - void _trySendMessage() { - var message = _messages.first; - if (message is HeadersMessage) { + void _trySendMessage(Message message) { + // Fast path for the common case where [message] is at the head of the + // queue (`removeFirst()` is O(1)); only fall back to the O(n) `remove()` + // when `_peekSendableMessage()` skipped past blocked [DataMessage]s. + if (identical(message, _messages.first)) { _messages.removeFirst(); + } else { + _messages.remove(message); + } + if (message is HeadersMessage) { _frameWriter.writeHeadersFrame( message.streamId, message.headers, endStream: message.endStream, ); } else if (message is PushPromiseMessage) { - _messages.removeFirst(); _frameWriter.writePushPromiseFrame( message.streamId, message.promisedStreamId, message.headers, ); } else if (message is DataMessage) { - _messages.removeFirst(); - if (_connectionWindow.peerWindowSize >= message.bytes.length) { _connectionWindow.decreaseWindow(message.bytes.length); _frameWriter.writeDataFrame( @@ -170,10 +214,8 @@ _messages.addFirst(tailMessage); } } else if (message is ResetStreamMessage) { - _messages.removeFirst(); _frameWriter.writeRstStreamFrame(message.streamId, message.errorCode); } else if (message is GoawayMessage) { - _messages.removeFirst(); _frameWriter.writeGoawayFrame( message.lastStreamId, message.errorCode,
diff --git a/pkgs/http2/test/client_test.dart b/pkgs/http2/test/client_test.dart index 19e208d..662476f 100644 --- a/pkgs/http2/test/client_test.dart +++ b/pkgs/http2/test/client_test.dart
@@ -1511,6 +1511,187 @@ await Future.wait([serverFun(), clientFun()], eagerError: true); }); + + clientTest( + 'headers-and-rst-not-blocked-or-reordered-on-exhausted-window', + ( + ClientTransportConnection client, + FrameWriter serverWriter, + StreamIterator<Frame> serverReader, + Future<Frame> Function() nextFrame, + ) async { + final handshakeDone = Completer<void>(); + final windowExhausted = Completer<void>(); + + Future<void> serverFun() async { + serverWriter.writeSettingsFrame([ + Setting(Setting.SETTINGS_INITIAL_WINDOW_SIZE, 200000), + ]); + expect(await nextFrame(), isA<SettingsFrame>()); + serverWriter.writeSettingsAckFrame(); + expect(await nextFrame(), isA<SettingsFrame>()); + handshakeDone.complete(); + + final req1 = await nextFrame() as HeadersFrame; + expect(req1.header.streamId, 1); + + var s1Bytes = 0; + while (s1Bytes < 65535) { + final data = await nextFrame() as DataFrame; + expect(data.header.streamId, 1); + expect(data.hasEndStreamFlag, isFalse); + s1Bytes += data.bytes.length; + } + expect(s1Bytes, 65535); + windowExhausted.complete(); + + // Even though the connection send window is exhausted by stream 1, + // stream 3's HEADERS and stream 5's HEADERS + RST_STREAM must not + // be head-of-line blocked, and stream 5's HEADERS must precede its + // RST_STREAM. + expect( + await nextFrame(), + isA<HeadersFrame>() + .having((f) => f.header.streamId, 'streamId', 3) + .having((f) => f.hasEndStreamFlag, 'endStream', isTrue), + ); + expect( + await nextFrame(), + isA<HeadersFrame>().having( + (f) => f.header.streamId, + 'streamId', + 5, + ), + ); + expect( + await nextFrame(), + isA<RstStreamFrame>() + .having((f) => f.header.streamId, 'streamId', 5) + .having((f) => f.errorCode, 'errorCode', ErrorCode.CANCEL), + ); + + serverWriter.writeHeadersFrame(3, [ + Header.ascii(':status', '200'), + ], endStream: true); + + // Grant connection window credit so stream 1's tail DATA flushes. + serverWriter.writeWindowUpdate(10000, streamId: 0); + final tail = await nextFrame() as DataFrame; + expect(tail.header.streamId, 1); + expect(tail.bytes.length, 70000 - 65535); + expect(tail.hasEndStreamFlag, isTrue); + + serverWriter.writeHeadersFrame(1, [ + Header.ascii(':status', '200'), + ], endStream: true); + + expect(await nextFrame(), isA<GoawayFrame>()); + expect(await serverReader.moveNext(), isFalse); + } + + Future<void> clientFun() async { + await handshakeDone.future; + final s1 = client.makeRequest([Header.ascii(':path', '/s1')]); + s1.sendData(List<int>.filled(70000, 1), endStream: true); + + await windowExhausted.future; + + final s2 = client.makeRequest([ + Header.ascii(':path', '/s2'), + ], endStream: true); + final s3 = client.makeRequest([Header.ascii(':path', '/s3')]); + s3.terminate(); + + await s2.incomingMessages.drain<void>(); + await s1.incomingMessages.drain<void>(); + await client.finish(); + } + + await Future.wait([serverFun(), clientFun()], eagerError: true); + }, + ); + + clientTest( + 'trailers-stay-behind-blocked-data-while-other-headers-bypass', + ( + ClientTransportConnection client, + FrameWriter serverWriter, + StreamIterator<Frame> serverReader, + Future<Frame> Function() nextFrame, + ) async { + final handshakeDone = Completer<void>(); + final windowExhausted = Completer<void>(); + + Future<void> serverFun() async { + serverWriter.writeSettingsFrame([ + Setting(Setting.SETTINGS_INITIAL_WINDOW_SIZE, 200000), + ]); + expect(await nextFrame(), isA<SettingsFrame>()); + serverWriter.writeSettingsAckFrame(); + expect(await nextFrame(), isA<SettingsFrame>()); + handshakeDone.complete(); + + final req1 = await nextFrame() as HeadersFrame; + expect(req1.header.streamId, 1); + var s1Bytes = 0; + while (s1Bytes < 65535) { + final data = await nextFrame() as DataFrame; + expect(data.header.streamId, 1); + s1Bytes += data.bytes.length; + } + expect(s1Bytes, 65535); + windowExhausted.complete(); + + // Stream 1 still has DATA and its trailers queued. Stream 3's + // HEADERS may pass them, stream 1's trailers may not. + expect( + await nextFrame(), + isA<HeadersFrame>().having( + (f) => f.header.streamId, + 'streamId', + 3, + ), + ); + serverWriter.writeWindowUpdate(10000, streamId: 0); + final tail = await nextFrame() as DataFrame; + expect(tail.header.streamId, 1); + expect(tail.bytes.length, 70000 - 65535); + expect(tail.hasEndStreamFlag, isFalse); + expect( + await nextFrame(), + isA<HeadersFrame>() + .having((f) => f.header.streamId, 'streamId', 1) + .having((f) => f.hasEndStreamFlag, 'endStream', isTrue), + ); + + for (final streamId in [1, 3]) { + serverWriter.writeHeadersFrame(streamId, [ + Header.ascii(':status', '200'), + ], endStream: true); + } + expect(await nextFrame(), isA<GoawayFrame>()); + expect(await serverReader.moveNext(), isFalse); + } + + Future<void> clientFun() async { + await handshakeDone.future; + final s1 = client.makeRequest([Header.ascii(':path', '/s1')]); + s1.sendData(List<int>.filled(70000, 1)); + s1.sendHeaders([Header.ascii('x-trailer', '1')], endStream: true); + + await windowExhausted.future; + final s2 = client.makeRequest([ + Header.ascii(':path', '/s2'), + ], endStream: true); + + await s2.incomingMessages.drain<void>(); + await s1.incomingMessages.drain<void>(); + await client.finish(); + } + + await Future.wait([serverFun(), clientFun()], eagerError: true); + }, + ); }); }); }