diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index 4620a000ad..21c8b3d488 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/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 diff --git a/pkgs/http2/lib/src/flowcontrol/stream_queues.dart b/pkgs/http2/lib/src/flowcontrol/stream_queues.dart index 0e44bde0e4..7f8fe4fbe2 100644 --- a/pkgs/http2/lib/src/flowcontrol/stream_queues.dart +++ b/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) { diff --git a/pkgs/http2/lib/src/streams/stream_handler.dart b/pkgs/http2/lib/src/streams/stream_handler.dart index 80e9f36615..0269025efd 100644 --- a/pkgs/http2/lib/src/streams/stream_handler.dart +++ b/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 ' diff --git a/pkgs/http2/test/transport_test.dart b/pkgs/http2/test/transport_test.dart index f4477156fc..939bd8480e 100644 --- a/pkgs/http2/test/transport_test.dart +++ b/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(); + 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(); + await uploadReceived.future; + + await client.finish(); + await server.finish(); + }, serverSettings: const ServerSettings(concurrentStreamLimit: 1)); }); }); }