fix(http2): prevent end-stream data from stalling under flow control

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)

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) {

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
 

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<ServerTransportStream> requests;
+
+  setUp(() async {
+    final listener = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
+    addTearDown(listener.close);
+    final accepted = Completer<Socket>();
+    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<void> 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<void>();
+    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);
+    },
+  );
+}

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 = <DataMessage>[];
+          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);