diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index 82432e860f..645be68588 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md @@ -1,3 +1,8 @@ +## 3.1.1-wip + +- Fix responses stalling when `sendData` is called with `endStream: true` and + the response body exceeds the peer's stream flow-control window. + ## 3.1.0 - Gracefully handle receiving headers on a stream that the client has canceled. (#1799) diff --git a/pkgs/http2/lib/src/flowcontrol/stream_queues.dart b/pkgs/http2/lib/src/flowcontrol/stream_queues.dart index b6d7d686aa..326289a816 100644 --- a/pkgs/http2/lib/src/flowcontrol/stream_queues.dart +++ b/pkgs/http2/lib/src/flowcontrol/stream_queues.dart @@ -71,13 +71,14 @@ class StreamMessageQueueOut extends Object void enqueueMessage(Message message) { if (message is! ResetStreamMessage) ensureNotClosingSync(() {}); if (!wasTerminated) { - if (message.endStream) startClosing(); - if (message is DataMessage) { toBeWrittenBytes += message.bytes.length; } _messages.addLast(message); + // Closing checks the queue immediately, so enqueue the final message + // before checking whether all pending data has been sent. + if (message.endStream) startClosing(); _trySendData(); if (_messages.isNotEmpty) { diff --git a/pkgs/http2/pubspec.yaml b/pkgs/http2/pubspec.yaml index 489afea61a..7bfe851518 100644 --- a/pkgs/http2/pubspec.yaml +++ b/pkgs/http2/pubspec.yaml @@ -1,5 +1,5 @@ name: http2 -version: 3.1.0 +version: 3.1.1-wip description: A HTTP/2 implementation in Dart. repository: https://github.com/dart-lang/http/tree/master/pkgs/http2 diff --git a/pkgs/http2/test/flow_control_end_stream_test.dart b/pkgs/http2/test/flow_control_end_stream_test.dart new file mode 100644 index 0000000000..7d378aff83 --- /dev/null +++ b/pkgs/http2/test/flow_control_end_stream_test.dart @@ -0,0 +1,87 @@ +// Copyright (c) 2026, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import 'dart:async'; +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:http2/transport.dart'; +import 'package:test/test.dart'; + +void main() { + late ClientTransportConnection client; + late ServerTransportConnection server; + late StreamIterator requests; + + setUp(() async { + final listener = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + addTearDown(listener.close); + final accepted = Completer(); + listener.listen(accepted.complete); + final clientSocket = await Socket.connect( + InternetAddress.loopbackIPv4, + listener.port, + ); + addTearDown(clientSocket.destroy); + final serverSocket = await accepted.future; + addTearDown(serverSocket.destroy); + client = ClientTransportConnection.viaSocket(clientSocket); + server = ServerTransportConnection.viaSocket(serverSocket); + addTearDown(() async { + await [client.terminate(), server.terminate()].wait; + }); + requests = StreamIterator(server.incomingStreams); + addTearDown(requests.cancel); + }); + + Future checkResponse(int length, {bool closeSink = false}) async { + final request = client.makeRequest([ + Header.ascii(':method', 'GET'), + Header.ascii(':path', '/'), + Header.ascii(':scheme', 'http'), + Header.ascii(':authority', 'localhost'), + ], endStream: true); + final body = Uint8List.fromList(List.generate(length, (i) => i % 251)); + final received = BytesBuilder(); + var endStreamMessages = 0; + final responseDone = request.incomingMessages.forEach((message) { + if (message is DataStreamMessage) received.add(message.bytes); + if (message.endStream) endStreamMessages++; + }); + expect(await requests.moveNext(), isTrue); + final response = requests.current; + await response.incomingMessages.drain(); + response.sendHeaders([Header.ascii(':status', '200')]); + if (closeSink) { + await response.outgoingMessages.addStream( + Stream.value(DataStreamMessage(body)), + ); + unawaited(response.outgoingMessages.close()); + } else { + response.sendData(body, endStream: true); + } + await responseDone.timeout( + const Duration(seconds: 5), + onTimeout: () => fail('Response stalled after ${received.length} bytes'), + ); + expect(received.takeBytes(), body); + // Empty DATA frames close the incoming stream without a stream message. + expect(endStreamMessages, closeSink || length == 0 ? 0 : 1); + await client.ping(); + await server.ping(); + } + + for (final length in [0, 1, 65535, 65536, 200000]) { + test('sendData with endStream delivers $length bytes over TCP', () async { + await checkResponse(length); + }); + } + + test( + 'closing the outgoing sink delivers a large response over TCP', + () async { + await checkResponse(200000, closeSink: true); + }, + ); +} diff --git a/pkgs/http2/test/src/flowcontrol/stream_queues_test.dart b/pkgs/http2/test/src/flowcontrol/stream_queues_test.dart index cbeb147a3a..82a077d842 100644 --- a/pkgs/http2/test/src/flowcontrol/stream_queues_test.dart +++ b/pkgs/http2/test/src/flowcontrol/stream_queues_test.dart @@ -5,6 +5,9 @@ import 'package:http2/src/async_utils/async_utils.dart'; import 'package:http2/src/flowcontrol/queue_messages.dart'; import 'package:http2/src/flowcontrol/stream_queues.dart'; +import 'package:http2/src/flowcontrol/window.dart'; +import 'package:http2/src/flowcontrol/window_handler.dart'; +import 'package:http2/src/frames/frames.dart'; import 'package:http2/transport.dart'; import 'package:mockito/mockito.dart'; import 'package:test/test.dart'; @@ -17,6 +20,145 @@ void main() { const BYTES = [1, 2, 3]; group('stream-message-queue-out', () { + for (final initialWindow in [0, 1]) { + test('end-stream waits for updates from $initialWindow', () async { + final sent = []; + final connectionQueue = MockConnectionMessageQueueOut(); + when(connectionQueue.enqueueMessage(any)).thenAnswer((invocation) { + sent.add(invocation.positionalArguments.single as DataMessage); + }); + final window = OutgoingStreamWindowHandler( + Window(initialSize: initialWindow), + ); + final queue = StreamMessageQueueOut( + STREAM_ID, + window, + connectionQueue, + ); + queue.bufferIndicator.bufferEmptyEvents.listen( + expectAsync1((_) { + // A synchronous drain callback must not reopen the closing queue. + expect(queue.isClosing, isTrue); + expect( + () => + queue.enqueueMessage(DataMessage(STREAM_ID, BYTES, false)), + throwsStateError, + ); + }), + ); + + queue.enqueueMessage(DataMessage(STREAM_ID, BYTES, true)); + expect(queue.isClosing, isTrue); + expect(queue.wasClosed, isFalse); + expect(queue.pendingMessages, 1); + expect(queue.writtenBytes, initialWindow); + expect(queue.bufferIndicator.wouldBuffer, isTrue); + expect( + () => queue.enqueueMessage(DataMessage(STREAM_ID, BYTES, false)), + throwsStateError, + ); + + for (var written = initialWindow; written < BYTES.length; written++) { + expect(queue.wasClosed, isFalse); + window.processWindowUpdate( + WindowUpdateFrame( + FrameHeader(4, FrameType.WINDOW_UPDATE, 0, STREAM_ID), + 1, + ), + ); + } + await queue.done; + expect(queue.wasClosed, isTrue); + expect(queue.pendingMessages, 0); + expect(queue.bufferIndicator.wouldBuffer, isFalse); + expect(queue.writtenBytes, BYTES.length); + expect(sent.expand((message) => message.bytes), BYTES); + expect(sent.take(sent.length - 1).every((m) => !m.endStream), isTrue); + expect(sent.last.endStream, isTrue); + }); + } + + test('end-stream headers close after reaching the connection queue', () { + final connectionQueue = MockConnectionMessageQueueOut(); + final window = OutgoingStreamWindowHandler(Window(initialSize: 0)); + final queue = StreamMessageQueueOut(STREAM_ID, window, connectionQueue); + when(connectionQueue.enqueueMessage(any)).thenAnswer((_) { + expect(queue.wasClosed, isFalse); + }); + + queue.enqueueMessage(HeadersMessage(STREAM_ID, [], true)); + expect(queue.wasClosed, isTrue); + expect(queue.pendingMessages, 0); + }); + + test('reset remains allowed after end-stream', () { + final connectionQueue = MockConnectionMessageQueueOut(); + when(connectionQueue.enqueueMessage(any)).thenReturn(null); + final window = OutgoingStreamWindowHandler(Window(initialSize: 0)); + final queue = StreamMessageQueueOut(STREAM_ID, window, connectionQueue); + queue.enqueueMessage(HeadersMessage(STREAM_ID, [], true)); + expect(queue.wasClosed, isTrue); + final reset = ResetStreamMessage(STREAM_ID, ErrorCode.CANCEL); + queue.enqueueMessage(reset); + verify(connectionQueue.enqueueMessage(reset)).called(1); + expect(queue.pendingMessages, 0); + }); + + test('empty end-stream data closes with an empty window', () async { + final connectionQueue = MockConnectionMessageQueueOut(); + when(connectionQueue.enqueueMessage(any)).thenReturn(null); + final window = OutgoingStreamWindowHandler(Window(initialSize: 0)); + final queue = StreamMessageQueueOut(STREAM_ID, window, connectionQueue); + + final subscription = queue.bufferIndicator.bufferEmptyEvents.listen( + expectAsync1((_) { + expect(queue.isClosing, isTrue); + expect( + () => queue.enqueueMessage(DataMessage(STREAM_ID, BYTES, false)), + throwsStateError, + ); + }), + ); + addTearDown(subscription.cancel); + + queue.enqueueMessage(DataMessage(STREAM_ID, [], true)); + await queue.done; + expect(queue.wasClosed, isTrue); + expect(queue.pendingMessages, 0); + final message = + verify(connectionQueue.enqueueMessage(captureAny)).captured.single + as DataMessage; + expect(message.bytes, isEmpty); + expect(message.endStream, isTrue); + }); + + for (final error in [null, StateError('transport failed')]) { + test('termination discards pending end-stream data: $error', () async { + final connectionQueue = MockConnectionMessageQueueOut(); + final window = OutgoingStreamWindowHandler(Window(initialSize: 0)); + final queue = StreamMessageQueueOut( + STREAM_ID, + window, + connectionQueue, + ); + + queue.enqueueMessage(DataMessage(STREAM_ID, BYTES, true)); + expect(queue.wasClosed, isFalse); + queue.terminate(error); + await queue.done; + expect(queue.wasTerminated, isTrue); + expect(queue.wasClosed, isTrue); + expect(queue.pendingMessages, 0); + window.processWindowUpdate( + WindowUpdateFrame( + FrameHeader(4, FrameType.WINDOW_UPDATE, 0, STREAM_ID), + BYTES.length, + ), + ); + verifyZeroInteractions(connectionQueue); + }); + } + test('window-big-enough', () { var connectionQueueMock = MockConnectionMessageQueueOut(); when(connectionQueueMock.enqueueMessage(any)).thenReturn(null);