From 0b1021f0cb134f2560f90bcd68c583c5cc6991c2 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 30 Jul 2026 08:57:36 +0100 Subject: [PATCH 01/27] feat(http2): add generic ClientPool with unit tests --- pkgs/http2/lib/src/client_pool.dart | 136 +++++++++++++++++++ pkgs/http2/test/client_pool_test.dart | 182 ++++++++++++++++++++++++++ 2 files changed, 318 insertions(+) create mode 100644 pkgs/http2/lib/src/client_pool.dart create mode 100644 pkgs/http2/test/client_pool_test.dart diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart new file mode 100644 index 0000000000..d6e857a5f2 --- /dev/null +++ b/pkgs/http2/lib/src/client_pool.dart @@ -0,0 +1,136 @@ +// 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'; + +class _PooledResource { + _PooledResource(this.future); + final Future future; + int inFlight = 0; + bool failed = false; +} + +/// A pool of resources of type [T]. +/// +/// Packs load onto the most-full resource under [maxConcurrentOperations] +/// (rather than spreading evenly across resources), opens a new resource +/// once existing ones are full, stops routing new work to a resource once an +/// operation on it throws, and garbage-collects idle resources past +/// [maxIdleResources]. +class ClientPool { + ClientPool( + Future Function() create, { + required this.maxConcurrentOperations, + required Future Function(T resource) destroy, + this.maxIdleResources = 1, + }) : _create = create, + _destroy = destroy; + + final Future Function() _create; + final Future Function(T resource) _destroy; + final int maxConcurrentOperations; + final int maxIdleResources; + + final _resources = <_PooledResource>[]; + var _terminated = false; + Completer? _drained; + + /// The number of resources currently in the pool. For testing. + int get size => _resources.length; + + /// The number of in-flight operations across every resource. For testing. + int get opCount => + _resources.fold(0, (total, resource) => total + resource.inFlight); + + /// Runs [operation] on an available (or newly created) resource. + Future run(Future Function(T resource) operation) async { + if (_terminated) { + throw StateError('This pool has already been terminated.'); + } + + final pooled = _acquire(); + pooled.inFlight++; + try { + return await operation(await pooled.future); + } catch (_) { + pooled.failed = true; + rethrow; + } finally { + pooled.inFlight--; + if (_terminated) { + _maybeCompleteDrain(); + } else { + await _collectIfIdle(pooled); + } + } + } + + // Synchronous (no `await`), so concurrent calls can't race each other + // into both creating a resource before either sees the other's. + _PooledResource _acquire() { + _PooledResource? selected; + for (final resource in _resources) { + if (resource.failed) continue; + if (resource.inFlight < maxConcurrentOperations && + (selected == null || resource.inFlight > selected.inFlight)) { + selected = resource; + } + } + if (selected != null) return selected; + + final resource = _PooledResource(_create()); + _resources.add(resource); + return resource; + } + + Future _collectIfIdle(_PooledResource resource) async { + if (resource.inFlight > 0) return; + if (!resource.failed && !_hasExcessIdleCapacity) return; + + _resources.remove(resource); + try { + await _destroy(await resource.future); + } catch (_) { + // Best-effort: a failure here must not shadow the caller's own + // request error, since this runs inside run()'s finally block. + } + } + + bool get _hasExcessIdleCapacity { + final idleCapacity = _resources.fold( + 0, + (total, resource) => + total + (maxConcurrentOperations - resource.inFlight), + ); + return idleCapacity > maxIdleResources * maxConcurrentOperations; + } + + void _maybeCompleteDrain() { + final drained = _drained; + if (drained != null && !drained.isCompleted && opCount == 0) { + drained.complete(); + } + } + + /// Waits for in-flight operations to finish, then destroys every + /// resource in the pool. No further operations can run afterward. + Future terminate() async { + _terminated = true; + + if (opCount > 0) { + _drained = Completer(); + await _drained!.future; + } + + for (final resource in _resources) { + try { + await _destroy(await resource.future); + } catch (_) { + // Best-effort: one resource failing to close shouldn't stop the + // rest from being destroyed. + } + } + _resources.clear(); + } +} diff --git a/pkgs/http2/test/client_pool_test.dart b/pkgs/http2/test/client_pool_test.dart new file mode 100644 index 0000000000..b8066c8ba7 --- /dev/null +++ b/pkgs/http2/test/client_pool_test.dart @@ -0,0 +1,182 @@ +// 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 'package:http2/src/client_pool.dart'; +import 'package:test/test.dart'; + +ClientPool _pool({ + required int maxConcurrentOperations, + int maxIdleResources = 1, +}) { + var nextId = 0; + return ClientPool( + () async => nextId++, + maxConcurrentOperations: maxConcurrentOperations, + maxIdleResources: maxIdleResources, + destroy: (_) async {}, + ); +} + +void main() { + group('client-pool-test', () { + test('creates-new-resources-as-needed', () { + final pool = _pool(maxConcurrentOperations: 2); + final completers = List.generate(3, (_) => Completer()); + + expect(pool.size, 0); + unawaited(pool.run((_) => completers[0].future)); + unawaited(pool.run((_) => completers[1].future)); + expect(pool.size, 1); + unawaited(pool.run((_) => completers[2].future)); + expect(pool.size, 2); + + for (final c in completers) { + c.complete(); + } + }); + + test('reuses-resource-with-remaining-capacity', () async { + final pool = _pool(maxConcurrentOperations: 2); + final completers = List.generate(3, (_) => Completer()); + + final first = pool.run((_) => completers[0].future); + unawaited(pool.run((_) => completers[1].future)); + expect(pool.size, 1); + + completers[0].complete(); + await first; + + unawaited(pool.run((_) => completers[2].future)); + expect(pool.size, 1); + + completers[1].complete(); + completers[2].complete(); + }); + + test('packs-load-onto-most-full-resource', () async { + final pool = _pool(maxConcurrentOperations: 2); + final completers = List.generate(4, (_) => Completer()); + final resourcesUsed = []; + + void run(int i) => unawaited( + pool.run((r) { + resourcesUsed.add(r); + return completers[i].future; + }), + ); + + run(0); + run(1); + run(2); // Resource 0 is full - this should open resource 1. + await Future.value(); + expect(resourcesUsed, [0, 0, 1]); + + completers[0].complete(); + await Future.value(); + + run(3); // Resource 0 has a free slot again and is the most-full option. + await Future.value(); + expect(resourcesUsed, [0, 0, 1, 0]); + + completers[1].complete(); + completers[2].complete(); + completers[3].complete(); + }); + + test('stops-reusing-resource-after-failure', () async { + final pool = _pool(maxConcurrentOperations: 10); + final resourcesUsed = []; + + await pool + .run((r) { + resourcesUsed.add(r); + return Future.error('boom'); + }) + .catchError((_) {}); + + await pool.run((r) { + resourcesUsed.add(r); + return Future.value(); + }); + + expect(resourcesUsed, [0, 1]); + }); + + test('garbage-collects-after-success', () async { + final pool = _pool(maxConcurrentOperations: 2, maxIdleResources: 0); + final completers = List.generate(4, (_) => Completer()); + + final ops = [ + pool.run((_) => completers[0].future), + pool.run((_) => completers[1].future), + pool.run((_) => completers[2].future), + pool.run((_) => completers[3].future), + ]; + expect(pool.size, 2); + + for (final c in completers) { + c.complete(); + } + await Future.wait(ops); + + expect(pool.size, 0); + }); + + test('garbage-collects-after-error', () async { + final pool = _pool(maxConcurrentOperations: 2, maxIdleResources: 0); + + final ops = List.generate( + 4, + (_) => pool.run((_) => Future.error('boom')).catchError((_) {}), + ); + await Future.wait(ops); + + expect(pool.size, 0); + }); + + test('keeps-idle-resources-up-to-max-idle-resources', () async { + final pool = _pool(maxConcurrentOperations: 1, maxIdleResources: 3); + final completers = List.generate(4, (_) => Completer()); + + final ops = [ + pool.run((_) => completers[0].future), + pool.run((_) => completers[1].future), + pool.run((_) => completers[2].future), + pool.run((_) => completers[3].future), + ]; + expect(pool.size, 4); + + for (final c in completers) { + c.complete(); + } + await Future.wait(ops); + + expect(pool.size, 3); + }); + + test('rejects-operations-after-terminate', () async { + final pool = _pool(maxConcurrentOperations: 1); + + await pool.terminate(); + + expect(() => pool.run((_) async {}), throwsA(isA())); + }); + + test('waits-for-in-flight-operations-before-terminating', () async { + final pool = _pool(maxConcurrentOperations: 1); + final completer = Completer(); + var terminated = false; + + unawaited(pool.run((_) => completer.future)); + final terminateOp = pool.terminate().then((_) => terminated = true); + + expect(terminated, isFalse); + completer.complete(); + await terminateOp; + expect(terminated, isTrue); + }); + }); +} From 1e7e56e4cf23d09ee92ce83c25daea9a3dffd9d2 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 30 Jul 2026 10:41:00 +0100 Subject: [PATCH 02/27] feat(http2): add Http2Client, a pooled multiplexed http.Client --- pkgs/http2/lib/src/http2_client.dart | 196 +++++++++++++++++++++++++ pkgs/http2/pubspec.yaml | 4 + pkgs/http2/test/http2_client_test.dart | 161 ++++++++++++++++++++ 3 files changed, 361 insertions(+) create mode 100644 pkgs/http2/lib/src/http2_client.dart create mode 100644 pkgs/http2/test/http2_client_test.dart diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart new file mode 100644 index 0000000000..2e4aee65c2 --- /dev/null +++ b/pkgs/http2/lib/src/http2_client.dart @@ -0,0 +1,196 @@ +// 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:convert'; +import 'dart:io'; + +import 'package:http/http.dart'; +import 'package:pool/pool.dart'; + +import '../transport.dart'; +import 'client_pool.dart'; + +/// A pooled, multiplexed `http.Client` backed by HTTP/2 connections. +/// +/// Every request is sent as its own HTTP/2 stream on a shared connection - +/// see [ClientPool] - dialed per `host:port` via +/// `SecureSocket.connect(..., supportedProtocols: ['h2'])`. Once a +/// connection's [maxStreamsPerConnection] concurrent streams are in use, a +/// new connection is dialed rather than queuing behind the existing one. +/// +/// Being multi-host makes this safe to use as a general-purpose transport - +/// for example as the `baseClient` passed to `googleapis_auth`'s client +/// helpers, whose credential negotiation (OAuth2 token endpoint, WIF/OIDC +/// token exchange) targets different hosts than the API calls that follow. +/// A connection dialed for one host is never reused for another. +/// +/// `onBadCertificate` is forwarded as-is to `SecureSocket.connect`: returning +/// `true` accepts a certificate that failed normal verification (expired, +/// self-signed, wrong host, ...). It exists for tests and trusted private +/// networks - do not use it to accept arbitrary certificates in production. +class Http2Client extends BaseClient { + Http2Client({ + this.maxStreamsPerConnection = 100, + this.maxIdleConnections = 1, + int maxConcurrentHandshakes = 50, + SecurityContext? context, + bool Function(X509Certificate certificate)? onBadCertificate, + }) : _context = context, + _onBadCertificate = onBadCertificate, + _handshakeGate = Pool(maxConcurrentHandshakes); + + /// The maximum number of concurrent HTTP/2 streams (i.e. requests) to + /// multiplex onto a single connection before dialing another. + final int maxStreamsPerConnection; + + /// The maximum number of idle connections to keep per host, per + /// [ClientPool.maxIdleResources]. + final int maxIdleConnections; + + final SecurityContext? _context; + + // Forwarded to `SecureSocket.connect` as-is: returning `true` accepts a + // certificate that failed normal verification (expired, self-signed, + // wrong host, ...). Intended for tests and trusted private networks - + // never use this to accept arbitrary certificates in production. + final bool Function(X509Certificate certificate)? _onBadCertificate; + + // Caps concurrent in-flight TCP+TLS handshakes across every host, + // independent of how many connections any one host's pool ends up + // needing. + final Pool _handshakeGate; + + final _pools = >{}; + + // Synchronous (no `await`), so concurrent requests to a new host:port + // can't race each other into creating two pools for the same key. + ClientPool _poolFor(Uri url) { + final key = '${url.host}:${url.port}'; + return _pools.putIfAbsent( + key, + () => ClientPool( + () => _handshakeGate.withResource(() => _dial(url.host, url.port)), + maxConcurrentOperations: maxStreamsPerConnection, + maxIdleResources: maxIdleConnections, + destroy: (transport) => transport.finish(), + ), + ); + } + + Future _dial(String host, int port) async { + final socket = await SecureSocket.connect( + host, + port, + context: _context, + onBadCertificate: _onBadCertificate, + supportedProtocols: ['h2'], + ); + if (socket.selectedProtocol != 'h2') { + socket.destroy(); + throw StateError( + 'Server did not negotiate HTTP/2 (got ${socket.selectedProtocol})', + ); + } + return ClientTransportConnection.viaSocket(socket); + } + + /// Sends [request] as a single HTTP/2 stream on [transport], translating + /// between `BaseRequest`/`StreamedResponse` and http2's frames. + Future _sendOverHttp2( + ClientTransportConnection transport, + BaseRequest request, + ) async { + final bodyBytes = await request.finalize().toBytes(); + final path = + request.url.hasQuery + ? '${request.url.path}?${request.url.query}' + : request.url.path; + + final stream = transport.makeRequest([ + Header.ascii(':method', request.method), + Header.ascii(':scheme', 'https'), + Header.ascii(':authority', request.url.host), + Header.ascii(':path', path), + // HTTP/2 requires lowercase header names (RFC 7540 8.1.2). + for (final entry in request.headers.entries) + Header.ascii(entry.key.toLowerCase(), entry.value), + ], endStream: bodyBytes.isEmpty); + + if (bodyBytes.isNotEmpty) stream.sendData(bodyBytes, endStream: true); + + final statusCompleter = Completer(); + late final StreamSubscription subscription; + final bodyController = StreamController>( + onCancel: () => subscription.cancel(), + ); + final responseHeaders = {}; + + subscription = stream.incomingMessages.listen( + (message) { + if (message is HeadersStreamMessage) { + for (final header in message.headers) { + final name = ascii.decode(header.name); + final value = ascii.decode(header.value); + if (name == ':status') { + if (!statusCompleter.isCompleted) { + statusCompleter.complete(int.parse(value)); + } + } else { + responseHeaders[name] = value; + } + } + } else if (message is DataStreamMessage) { + bodyController.add(message.bytes); + } + }, + onDone: () { + if (!statusCompleter.isCompleted) { + statusCompleter.completeError( + StateError('Stream closed before a response status was received'), + ); + } + if (!bodyController.isClosed) bodyController.close(); + }, + onError: (Object error, StackTrace stackTrace) { + if (!statusCompleter.isCompleted) { + statusCompleter.completeError(error, stackTrace); + } + bodyController.addError(error, stackTrace); + if (!bodyController.isClosed) bodyController.close(); + }, + cancelOnError: true, + ); + + final statusCode = await statusCompleter.future; + return StreamedResponse( + bodyController.stream, + statusCode, + headers: responseHeaders, + request: request, + ); + } + + /// The number of connections currently pooled, across every host. + int get connectionCount => + _pools.values.fold(0, (total, pool) => total + pool.size); + + @override + Future send(BaseRequest request) => _poolFor( + request.url, + ).run((transport) => _sendOverHttp2(transport, request)); + + /// Waits for in-flight requests to finish, then closes every connection. + /// + /// Unlike [close] (constrained by `http.Client`'s synchronous signature), + /// this can be awaited by callers who hold a concrete [Http2Client]. + Future terminate() async { + for (final pool in _pools.values) { + await pool.terminate(); + } + } + + @override + void close() => unawaited(terminate()); +} diff --git a/pkgs/http2/pubspec.yaml b/pkgs/http2/pubspec.yaml index e65f0a31a4..afa146498e 100644 --- a/pkgs/http2/pubspec.yaml +++ b/pkgs/http2/pubspec.yaml @@ -11,6 +11,10 @@ topics: environment: sdk: ^3.7.0 +dependencies: + http: ^1.5.0 + pool: ^1.5.0 + dev_dependencies: build_runner: ^2.4.15 dart_flutter_team_lints: ^3.5.1 diff --git a/pkgs/http2/test/http2_client_test.dart b/pkgs/http2/test/http2_client_test.dart new file mode 100644 index 0000000000..83d2405fe0 --- /dev/null +++ b/pkgs/http2/test/http2_client_test.dart @@ -0,0 +1,161 @@ +// 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:convert' show ascii; +import 'dart:io'; + +import 'package:http2/multiprotocol_server.dart'; +import 'package:http2/src/http2_client.dart'; +import 'package:http2/transport.dart'; +import 'package:test/test.dart'; + +SecurityContext _serverContext() => + SecurityContext() + ..useCertificateChain('test/certificates/server_chain.pem') + ..usePrivateKey('test/certificates/server_key.pem', password: 'dartdart'); + +Future _bind() => + MultiProtocolHttpServer.bind('localhost', 0, _serverContext()); + +// test/certificates/server_chain.pem is long expired and self-signed, same +// as every other test in this package that dials it - see +// multiprotocol_server_test.dart. Bypassing verification here is just for +// this fixture; Http2Client itself defaults to real verification. +Http2Client _testClient({int maxStreamsPerConnection = 100}) => Http2Client( + maxStreamsPerConnection: maxStreamsPerConnection, + onBadCertificate: (_) => true, +); + +/// Replies with [body] after waiting on [delay], if given. +void Function(ServerTransportStream) _respondWith( + String body, { + Future? delay, +}) { + return (stream) async { + final subscription = StreamIterator(stream.incomingMessages); + await subscription.moveNext(); // Consume the request headers. + while (await subscription.moveNext()) {} // Drain any request body. + + if (delay != null) await delay; + + stream.outgoingMessages.add( + HeadersStreamMessage([Header.ascii(':status', '200')]), + ); + stream.outgoingMessages.add(DataStreamMessage(ascii.encode(body))); + await stream.outgoingMessages.close(); + }; +} + +void main() { + group('http2-client-test', () { + test('sends-request-and-receives-response', () async { + final server = await _bind(); + server.startServing( + (request) {}, + expectAsync1(_respondWith('hello'), count: 1), + ); + + final client = _testClient(); + final response = await client.get( + Uri.parse('https://localhost:${server.port}/'), + ); + + expect(response.statusCode, 200); + expect(response.body, 'hello'); + + await client.terminate(); + await server.close(); + }); + + test('pools-connections-per-host-and-port', () async { + final serverA = await _bind(); + final serverB = await _bind(); + serverA.startServing( + (request) {}, + expectAsync1(_respondWith('a'), count: 1), + ); + serverB.startServing( + (request) {}, + expectAsync1(_respondWith('b'), count: 1), + ); + + final client = _testClient(); + await Future.wait([ + client.get(Uri.parse('https://localhost:${serverA.port}/')), + client.get(Uri.parse('https://localhost:${serverB.port}/')), + ]); + + expect(client.connectionCount, 2); + + await client.terminate(); + await Future.wait([serverA.close(), serverB.close()]); + }); + + test('exceeding-max-streams-per-connection-opens-new-connection', () async { + final server = await _bind(); + final releaseA = Completer(); + final releaseB = Completer(); + var requestNr = 0; + server.startServing( + (request) {}, + expectAsync1((stream) { + final release = requestNr++ == 0 ? releaseA : releaseB; + return _respondWith('r', delay: release.future)(stream); + }, count: 2), + ); + + final client = _testClient(maxStreamsPerConnection: 1); + final requestA = client.get( + Uri.parse('https://localhost:${server.port}/a'), + ); + // Give the pool a chance to dial and dispatch the first request + // before the second one arrives, so it's guaranteed to land on an + // already-full connection rather than racing to share it. + await Future.delayed(const Duration(milliseconds: 50)); + final requestB = client.get( + Uri.parse('https://localhost:${server.port}/b'), + ); + + await Future.delayed(const Duration(milliseconds: 50)); + expect(client.connectionCount, 2); + + releaseA.complete(); + releaseB.complete(); + await Future.wait([requestA, requestB]); + + await client.terminate(); + await server.close(); + }); + + test('terminate-waits-for-in-flight-request', () async { + final server = await _bind(); + final release = Completer(); + server.startServing( + (request) {}, + expectAsync1(_respondWith('done', delay: release.future), count: 1), + ); + + final client = _testClient(); + final request = client.get( + Uri.parse('https://localhost:${server.port}/'), + ); + + var terminated = false; + final terminateFuture = client.terminate().then((_) { + terminated = true; + }); + + await Future.delayed(const Duration(milliseconds: 50)); + expect(terminated, isFalse); + + release.complete(); + await request; + await terminateFuture; + expect(terminated, isTrue); + + await server.close(); + }); + }); +} From 894b94cb77eee08a9d0f83e493858810965bb61e Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 30 Jul 2026 11:11:02 +0100 Subject: [PATCH 03/27] feat(http2): expose Http2Client via package:http2/client.dart --- pkgs/http2/README.md | 24 +++++++++++++++++++ pkgs/http2/example/pooled_client.dart | 34 +++++++++++++++++++++++++++ pkgs/http2/lib/client.dart | 13 ++++++++++ 3 files changed, 71 insertions(+) create mode 100644 pkgs/http2/example/pooled_client.dart create mode 100644 pkgs/http2/lib/client.dart diff --git a/pkgs/http2/README.md b/pkgs/http2/README.md index edb75c99d6..fce690f59e 100644 --- a/pkgs/http2/README.md +++ b/pkgs/http2/README.md @@ -53,3 +53,27 @@ Future main() async { An example with better error handling is available [here][example]. See the [API docs][api] for more details. + +## Pooled `http.Client` + +`package:http2/client.dart` provides `Http2Client`, a `package:http` +`Client` that pools and multiplexes requests over shared HTTP/2 connections +instead of opening one connection per request. This is useful for workloads +that send many concurrent requests to the same host or hosts, where +`dart:io`'s `HttpClient` (HTTP/1.1 only) would otherwise open a new TCP+TLS +connection per request. + +```dart +import 'package:http2/client.dart'; + +Future main() async { + final client = Http2Client(); + final response = await client.get(Uri.parse('https://example.com/')); + print(response.body); + await client.terminate(); +} +``` + +A connection is dialed per `host:port` as needed, so a single `Http2Client` +is safe to reuse across requests to different hosts. See the example +[here](example/pooled_client.dart). diff --git a/pkgs/http2/example/pooled_client.dart b/pkgs/http2/example/pooled_client.dart new file mode 100644 index 0000000000..97b1951a0e --- /dev/null +++ b/pkgs/http2/example/pooled_client.dart @@ -0,0 +1,34 @@ +// 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:io'; + +import 'package:http2/client.dart'; + +/// Sends several concurrent requests through a single [Http2Client], +/// demonstrating that they share pooled HTTP/2 connections instead of each +/// opening their own. +void main(List args) async { + if (args.length != 1) { + print('Usage: dart pooled_client.dart '); + exit(1); + } + + final uri = Uri.parse(args[0]); + final client = Http2Client(); + + try { + final responses = await Future.wait( + List.generate(5, (_) => client.get(uri)), + ); + for (final response in responses) { + print('${response.statusCode}: ${response.body.length} bytes'); + } + print('Connections used: ${client.connectionCount}'); + } finally { + // Waits for the requests above to finish before closing every + // connection - see Http2Client.terminate(). + await client.terminate(); + } +} diff --git a/pkgs/http2/lib/client.dart b/pkgs/http2/lib/client.dart new file mode 100644 index 0000000000..90759b1994 --- /dev/null +++ b/pkgs/http2/lib/client.dart @@ -0,0 +1,13 @@ +// 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. + +/// A pooled, multiplexed `package:http` `Client` backed by HTTP/2 +/// connections. +/// +/// See [Http2Client]. +library; + +import 'src/http2_client.dart' show Http2Client; + +export 'src/http2_client.dart' show Http2Client; From 4a874cb4ba4d14923c4e314a375a8c82da8422da Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 30 Jul 2026 11:12:38 +0100 Subject: [PATCH 04/27] chore(http2): changelog entry for Http2Client --- pkgs/http2/CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index b98deeeefa..ddb45d513c 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md @@ -1,6 +1,8 @@ ## 3.0.1-wip - Gracefully handle receiving headers on a stream that the client has canceled. (#1799) +- Add `Http2Client` (`package:http2/client.dart`), a pooled, multiplexed + `package:http` `Client` backed by HTTP/2 connections. ## 3.0.0 From 258830f01d1f53b6f27c74306b7e32616087937a Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 30 Jul 2026 13:36:10 +0100 Subject: [PATCH 05/27] fix(http2): retry once when a pooled connection was closed by the peer --- pkgs/http2/lib/src/http2_client.dart | 30 ++++++++++++-- pkgs/http2/test/http2_client_test.dart | 54 ++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 4 deletions(-) diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 2e4aee65c2..4a1fc8f363 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -101,8 +101,10 @@ class Http2Client extends BaseClient { Future _sendOverHttp2( ClientTransportConnection transport, BaseRequest request, + List bodyBytes, ) async { - final bodyBytes = await request.finalize().toBytes(); + if (!transport.isOpen) throw const _ConnectionClosedByPeer(); + final path = request.url.hasQuery ? '${request.url.path}?${request.url.query}' @@ -177,9 +179,19 @@ class Http2Client extends BaseClient { _pools.values.fold(0, (total, pool) => total + pool.size); @override - Future send(BaseRequest request) => _poolFor( - request.url, - ).run((transport) => _sendOverHttp2(transport, request)); + Future send(BaseRequest request) { + List? bodyBytes; + Future attempt() => + _poolFor(request.url).run((transport) async { + bodyBytes ??= await request.finalize().toBytes(); + return _sendOverHttp2(transport, request, bodyBytes!); + }); + + return attempt().catchError( + (Object _) => attempt(), + test: (error) => error is _ConnectionClosedByPeer, + ); + } /// Waits for in-flight requests to finish, then closes every connection. /// @@ -194,3 +206,13 @@ class Http2Client extends BaseClient { @override void close() => unawaited(terminate()); } + +/// Thrown by [Http2Client._sendOverHttp2] when a pooled connection turns +/// out to have already been closed by the peer (e.g. a graceful `GOAWAY`) +/// before any bytes were written for this request. [Http2Client.send] +/// catches this and retries once on whatever the pool dials next - +/// [ClientPool.run] has already marked the dead connection failed and +/// evicted it by the time the retry runs. +class _ConnectionClosedByPeer implements Exception { + const _ConnectionClosedByPeer(); +} diff --git a/pkgs/http2/test/http2_client_test.dart b/pkgs/http2/test/http2_client_test.dart index 83d2405fe0..717229ba79 100644 --- a/pkgs/http2/test/http2_client_test.dart +++ b/pkgs/http2/test/http2_client_test.dart @@ -28,6 +28,37 @@ Http2Client _testClient({int maxStreamsPerConnection = 100}) => Http2Client( onBadCertificate: (_) => true, ); +/// A minimal HTTP/2-only server that (unlike [MultiProtocolHttpServer]) +/// exposes each accepted [ServerTransportConnection], so a test can finish +/// one connection gracefully while the server keeps listening for new ones. +class _RawHttp2Server { + _RawHttp2Server._(this._socket) { + _socket.listen((socket) { + final connection = ServerTransportConnection.viaSocket(socket); + connections.add(connection); + connection.incomingStreams.listen(_respondWith('ok')); + }); + } + + static Future<_RawHttp2Server> bind() async { + final context = _serverContext()..setAlpnProtocols(['h2'], true); + final socket = await SecureServerSocket.bind('localhost', 0, context); + return _RawHttp2Server._(socket); + } + + final SecureServerSocket _socket; + final connections = []; + + int get port => _socket.port; + + Future close() async { + await _socket.close(); + for (final connection in connections) { + await connection.terminate(); + } + } +} + /// Replies with [body] after waiting on [delay], if given. void Function(ServerTransportStream) _respondWith( String body, { @@ -129,6 +160,29 @@ void main() { await server.close(); }); + test('retries-once-when-pooled-connection-was-closed-by-peer', () async { + final server = await _RawHttp2Server.bind(); + final client = _testClient(); + + final r1 = await client.get( + Uri.parse('https://localhost:${server.port}/'), + ); + expect(r1.statusCode, 200); + expect(client.connectionCount, 1); + + await server.connections.single.finish(); + await Future.delayed(const Duration(milliseconds: 200)); + + final r2 = await client.get( + Uri.parse('https://localhost:${server.port}/'), + ); + expect(r2.statusCode, 200); + expect(r2.body, 'ok'); + + await client.terminate(); + await server.close(); + }); + test('terminate-waits-for-in-flight-request', () async { final server = await _bind(); final release = Completer(); From 2c8ae6cfd578b5633db4e76ddee95b0909e62bfb Mon Sep 17 00:00:00 2001 From: demolaf Date: Mon, 3 Aug 2026 17:08:05 +0100 Subject: [PATCH 06/27] fix(http2): default empty :path to / and strip HTTP/1.x connection-specific headers --- pkgs/http2/lib/src/http2_client.dart | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 4a1fc8f363..aa08253352 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -105,19 +105,23 @@ class Http2Client extends BaseClient { ) async { if (!transport.isOpen) throw const _ConnectionClosedByPeer(); + // RFC 7540 8.1.2.3: ":path" must not be empty. + final rawPath = request.url.path.isEmpty ? '/' : request.url.path; final path = - request.url.hasQuery - ? '${request.url.path}?${request.url.query}' - : request.url.path; + request.url.hasQuery ? '$rawPath?${request.url.query}' : rawPath; final stream = transport.makeRequest([ Header.ascii(':method', request.method), Header.ascii(':scheme', 'https'), Header.ascii(':authority', request.url.host), Header.ascii(':path', path), - // HTTP/2 requires lowercase header names (RFC 7540 8.1.2). + // HTTP/2 requires lowercase header names (RFC 7540 8.1.2), and forbids + // connection-specific header fields (RFC 7540 8.1.2.2) - which a + // request built for an HTTP/1.1-oriented client might still set. + // `host` is dropped too, since `:authority` already carries it. for (final entry in request.headers.entries) - Header.ascii(entry.key.toLowerCase(), entry.value), + if (!_connectionSpecificHeaders.contains(entry.key.toLowerCase())) + Header.ascii(entry.key.toLowerCase(), entry.value), ], endStream: bodyBytes.isEmpty); if (bodyBytes.isNotEmpty) stream.sendData(bodyBytes, endStream: true); @@ -216,3 +220,14 @@ class Http2Client extends BaseClient { class _ConnectionClosedByPeer implements Exception { const _ConnectionClosedByPeer(); } + +/// Header fields forbidden on an HTTP/2 stream (RFC 7540 8.1.2.2), plus +/// `host` since `:authority` already carries what it would. +const _connectionSpecificHeaders = { + 'connection', + 'keep-alive', + 'proxy-connection', + 'transfer-encoding', + 'upgrade', + 'host', +}; From 2376148d3eb04aba551473a2a31913dc517bf985 Mon Sep 17 00:00:00 2001 From: demolaf Date: Mon, 3 Aug 2026 17:08:40 +0100 Subject: [PATCH 07/27] fix(http2): throw ClientException instead of an internal pool error after terminate() --- pkgs/http2/lib/src/http2_client.dart | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index aa08253352..a1936a5f98 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -63,6 +63,7 @@ class Http2Client extends BaseClient { final Pool _handshakeGate; final _pools = >{}; + var _closed = false; // Synchronous (no `await`), so concurrent requests to a new host:port // can't race each other into creating two pools for the same key. @@ -184,6 +185,13 @@ class Http2Client extends BaseClient { @override Future send(BaseRequest request) { + if (_closed) { + throw ClientException( + 'HTTP request failed. Client is already closed.', + request.url, + ); + } + List? bodyBytes; Future attempt() => _poolFor(request.url).run((transport) async { @@ -202,6 +210,7 @@ class Http2Client extends BaseClient { /// Unlike [close] (constrained by `http.Client`'s synchronous signature), /// this can be awaited by callers who hold a concrete [Http2Client]. Future terminate() async { + _closed = true; for (final pool in _pools.values) { await pool.terminate(); } From 6955e303cf077473f210168e3008f89e916a243d Mon Sep 17 00:00:00 2001 From: demolaf Date: Mon, 3 Aug 2026 17:15:05 +0100 Subject: [PATCH 08/27] fix(http2): exclude failed resources from idle-capacity calculation --- pkgs/http2/lib/src/client_pool.dart | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index d6e857a5f2..f78dd3026a 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -98,10 +98,13 @@ class ClientPool { } bool get _hasExcessIdleCapacity { + // Failed resources are never routed new work by _acquire(), so they + // contribute no real idle capacity. final idleCapacity = _resources.fold( 0, (total, resource) => - total + (maxConcurrentOperations - resource.inFlight), + total + + (resource.failed ? 0 : maxConcurrentOperations - resource.inFlight), ); return idleCapacity > maxIdleResources * maxConcurrentOperations; } From 5d1a7f44f27f9b597c100e2163133ac956b44a94 Mon Sep 17 00:00:00 2001 From: demolaf Date: Mon, 3 Aug 2026 17:15:40 +0100 Subject: [PATCH 09/27] chore(http2): mark ClientPool.opCount @visibleForTesting --- pkgs/http2/lib/src/client_pool.dart | 3 +++ pkgs/http2/pubspec.yaml | 1 + 2 files changed, 4 insertions(+) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index f78dd3026a..f0d0e35912 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -4,6 +4,8 @@ import 'dart:async'; +import 'package:meta/meta.dart'; + class _PooledResource { _PooledResource(this.future); final Future future; @@ -40,6 +42,7 @@ class ClientPool { int get size => _resources.length; /// The number of in-flight operations across every resource. For testing. + @visibleForTesting int get opCount => _resources.fold(0, (total, resource) => total + resource.inFlight); diff --git a/pkgs/http2/pubspec.yaml b/pkgs/http2/pubspec.yaml index afa146498e..4d61be5afe 100644 --- a/pkgs/http2/pubspec.yaml +++ b/pkgs/http2/pubspec.yaml @@ -13,6 +13,7 @@ environment: dependencies: http: ^1.5.0 + meta: ^1.15.0 pool: ^1.5.0 dev_dependencies: From 059cf7655dbb6dda658ed4121cf59d5560d372a1 Mon Sep 17 00:00:00 2001 From: demolaf Date: Tue, 4 Aug 2026 10:17:37 +0100 Subject: [PATCH 10/27] docs(http2): ClientPool.size is used by production code, not just tests --- pkgs/http2/lib/src/client_pool.dart | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index f0d0e35912..3ac48a745f 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -38,7 +38,7 @@ class ClientPool { var _terminated = false; Completer? _drained; - /// The number of resources currently in the pool. For testing. + /// The number of resources currently in the pool. int get size => _resources.length; /// The number of in-flight operations across every resource. For testing. From 84c0d66cac3e1dfd3ecda8c435c46fc9aed0e064 Mon Sep 17 00:00:00 2001 From: demolaf Date: Tue, 4 Aug 2026 12:49:40 +0100 Subject: [PATCH 11/27] refactor(http2): use collection's .sum instead of manual fold sums --- pkgs/http2/lib/src/client_pool.dart | 19 +++++++++++-------- pkgs/http2/lib/src/http2_client.dart | 4 ++-- pkgs/http2/pubspec.yaml | 1 + 3 files changed, 14 insertions(+), 10 deletions(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index 3ac48a745f..4d14f1e20a 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -4,6 +4,7 @@ import 'dart:async'; +import 'package:collection/collection.dart'; import 'package:meta/meta.dart'; class _PooledResource { @@ -43,8 +44,7 @@ class ClientPool { /// The number of in-flight operations across every resource. For testing. @visibleForTesting - int get opCount => - _resources.fold(0, (total, resource) => total + resource.inFlight); + int get opCount => _resources.map((resource) => resource.inFlight).sum; /// Runs [operation] on an available (or newly created) resource. Future run(Future Function(T resource) operation) async { @@ -103,12 +103,15 @@ class ClientPool { bool get _hasExcessIdleCapacity { // Failed resources are never routed new work by _acquire(), so they // contribute no real idle capacity. - final idleCapacity = _resources.fold( - 0, - (total, resource) => - total + - (resource.failed ? 0 : maxConcurrentOperations - resource.inFlight), - ); + final idleCapacity = + _resources + .map( + (resource) => + resource.failed + ? 0 + : maxConcurrentOperations - resource.inFlight, + ) + .sum; return idleCapacity > maxIdleResources * maxConcurrentOperations; } diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index a1936a5f98..82875a88e3 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -6,6 +6,7 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; +import 'package:collection/collection.dart'; import 'package:http/http.dart'; import 'package:pool/pool.dart'; @@ -180,8 +181,7 @@ class Http2Client extends BaseClient { } /// The number of connections currently pooled, across every host. - int get connectionCount => - _pools.values.fold(0, (total, pool) => total + pool.size); + int get connectionCount => _pools.values.map((pool) => pool.size).sum; @override Future send(BaseRequest request) { diff --git a/pkgs/http2/pubspec.yaml b/pkgs/http2/pubspec.yaml index 4d61be5afe..a2e21c0670 100644 --- a/pkgs/http2/pubspec.yaml +++ b/pkgs/http2/pubspec.yaml @@ -12,6 +12,7 @@ environment: sdk: ^3.7.0 dependencies: + collection: ^1.19.0 http: ^1.5.0 meta: ^1.15.0 pool: ^1.5.0 From 0ca6c5c9b6c5e222e591a2745e72011e4f37ebf2 Mon Sep 17 00:00:00 2001 From: demolaf Date: Tue, 4 Aug 2026 12:53:28 +0100 Subject: [PATCH 12/27] style(http2): use braces for single-statement ifs in _sendOverHttp2 --- pkgs/http2/lib/src/http2_client.dart | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 82875a88e3..c5bbd6d3d9 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -159,14 +159,18 @@ class Http2Client extends BaseClient { StateError('Stream closed before a response status was received'), ); } - if (!bodyController.isClosed) bodyController.close(); + if (!bodyController.isClosed) { + bodyController.close(); + } }, onError: (Object error, StackTrace stackTrace) { if (!statusCompleter.isCompleted) { statusCompleter.completeError(error, stackTrace); } bodyController.addError(error, stackTrace); - if (!bodyController.isClosed) bodyController.close(); + if (!bodyController.isClosed) { + bodyController.close(); + } }, cancelOnError: true, ); From 327997151652d064b48a1eb47677f851de5c9ad1 Mon Sep 17 00:00:00 2001 From: demolaf Date: Tue, 4 Aug 2026 14:13:32 +0100 Subject: [PATCH 13/27] style(http2): use a case pattern for _maybeCompleteDrain's null check --- pkgs/http2/lib/src/client_pool.dart | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index 4d14f1e20a..27b28c8a16 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -116,8 +116,8 @@ class ClientPool { } void _maybeCompleteDrain() { - final drained = _drained; - if (drained != null && !drained.isCompleted && opCount == 0) { + if (_drained case final drained? + when !drained.isCompleted && opCount == 0) { drained.complete(); } } From 659e4475d856f7897b5cfd75d62609b9f5e7d1e9 Mon Sep 17 00:00:00 2001 From: demolaf Date: Tue, 4 Aug 2026 14:34:04 +0100 Subject: [PATCH 14/27] test(http2): drop redundant comment on self-signed test cert bypass --- pkgs/http2/test/http2_client_test.dart | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/pkgs/http2/test/http2_client_test.dart b/pkgs/http2/test/http2_client_test.dart index 717229ba79..471fb64513 100644 --- a/pkgs/http2/test/http2_client_test.dart +++ b/pkgs/http2/test/http2_client_test.dart @@ -19,10 +19,6 @@ SecurityContext _serverContext() => Future _bind() => MultiProtocolHttpServer.bind('localhost', 0, _serverContext()); -// test/certificates/server_chain.pem is long expired and self-signed, same -// as every other test in this package that dials it - see -// multiprotocol_server_test.dart. Bypassing verification here is just for -// this fixture; Http2Client itself defaults to real verification. Http2Client _testClient({int maxStreamsPerConnection = 100}) => Http2Client( maxStreamsPerConnection: maxStreamsPerConnection, onBadCertificate: (_) => true, @@ -169,7 +165,7 @@ void main() { ); expect(r1.statusCode, 200); expect(client.connectionCount, 1); - + await server.connections.single.finish(); await Future.delayed(const Duration(milliseconds: 200)); From 33631c713da709af7fbe1b048d10977ea7ebff10 Mon Sep 17 00:00:00 2001 From: demolaf Date: Tue, 4 Aug 2026 19:11:10 +0100 Subject: [PATCH 15/27] feat(http2): expose peer's SETTINGS_MAX_CONCURRENT_STREAMS via peerMaxConcurrentStreams --- pkgs/http2/CHANGELOG.md | 2 ++ pkgs/http2/lib/src/connection.dart | 4 ++++ pkgs/http2/lib/transport.dart | 5 +++++ 3 files changed, 11 insertions(+) diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index ddb45d513c..5dc9db4b1e 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md @@ -3,6 +3,8 @@ - Gracefully handle receiving headers on a stream that the client has canceled. (#1799) - Add `Http2Client` (`package:http2/client.dart`), a pooled, multiplexed `package:http` `Client` backed by HTTP/2 connections. +- Add `ClientTransportConnection.peerMaxConcurrentStreams`, exposing the + peer's most recently advertised `SETTINGS_MAX_CONCURRENT_STREAMS`. ## 3.0.0 diff --git a/pkgs/http2/lib/src/connection.dart b/pkgs/http2/lib/src/connection.dart index 6a9f3fc9de..eb866cb975 100644 --- a/pkgs/http2/lib/src/connection.dart +++ b/pkgs/http2/lib/src/connection.dart @@ -513,6 +513,10 @@ class ClientConnection extends Connection implements ClientTransportConnection { bool get isOpen => !_state.isFinishing && !_state.isTerminated && _streams.canOpenStream; + @override + int? get peerMaxConcurrentStreams => + _settingsHandler.peerSettings.maxConcurrentStreams; + @override ClientTransportStream makeRequest( List
headers, { diff --git a/pkgs/http2/lib/transport.dart b/pkgs/http2/lib/transport.dart index 4584e71bc6..ae57fa89f0 100644 --- a/pkgs/http2/lib/transport.dart +++ b/pkgs/http2/lib/transport.dart @@ -97,6 +97,11 @@ abstract class ClientTransportConnection extends TransportConnection { /// via [makeRequest]. bool get isOpen; + /// The maximum number of concurrent streams the peer currently allows, + /// per its most recent SETTINGS_MAX_CONCURRENT_STREAMS (RFC 7540 6.5.2), + /// or `null` if the peer hasn't advertised a limit (unlimited). + int? get peerMaxConcurrentStreams; + /// Creates a new outgoing stream. ClientTransportStream makeRequest( List
headers, { From 168d1e94ca6f398d54bdf6c693d9da08e3d10c8d Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 16:43:08 +0100 Subject: [PATCH 16/27] test(http2): add client conformance tests for Http2Client --- pkgs/http2/pubspec.yaml | 6 + pkgs/http2/test/client_conformance_test.dart | 285 +++++++++++++++++++ 2 files changed, 291 insertions(+) create mode 100644 pkgs/http2/test/client_conformance_test.dart diff --git a/pkgs/http2/pubspec.yaml b/pkgs/http2/pubspec.yaml index a2e21c0670..a14707ec97 100644 --- a/pkgs/http2/pubspec.yaml +++ b/pkgs/http2/pubspec.yaml @@ -20,5 +20,11 @@ dependencies: dev_dependencies: build_runner: ^2.4.15 dart_flutter_team_lints: ^3.5.1 + http_client_conformance_tests: + path: ../http_client_conformance_tests/ mockito: ^5.4.5 test: ^1.25.15 + +dependency_overrides: + http: + path: ../http/ diff --git a/pkgs/http2/test/client_conformance_test.dart b/pkgs/http2/test/client_conformance_test.dart new file mode 100644 index 0000000000..bb4f291206 --- /dev/null +++ b/pkgs/http2/test/client_conformance_test.dart @@ -0,0 +1,285 @@ +// 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:convert'; +import 'dart:io'; + +import 'package:http/http.dart' as http; +import 'package:http/io_client.dart'; +import 'package:http2/src/http2_client.dart'; +import 'package:http2/transport.dart'; +import 'package:http_client_conformance_tests/http_client_conformance_tests.dart'; +import 'package:test/test.dart'; + +class Http2ProxyServer { + final SecureServerSocket _socket; + final List _connections = []; + final IOClient _httpClient = IOClient(); + + Http2ProxyServer._(this._socket) { + _socket.listen((socket) { + final connection = ServerTransportConnection.viaSocket(socket); + _connections.add(connection); + connection.incomingStreams.listen(_handleStream); + }); + } + + static Future start() async { + final context = + SecurityContext() + ..useCertificateChain('test/certificates/server_chain.pem') + ..usePrivateKey( + 'test/certificates/server_key.pem', + password: 'dartdart', + ) + ..setAlpnProtocols(['h2'], true); + final socket = await SecureServerSocket.bind('localhost', 0, context); + return Http2ProxyServer._(socket); + } + + int get port => _socket.port; + + Future _handleStream(ServerTransportStream stream) async { + try { + final messages = StreamIterator(stream.incomingMessages); + if (!await messages.moveNext()) return; + + final headersMsg = messages.current as HeadersStreamMessage; + String? method; + String? path; + int? targetPort; + final headers = {}; + + for (final header in headersMsg.headers) { + final name = ascii.decode(header.name); + final value = ascii.decode(header.value); + if (name == ':method') { + method = value; + } else if (name == ':path') { + path = value; + } else if (name == 'x-target-port') { + targetPort = int.parse(value); + } else if (!name.startsWith(':')) { + headers[name] = value; + } + } + + if (method == null || path == null || targetPort == null) { + stream.outgoingMessages.add( + HeadersStreamMessage([Header.ascii(':status', '400')]), + ); + await stream.outgoingMessages.close(); + return; + } + + // Collect body + final bodyBytes = []; + while (await messages.moveNext()) { + final msg = messages.current; + if (msg is DataStreamMessage) { + bodyBytes.addAll(msg.bytes); + } + } + + // Forward to HTTP/1.1 server + final targetUri = Uri.parse('http://localhost:$targetPort$path'); + final httpRequest = http.Request(method, targetUri); + headers.forEach((k, v) { + httpRequest.headers[k] = v; + }); + httpRequest.bodyBytes = bodyBytes; + + final httpResponse = await _httpClient.send(httpRequest); + + // Send response headers + final responseHeaders =
[ + Header.ascii(':status', httpResponse.statusCode.toString()), + ]; + httpResponse.headers.forEach((k, v) { + responseHeaders.add(Header.ascii(k.toLowerCase(), v)); + }); + stream.outgoingMessages.add(HeadersStreamMessage(responseHeaders)); + + // Send response body + await for (final chunk in httpResponse.stream) { + stream.outgoingMessages.add(DataStreamMessage(chunk)); + } + await stream.outgoingMessages.close(); + } catch (e) { + print('Proxy error: $e'); + stream.terminate(); + } + } + + Future close() async { + await _socket.close(); + for (final conn in _connections) { + await conn.terminate(); + } + _httpClient.close(); + } +} + +class ProxyRequest extends http.BaseRequest implements http.Abortable { + final http.BaseRequest _original; + @override + final Uri url; + + ProxyRequest(this._original, this.url) : super(_original.method, url); + + @override + Future? get abortTrigger => + _original is http.Abortable ? _original.abortTrigger : null; + + @override + Map get headers => _original.headers; + + @override + int? get contentLength => _original.contentLength; + @override + set contentLength(int? value) => _original.contentLength = value; + + @override + bool get followRedirects => _original.followRedirects; + @override + set followRedirects(bool value) => _original.followRedirects = value; + + @override + int get maxRedirects => _original.maxRedirects; + @override + set maxRedirects(int value) => _original.maxRedirects = value; + + @override + bool get persistentConnection => _original.persistentConnection; + @override + set persistentConnection(bool value) => + _original.persistentConnection = value; + + @override + http.ByteStream finalize() { + super.finalize(); + return _original.finalize(); + } +} + +class ConformanceProxyClient extends http.BaseClient { + final Http2Client _inner; + final int _proxyPort; + + ConformanceProxyClient(this._inner, this._proxyPort); + + @override + Future send(http.BaseRequest request) async { + final targetPort = request.url.port; + final proxyUrl = request.url.replace( + scheme: 'https', + host: 'localhost', + port: _proxyPort, + ); + + request.headers['x-target-port'] = targetPort.toString(); + final proxyRequest = ProxyRequest(request, proxyUrl); + + try { + final response = await _inner.send(proxyRequest); + return http.StreamedResponse( + response.stream, + response.statusCode, + contentLength: response.contentLength, + headers: response.headers, + isRedirect: response.isRedirect, + persistentConnection: response.persistentConnection, + reasonPhrase: response.reasonPhrase, + request: request, + ); + } on http.ClientException catch (e) { + if (e is http.RequestAbortedException) { + throw http.RequestAbortedException(request.url); + } + throw http.ClientException(e.message, request.url); + } + } + + @override + void close() { + _inner.close(); + } +} + +/// [Http2Client] only supports HTTP/2 over TLS (HTTPS). However, the standard +/// servers started by http_client_conformance_tests only support unencrypted +/// HTTP/1.1. +/// To bridge this protocol gap, we run a local HTTP/2 proxy server +/// ([Http2ProxyServer]) in-process. [ConformanceProxyClient] wraps +/// [Http2Client] and rewrites the destination URI of all outgoing requests to +/// point to the local proxy server, attaching a custom `x-target-port` header +/// to specify the target HTTP/1.1 server port. The proxy server then forwards +/// the request over HTTP/1.1 and returns the response to [Http2Client] over +/// HTTP/2. +void main() { + late final Http2ProxyServer proxy; + + setUpAll(() async { + proxy = await Http2ProxyServer.start(); + }); + + tearDownAll(() async { + await proxy.close(); + }); + + ConformanceProxyClient clientFactory() => ConformanceProxyClient( + Http2Client(onBadCertificate: (_) => true), + proxy.port, + ); + + testRequestBody(clientFactory); + + // TODO: Implement request body streaming support in Http2Client. + // Currently Http2Client reads the entire request body into memory before + // sending. + testRequestBodyStreamed(clientFactory, canStreamRequestBody: false); + + testResponseBody(clientFactory); + // TODO: Re-enable once request abort support is implemented in Http2Client. + // testResponseBodyStreamed(clientFactory); + testRequestHeaders(clientFactory); + testRequestMethods(clientFactory, preservesMethodCase: false); + + testResponseHeaders( + clientFactory, + // HTTP/2 explicitly forbids folded headers (RFC 7540 Section 8.1.2.6). + supportsFoldedHeaders: false, + // HTTP/2 does not allow NUL characters inside header names or values. + correctlyHandlesNullHeaderValues: false, + ); + + testResponseStatusLine(clientFactory); + + // TODO: Implement redirect-following support in Http2Client. + // testRedirect(clientFactory); + + testServerErrors(clientFactory); + testCompressedResponseBody(clientFactory); + testMultipleClients(clientFactory); + testMultipartRequests(clientFactory, supportsMultipartRequest: true); + testClose(clientFactory); + + // TODO: Support running client conformance tests in isolates. + // Currently we set `canWorkInIsolates` to false because the proxy server uses + // `SecureServerSocket`, which cannot be sent across isolates. + testIsolate(clientFactory, canWorkInIsolates: false); + + // TODO: Support cookies in Http2Client. + testRequestCookies(clientFactory, canSendCookieHeaders: false); + testResponseCookies(clientFactory, canReceiveSetCookieHeaders: false); + + // TODO: Implement request abort support in Http2Client. + // testAbort( + // clientFactory, + // supportsAbort: true, + // canStreamRequestBody: false, + // canStreamResponseBody: true, + // ); +} From 832cae082635cea4d9f7b286a623142418bca3d6 Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 16:43:47 +0100 Subject: [PATCH 17/27] fix(http2): cap streams per connection by the server's advertised limit --- pkgs/http2/lib/src/client_pool.dart | 61 +++++++++--- pkgs/http2/lib/src/http2_client.dart | 6 ++ pkgs/http2/test/client_pool_test.dart | 126 ++++++++++++++++++++++++- pkgs/http2/test/http2_client_test.dart | 67 +++++++++++-- 4 files changed, 238 insertions(+), 22 deletions(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index 27b28c8a16..f0bd33e00d 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -3,6 +3,7 @@ // BSD-style license that can be found in the LICENSE file. import 'dart:async'; +import 'dart:math'; import 'package:collection/collection.dart'; import 'package:meta/meta.dart'; @@ -12,26 +13,41 @@ class _PooledResource { final Future future; int inFlight = 0; bool failed = false; + + /// The resolved value of [future], once available - kept so scheduling can + /// consult the resource itself from paths that are synchronous by design. + T? value; } /// A pool of resources of type [T]. /// -/// Packs load onto the most-full resource under [maxConcurrentOperations] -/// (rather than spreading evenly across resources), opens a new resource -/// once existing ones are full, stops routing new work to a resource once an -/// operation on it throws, and garbage-collects idle resources past -/// [maxIdleResources]. +/// Packs load onto the most-full resource under its capacity (rather than +/// spreading evenly across resources), opens a new resource once existing +/// ones are full, stops routing new work to a resource once an operation on +/// it throws, and garbage-collects idle resources past [maxIdleResources]. +/// +/// Capacity is [maxConcurrentOperations], lowered to whatever limit a +/// resource reports for itself via `concurrencyLimitOf`. class ClientPool { ClientPool( Future Function() create, { required this.maxConcurrentOperations, required Future Function(T resource) destroy, this.maxIdleResources = 1, + int? Function(T resource)? concurrencyLimitOf, }) : _create = create, - _destroy = destroy; + _destroy = destroy, + _concurrencyLimitOf = concurrencyLimitOf; final Future Function() _create; final Future Function(T resource) _destroy; + + /// Reports a resource's own concurrency limit, or `null` if it imposes none. + /// + /// Consulted on every scheduling decision rather than cached, so a limit + /// the resource revises over its lifetime is picked up. + final int? Function(T resource)? _concurrencyLimitOf; + final int maxConcurrentOperations; final int maxIdleResources; @@ -55,7 +71,8 @@ class ClientPool { final pooled = _acquire(); pooled.inFlight++; try { - return await operation(await pooled.future); + final resource = pooled.value ??= await pooled.future; + return await operation(resource); } catch (_) { pooled.failed = true; rethrow; @@ -75,7 +92,7 @@ class ClientPool { _PooledResource? selected; for (final resource in _resources) { if (resource.failed) continue; - if (resource.inFlight < maxConcurrentOperations && + if (resource.inFlight < _capacityOf(resource) && (selected == null || resource.inFlight > selected.inFlight)) { selected = resource; } @@ -87,9 +104,23 @@ class ClientPool { return resource; } + /// How many operations [resource] may run at once. + /// + /// A resource that hasn't been created yet, or that reports no limit of its + /// own, is held to [maxConcurrentOperations]. + int _capacityOf(_PooledResource resource) { + final value = resource.value; + final limitOf = _concurrencyLimitOf; + if (value == null || limitOf == null) return maxConcurrentOperations; + final limit = limitOf(value); + return limit == null + ? maxConcurrentOperations + : min(maxConcurrentOperations, limit); + } + Future _collectIfIdle(_PooledResource resource) async { if (resource.inFlight > 0) return; - if (!resource.failed && !_hasExcessIdleCapacity) return; + if (!resource.failed && !_hasExcessIdleCapacity(resource)) return; _resources.remove(resource); try { @@ -100,19 +131,19 @@ class ClientPool { } } - bool get _hasExcessIdleCapacity { + // Measured in [resource]'s own capacity, so that resources holding fewer + // operations than [maxConcurrentOperations] aren't collected as excess the + // moment they drain - which would churn a connection per operation. + bool _hasExcessIdleCapacity(_PooledResource resource) { // Failed resources are never routed new work by _acquire(), so they // contribute no real idle capacity. final idleCapacity = _resources .map( - (resource) => - resource.failed - ? 0 - : maxConcurrentOperations - resource.inFlight, + (other) => other.failed ? 0 : _capacityOf(other) - other.inFlight, ) .sum; - return idleCapacity > maxIdleResources * maxConcurrentOperations; + return idleCapacity > maxIdleResources * _capacityOf(resource); } void _maybeCompleteDrain() { diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index c5bbd6d3d9..61ced7c7a9 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -44,6 +44,9 @@ class Http2Client extends BaseClient { /// The maximum number of concurrent HTTP/2 streams (i.e. requests) to /// multiplex onto a single connection before dialing another. + /// + /// An upper bound only: if a server says it accepts fewer concurrent + /// streams than this, that smaller number is used for its connections. final int maxStreamsPerConnection; /// The maximum number of idle connections to keep per host, per @@ -77,6 +80,9 @@ class Http2Client extends BaseClient { maxConcurrentOperations: maxStreamsPerConnection, maxIdleResources: maxIdleConnections, destroy: (transport) => transport.finish(), + // Never open more streams on a connection than its server allows, + // which it can revise at any point by sending a new SETTINGS frame. + concurrencyLimitOf: (transport) => transport.peerMaxConcurrentStreams, ), ); } diff --git a/pkgs/http2/test/client_pool_test.dart b/pkgs/http2/test/client_pool_test.dart index b8066c8ba7..9655316da1 100644 --- a/pkgs/http2/test/client_pool_test.dart +++ b/pkgs/http2/test/client_pool_test.dart @@ -10,13 +10,16 @@ import 'package:test/test.dart'; ClientPool _pool({ required int maxConcurrentOperations, int maxIdleResources = 1, + int? Function(int resource)? concurrencyLimitOf, + List? destroyed, }) { var nextId = 0; return ClientPool( () async => nextId++, maxConcurrentOperations: maxConcurrentOperations, maxIdleResources: maxIdleResources, - destroy: (_) async {}, + destroy: (resource) async => destroyed?.add(resource), + concurrencyLimitOf: concurrencyLimitOf, ); } @@ -157,6 +160,127 @@ void main() { expect(pool.size, 3); }); + test('honours-a-resources-own-lower-concurrency-limit', () async { + // The pool would allow 10 per resource; each resource only allows 2. + final pool = _pool( + maxConcurrentOperations: 10, + concurrencyLimitOf: (_) => 2, + ); + final completers = List.generate(3, (_) => Completer()); + final resourcesUsed = []; + + void run(int i) => unawaited( + pool.run((r) { + resourcesUsed.add(r); + return completers[i].future; + }), + ); + + run(0); + // The limit is only visible once resource 0 has been created, so let it + // resolve before dispatching work that has to respect it. + await pool.run((_) async {}); + run(1); + run(2); // Resource 0 is at its own limit of 2 - this opens resource 1. + await Future.value(); + + expect(resourcesUsed, [0, 0, 1]); + expect(pool.size, 2); + + for (final c in completers) { + c.complete(); + } + }); + + test('ignores-a-resource-limit-above-max-concurrent-operations', () async { + // A resource permitting more than the pool does must not raise the cap. + final pool = _pool( + maxConcurrentOperations: 1, + concurrencyLimitOf: (_) => 1000, + ); + final completers = List.generate(2, (_) => Completer()); + + unawaited(pool.run((_) => completers[0].future)); + await Future.value(); + unawaited(pool.run((_) => completers[1].future)); + await Future.value(); + + expect(pool.size, 2); + + for (final c in completers) { + c.complete(); + } + }); + + test('re-reads-a-resource-limit-that-changes', () async { + // Stands in for a server revising SETTINGS_MAX_CONCURRENT_STREAMS: the + // limit starts at 2 and drops to 1. + var limit = 2; + final pool = _pool( + maxConcurrentOperations: 10, + concurrencyLimitOf: (_) => limit, + ); + final completers = List.generate(3, (_) => Completer()); + final resourcesUsed = []; + + void run(int i) => unawaited( + pool.run((r) { + resourcesUsed.add(r); + return completers[i].future; + }), + ); + + run(0); + await pool.run((_) async {}); // Let resource 0 resolve. + run(1); // Still within the limit of 2, so resource 0 is reused. + await Future.value(); + expect(resourcesUsed, [0, 0]); + + limit = 1; + run(2); // Resource 0 is now over its lowered limit, so a new one opens. + await Future.value(); + expect(resourcesUsed, [0, 0, 1]); + + for (final c in completers) { + c.complete(); + } + }); + + test('keeps-a-limited-resource-that-is-merely-full', () async { + // Regression test for collecting a healthy resource as "excess idle" + // because its real capacity is below maxConcurrentOperations, which + // would churn one resource per operation. + final destroyed = []; + final pool = _pool( + maxConcurrentOperations: 10, + concurrencyLimitOf: (_) => 1, + destroyed: destroyed, + ); + final completers = List.generate(2, (_) => Completer()); + + final first = pool.run((_) => completers[0].future); + await Future.value(); + final second = pool.run((_) => completers[1].future); + await Future.value(); + expect(pool.size, 2); + + completers[0].complete(); + await first; + // Resource 1 is still busy, so resource 0 going idle doesn't make it + // excess - measuring its spare capacity against maxConcurrentOperations + // rather than its own limit of 1 would collect it here. + expect(destroyed, isEmpty); + expect(pool.size, 2); + + completers[1].complete(); + await second; + + // maxIdleResources is 1 and each resource holds 1, so exactly one of + // the two is excess - not both, and not one per operation. + expect(destroyed, hasLength(1)); + expect(pool.size, 1); + }); + test('rejects-operations-after-terminate', () async { final pool = _pool(maxConcurrentOperations: 1); diff --git a/pkgs/http2/test/http2_client_test.dart b/pkgs/http2/test/http2_client_test.dart index 471fb64513..580e1b47fd 100644 --- a/pkgs/http2/test/http2_client_test.dart +++ b/pkgs/http2/test/http2_client_test.dart @@ -19,8 +19,12 @@ SecurityContext _serverContext() => Future _bind() => MultiProtocolHttpServer.bind('localhost', 0, _serverContext()); -Http2Client _testClient({int maxStreamsPerConnection = 100}) => Http2Client( +Http2Client _testClient({ + int maxStreamsPerConnection = 100, + int maxIdleConnections = 1, +}) => Http2Client( maxStreamsPerConnection: maxStreamsPerConnection, + maxIdleConnections: maxIdleConnections, onBadCertificate: (_) => true, ); @@ -28,21 +32,33 @@ Http2Client _testClient({int maxStreamsPerConnection = 100}) => Http2Client( /// exposes each accepted [ServerTransportConnection], so a test can finish /// one connection gracefully while the server keeps listening for new ones. class _RawHttp2Server { - _RawHttp2Server._(this._socket) { + _RawHttp2Server._(this._socket, this._settings, this._responseDelay) { _socket.listen((socket) { - final connection = ServerTransportConnection.viaSocket(socket); + final connection = ServerTransportConnection.viaSocket( + socket, + settings: _settings, + ); connections.add(connection); - connection.incomingStreams.listen(_respondWith('ok')); + connection.incomingStreams.listen( + _respondWith('ok', delay: _responseDelay), + ); }); } - static Future<_RawHttp2Server> bind() async { + /// [settings] defaults to the same value `ServerTransportConnection` would + /// have applied on its own, so callers that don't care are unaffected. + static Future<_RawHttp2Server> bind({ + ServerSettings settings = const ServerSettings(concurrentStreamLimit: 1000), + Future? responseDelay, + }) async { final context = _serverContext()..setAlpnProtocols(['h2'], true); final socket = await SecureServerSocket.bind('localhost', 0, context); - return _RawHttp2Server._(socket); + return _RawHttp2Server._(socket, settings, responseDelay); } final SecureServerSocket _socket; + final ServerSettings _settings; + final Future? _responseDelay; final connections = []; int get port => _socket.port; @@ -179,6 +195,45 @@ void main() { await server.close(); }); + test('respects-server-advertised-max-concurrent-streams', () async { + // The server allows a single concurrent stream, well below the 100 this + // client would otherwise be willing to multiplex onto one connection. + final release = Completer(); + final server = await _RawHttp2Server.bind( + settings: const ServerSettings(concurrentStreamLimit: 1), + responseDelay: release.future, + ); + // Idle connections are kept generously, so the assertion below reflects + // whether a connection was evicted as *failed* rather than as excess. + final client = _testClient( + maxStreamsPerConnection: 100, + maxIdleConnections: 5, + ); + + final requestA = client.get( + Uri.parse('https://localhost:${server.port}/a'), + ); + // Let the first request dial and occupy the server's only stream slot, + // which also gives the peer's SETTINGS frame time to arrive. + await Future.delayed(const Duration(milliseconds: 100)); + final requestB = client.get( + Uri.parse('https://localhost:${server.port}/b'), + ); + await Future.delayed(const Duration(milliseconds: 100)); + + release.complete(); + final responses = await Future.wait([requestA, requestB]); + expect(responses.map((r) => r.statusCode), everyElement(200)); + expect(server.connections, hasLength(2)); + + // Both connections are healthy and idle, so both should survive: the + // first was merely at the server's stream limit, not broken. + expect(client.connectionCount, 2); + + await client.terminate(); + await server.close(); + }); + test('terminate-waits-for-in-flight-request', () async { final server = await _bind(); final release = Completer(); From d8d0ba71a0117be0bdb43b8923d4f5ea34a0c645 Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 18:39:24 +0100 Subject: [PATCH 18/27] fix(http2): correct response translation and error types in Http2Client --- pkgs/http2/lib/src/http2_client.dart | 148 +++++++++++++++++-- pkgs/http2/test/client_conformance_test.dart | 12 +- 2 files changed, 138 insertions(+), 22 deletions(-) diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 61ced7c7a9..070b48d06a 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -121,7 +121,7 @@ class Http2Client extends BaseClient { final stream = transport.makeRequest([ Header.ascii(':method', request.method), Header.ascii(':scheme', 'https'), - Header.ascii(':authority', request.url.host), + Header.ascii(':authority', _authorityOf(request.url)), Header.ascii(':path', path), // HTTP/2 requires lowercase header names (RFC 7540 8.1.2), and forbids // connection-specific header fields (RFC 7540 8.1.2.2) - which a @@ -145,14 +145,36 @@ class Http2Client extends BaseClient { (message) { if (message is HeadersStreamMessage) { for (final header in message.headers) { - final name = ascii.decode(header.name); - final value = ascii.decode(header.value); + // Not `ascii`: RFC 9113 permits arbitrary octets in field values, + // and a decode failure here would be thrown into this handler, + // where it becomes an uncaught async error rather than failing + // the request. + final name = latin1.decode(header.name); + final value = latin1.decode(header.value); if (name == ':status') { - if (!statusCompleter.isCompleted) { - statusCompleter.complete(int.parse(value)); + final status = int.tryParse(value); + if (status == null) { + if (!statusCompleter.isCompleted) { + statusCompleter.completeError( + ClientException( + 'Invalid HTTP/2 ":status" value "$value"', + request.url, + ), + ); + } + } else if (status >= 200 && !statusCompleter.isCompleted) { + // A 1xx is informational (RFC 9113 8.1) - the real response + // arrives in a later HEADERS frame. + statusCompleter.complete(status); } } else { - responseHeaders[name] = value; + // Repeated fields are joined, per the `package:http` convention; + // overwriting would silently drop e.g. a second `set-cookie`. + responseHeaders.update( + name, + (existing) => '$existing, $value', + ifAbsent: () => value, + ); } } } else if (message is DataStreamMessage) { @@ -162,7 +184,10 @@ class Http2Client extends BaseClient { onDone: () { if (!statusCompleter.isCompleted) { statusCompleter.completeError( - StateError('Stream closed before a response status was received'), + ClientException( + 'Stream closed before a response status was received', + request.url, + ), ); } if (!bodyController.isClosed) { @@ -170,10 +195,18 @@ class Http2Client extends BaseClient { } }, onError: (Object error, StackTrace stackTrace) { + // Wrapped here rather than in send(): by the time the body errors, + // send()'s future has usually already completed with the headers, so + // this is the only place a body-stream error can be given the type + // `Client` callers are promised. + final failure = + error is ClientException + ? error + : ClientException('$error', request.url); if (!statusCompleter.isCompleted) { - statusCompleter.completeError(error, stackTrace); + statusCompleter.completeError(failure, stackTrace); } - bodyController.addError(error, stackTrace); + bodyController.addError(failure, stackTrace); if (!bodyController.isClosed) { bodyController.close(); } @@ -185,7 +218,11 @@ class Http2Client extends BaseClient { return StreamedResponse( bodyController.stream, statusCode, - headers: responseHeaders, + contentLength: int.tryParse(responseHeaders['content-length'] ?? ''), + // Snapshotted: trailer HEADERS frames keep adding to responseHeaders + // after this response has been handed to the caller. + headers: Map.unmodifiable(responseHeaders), + reasonPhrase: _reasonPhrases[statusCode], request: request, ); } @@ -194,13 +231,19 @@ class Http2Client extends BaseClient { int get connectionCount => _pools.values.map((pool) => pool.size).sum; @override - Future send(BaseRequest request) { + Future send(BaseRequest request) async { if (_closed) { throw ClientException( 'HTTP request failed. Client is already closed.', request.url, ); } + if (request.url.scheme != 'https') { + throw ClientException( + 'Http2Client only supports https (got "${request.url.scheme}").', + request.url, + ); + } List? bodyBytes; Future attempt() => @@ -209,10 +252,21 @@ class Http2Client extends BaseClient { return _sendOverHttp2(transport, request, bodyBytes!); }); - return attempt().catchError( - (Object _) => attempt(), - test: (error) => error is _ConnectionClosedByPeer, - ); + return attempt() + .catchError( + (Object _) => attempt(), + test: (error) => error is _ConnectionClosedByPeer, + ) + // `Client` promises ClientException; without this a caller can see a + // SocketException, a HandshakeException, or one of our own internal + // error types. + .catchError( + (Object error, StackTrace stackTrace) => Error.throwWithStackTrace( + ClientException('$error', request.url), + stackTrace, + ), + test: (error) => error is! ClientException, + ); } /// Waits for in-flight requests to finish, then closes every connection. @@ -238,8 +292,72 @@ class Http2Client extends BaseClient { /// evicted it by the time the retry runs. class _ConnectionClosedByPeer implements Exception { const _ConnectionClosedByPeer(); + + @override + String toString() => + 'The pooled HTTP/2 connection was closed by the peer before this ' + 'request could be sent.'; } +/// HTTP/2 carries no reason phrase (RFC 9113 8.3.2 dropped it as redundant +/// with the status code), so one is derived from the status instead - the same +/// approach `package:cupertino_http` takes for NSURLSession. +const _reasonPhrases = { + 100: 'Continue', + 101: 'Switching Protocols', + 200: 'OK', + 201: 'Created', + 202: 'Accepted', + 203: 'Non-Authoritative Information', + 204: 'No Content', + 205: 'Reset Content', + 206: 'Partial Content', + 300: 'Multiple Choices', + 301: 'Moved Permanently', + 302: 'Found', + 303: 'See Other', + 304: 'Not Modified', + 305: 'Use Proxy', + 307: 'Temporary Redirect', + 308: 'Permanent Redirect', + 400: 'Bad Request', + 401: 'Unauthorized', + 402: 'Payment Required', + 403: 'Forbidden', + 404: 'Not Found', + 405: 'Method Not Allowed', + 406: 'Not Acceptable', + 407: 'Proxy Authentication Required', + 408: 'Request Time-out', + 409: 'Conflict', + 410: 'Gone', + 411: 'Length Required', + 412: 'Precondition Failed', + 413: 'Request Entity Too Large', + 414: 'Request-URI Too Long', + 415: 'Unsupported Media Type', + 416: 'Requested range not satisfiable', + 417: 'Expectation Failed', + 421: 'Misdirected Request', + 422: 'Unprocessable Entity', + 426: 'Upgrade Required', + 428: 'Precondition Required', + 429: 'Too Many Requests', + 431: 'Request Header Fields Too Large', + 500: 'Internal Server Error', + 501: 'Not Implemented', + 502: 'Bad Gateway', + 503: 'Service Unavailable', + 504: 'Gateway Time-out', + 505: 'Http Version not supported', + 511: 'Network Authentication Required', +}; + +/// RFC 9113 8.3.1: ":authority" carries the port unless it is the default for +/// the scheme - which is always https here, so 443. +String _authorityOf(Uri url) => + url.port == 443 ? url.host : '${url.host}:${url.port}'; + /// Header fields forbidden on an HTTP/2 stream (RFC 7540 8.1.2.2), plus /// `host` since `:authority` already carries what it would. const _connectionSpecificHeaders = { diff --git a/pkgs/http2/test/client_conformance_test.dart b/pkgs/http2/test/client_conformance_test.dart index bb4f291206..0294ffe77c 100644 --- a/pkgs/http2/test/client_conformance_test.dart +++ b/pkgs/http2/test/client_conformance_test.dart @@ -123,11 +123,10 @@ class Http2ProxyServer { } class ProxyRequest extends http.BaseRequest implements http.Abortable { - final http.BaseRequest _original; - @override - final Uri url; + // `url` is not redeclared here - BaseRequest's own constructor stores it. + ProxyRequest(this._original, Uri url) : super(_original.method, url); - ProxyRequest(this._original, this.url) : super(_original.method, url); + final http.BaseRequest _original; @override Future? get abortTrigger => @@ -271,9 +270,8 @@ void main() { // `SecureServerSocket`, which cannot be sent across isolates. testIsolate(clientFactory, canWorkInIsolates: false); - // TODO: Support cookies in Http2Client. - testRequestCookies(clientFactory, canSendCookieHeaders: false); - testResponseCookies(clientFactory, canReceiveSetCookieHeaders: false); + testRequestCookies(clientFactory, canSendCookieHeaders: true); + testResponseCookies(clientFactory, canReceiveSetCookieHeaders: true); // TODO: Implement request abort support in Http2Client. // testAbort( From a47fe1b590a6c6add70e461c73020b04729c3fe3 Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 18:42:16 +0100 Subject: [PATCH 19/27] fix(http2): make ClientPool.terminate idempotent and snapshot-safe --- pkgs/http2/lib/src/client_pool.dart | 14 ++++++-- pkgs/http2/lib/src/http2_client.dart | 8 +++-- pkgs/http2/test/client_pool_test.dart | 50 +++++++++++++++++++++++++++ 3 files changed, 66 insertions(+), 6 deletions(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index f0bd33e00d..4d0cd9eb5a 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -54,6 +54,7 @@ class ClientPool { final _resources = <_PooledResource>[]; var _terminated = false; Completer? _drained; + Future? _termination; /// The number of resources currently in the pool. int get size => _resources.length; @@ -155,7 +156,11 @@ class ClientPool { /// Waits for in-flight operations to finish, then destroys every /// resource in the pool. No further operations can run afterward. - Future terminate() async { + /// + /// Idempotent: concurrent and repeated calls all observe the same shutdown. + Future terminate() => _termination ??= _terminate(); + + Future _terminate() async { _terminated = true; if (opCount > 0) { @@ -163,7 +168,11 @@ class ClientPool { await _drained!.future; } - for (final resource in _resources) { + // Snapshotted and cleared before any `await`, so nothing is iterating + // _resources across a suspension point. + final resources = _resources.toList(); + _resources.clear(); + for (final resource in resources) { try { await _destroy(await resource.future); } catch (_) { @@ -171,6 +180,5 @@ class ClientPool { // rest from being destroyed. } } - _resources.clear(); } } diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 070b48d06a..3c8aa944c8 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -275,9 +275,11 @@ class Http2Client extends BaseClient { /// this can be awaited by callers who hold a concrete [Http2Client]. Future terminate() async { _closed = true; - for (final pool in _pools.values) { - await pool.terminate(); - } + // Snapshotted: a request already past the _closed check above - or its + // retry - can still add a pool for a new host while this is awaiting. + final pools = _pools.values.toList(); + _pools.clear(); + await Future.wait(pools.map((pool) => pool.terminate())); } @override diff --git a/pkgs/http2/test/client_pool_test.dart b/pkgs/http2/test/client_pool_test.dart index 9655316da1..b89e1c7b6c 100644 --- a/pkgs/http2/test/client_pool_test.dart +++ b/pkgs/http2/test/client_pool_test.dart @@ -289,6 +289,56 @@ void main() { expect(() => pool.run((_) async {}), throwsA(isA())); }); + test('terminate-is-idempotent', () async { + final destroyed = []; + final pool = _pool(maxConcurrentOperations: 2, destroyed: destroyed); + final completer = Completer(); + + unawaited(pool.run((_) => completer.future)); + + // Both callers must observe the same shutdown. Before, the second call + // replaced the completer the first was waiting on, so the first hung. + final first = pool.terminate(); + final second = pool.terminate(); + completer.complete(); + await Future.wait([first, second]); + + expect(destroyed, [0]); // Destroyed once, not once per terminate() call. + expect(pool.size, 0); + }); + + test('terminate-twice-does-not-throw-concurrent-modification', () async { + final pool = _pool(maxConcurrentOperations: 1); + final completers = List.generate(2, (_) => Completer()); + + final ops = [ + pool.run((_) => completers[0].future), + pool.run((_) => completers[1].future), + ]; + await Future.value(); + expect(pool.size, 2); + + final terminations = [pool.terminate(), pool.terminate()]; + for (final c in completers) { + c.complete(); + } + await Future.wait(ops); + + // Before, the two calls iterated _resources while the other cleared it. + await expectLater(Future.wait(terminations), completes); + }); + + test('terminate-after-terminate-returns-immediately', () async { + final destroyed = []; + final pool = _pool(maxConcurrentOperations: 1, destroyed: destroyed); + + await pool.run((_) async {}); + await pool.terminate(); + await pool.terminate(); + + expect(destroyed, hasLength(lessThanOrEqualTo(1))); + }); + test('waits-for-in-flight-operations-before-terminating', () async { final pool = _pool(maxConcurrentOperations: 1); final completer = Completer(); From bf3dfb4d496d32307f49f12eb3ec79551c65b159 Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 18:44:48 +0100 Subject: [PATCH 20/27] fix(http2): don't couple an operation to its resource's teardown --- pkgs/http2/lib/src/client_pool.dart | 41 ++++++++++++++++-------- pkgs/http2/test/client_pool_test.dart | 46 ++++++++++++++++++++++++++- 2 files changed, 72 insertions(+), 15 deletions(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index 4d0cd9eb5a..42152ef22a 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -52,6 +52,7 @@ class ClientPool { final int maxIdleResources; final _resources = <_PooledResource>[]; + final _pendingDestroys = >{}; var _terminated = false; Completer? _drained; Future? _termination; @@ -82,7 +83,7 @@ class ClientPool { if (_terminated) { _maybeCompleteDrain(); } else { - await _collectIfIdle(pooled); + _collectIfIdle(pooled); } } } @@ -119,17 +120,31 @@ class ClientPool { : min(maxConcurrentOperations, limit); } - Future _collectIfIdle(_PooledResource resource) async { + void _collectIfIdle(_PooledResource resource) { if (resource.inFlight > 0) return; if (!resource.failed && !_hasExcessIdleCapacity(resource)) return; _resources.remove(resource); - try { - await _destroy(await resource.future); - } catch (_) { - // Best-effort: a failure here must not shadow the caller's own - // request error, since this runs inside run()'s finally block. - } + _startDestroy(resource); + } + + /// Starts destroying [resource] without waiting for it, so that an operation + /// is never held up by its resource's teardown - `destroy` may itself wait + /// on unrelated work still running on that resource. + /// + /// Tracked in [_pendingDestroys] so [terminate] can still promise that every + /// resource has actually been destroyed by the time it completes. + void _startDestroy(_PooledResource resource) { + final value = resource.value; + // Called synchronously when the value is already known, so a resource is + // observably destroyed as soon as it leaves the pool. + final done = + value != null ? _destroy(value) : resource.future.then(_destroy); + // Best-effort: a failure here must neither shadow the caller's own request + // error nor surface as an unhandled async error. + final tracked = done.catchError((Object _) {}); + _pendingDestroys.add(tracked); + unawaited(tracked.whenComplete(() => _pendingDestroys.remove(tracked))); } // Measured in [resource]'s own capacity, so that resources holding fewer @@ -173,12 +188,10 @@ class ClientPool { final resources = _resources.toList(); _resources.clear(); for (final resource in resources) { - try { - await _destroy(await resource.future); - } catch (_) { - // Best-effort: one resource failing to close shouldn't stop the - // rest from being destroyed. - } + _startDestroy(resource); } + // Includes destroys started earlier by _collectIfIdle, so that when this + // completes every resource really has been destroyed. + await Future.wait(_pendingDestroys.toList()); } } diff --git a/pkgs/http2/test/client_pool_test.dart b/pkgs/http2/test/client_pool_test.dart index b89e1c7b6c..3886a608c0 100644 --- a/pkgs/http2/test/client_pool_test.dart +++ b/pkgs/http2/test/client_pool_test.dart @@ -12,13 +12,19 @@ ClientPool _pool({ int maxIdleResources = 1, int? Function(int resource)? concurrencyLimitOf, List? destroyed, + // When given, a resource's destroy doesn't finish until this does - stands + // in for `transport.finish()`, which waits on the connection's streams. + Future? destroyGate, }) { var nextId = 0; return ClientPool( () async => nextId++, maxConcurrentOperations: maxConcurrentOperations, maxIdleResources: maxIdleResources, - destroy: (resource) async => destroyed?.add(resource), + destroy: (resource) async { + destroyed?.add(resource); + if (destroyGate != null) await destroyGate; + }, concurrencyLimitOf: concurrencyLimitOf, ); } @@ -289,6 +295,44 @@ void main() { expect(() => pool.run((_) async {}), throwsA(isA())); }); + test('does-not-couple-an-operation-to-a-slow-destroy', () async { + // `destroy` never finishes. An operation that happens to trigger + // collection must still complete - before, run()'s finally awaited the + // destroy, so the operation's own future never settled. + final pool = _pool( + maxConcurrentOperations: 1, + maxIdleResources: 0, + destroyGate: Completer().future, + ); + + await expectLater(pool.run((_) async => 'done'), completion('done')); + }); + + test('terminate-waits-for-a-destroy-started-by-collection', () async { + final gate = Completer(); + final destroyed = []; + final pool = _pool( + maxConcurrentOperations: 1, + maxIdleResources: 0, + destroyed: destroyed, + destroyGate: gate.future, + ); + + // Completing this op collects the resource, starting a destroy that + // won't finish until the gate does. + await pool.run((_) async {}); + expect(destroyed, [0]); + + var terminated = false; + final termination = pool.terminate().then((_) => terminated = true); + await Future.value(); + expect(terminated, isFalse, reason: 'destroy has not finished yet'); + + gate.complete(); + await termination; + expect(terminated, isTrue); + }); + test('terminate-is-idempotent', () async { final destroyed = []; final pool = _pool(maxConcurrentOperations: 2, destroyed: destroyed); From 5e0b38dffbfaf6fae63d03d1c195e1442c9d36a9 Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 19:37:55 +0100 Subject: [PATCH 21/27] feat(http2): give ClientPool an explicit lease API --- pkgs/http2/lib/src/client_pool.dart | 70 +++++++++++++++++++++++---- pkgs/http2/test/client_pool_test.dart | 35 ++++++++++++++ 2 files changed, 95 insertions(+), 10 deletions(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index 42152ef22a..a8bc96f597 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -19,6 +19,34 @@ class _PooledResource { T? value; } +/// A claim on one concurrency slot of a pooled resource. +/// +/// Held from [ClientPool.acquire] until [release], which lets a slot outlive +/// the future that produced it - an HTTP/2 response, for instance, is returned +/// as soon as its headers arrive but keeps its stream open until the body ends. +class PoolLease { + PoolLease._(this._pool, this._resource, this.value); + + final ClientPool _pool; + final _PooledResource _resource; + + /// The resource this slot was claimed on. + final T value; + + var _released = false; + + /// Stops the pool routing new work to this resource. + void markFailed() => _resource.failed = true; + + /// Gives the slot back. Idempotent, so it is safe to call from several + /// terminal paths that may race. + void release() { + if (_released) return; + _released = true; + _pool._release(_resource); + } +} + /// A pool of resources of type [T]. /// /// Packs load onto the most-full resource under its capacity (rather than @@ -64,27 +92,49 @@ class ClientPool { @visibleForTesting int get opCount => _resources.map((resource) => resource.inFlight).sum; - /// Runs [operation] on an available (or newly created) resource. - Future run(Future Function(T resource) operation) async { + /// Claims a slot on an available (or newly created) resource. + /// + /// The caller owns the returned lease and must [PoolLease.release] it on + /// every path, errors included, or the slot leaks and [terminate] never + /// drains. Prefer [run] unless the slot has to outlive the future that + /// produced whatever the caller is returning. + Future> acquire() async { if (_terminated) { throw StateError('This pool has already been terminated.'); } final pooled = _acquire(); pooled.inFlight++; + final T value; try { - final resource = pooled.value ??= await pooled.future; - return await operation(resource); + value = pooled.value ??= await pooled.future; } catch (_) { pooled.failed = true; + _release(pooled); + rethrow; + } + return PoolLease._(this, pooled, value); + } + + void _release(_PooledResource resource) { + resource.inFlight--; + if (_terminated) { + _maybeCompleteDrain(); + } else { + _collectIfIdle(resource); + } + } + + /// Runs [operation] on an available (or newly created) resource. + Future run(Future Function(T resource) operation) async { + final lease = await acquire(); + try { + return await operation(lease.value); + } catch (_) { + lease.markFailed(); rethrow; } finally { - pooled.inFlight--; - if (_terminated) { - _maybeCompleteDrain(); - } else { - _collectIfIdle(pooled); - } + lease.release(); } } diff --git a/pkgs/http2/test/client_pool_test.dart b/pkgs/http2/test/client_pool_test.dart index 3886a608c0..19ff3a24c3 100644 --- a/pkgs/http2/test/client_pool_test.dart +++ b/pkgs/http2/test/client_pool_test.dart @@ -295,6 +295,41 @@ void main() { expect(() => pool.run((_) async {}), throwsA(isA())); }); + test('a-held-lease-keeps-its-slot', () async { + final pool = _pool(maxConcurrentOperations: 2); + + final lease = await pool.acquire(); + expect(pool.opCount, 1); + + lease.release(); + expect(pool.opCount, 0); + }); + + test('releasing-a-lease-twice-is-a-no-op', () async { + final pool = _pool(maxConcurrentOperations: 2); + + final lease = await pool.acquire(); + lease.release(); + lease.release(); + + // Not -1: release() is called from several terminal paths that can race. + expect(pool.opCount, 0); + }); + + test('a-failed-lease-stops-the-resource-being-reused', () async { + final pool = _pool(maxConcurrentOperations: 10); + final resourcesUsed = []; + + final lease = await pool.acquire(); + resourcesUsed.add(lease.value); + lease.markFailed(); + lease.release(); + + await pool.run((r) async => resourcesUsed.add(r)); + + expect(resourcesUsed, [0, 1]); + }); + test('does-not-couple-an-operation-to-a-slow-destroy', () async { // `destroy` never finishes. An operation that happens to trigger // collection must still complete - before, run()'s finally awaited the From 0ae4a1720236b748219d2f8c763c8b0e72eff6ad Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 19:53:12 +0100 Subject: [PATCH 22/27] fix(http2): hold a pool slot until the response body completes --- pkgs/http2/lib/src/http2_client.dart | 55 ++++++++++-- pkgs/http2/test/http2_client_test.dart | 112 +++++++++++++++++++++++-- 2 files changed, 153 insertions(+), 14 deletions(-) diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 3c8aa944c8..72171a4fa2 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -104,13 +104,18 @@ class Http2Client extends BaseClient { return ClientTransportConnection.viaSocket(socket); } - /// Sends [request] as a single HTTP/2 stream on [transport], translating - /// between `BaseRequest`/`StreamedResponse` and http2's frames. + /// Sends [request] as a single HTTP/2 stream on [lease]'s connection, + /// translating between `BaseRequest`/`StreamedResponse` and http2's frames. + /// + /// Returns once the response headers arrive, but takes ownership of [lease]: + /// the stream stays open while the body is delivered, so the slot is only + /// released once that stream reaches a terminal state. Future _sendOverHttp2( - ClientTransportConnection transport, + PoolLease lease, BaseRequest request, List bodyBytes, ) async { + final transport = lease.value; if (!transport.isOpen) throw const _ConnectionClosedByPeer(); // RFC 7540 8.1.2.3: ":path" must not be empty. @@ -137,7 +142,15 @@ class Http2Client extends BaseClient { final statusCompleter = Completer(); late final StreamSubscription subscription; final bodyController = StreamController>( - onCancel: () => subscription.cancel(), + onCancel: () { + // The consumer abandoned the body, so the stream is done with. + lease.release(); + // Reset it rather than just unsubscribing: an abandoned response is an + // abnormal end for the h2 stream, and without RST_STREAM neither end + // frees it - leaving `finish()` waiting on it forever. + stream.terminate(); + return subscription.cancel(); + }, ); final responseHeaders = {}; @@ -193,6 +206,8 @@ class Http2Client extends BaseClient { if (!bodyController.isClosed) { bodyController.close(); } + // The h2 stream has ended, so its slot can be reused. + lease.release(); }, onError: (Object error, StackTrace stackTrace) { // Wrapped here rather than in send(): by the time the body errors, @@ -210,6 +225,8 @@ class Http2Client extends BaseClient { if (!bodyController.isClosed) { bodyController.close(); } + // RST_STREAM or a connection error - the stream is over either way. + lease.release(); }, cancelOnError: true, ); @@ -245,12 +262,32 @@ class Http2Client extends BaseClient { ); } + // Finalized at most once, including across the retry below. List? bodyBytes; - Future attempt() => - _poolFor(request.url).run((transport) async { - bodyBytes ??= await request.finalize().toBytes(); - return _sendOverHttp2(transport, request, bodyBytes!); - }); + + Future attempt() async { + // Called before any `await` in this function, so the slot is reserved + // synchronously and a concurrent terminate() sees this request as + // in-flight rather than racing ahead of it. + final lease = await _poolFor(request.url).acquire(); + try { + bodyBytes ??= await request.finalize().toBytes(); + } catch (_) { + // Failing to read the caller's body says nothing about the + // connection, so release the slot without condemning it. + lease.release(); + rethrow; + } + try { + // Returns at the response headers; _sendOverHttp2 owns the lease from + // here and releases it when the h2 stream ends. + return await _sendOverHttp2(lease, request, bodyBytes!); + } catch (_) { + lease.markFailed(); + lease.release(); + rethrow; + } + } return attempt() .catchError( diff --git a/pkgs/http2/test/http2_client_test.dart b/pkgs/http2/test/http2_client_test.dart index 580e1b47fd..feb3bf0cf5 100644 --- a/pkgs/http2/test/http2_client_test.dart +++ b/pkgs/http2/test/http2_client_test.dart @@ -6,6 +6,7 @@ import 'dart:async'; import 'dart:convert' show ascii; import 'dart:io'; +import 'package:http/http.dart' show ClientException, Request; import 'package:http2/multiprotocol_server.dart'; import 'package:http2/src/http2_client.dart'; import 'package:http2/transport.dart'; @@ -32,7 +33,12 @@ Http2Client _testClient({ /// exposes each accepted [ServerTransportConnection], so a test can finish /// one connection gracefully while the server keeps listening for new ones. class _RawHttp2Server { - _RawHttp2Server._(this._socket, this._settings, this._responseDelay) { + _RawHttp2Server._( + this._socket, + this._settings, + this._responseDelay, + this._bodyGate, + ) { _socket.listen((socket) { final connection = ServerTransportConnection.viaSocket( socket, @@ -40,7 +46,7 @@ class _RawHttp2Server { ); connections.add(connection); connection.incomingStreams.listen( - _respondWith('ok', delay: _responseDelay), + _respondWith('ok', delay: _responseDelay, bodyGate: _bodyGate), ); }); } @@ -50,15 +56,17 @@ class _RawHttp2Server { static Future<_RawHttp2Server> bind({ ServerSettings settings = const ServerSettings(concurrentStreamLimit: 1000), Future? responseDelay, + Future? bodyGate, }) async { final context = _serverContext()..setAlpnProtocols(['h2'], true); final socket = await SecureServerSocket.bind('localhost', 0, context); - return _RawHttp2Server._(socket, settings, responseDelay); + return _RawHttp2Server._(socket, settings, responseDelay, bodyGate); } final SecureServerSocket _socket; final ServerSettings _settings; final Future? _responseDelay; + final Future? _bodyGate; final connections = []; int get port => _socket.port; @@ -72,9 +80,14 @@ class _RawHttp2Server { } /// Replies with [body] after waiting on [delay], if given. +/// +/// [bodyGate] holds the response open *after* its headers have been sent, so a +/// test can observe a request whose headers have arrived but whose stream is +/// still open. void Function(ServerTransportStream) _respondWith( String body, { Future? delay, + Future? bodyGate, }) { return (stream) async { final subscription = StreamIterator(stream.incomingMessages); @@ -86,8 +99,14 @@ void Function(ServerTransportStream) _respondWith( stream.outgoingMessages.add( HeadersStreamMessage([Header.ascii(':status', '200')]), ); - stream.outgoingMessages.add(DataStreamMessage(ascii.encode(body))); - await stream.outgoingMessages.close(); + if (bodyGate != null) await bodyGate; + try { + stream.outgoingMessages.add(DataStreamMessage(ascii.encode(body))); + await stream.outgoingMessages.close(); + } catch (_) { + // While the body was gated the client may have reset this stream, which + // a real server would likewise discover only on its next write. + } }; } @@ -234,6 +253,89 @@ void main() { await server.close(); }); + test('holds-a-pool-slot-until-the-response-body-completes', () async { + // Both responses send headers and then stall, so each request's h2 + // stream is still open once send() has returned. + final gate = Completer(); + final server = await _bind(); + server.startServing( + (request) {}, + expectAsync1(_respondWith('ok', bodyGate: gate.future), count: 2), + ); + + final client = _testClient(maxStreamsPerConnection: 1); + final url = Uri.parse('https://localhost:${server.port}/'); + final first = await client.send(Request('GET', url)); + final second = await client.send(Request('GET', url)); + + // The first request still occupies its connection's only slot, so the + // second must have been given a connection of its own. + expect(client.connectionCount, 2); + + gate.complete(); + expect(await first.stream.bytesToString(), 'ok'); + expect(await second.stream.bytesToString(), 'ok'); + + await client.terminate(); + await server.close(); + }); + + test('releases-the-slot-when-the-response-body-is-cancelled', () async { + // Only the first response is held open; the second answers normally. + final gate = Completer(); + final server = await _bind(); + var streamNr = 0; + server.startServing( + (request) {}, + expectAsync1((stream) { + final held = streamNr++ == 0 ? gate.future : null; + return _respondWith('ok', bodyGate: held)(stream); + }, count: 2), + ); + + final client = _testClient(maxStreamsPerConnection: 1); + final url = Uri.parse('https://localhost:${server.port}/'); + + final first = await client.send(Request('GET', url)); + // Abandoning the body must hand the slot back... + await first.stream.listen((_) {}).cancel(); + + // ...so this reuses the connection rather than dialing another. + final second = await client.get(url); + expect(second.statusCode, 200); + expect(client.connectionCount, 1); + + gate.complete(); + await client.terminate(); + await server.close(); + }); + + test('releases-the-slot-when-the-response-body-errors', () async { + // The body is held open, so the reset below lands mid-response rather + // than after the stream has already finished. + final gate = Completer(); + final server = await _RawHttp2Server.bind(bodyGate: gate.future); + final client = _testClient(maxStreamsPerConnection: 1); + final url = Uri.parse('https://localhost:${server.port}/'); + + final first = await client.send(Request('GET', url)); + await server.connections.single.terminate(); + await expectLater( + first.stream.drain(), + throwsA(isA()), + ); + + // That stream is dead; let anything dialed from here answer normally. + gate.complete(); + + // The slot was handed back, so a further request can proceed. + final second = await client.get(url); + expect(second.statusCode, 200); + + await client.terminate(); + await server.close(); + }); + test('terminate-waits-for-in-flight-request', () async { final server = await _bind(); final release = Completer(); From a1fef44fa913c602b76728f885ec5b9362f314a7 Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 19:59:43 +0100 Subject: [PATCH 23/27] fix(http2): don't multiplex onto a connection whose stream limit is unknown --- pkgs/http2/lib/src/client_pool.dart | 33 ++++++--- pkgs/http2/lib/src/http2_client.dart | 52 +++++++++++++- pkgs/http2/test/client_pool_test.dart | 66 +++++++++++++++++- pkgs/http2/test/http2_client_test.dart | 96 ++++++++++++++++++++++++++ 4 files changed, 235 insertions(+), 12 deletions(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index a8bc96f597..ea095043af 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -103,17 +103,28 @@ class ClientPool { throw StateError('This pool has already been terminated.'); } - final pooled = _acquire(); - pooled.inFlight++; - final T value; - try { - value = pooled.value ??= await pooled.future; - } catch (_) { - pooled.failed = true; + while (true) { + final pooled = _acquire(); + pooled.inFlight++; + final T value; + try { + value = pooled.value ??= await pooled.future; + } catch (_) { + pooled.failed = true; + _release(pooled); + rethrow; + } + + // A resource's own limit can only be read once it exists, so everything + // admitted while it was being created was admitted against the nominal + // cap. If that over-committed it, hand this slot back and pick again - + // the resource is resolved from here on, so the next pass either finds + // room elsewhere or creates a resource, and progress is guaranteed. + if (pooled.inFlight <= _capacityOf(pooled)) { + return PoolLease._(this, pooled, value); + } _release(pooled); - rethrow; } - return PoolLease._(this, pooled, value); } void _release(_PooledResource resource) { @@ -165,9 +176,11 @@ class ClientPool { final limitOf = _concurrencyLimitOf; if (value == null || limitOf == null) return maxConcurrentOperations; final limit = limitOf(value); + // Floored at one: a resource reporting zero would otherwise be unusable, + // and acquire()'s confirm loop would spin looking for room for it. return limit == null ? maxConcurrentOperations - : min(maxConcurrentOperations, limit); + : max(1, min(maxConcurrentOperations, limit)); } void _collectIfIdle(_PooledResource resource) { diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 72171a4fa2..7a3c04a192 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -5,6 +5,7 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; +import 'dart:typed_data'; import 'package:collection/collection.dart'; import 'package:http/http.dart'; @@ -35,6 +36,7 @@ class Http2Client extends BaseClient { Http2Client({ this.maxStreamsPerConnection = 100, this.maxIdleConnections = 1, + this.settingsTimeout = const Duration(seconds: 10), int maxConcurrentHandshakes = 50, SecurityContext? context, bool Function(X509Certificate certificate)? onBadCertificate, @@ -53,6 +55,14 @@ class Http2Client extends BaseClient { /// [ClientPool.maxIdleResources]. final int maxIdleConnections; + /// How long to wait for a freshly dialed connection's peer to send its + /// mandatory initial SETTINGS frame (RFC 7540 3.5). + /// + /// Until it arrives the peer's stream limit is unknown, so the connection + /// can't be safely multiplexed onto; a peer that never sends one is + /// abandoned rather than waited on forever. + final Duration settingsTimeout; + final SecurityContext? _context; // Forwarded to `SecureSocket.connect` as-is: returning `true` accepts a @@ -101,7 +111,47 @@ class Http2Client extends BaseClient { 'Server did not negotiate HTTP/2 (got ${socket.selectedProtocol})', ); } - return ClientTransportConnection.viaSocket(socket); + + // The peer's stream limit isn't knowable until its initial SETTINGS frame + // arrives, and multiplexing onto the connection before then risks + // exceeding a limit we haven't been told about yet. Wait for it - but + // `onInitialPeerSettingsReceived` is only ever completed successfully, so + // it can't be awaited on its own: a peer that connects and then goes + // quiet would hang the dial forever. Watch the byte stream for the + // connection dying, and cap the wait. + final died = Completer(); + final incoming = socket.transform( + StreamTransformer>.fromHandlers( + handleDone: (sink) { + if (!died.isCompleted) died.complete(); + sink.close(); + }, + handleError: (error, stackTrace, sink) { + if (!died.isCompleted) died.completeError(error, stackTrace); + sink.addError(error, stackTrace); + }, + ), + ); + // The connection usually outlives this race, leaving `died` unhandled. + unawaited(died.future.catchError((Object _) {})); + + final transport = ClientTransportConnection.viaStreams(incoming, socket); + try { + await Future.any([ + transport.onInitialPeerSettingsReceived, + died.future.then( + (_) => + throw ClientException( + 'The connection closed before the peer sent its initial ' + 'SETTINGS frame.', + ), + ), + ]).timeout(settingsTimeout); + } catch (_) { + await transport.terminate(); + rethrow; + } + return transport; } /// Sends [request] as a single HTTP/2 stream on [lease]'s connection, diff --git a/pkgs/http2/test/client_pool_test.dart b/pkgs/http2/test/client_pool_test.dart index 19ff3a24c3..91f59d8ef7 100644 --- a/pkgs/http2/test/client_pool_test.dart +++ b/pkgs/http2/test/client_pool_test.dart @@ -3,6 +3,7 @@ // BSD-style license that can be found in the LICENSE file. import 'dart:async'; +import 'dart:math'; import 'package:http2/src/client_pool.dart'; import 'package:test/test.dart'; @@ -15,10 +16,18 @@ ClientPool _pool({ // When given, a resource's destroy doesn't finish until this does - stands // in for `transport.finish()`, which waits on the connection's streams. Future? destroyGate, + // When given, creating a resource doesn't finish until this does, which is + // how a test controls the window where a resource exists but its own limit + // isn't knowable yet. + Future? dialGate, }) { var nextId = 0; return ClientPool( - () async => nextId++, + () async { + final id = nextId++; + if (dialGate != null) await dialGate; + return id; + }, maxConcurrentOperations: maxConcurrentOperations, maxIdleResources: maxIdleResources, destroy: (resource) async { @@ -295,6 +304,61 @@ void main() { expect(() => pool.run((_) async {}), throwsA(isA())); }); + test( + 'does-not-over-commit-a-resource-whose-limit-is-not-known-yet', + () async { + // Every resource only allows one operation, but that isn't knowable + // until it has been created - and the pool would otherwise admit work + // against the full nominal cap in the meantime. + final dialGate = Completer(); + final workGate = Completer(); + final pool = _pool( + maxConcurrentOperations: 100, + maxIdleResources: 10, + concurrencyLimitOf: (_) => 1, + dialGate: dialGate.future, + ); + + final inFlight = {}; + final peak = {}; + final ops = List.generate( + 8, + (_) => pool.run((r) async { + final now = (inFlight[r] ?? 0) + 1; + inFlight[r] = now; + peak[r] = max(peak[r] ?? 0, now); + await workGate.future; + inFlight[r] = inFlight[r]! - 1; + }), + ); + + // All eight are admitted before anything has been created, so this is + // where over-commitment would happen. + dialGate.complete(); + await pumpEventQueue(); + + expect( + peak.values, + everyElement(1), + reason: 'no resource may run more than its own limit of 1', + ); + expect(peak, hasLength(8)); + + workGate.complete(); + await Future.wait(ops); + }, + ); + + test('tolerates-a-resource-that-reports-a-zero-limit', () async { + // A limit of 0 must not make acquire() spin forever looking for room. + final pool = _pool( + maxConcurrentOperations: 10, + concurrencyLimitOf: (_) => 0, + ); + + await expectLater(pool.run((r) async => r), completion(0)); + }); + test('a-held-lease-keeps-its-slot', () async { final pool = _pool(maxConcurrentOperations: 2); diff --git a/pkgs/http2/test/http2_client_test.dart b/pkgs/http2/test/http2_client_test.dart index feb3bf0cf5..f299a3ce10 100644 --- a/pkgs/http2/test/http2_client_test.dart +++ b/pkgs/http2/test/http2_client_test.dart @@ -5,6 +5,7 @@ import 'dart:async'; import 'dart:convert' show ascii; import 'dart:io'; +import 'dart:math'; import 'package:http/http.dart' show ClientException, Request; import 'package:http2/multiprotocol_server.dart'; @@ -336,6 +337,101 @@ void main() { await server.close(); }); + test('does-not-exceed-the-server-stream-limit-on-a-cold-burst', () async { + // A burst arrives before any connection exists, so every request is + // admitted before the server's limit of 2 can possibly be known. + const streamLimit = 2; + const requestCount = 12; + final release = Completer(); + final context = _serverContext()..setAlpnProtocols(['h2'], true); + final socket = await SecureServerSocket.bind('localhost', 0, context); + + final active = {}; + final peak = {}; + socket.listen((raw) { + final connection = ServerTransportConnection.viaSocket( + raw, + settings: const ServerSettings(concurrentStreamLimit: streamLimit), + ); + connection.incomingStreams.listen((stream) async { + final now = (active[connection] ?? 0) + 1; + active[connection] = now; + peak[connection] = max(peak[connection] ?? 0, now); + + final messages = StreamIterator(stream.incomingMessages); + await messages.moveNext(); + while (await messages.moveNext()) {} + await release.future; + stream.outgoingMessages.add( + HeadersStreamMessage([Header.ascii(':status', '200')]), + ); + stream.outgoingMessages.add(DataStreamMessage(ascii.encode('ok'))); + await stream.outgoingMessages.close(); + + active[connection] = active[connection]! - 1; + }); + }); + + final client = _testClient(maxStreamsPerConnection: 100); + final url = Uri.parse('https://localhost:${socket.port}/'); + final requests = List.generate(requestCount, (_) => client.get(url)); + + // Let every request reach a connection before any of them completes. + await pumpEventQueue(); + release.complete(); + final responses = await Future.wait(requests); + + expect(responses.map((r) => r.statusCode), everyElement(200)); + expect( + peak.values, + everyElement(lessThanOrEqualTo(streamLimit)), + reason: 'no connection may carry more streams than the server allows', + ); + + await client.terminate(); + await socket.close(); + }); + + test('fails-the-dial-when-the-peer-closes-before-settings', () async { + // Negotiates h2 and then hangs up without sending its mandatory initial + // SETTINGS frame (RFC 7540 3.5). + final context = _serverContext()..setAlpnProtocols(['h2'], true); + final socket = await SecureServerSocket.bind('localhost', 0, context); + socket.listen((connection) => connection.destroy()); + + final client = _testClient(); + await expectLater( + client.get(Uri.parse('https://localhost:${socket.port}/')), + throwsA(isA()), + ); + + await client.terminate(); + await socket.close(); + }); + + test('fails-the-dial-when-the-peer-never-sends-settings', () async { + // Accepts and then stays silent, so only the timeout can end this. + final context = _serverContext()..setAlpnProtocols(['h2'], true); + final socket = await SecureServerSocket.bind('localhost', 0, context); + final held = []; + socket.listen(held.add); + + final client = Http2Client( + onBadCertificate: (_) => true, + settingsTimeout: const Duration(milliseconds: 200), + ); + await expectLater( + client.get(Uri.parse('https://localhost:${socket.port}/')), + throwsA(isA()), + ); + + await client.terminate(); + for (final connection in held) { + connection.destroy(); + } + await socket.close(); + }); + test('terminate-waits-for-in-flight-request', () async { final server = await _bind(); final release = Completer(); From 92987b65223a285723126409a979c073f83cc0fe Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 20:01:16 +0100 Subject: [PATCH 24/27] docs(http2): record the pooling behaviour changes and known isOpen race --- pkgs/http2/CHANGELOG.md | 5 ++++- pkgs/http2/lib/src/http2_client.dart | 16 +++++++++++++--- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index 5dc9db4b1e..29b86bb96f 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md @@ -2,7 +2,10 @@ - Gracefully handle receiving headers on a stream that the client has canceled. (#1799) - Add `Http2Client` (`package:http2/client.dart`), a pooled, multiplexed - `package:http` `Client` backed by HTTP/2 connections. + `package:http` `Client` backed by HTTP/2 connections. A connection is never + given more concurrent streams than its server allows, and is not used at all + until that server has sent its initial SETTINGS frame - see + `Http2Client.settingsTimeout` for how long that is waited for. - Add `ClientTransportConnection.peerMaxConcurrentStreams`, exposing the peer's most recently advertised `SETTINGS_MAX_CONCURRENT_STREAMS`. diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 7a3c04a192..4fa74a46f9 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -18,9 +18,10 @@ import 'client_pool.dart'; /// /// Every request is sent as its own HTTP/2 stream on a shared connection - /// see [ClientPool] - dialed per `host:port` via -/// `SecureSocket.connect(..., supportedProtocols: ['h2'])`. Once a -/// connection's [maxStreamsPerConnection] concurrent streams are in use, a -/// new connection is dialed rather than queuing behind the existing one. +/// `SecureSocket.connect(..., supportedProtocols: ['h2'])`. Once a connection +/// is carrying as many concurrent streams as it may - the lower of +/// [maxStreamsPerConnection] and the server's own advertised limit - a new +/// connection is dialed rather than queuing behind the existing one. /// /// Being multi-host makes this safe to use as a general-purpose transport - /// for example as the `baseClient` passed to `googleapis_auth`'s client @@ -166,6 +167,11 @@ class Http2Client extends BaseClient { List bodyBytes, ) async { final transport = lease.value; + // `isOpen` conflates "the peer went away" with "this connection is + // momentarily at its stream limit", so a healthy connection can still be + // condemned here in the few microtasks between a lease being released and + // http2 retiring the stream it belonged to. Rare, and costs one retry; + // separating the two would mean splitting `isOpen` in the public API. if (!transport.isOpen) throw const _ConnectionClosedByPeer(); // RFC 7540 8.1.2.3: ":path" must not be empty. @@ -360,6 +366,10 @@ class Http2Client extends BaseClient { /// /// Unlike [close] (constrained by `http.Client`'s synchronous signature), /// this can be awaited by callers who hold a concrete [Http2Client]. + /// + /// A request counts as in-flight until its response body ends or is + /// cancelled, so a caller holding a response it never reads will hold this + /// up. [close] does not await this, so it can never block on that. Future terminate() async { _closed = true; // Snapshotted: a request already past the _closed check above - or its From ff10a97dda3917acc67fa93b8a5f950b10229d65 Mon Sep 17 00:00:00 2001 From: demolaf Date: Wed, 5 Aug 2026 20:53:04 +0100 Subject: [PATCH 25/27] style(http2): drop explanatory inline comments --- pkgs/http2/lib/src/client_pool.dart | 20 ---------- pkgs/http2/lib/src/http2_client.dart | 53 -------------------------- pkgs/http2/test/client_pool_test.dart | 34 ----------------- pkgs/http2/test/http2_client_test.dart | 33 +--------------- 4 files changed, 1 insertion(+), 139 deletions(-) diff --git a/pkgs/http2/lib/src/client_pool.dart b/pkgs/http2/lib/src/client_pool.dart index ea095043af..b33e99c3f9 100644 --- a/pkgs/http2/lib/src/client_pool.dart +++ b/pkgs/http2/lib/src/client_pool.dart @@ -115,11 +115,6 @@ class ClientPool { rethrow; } - // A resource's own limit can only be read once it exists, so everything - // admitted while it was being created was admitted against the nominal - // cap. If that over-committed it, hand this slot back and pick again - - // the resource is resolved from here on, so the next pass either finds - // room elsewhere or creates a resource, and progress is guaranteed. if (pooled.inFlight <= _capacityOf(pooled)) { return PoolLease._(this, pooled, value); } @@ -176,8 +171,6 @@ class ClientPool { final limitOf = _concurrencyLimitOf; if (value == null || limitOf == null) return maxConcurrentOperations; final limit = limitOf(value); - // Floored at one: a resource reporting zero would otherwise be unusable, - // and acquire()'s confirm loop would spin looking for room for it. return limit == null ? maxConcurrentOperations : max(1, min(maxConcurrentOperations, limit)); @@ -199,23 +192,14 @@ class ClientPool { /// resource has actually been destroyed by the time it completes. void _startDestroy(_PooledResource resource) { final value = resource.value; - // Called synchronously when the value is already known, so a resource is - // observably destroyed as soon as it leaves the pool. final done = value != null ? _destroy(value) : resource.future.then(_destroy); - // Best-effort: a failure here must neither shadow the caller's own request - // error nor surface as an unhandled async error. final tracked = done.catchError((Object _) {}); _pendingDestroys.add(tracked); unawaited(tracked.whenComplete(() => _pendingDestroys.remove(tracked))); } - // Measured in [resource]'s own capacity, so that resources holding fewer - // operations than [maxConcurrentOperations] aren't collected as excess the - // moment they drain - which would churn a connection per operation. bool _hasExcessIdleCapacity(_PooledResource resource) { - // Failed resources are never routed new work by _acquire(), so they - // contribute no real idle capacity. final idleCapacity = _resources .map( @@ -246,15 +230,11 @@ class ClientPool { await _drained!.future; } - // Snapshotted and cleared before any `await`, so nothing is iterating - // _resources across a suspension point. final resources = _resources.toList(); _resources.clear(); for (final resource in resources) { _startDestroy(resource); } - // Includes destroys started earlier by _collectIfIdle, so that when this - // completes every resource really has been destroyed. await Future.wait(_pendingDestroys.toList()); } } diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index 4fa74a46f9..f785e15a94 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -91,8 +91,6 @@ class Http2Client extends BaseClient { maxConcurrentOperations: maxStreamsPerConnection, maxIdleResources: maxIdleConnections, destroy: (transport) => transport.finish(), - // Never open more streams on a connection than its server allows, - // which it can revise at any point by sending a new SETTINGS frame. concurrencyLimitOf: (transport) => transport.peerMaxConcurrentStreams, ), ); @@ -113,13 +111,6 @@ class Http2Client extends BaseClient { ); } - // The peer's stream limit isn't knowable until its initial SETTINGS frame - // arrives, and multiplexing onto the connection before then risks - // exceeding a limit we haven't been told about yet. Wait for it - but - // `onInitialPeerSettingsReceived` is only ever completed successfully, so - // it can't be awaited on its own: a peer that connects and then goes - // quiet would hang the dial forever. Watch the byte stream for the - // connection dying, and cap the wait. final died = Completer(); final incoming = socket.transform( StreamTransformer>.fromHandlers( @@ -133,7 +124,6 @@ class Http2Client extends BaseClient { }, ), ); - // The connection usually outlives this race, leaving `died` unhandled. unawaited(died.future.catchError((Object _) {})); final transport = ClientTransportConnection.viaStreams(incoming, socket); @@ -167,14 +157,8 @@ class Http2Client extends BaseClient { List bodyBytes, ) async { final transport = lease.value; - // `isOpen` conflates "the peer went away" with "this connection is - // momentarily at its stream limit", so a healthy connection can still be - // condemned here in the few microtasks between a lease being released and - // http2 retiring the stream it belonged to. Rare, and costs one retry; - // separating the two would mean splitting `isOpen` in the public API. if (!transport.isOpen) throw const _ConnectionClosedByPeer(); - // RFC 7540 8.1.2.3: ":path" must not be empty. final rawPath = request.url.path.isEmpty ? '/' : request.url.path; final path = request.url.hasQuery ? '$rawPath?${request.url.query}' : rawPath; @@ -184,10 +168,6 @@ class Http2Client extends BaseClient { Header.ascii(':scheme', 'https'), Header.ascii(':authority', _authorityOf(request.url)), Header.ascii(':path', path), - // HTTP/2 requires lowercase header names (RFC 7540 8.1.2), and forbids - // connection-specific header fields (RFC 7540 8.1.2.2) - which a - // request built for an HTTP/1.1-oriented client might still set. - // `host` is dropped too, since `:authority` already carries it. for (final entry in request.headers.entries) if (!_connectionSpecificHeaders.contains(entry.key.toLowerCase())) Header.ascii(entry.key.toLowerCase(), entry.value), @@ -199,11 +179,7 @@ class Http2Client extends BaseClient { late final StreamSubscription subscription; final bodyController = StreamController>( onCancel: () { - // The consumer abandoned the body, so the stream is done with. lease.release(); - // Reset it rather than just unsubscribing: an abandoned response is an - // abnormal end for the h2 stream, and without RST_STREAM neither end - // frees it - leaving `finish()` waiting on it forever. stream.terminate(); return subscription.cancel(); }, @@ -214,10 +190,6 @@ class Http2Client extends BaseClient { (message) { if (message is HeadersStreamMessage) { for (final header in message.headers) { - // Not `ascii`: RFC 9113 permits arbitrary octets in field values, - // and a decode failure here would be thrown into this handler, - // where it becomes an uncaught async error rather than failing - // the request. final name = latin1.decode(header.name); final value = latin1.decode(header.value); if (name == ':status') { @@ -232,13 +204,9 @@ class Http2Client extends BaseClient { ); } } else if (status >= 200 && !statusCompleter.isCompleted) { - // A 1xx is informational (RFC 9113 8.1) - the real response - // arrives in a later HEADERS frame. statusCompleter.complete(status); } } else { - // Repeated fields are joined, per the `package:http` convention; - // overwriting would silently drop e.g. a second `set-cookie`. responseHeaders.update( name, (existing) => '$existing, $value', @@ -262,14 +230,9 @@ class Http2Client extends BaseClient { if (!bodyController.isClosed) { bodyController.close(); } - // The h2 stream has ended, so its slot can be reused. lease.release(); }, onError: (Object error, StackTrace stackTrace) { - // Wrapped here rather than in send(): by the time the body errors, - // send()'s future has usually already completed with the headers, so - // this is the only place a body-stream error can be given the type - // `Client` callers are promised. final failure = error is ClientException ? error @@ -281,7 +244,6 @@ class Http2Client extends BaseClient { if (!bodyController.isClosed) { bodyController.close(); } - // RST_STREAM or a connection error - the stream is over either way. lease.release(); }, cancelOnError: true, @@ -292,8 +254,6 @@ class Http2Client extends BaseClient { bodyController.stream, statusCode, contentLength: int.tryParse(responseHeaders['content-length'] ?? ''), - // Snapshotted: trailer HEADERS frames keep adding to responseHeaders - // after this response has been handed to the caller. headers: Map.unmodifiable(responseHeaders), reasonPhrase: _reasonPhrases[statusCode], request: request, @@ -318,25 +278,17 @@ class Http2Client extends BaseClient { ); } - // Finalized at most once, including across the retry below. List? bodyBytes; Future attempt() async { - // Called before any `await` in this function, so the slot is reserved - // synchronously and a concurrent terminate() sees this request as - // in-flight rather than racing ahead of it. final lease = await _poolFor(request.url).acquire(); try { bodyBytes ??= await request.finalize().toBytes(); } catch (_) { - // Failing to read the caller's body says nothing about the - // connection, so release the slot without condemning it. lease.release(); rethrow; } try { - // Returns at the response headers; _sendOverHttp2 owns the lease from - // here and releases it when the h2 stream ends. return await _sendOverHttp2(lease, request, bodyBytes!); } catch (_) { lease.markFailed(); @@ -350,9 +302,6 @@ class Http2Client extends BaseClient { (Object _) => attempt(), test: (error) => error is _ConnectionClosedByPeer, ) - // `Client` promises ClientException; without this a caller can see a - // SocketException, a HandshakeException, or one of our own internal - // error types. .catchError( (Object error, StackTrace stackTrace) => Error.throwWithStackTrace( ClientException('$error', request.url), @@ -372,8 +321,6 @@ class Http2Client extends BaseClient { /// up. [close] does not await this, so it can never block on that. Future terminate() async { _closed = true; - // Snapshotted: a request already past the _closed check above - or its - // retry - can still add a pool for a new host while this is awaiting. final pools = _pools.values.toList(); _pools.clear(); await Future.wait(pools.map((pool) => pool.terminate())); diff --git a/pkgs/http2/test/client_pool_test.dart b/pkgs/http2/test/client_pool_test.dart index 91f59d8ef7..a970b7ac10 100644 --- a/pkgs/http2/test/client_pool_test.dart +++ b/pkgs/http2/test/client_pool_test.dart @@ -13,12 +13,7 @@ ClientPool _pool({ int maxIdleResources = 1, int? Function(int resource)? concurrencyLimitOf, List? destroyed, - // When given, a resource's destroy doesn't finish until this does - stands - // in for `transport.finish()`, which waits on the connection's streams. Future? destroyGate, - // When given, creating a resource doesn't finish until this does, which is - // how a test controls the window where a resource exists but its own limit - // isn't knowable yet. Future? dialGate, }) { var nextId = 0; @@ -176,7 +171,6 @@ void main() { }); test('honours-a-resources-own-lower-concurrency-limit', () async { - // The pool would allow 10 per resource; each resource only allows 2. final pool = _pool( maxConcurrentOperations: 10, concurrencyLimitOf: (_) => 2, @@ -192,8 +186,6 @@ void main() { ); run(0); - // The limit is only visible once resource 0 has been created, so let it - // resolve before dispatching work that has to respect it. await pool.run((_) async {}); run(1); run(2); // Resource 0 is at its own limit of 2 - this opens resource 1. @@ -208,7 +200,6 @@ void main() { }); test('ignores-a-resource-limit-above-max-concurrent-operations', () async { - // A resource permitting more than the pool does must not raise the cap. final pool = _pool( maxConcurrentOperations: 1, concurrencyLimitOf: (_) => 1000, @@ -228,8 +219,6 @@ void main() { }); test('re-reads-a-resource-limit-that-changes', () async { - // Stands in for a server revising SETTINGS_MAX_CONCURRENT_STREAMS: the - // limit starts at 2 and drops to 1. var limit = 2; final pool = _pool( maxConcurrentOperations: 10, @@ -262,9 +251,6 @@ void main() { }); test('keeps-a-limited-resource-that-is-merely-full', () async { - // Regression test for collecting a healthy resource as "excess idle" - // because its real capacity is below maxConcurrentOperations, which - // would churn one resource per operation. final destroyed = []; final pool = _pool( maxConcurrentOperations: 10, @@ -281,17 +267,12 @@ void main() { completers[0].complete(); await first; - // Resource 1 is still busy, so resource 0 going idle doesn't make it - // excess - measuring its spare capacity against maxConcurrentOperations - // rather than its own limit of 1 would collect it here. expect(destroyed, isEmpty); expect(pool.size, 2); completers[1].complete(); await second; - // maxIdleResources is 1 and each resource holds 1, so exactly one of - // the two is excess - not both, and not one per operation. expect(destroyed, hasLength(1)); expect(pool.size, 1); }); @@ -307,9 +288,6 @@ void main() { test( 'does-not-over-commit-a-resource-whose-limit-is-not-known-yet', () async { - // Every resource only allows one operation, but that isn't knowable - // until it has been created - and the pool would otherwise admit work - // against the full nominal cap in the meantime. final dialGate = Completer(); final workGate = Completer(); final pool = _pool( @@ -332,8 +310,6 @@ void main() { }), ); - // All eight are admitted before anything has been created, so this is - // where over-commitment would happen. dialGate.complete(); await pumpEventQueue(); @@ -350,7 +326,6 @@ void main() { ); test('tolerates-a-resource-that-reports-a-zero-limit', () async { - // A limit of 0 must not make acquire() spin forever looking for room. final pool = _pool( maxConcurrentOperations: 10, concurrencyLimitOf: (_) => 0, @@ -376,7 +351,6 @@ void main() { lease.release(); lease.release(); - // Not -1: release() is called from several terminal paths that can race. expect(pool.opCount, 0); }); @@ -395,9 +369,6 @@ void main() { }); test('does-not-couple-an-operation-to-a-slow-destroy', () async { - // `destroy` never finishes. An operation that happens to trigger - // collection must still complete - before, run()'s finally awaited the - // destroy, so the operation's own future never settled. final pool = _pool( maxConcurrentOperations: 1, maxIdleResources: 0, @@ -417,8 +388,6 @@ void main() { destroyGate: gate.future, ); - // Completing this op collects the resource, starting a destroy that - // won't finish until the gate does. await pool.run((_) async {}); expect(destroyed, [0]); @@ -439,8 +408,6 @@ void main() { unawaited(pool.run((_) => completer.future)); - // Both callers must observe the same shutdown. Before, the second call - // replaced the completer the first was waiting on, so the first hung. final first = pool.terminate(); final second = pool.terminate(); completer.complete(); @@ -467,7 +434,6 @@ void main() { } await Future.wait(ops); - // Before, the two calls iterated _resources while the other cleared it. await expectLater(Future.wait(terminations), completes); }); diff --git a/pkgs/http2/test/http2_client_test.dart b/pkgs/http2/test/http2_client_test.dart index f299a3ce10..3581d4c839 100644 --- a/pkgs/http2/test/http2_client_test.dart +++ b/pkgs/http2/test/http2_client_test.dart @@ -104,10 +104,7 @@ void Function(ServerTransportStream) _respondWith( try { stream.outgoingMessages.add(DataStreamMessage(ascii.encode(body))); await stream.outgoingMessages.close(); - } catch (_) { - // While the body was gated the client may have reset this stream, which - // a real server would likewise discover only on its next write. - } + } catch (_) {} }; } @@ -173,9 +170,6 @@ void main() { final requestA = client.get( Uri.parse('https://localhost:${server.port}/a'), ); - // Give the pool a chance to dial and dispatch the first request - // before the second one arrives, so it's guaranteed to land on an - // already-full connection rather than racing to share it. await Future.delayed(const Duration(milliseconds: 50)); final requestB = client.get( Uri.parse('https://localhost:${server.port}/b'), @@ -216,15 +210,11 @@ void main() { }); test('respects-server-advertised-max-concurrent-streams', () async { - // The server allows a single concurrent stream, well below the 100 this - // client would otherwise be willing to multiplex onto one connection. final release = Completer(); final server = await _RawHttp2Server.bind( settings: const ServerSettings(concurrentStreamLimit: 1), responseDelay: release.future, ); - // Idle connections are kept generously, so the assertion below reflects - // whether a connection was evicted as *failed* rather than as excess. final client = _testClient( maxStreamsPerConnection: 100, maxIdleConnections: 5, @@ -233,8 +223,6 @@ void main() { final requestA = client.get( Uri.parse('https://localhost:${server.port}/a'), ); - // Let the first request dial and occupy the server's only stream slot, - // which also gives the peer's SETTINGS frame time to arrive. await Future.delayed(const Duration(milliseconds: 100)); final requestB = client.get( Uri.parse('https://localhost:${server.port}/b'), @@ -246,8 +234,6 @@ void main() { expect(responses.map((r) => r.statusCode), everyElement(200)); expect(server.connections, hasLength(2)); - // Both connections are healthy and idle, so both should survive: the - // first was merely at the server's stream limit, not broken. expect(client.connectionCount, 2); await client.terminate(); @@ -255,8 +241,6 @@ void main() { }); test('holds-a-pool-slot-until-the-response-body-completes', () async { - // Both responses send headers and then stall, so each request's h2 - // stream is still open once send() has returned. final gate = Completer(); final server = await _bind(); server.startServing( @@ -269,8 +253,6 @@ void main() { final first = await client.send(Request('GET', url)); final second = await client.send(Request('GET', url)); - // The first request still occupies its connection's only slot, so the - // second must have been given a connection of its own. expect(client.connectionCount, 2); gate.complete(); @@ -282,7 +264,6 @@ void main() { }); test('releases-the-slot-when-the-response-body-is-cancelled', () async { - // Only the first response is held open; the second answers normally. final gate = Completer(); final server = await _bind(); var streamNr = 0; @@ -298,10 +279,8 @@ void main() { final url = Uri.parse('https://localhost:${server.port}/'); final first = await client.send(Request('GET', url)); - // Abandoning the body must hand the slot back... await first.stream.listen((_) {}).cancel(); - // ...so this reuses the connection rather than dialing another. final second = await client.get(url); expect(second.statusCode, 200); expect(client.connectionCount, 1); @@ -312,8 +291,6 @@ void main() { }); test('releases-the-slot-when-the-response-body-errors', () async { - // The body is held open, so the reset below lands mid-response rather - // than after the stream has already finished. final gate = Completer(); final server = await _RawHttp2Server.bind(bodyGate: gate.future); final client = _testClient(maxStreamsPerConnection: 1); @@ -326,10 +303,8 @@ void main() { throwsA(isA()), ); - // That stream is dead; let anything dialed from here answer normally. gate.complete(); - // The slot was handed back, so a further request can proceed. final second = await client.get(url); expect(second.statusCode, 200); @@ -338,8 +313,6 @@ void main() { }); test('does-not-exceed-the-server-stream-limit-on-a-cold-burst', () async { - // A burst arrives before any connection exists, so every request is - // admitted before the server's limit of 2 can possibly be known. const streamLimit = 2; const requestCount = 12; final release = Completer(); @@ -376,7 +349,6 @@ void main() { final url = Uri.parse('https://localhost:${socket.port}/'); final requests = List.generate(requestCount, (_) => client.get(url)); - // Let every request reach a connection before any of them completes. await pumpEventQueue(); release.complete(); final responses = await Future.wait(requests); @@ -393,8 +365,6 @@ void main() { }); test('fails-the-dial-when-the-peer-closes-before-settings', () async { - // Negotiates h2 and then hangs up without sending its mandatory initial - // SETTINGS frame (RFC 7540 3.5). final context = _serverContext()..setAlpnProtocols(['h2'], true); final socket = await SecureServerSocket.bind('localhost', 0, context); socket.listen((connection) => connection.destroy()); @@ -410,7 +380,6 @@ void main() { }); test('fails-the-dial-when-the-peer-never-sends-settings', () async { - // Accepts and then stays silent, so only the timeout can end this. final context = _serverContext()..setAlpnProtocols(['h2'], true); final socket = await SecureServerSocket.bind('localhost', 0, context); final held = []; From 7557351c9130789c451cb27039e473381a9e4ffa Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 6 Aug 2026 09:36:38 +0100 Subject: [PATCH 26/27] refactor(http2): keep the peer stream limit off the public transport API --- pkgs/http2/CHANGELOG.md | 2 -- pkgs/http2/lib/src/connection.dart | 8 +++++++- pkgs/http2/lib/src/http2_client.dart | 17 +++++++++++------ pkgs/http2/lib/transport.dart | 5 ----- 4 files changed, 18 insertions(+), 14 deletions(-) diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index 29b86bb96f..ca6fed69b8 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md @@ -6,8 +6,6 @@ given more concurrent streams than its server allows, and is not used at all until that server has sent its initial SETTINGS frame - see `Http2Client.settingsTimeout` for how long that is waited for. -- Add `ClientTransportConnection.peerMaxConcurrentStreams`, exposing the - peer's most recently advertised `SETTINGS_MAX_CONCURRENT_STREAMS`. ## 3.0.0 diff --git a/pkgs/http2/lib/src/connection.dart b/pkgs/http2/lib/src/connection.dart index eb866cb975..355b5c3b99 100644 --- a/pkgs/http2/lib/src/connection.dart +++ b/pkgs/http2/lib/src/connection.dart @@ -513,7 +513,13 @@ class ClientConnection extends Connection implements ClientTransportConnection { bool get isOpen => !_state.isFinishing && !_state.isTerminated && _streams.canOpenStream; - @override + /// The maximum number of concurrent streams the peer currently allows, per + /// its most recent SETTINGS_MAX_CONCURRENT_STREAMS (RFC 7540 6.5.2), or + /// `null` if it hasn't advertised a limit. + /// + /// Deliberately not on [ClientTransportConnection]: that class can only be + /// subtyped with `implements`, so adding a member to it would break every + /// downstream implementation. int? get peerMaxConcurrentStreams => _settingsHandler.peerSettings.maxConcurrentStreams; diff --git a/pkgs/http2/lib/src/http2_client.dart b/pkgs/http2/lib/src/http2_client.dart index f785e15a94..7fb385c75c 100644 --- a/pkgs/http2/lib/src/http2_client.dart +++ b/pkgs/http2/lib/src/http2_client.dart @@ -13,6 +13,7 @@ import 'package:pool/pool.dart'; import '../transport.dart'; import 'client_pool.dart'; +import 'connection.dart'; /// A pooled, multiplexed `http.Client` backed by HTTP/2 connections. /// @@ -77,16 +78,16 @@ class Http2Client extends BaseClient { // needing. final Pool _handshakeGate; - final _pools = >{}; + final _pools = >{}; var _closed = false; // Synchronous (no `await`), so concurrent requests to a new host:port // can't race each other into creating two pools for the same key. - ClientPool _poolFor(Uri url) { + ClientPool _poolFor(Uri url) { final key = '${url.host}:${url.port}'; return _pools.putIfAbsent( key, - () => ClientPool( + () => ClientPool( () => _handshakeGate.withResource(() => _dial(url.host, url.port)), maxConcurrentOperations: maxStreamsPerConnection, maxIdleResources: maxIdleConnections, @@ -96,7 +97,7 @@ class Http2Client extends BaseClient { ); } - Future _dial(String host, int port) async { + Future _dial(String host, int port) async { final socket = await SecureSocket.connect( host, port, @@ -126,7 +127,11 @@ class Http2Client extends BaseClient { ); unawaited(died.future.catchError((Object _) {})); - final transport = ClientTransportConnection.viaStreams(incoming, socket); + final transport = ClientConnection( + incoming, + socket, + const ClientSettings(), + ); try { await Future.any([ transport.onInitialPeerSettingsReceived, @@ -152,7 +157,7 @@ class Http2Client extends BaseClient { /// the stream stays open while the body is delivered, so the slot is only /// released once that stream reaches a terminal state. Future _sendOverHttp2( - PoolLease lease, + PoolLease lease, BaseRequest request, List bodyBytes, ) async { diff --git a/pkgs/http2/lib/transport.dart b/pkgs/http2/lib/transport.dart index ae57fa89f0..4584e71bc6 100644 --- a/pkgs/http2/lib/transport.dart +++ b/pkgs/http2/lib/transport.dart @@ -97,11 +97,6 @@ abstract class ClientTransportConnection extends TransportConnection { /// via [makeRequest]. bool get isOpen; - /// The maximum number of concurrent streams the peer currently allows, - /// per its most recent SETTINGS_MAX_CONCURRENT_STREAMS (RFC 7540 6.5.2), - /// or `null` if the peer hasn't advertised a limit (unlimited). - int? get peerMaxConcurrentStreams; - /// Creates a new outgoing stream. ClientTransportStream makeRequest( List
headers, { From 39b75f936d1bcbeb2ff42e5210f600b0efb11f0f Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 6 Aug 2026 11:36:35 +0100 Subject: [PATCH 27/27] updates --- pkgs/http2/CHANGELOG.md | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/pkgs/http2/CHANGELOG.md b/pkgs/http2/CHANGELOG.md index ca6fed69b8..ddb45d513c 100644 --- a/pkgs/http2/CHANGELOG.md +++ b/pkgs/http2/CHANGELOG.md @@ -2,10 +2,7 @@ - Gracefully handle receiving headers on a stream that the client has canceled. (#1799) - Add `Http2Client` (`package:http2/client.dart`), a pooled, multiplexed - `package:http` `Client` backed by HTTP/2 connections. A connection is never - given more concurrent streams than its server allows, and is not used at all - until that server has sent its initial SETTINGS frame - see - `Http2Client.settingsTimeout` for how long that is waited for. + `package:http` `Client` backed by HTTP/2 connections. ## 3.0.0