fix(http2): replenish connection receive window when stream closes with unconsumed data

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
 

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.

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();
     }
   }

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

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<int>(
+      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);
+  });
 }