fix(http2): keep flow-controlled end-stream DATA alive until it is flushed (#1998)

pkgs/http2/CHANGELOG.md

@@ -14,6 +14,8 @@
 - Ignore `PRIORITY` and unknown extension frames on open, idle, and closed
   streams, and treat `SETTINGS`, `PING`, and `GOAWAY` frames with a non-zero
   stream ID as a connection-level `PROTOCOL_ERROR`.
+- Fix `sendData(..., endStream: true)` stalling forever when the data exceeds
+  the peer's flow-control window (#1998).
 
 ## 3.1.0
 

pkgs/http2/lib/src/flowcontrol/stream_queues.dart

@@ -71,13 +71,12 @@ 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);
+      if (message.endStream) startClosing();
       _trySendData();
 
       if (_messages.isNotEmpty) {

pkgs/http2/lib/src/streams/stream_handler.dart

@@ -891,12 +891,13 @@ class StreamHandler extends Object with TerminatableMixin, ClosableMixin {
       _changeState(stream, StreamState.HalfClosedRemote);
     } else if (stream.state == StreamState.HalfClosedLocal) {
       _changeState(stream, StreamState.Closed);
-      // TODO: We have to make sure that we
-      //   - remove the stream for data structures which only care about the
-      //     state
-      //   - keep the stream in data structures which need to be emptied
-      //     (e.g. MessageQueues which are not empty yet).
-      _openStreams.remove(stream.id);
+      // Only remove the stream from [_openStreams] once all outgoing messages
+      // have been sent; if outgoing DATA is still waiting on flow-control
+      // credit, keep the stream registered so incoming WINDOW_UPDATE frames
+      // reach [stream.windowHandler].
+      if (stream.outgoingQueue.wasClosed) {
+        _openStreams.remove(stream.id);
+      }
     } else {
       throw StateError(
         'Got an end-of-stream from the remote end, but this stream is '

pkgs/http2/test/transport_test.dart

@@ -627,6 +627,58 @@ void main() {
       ) async {
         await testWindowSize(client, server, 8096);
       }, clientSettings: const ClientSettings(streamWindowSize: 8096));
+
+      transportTest('send-data-exceeding-window-with-end-stream-completes', (
+        ClientTransportConnection client,
+        ServerTransportConnection server,
+      ) async {
+        final uploadReceived = Completer<void>();
+        server.incomingStreams.listen((ServerTransportStream s) async {
+          var requestBytes = 0;
+          String? path;
+          await for (final m in s.incomingMessages) {
+            if (m is HeadersStreamMessage && path == null) {
+              path = String.fromCharCodes(m.headers.first.value);
+              if (path == '/download') {
+                s.sendHeaders([Header.ascii(':status', '200')]);
+                s.sendData(Uint8List(100000), endStream: true);
+              } else if (path == '/early-response-upload') {
+                s.sendHeaders([
+                  Header.ascii(':status', '200'),
+                ], endStream: true);
+              }
+            } else if (m is DataStreamMessage) {
+              requestBytes += m.bytes.length;
+            }
+          }
+          if (path == '/early-response-upload') {
+            expect(requestBytes, 100000);
+            uploadReceived.complete();
+          }
+        });
+
+        // 1. Server sendData(> 65535, endStream: true)
+        final download = client.makeRequest([
+          Header.ascii(':path', '/download'),
+        ], endStream: true);
+        var downloadBytes = 0;
+        await for (final m in download.incomingMessages) {
+          if (m is DataStreamMessage) downloadBytes += m.bytes.length;
+        }
+        expect(downloadBytes, 100000);
+
+        // 2. Client sendData(> 65535, endStream: true) when server sends
+        // END_STREAM before the upload finishes draining.
+        final upload = client.makeRequest([
+          Header.ascii(':path', '/early-response-upload'),
+        ]);
+        upload.sendData(Uint8List(100000), endStream: true);
+        await upload.incomingMessages.drain<void>();
+        await uploadReceived.future;
+
+        await client.finish();
+        await server.finish();
+      }, serverSettings: const ServerSettings(concurrentStreamLimit: 1));
     });
   });
 }