From c5ed523e4d82116c97d572be800445fae315793b Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 10:01:52 +0100 Subject: [PATCH 01/10] feat(firestore): use pooled HTTP/2 connections by default for all requests --- .../lib/src/firestore_http_client.dart | 208 +++++++++++++++++- packages/google_cloud_firestore/pubspec.yaml | 2 + 2 files changed, 205 insertions(+), 5 deletions(-) diff --git a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart index f41c86cb..1dd90ff6 100644 --- a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart +++ b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart @@ -13,13 +13,17 @@ // limitations under the License. import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; import 'package:google_cloud/constants.dart' as google_cloud; import 'package:google_cloud/google_cloud.dart' as google_cloud; import 'package:google_cloud_firestore_v1/firestore.dart' as firestore_v1; import 'package:googleapis_auth/auth_io.dart' as googleapis_auth; import 'package:http/http.dart'; +import 'package:http2/transport.dart' hide Settings; import 'package:meta/meta.dart'; +import 'package:pool/pool.dart'; import '../google_cloud_firestore.dart'; import 'environment.dart'; @@ -77,6 +81,191 @@ class EmulatorClient extends BaseClient implements googleapis_auth.AuthClient { void close() => client.close(); } +/// 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), + for (final entry in request.headers.entries) + Header.ascii(entry.key, entry.value), + ], endStream: bodyBytes.isEmpty); + + if (bodyBytes.isNotEmpty) stream.sendData(bodyBytes, endStream: true); + + final statusCompleter = Completer(); + final bodyController = StreamController>(); + final responseHeaders = {}; + + 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 (!bodyController.isClosed) bodyController.close(); + }, + onError: (Object error, StackTrace stackTrace) { + if (!statusCompleter.isCompleted) { + statusCompleter.completeError(error, stackTrace); + } + bodyController.addError(error, stackTrace); + }, + cancelOnError: true, + ); + + final statusCode = await statusCompleter.future; + return StreamedResponse( + bodyController.stream, + statusCode, + headers: responseHeaders, + request: request, + ); +} + +class _PooledConnection { + _PooledConnection(this.transportFuture); + final Future transportFuture; + int inFlight = 0; + + /// Set once a send() on this connection throws - matches Node + /// ClientPool's RST_STREAM handling: stop routing NEW work here, but let + /// requests already in flight finish. Actually removing/closing failed + /// connections is deferred (see [Http2ClientPool] doc comment) - this + /// pass only stops reusing them. + bool failed = false; +} + +/// Sends every request as a stream on a pool of shared HTTP/2 connections +/// to [host], mirroring nodejs-firestore's `ClientPool` +/// (dev/src/pool.ts): packs load onto the most-full connection under +/// [maxStreamsPerConnection] (to keep others idle) rather than spreading +/// evenly, opens a new connection once all existing ones are full, and +/// stops routing new work to a connection once it's shown a +/// connection-level failure. +/// +/// Known gaps vs. Node's ClientPool, deferred to a follow-up: +/// - No idle-capacity garbage collection - connections opened during a +/// burst are never closed once traffic quiets down. +/// - No graceful terminate() draining - close() does not wait for +/// in-flight requests before finishing connections. +/// +/// Pure transport: no auth. Wrapped with googleapis_auth's client helpers +/// below (see [FirestoreHttpClient._createClient]) for a refreshing +/// AuthClient. +class Http2ClientPool extends BaseClient { + Http2ClientPool(this.host, {this.maxStreamsPerConnection = 100}) + : _handshakeGate = Pool(_maxConcurrentHandshakes); + + final String host; + final int maxStreamsPerConnection; + + // Caps concurrent in-flight TCP+TLS handshakes, independent of how many + // total connections the pool ends up needing - protects against a huge, + // unpredictable production burst opening hundreds of handshakes at once. + static const _maxConcurrentHandshakes = 50; + final Pool _handshakeGate; + + final List<_PooledConnection> _connections = []; + + Future _openConnection() { + return _handshakeGate.withResource(() async { + final socket = await SecureSocket.connect( + host, + 443, + supportedProtocols: ['h2'], + ); + if (socket.selectedProtocol != 'h2') { + throw StateError( + 'Server did not negotiate HTTP/2 (got ${socket.selectedProtocol})', + ); + } + return ClientTransportConnection.viaSocket(socket); + }); + } + + /// Picks the most-full connection under capacity, or reserves a new one. + /// + /// Deliberately synchronous (no `await`), so it runs atomically with + /// respect to other concurrent calls without needing a lock - otherwise + /// many callers arriving before the first connection finishes dialing + /// would all see an empty pool and each open their own (a thundering + /// herd), instead of piling onto the one already being established. + /// + /// "Most-full" (not least-loaded) selection matches Node ClientPool's + /// intent: pack load onto fewer connections so others stay idle and + /// become eligible for future cleanup, rather than spreading evenly. + _PooledConnection _acquireConnection() { + _PooledConnection? selected; + for (final conn in _connections) { + if (conn.failed) continue; + if (conn.inFlight < maxStreamsPerConnection && + (selected == null || conn.inFlight > selected.inFlight)) { + selected = conn; + } + } + if (selected != null) return selected; + + final pooled = _PooledConnection(_openConnection()); + _connections.add(pooled); + return pooled; + } + + /// The number of connections currently in the pool (including any that + /// have been marked [_PooledConnection.failed] but not yet cleaned up). + int get connectionCount => _connections.length; + + @override + Future send(BaseRequest request) async { + final conn = _acquireConnection(); + conn.inFlight++; + try { + final transport = await conn.transportFuture; + return await _sendOverHttp2(transport, request); + } catch (e) { + // Broader than Node's RST_STREAM-specific regex match - our own + // testing surfaced multiple distinct http2-level error shapes + // (REFUSED_STREAM, "forcefully terminated"/CONNECT_ERROR), so any + // thrown error here is treated as a signal this connection is no + // longer trustworthy for new work. + conn.failed = true; + rethrow; + } finally { + conn.inFlight--; + } + } + + @override + void close() { + for (final conn in _connections) { + conn.transportFuture.then((t) => t.finish()); + } + } +} + /// HTTP client wrapper for Firestore API operations. /// /// Provides authenticated API access with automatic project ID discovery. @@ -157,21 +346,30 @@ class FirestoreHttpClient { /// Creates the appropriate HTTP client based on emulator configuration. Future _createClient() async { if (_isUsingEmulator) { - // Emulator: Create unauthenticated client. + // Emulator: Create unauthenticated client. The emulator has no TLS + // and doesn't negotiate HTTP/2, so this stays on plain HTTP/1.1. return EmulatorClient(Client()); } - // Production: Create authenticated client. + // Production: every request multiplexes over a pool of shared HTTP/2 + // connections instead of dart:io's default one-connection-per-request + // HTTP/1.1 behavior. googleapis_auth wraps this transport with its + // token-refresh logic - the pool itself has no auth concept. + final transport = Http2ClientPool(_settings.host ?? 'firestore.googleapis.com'); + final serviceAccountCreds = credential.serviceAccountCredentials; if (serviceAccountCreds != null) { - return googleapis_auth.clientViaServiceAccount(serviceAccountCreds, [ - 'https://www.googleapis.com/auth/cloud-platform', - ]); + return googleapis_auth.clientViaServiceAccount( + serviceAccountCreds, + ['https://www.googleapis.com/auth/cloud-platform'], + baseClient: transport, + ); } // Fall back to Application Default Credentials return googleapis_auth.clientViaApplicationDefaultCredentials( scopes: ['https://www.googleapis.com/auth/cloud-platform'], + baseClient: transport, ); } diff --git a/packages/google_cloud_firestore/pubspec.yaml b/packages/google_cloud_firestore/pubspec.yaml index 0b8ff8db..1ebcbcb8 100644 --- a/packages/google_cloud_firestore/pubspec.yaml +++ b/packages/google_cloud_firestore/pubspec.yaml @@ -19,8 +19,10 @@ dependencies: google_cloud_type: ^0.5.2 googleapis_auth: ^2.3.3 http: ^1.6.0 + http2: ^2.3.1 intl: ^0.20.0 meta: ^1.17.0 + pool: ^1.5.1 retry: ^3.1.2 dev_dependencies: From 3a0862572579b1720d26b43e5c0d83cea3f84bd3 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 11:12:59 +0100 Subject: [PATCH 02/10] fix(firestore): keep emulator on plain HTTP/1.1, matching Node's REST-mode behavior --- .../lib/src/firestore_http_client.dart | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart index 1dd90ff6..1ad6146f 100644 --- a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart +++ b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart @@ -346,8 +346,12 @@ class FirestoreHttpClient { /// Creates the appropriate HTTP client based on emulator configuration. Future _createClient() async { if (_isUsingEmulator) { - // Emulator: Create unauthenticated client. The emulator has no TLS - // and doesn't negotiate HTTP/2, so this stays on plain HTTP/1.1. + // Emulator: plain HTTP/1.1, matching nodejs-firestore's own REST-mode + // behavior for the emulator (it forces `protocol: 'http'` when ssl is + // false in REST fallback - see index.ts's clientFactory). Node's + // gRPC-mode client does use HTTP/2 for the emulator, but that's a + // separate transport with no equivalent here - our SDK is REST-only, + // so its REST-mode behavior is the correct comparison, not gRPC's. return EmulatorClient(Client()); } From 35f430877dc38278376487c33a8db682f92cc26a Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 16:22:50 +0100 Subject: [PATCH 03/10] test(firestore): add unit tests for ClientPool --- .../test/http_client_test.dart | 191 ++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 packages/google_cloud_firestore/test/http_client_test.dart diff --git a/packages/google_cloud_firestore/test/http_client_test.dart b/packages/google_cloud_firestore/test/http_client_test.dart new file mode 100644 index 00000000..62bbb4c0 --- /dev/null +++ b/packages/google_cloud_firestore/test/http_client_test.dart @@ -0,0 +1,191 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import 'dart:async'; + +import 'package:google_cloud_firestore/src/firestore_http_client.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('ClientPool', () { + 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('re-uses a 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 the 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 a resource after it fails', () 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 a pool of idle resources up to maxIdleResources', () 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 0f0f33ec6c10567da04f04fb026bc059609a6389 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 16:26:11 +0100 Subject: [PATCH 04/10] fix(firestore): close pooled HTTP/2 connections on shutdown and complete ClientPool Node parity --- .../lib/src/firestore_http_client.dart | 264 ++++++++++-------- 1 file changed, 154 insertions(+), 110 deletions(-) diff --git a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart index 1ad6146f..0b67d160 100644 --- a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart +++ b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart @@ -146,124 +146,169 @@ Future _sendOverHttp2( ); } -class _PooledConnection { - _PooledConnection(this.transportFuture); - final Future transportFuture; +class _PooledResource { + _PooledResource(this.future); + final Future future; int inFlight = 0; - - /// Set once a send() on this connection throws - matches Node - /// ClientPool's RST_STREAM handling: stop routing NEW work here, but let - /// requests already in flight finish. Actually removing/closing failed - /// connections is deferred (see [Http2ClientPool] doc comment) - this - /// pass only stops reusing them. bool failed = false; } -/// Sends every request as a stream on a pool of shared HTTP/2 connections -/// to [host], mirroring nodejs-firestore's `ClientPool` -/// (dev/src/pool.ts): packs load onto the most-full connection under -/// [maxStreamsPerConnection] (to keep others idle) rather than spreading -/// evenly, opens a new connection once all existing ones are full, and -/// stops routing new work to a connection once it's shown a -/// connection-level failure. -/// -/// Known gaps vs. Node's ClientPool, deferred to a follow-up: -/// - No idle-capacity garbage collection - connections opened during a -/// burst are never closed once traffic quiets down. -/// - No graceful terminate() draining - close() does not wait for -/// in-flight requests before finishing connections. -/// -/// Pure transport: no auth. Wrapped with googleapis_auth's client helpers -/// below (see [FirestoreHttpClient._createClient]) for a refreshing -/// AuthClient. -class Http2ClientPool extends BaseClient { - Http2ClientPool(this.host, {this.maxStreamsPerConnection = 100}) - : _handshakeGate = Pool(_maxConcurrentHandshakes); - - final String host; - final int maxStreamsPerConnection; - - // Caps concurrent in-flight TCP+TLS handshakes, independent of how many - // total connections the pool ends up needing - protects against a huge, - // unpredictable production burst opening hundreds of handshakes at once. - static const _maxConcurrentHandshakes = 50; - final Pool _handshakeGate; - - final List<_PooledConnection> _connections = []; +/// A pool of resources of type [T], mirroring nodejs-firestore's +/// `ClientPool` (dev/src/pool.ts): packs load onto the most-full resource +/// under [maxConcurrentOperations] instead of spreading evenly, opens a +/// new resource once existing ones are full, retires a resource once +/// [run] reports it failed, and garbage collects idle resources past +/// [maxIdleResources]. +@internal +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; + + @visibleForTesting + int get size => _resources.length; + + @visibleForTesting + 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.'); + } - Future _openConnection() { - return _handshakeGate.withResource(() async { - final socket = await SecureSocket.connect( - host, - 443, - supportedProtocols: ['h2'], - ); - if (socket.selectedProtocol != 'h2') { - throw StateError( - 'Server did not negotiate HTTP/2 (got ${socket.selectedProtocol})', - ); + 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); } - return ClientTransportConnection.viaSocket(socket); - }); + } } - /// Picks the most-full connection under capacity, or reserves a new one. - /// - /// Deliberately synchronous (no `await`), so it runs atomically with - /// respect to other concurrent calls without needing a lock - otherwise - /// many callers arriving before the first connection finishes dialing - /// would all see an empty pool and each open their own (a thundering - /// herd), instead of piling onto the one already being established. - /// - /// "Most-full" (not least-loaded) selection matches Node ClientPool's - /// intent: pack load onto fewer connections so others stay idle and - /// become eligible for future cleanup, rather than spreading evenly. - _PooledConnection _acquireConnection() { - _PooledConnection? selected; - for (final conn in _connections) { - if (conn.failed) continue; - if (conn.inFlight < maxStreamsPerConnection && - (selected == null || conn.inFlight > selected.inFlight)) { - selected = conn; + // 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 pooled = _PooledConnection(_openConnection()); - _connections.add(pooled); - return pooled; + final resource = _PooledResource(_create()); + _resources.add(resource); + return resource; } - /// The number of connections currently in the pool (including any that - /// have been marked [_PooledConnection.failed] but not yet cleaned up). - int get connectionCount => _connections.length; + Future _collectIfIdle(_PooledResource resource) async { + if (resource.inFlight > 0) return; + if (!resource.failed && !_hasExcessIdleCapacity) return; - @override - Future send(BaseRequest request) async { - final conn = _acquireConnection(); - conn.inFlight++; - try { - final transport = await conn.transportFuture; - return await _sendOverHttp2(transport, request); - } catch (e) { - // Broader than Node's RST_STREAM-specific regex match - our own - // testing surfaced multiple distinct http2-level error shapes - // (REFUSED_STREAM, "forcefully terminated"/CONNECT_ERROR), so any - // thrown error here is treated as a signal this connection is no - // longer trustworthy for new work. - conn.failed = true; - rethrow; - } finally { - conn.inFlight--; + _resources.remove(resource); + await _destroy(await resource.future); + } + + 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(); } } - @override - void close() { - for (final conn in _connections) { - conn.transportFuture.then((t) => t.finish()); + /// 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) { + await _destroy(await resource.future); + } + _resources.clear(); + } +} + +/// Sends every request as a stream on a pool of shared HTTP/2 connections +/// to [host] - see [ClientPool]. Pure transport, no auth: wrapped with +/// googleapis_auth's client helpers in [FirestoreHttpClient._createClient] +/// for a refreshing [googleapis_auth.AuthClient]. +class Http2ClientPool extends BaseClient { + Http2ClientPool(this.host, {this.maxStreamsPerConnection = 100}) + : _pool = ClientPool( + () => _handshakeGate.withResource(() => _dial(host)), + maxConcurrentOperations: maxStreamsPerConnection, + destroy: (transport) => transport.finish(), + ); + + final String host; + final int maxStreamsPerConnection; + final ClientPool _pool; + + // Caps concurrent in-flight TCP+TLS handshakes across all pools, + // independent of how many total connections any one pool needs. + static final _handshakeGate = Pool(50); + + static Future _dial(String host) async { + final socket = await SecureSocket.connect( + host, + 443, + supportedProtocols: ['h2'], + ); + if (socket.selectedProtocol != 'h2') { + throw StateError( + 'Server did not negotiate HTTP/2 (got ${socket.selectedProtocol})', + ); } + return ClientTransportConnection.viaSocket(socket); } + + /// The number of connections currently in the pool. + int get connectionCount => _pool.size; + + @override + Future send(BaseRequest request) => + _pool.run((transport) => _sendOverHttp2(transport, request)); + + @override + void close() => unawaited(_pool.terminate()); } /// HTTP client wrapper for Firestore API operations. @@ -343,23 +388,22 @@ class FirestoreHttpClient { /// Lazy-initialized HTTP client that's cached for reuse. late final Future _client = _createClient(); + // googleapis_auth never closes a caller-supplied baseClient, so + // close() must reach this transport itself. + Http2ClientPool? _transport; + /// Creates the appropriate HTTP client based on emulator configuration. Future _createClient() async { if (_isUsingEmulator) { - // Emulator: plain HTTP/1.1, matching nodejs-firestore's own REST-mode - // behavior for the emulator (it forces `protocol: 'http'` when ssl is - // false in REST fallback - see index.ts's clientFactory). Node's - // gRPC-mode client does use HTTP/2 for the emulator, but that's a - // separate transport with no equivalent here - our SDK is REST-only, - // so its REST-mode behavior is the correct comparison, not gRPC's. + // Emulator: plain HTTP/1.1, matching nodejs-firestore's REST-mode + // behavior (its gRPC mode uses HTTP/2, but this SDK is REST-only). return EmulatorClient(Client()); } - // Production: every request multiplexes over a pool of shared HTTP/2 - // connections instead of dart:io's default one-connection-per-request - // HTTP/1.1 behavior. googleapis_auth wraps this transport with its - // token-refresh logic - the pool itself has no auth concept. - final transport = Http2ClientPool(_settings.host ?? 'firestore.googleapis.com'); + final transport = Http2ClientPool( + _settings.host ?? 'firestore.googleapis.com', + ); + _transport = transport; final serviceAccountCreds = credential.serviceAccountCredentials; if (serviceAccountCreds != null) { @@ -370,7 +414,6 @@ class FirestoreHttpClient { ); } - // Fall back to Application Default Credentials return googleapis_auth.clientViaApplicationDefaultCredentials( scopes: ['https://www.googleapis.com/auth/cloud-platform'], baseClient: transport, @@ -406,5 +449,6 @@ class FirestoreHttpClient { Future close() async { final client = await _client; client.close(); + _transport?.close(); } } From 016658f5416e9caf723a461f67e12d18b96bcabb Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 16:28:45 +0100 Subject: [PATCH 05/10] chore: update CHANGELOG --- packages/google_cloud_firestore/CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/google_cloud_firestore/CHANGELOG.md b/packages/google_cloud_firestore/CHANGELOG.md index 7114b835..559aaae4 100644 --- a/packages/google_cloud_firestore/CHANGELOG.md +++ b/packages/google_cloud_firestore/CHANGELOG.md @@ -5,6 +5,7 @@ - Updated `Transaction.delete` and `Transaction.update` type constraints to accept `DocumentReference`. (thanks to @Levin-Me) - Made `Timestamp` encodable by adding `toJson` method. (thanks to @OutdatedGuy) - Update dependency `googleapis_auth: ^2.3.3` to fix `auth/insufficient-permission` errors with Application Default Credentials that have no quota project set. +- Fixed slow and failing requests under high concurrency by pooling shared HTTP/2 connections by default instead of opening a new HTTP/1.1 connection per request. ## 0.5.2 From 9839fc6b264a69dcfef693d7183377298d2bea09 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 16:47:56 +0100 Subject: [PATCH 06/10] fix(firestore): lowercase HTTP/2 header names, avoid hangs/leaks in the transport and pool shutdown --- .../lib/src/firestore_http_client.dart | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart index 0b67d160..527d0a9e 100644 --- a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart +++ b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart @@ -98,7 +98,7 @@ Future _sendOverHttp2( Header.ascii(':authority', request.url.host), Header.ascii(':path', path), for (final entry in request.headers.entries) - Header.ascii(entry.key, entry.value), + Header.ascii(entry.key.toLowerCase(), entry.value), ], endStream: bodyBytes.isEmpty); if (bodyBytes.isNotEmpty) stream.sendData(bodyBytes, endStream: true); @@ -126,6 +126,11 @@ Future _sendOverHttp2( } }, 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) { @@ -133,6 +138,7 @@ Future _sendOverHttp2( statusCompleter.completeError(error, stackTrace); } bodyController.addError(error, stackTrace); + if (!bodyController.isClosed) bodyController.close(); }, cancelOnError: true, ); @@ -260,7 +266,12 @@ class ClientPool { } for (final resource in _resources) { - await _destroy(await resource.future); + try { + await _destroy(await resource.future); + } catch (_) { + // Best-effort: one resource failing to close shouldn't stop the + // rest from being destroyed. + } } _resources.clear(); } @@ -293,6 +304,7 @@ class Http2ClientPool extends BaseClient { supportedProtocols: ['h2'], ); if (socket.selectedProtocol != 'h2') { + socket.destroy(); throw StateError( 'Server did not negotiate HTTP/2 (got ${socket.selectedProtocol})', ); From ff058af34062cac3c5293f68030e1674c8490420 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 18:01:46 +0100 Subject: [PATCH 07/10] fix(firestore): cancel HTTP/2 stream subscription on early cancel, make pool shutdown awaitable, guard cleanup errors --- .../lib/src/firestore_http_client.dart | 24 +++++++++++++++---- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart index 527d0a9e..97e1edf7 100644 --- a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart +++ b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart @@ -104,10 +104,13 @@ Future _sendOverHttp2( if (bodyBytes.isNotEmpty) stream.sendData(bodyBytes, endStream: true); final statusCompleter = Completer(); - final bodyController = StreamController>(); + late final StreamSubscription subscription; + final bodyController = StreamController>( + onCancel: () => subscription.cancel(), + ); final responseHeaders = {}; - stream.incomingMessages.listen( + subscription = stream.incomingMessages.listen( (message) { if (message is HeadersStreamMessage) { for (final header in message.headers) { @@ -237,7 +240,12 @@ class ClientPool { if (!resource.failed && !_hasExcessIdleCapacity) return; _resources.remove(resource); - await _destroy(await resource.future); + 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 { @@ -319,8 +327,14 @@ class Http2ClientPool extends BaseClient { Future send(BaseRequest request) => _pool.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 [Http2ClientPool]. + Future terminate() => _pool.terminate(); + @override - void close() => unawaited(_pool.terminate()); + void close() => unawaited(terminate()); } /// HTTP client wrapper for Firestore API operations. @@ -461,6 +475,6 @@ class FirestoreHttpClient { Future close() async { final client = await _client; client.close(); - _transport?.close(); + await _transport?.terminate(); } } From aa4685fa3012305a10686a15983c0dfc61c18b25 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 19:47:57 +0100 Subject: [PATCH 08/10] fix(firestore): route non-Firestore-host requests (credential negotiation, WIF/OIDC) away from the HTTP/2 pool --- .../lib/src/firestore_http_client.dart | 38 +++++++++++++++++-- 1 file changed, 34 insertions(+), 4 deletions(-) diff --git a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart index 97e1edf7..29320c56 100644 --- a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart +++ b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart @@ -337,6 +337,37 @@ class Http2ClientPool extends BaseClient { void close() => unawaited(terminate()); } +/// Routes requests to [pooledHost] through [pooled]; everything else - +/// credential negotiation, WIF/OIDC token exchange, metadata server calls - +/// falls through to [fallback], since those target hosts the single-host +/// [pooled] transport was never dialed for. +class _HostRoutingClient extends BaseClient { + _HostRoutingClient(this.pooledHost, this.pooled, this.fallback); + + final String pooledHost; + final Http2ClientPool pooled; + final Client fallback; + + @override + Future send(BaseRequest request) { + if (request.url.host == pooledHost) { + return pooled.send(request); + } + return fallback.send(request); + } + + @override + void close() { + pooled.close(); + fallback.close(); + } + + Future terminate() async { + await pooled.terminate(); + fallback.close(); + } +} + /// HTTP client wrapper for Firestore API operations. /// /// Provides authenticated API access with automatic project ID discovery. @@ -416,7 +447,7 @@ class FirestoreHttpClient { // googleapis_auth never closes a caller-supplied baseClient, so // close() must reach this transport itself. - Http2ClientPool? _transport; + _HostRoutingClient? _transport; /// Creates the appropriate HTTP client based on emulator configuration. Future _createClient() async { @@ -426,9 +457,8 @@ class FirestoreHttpClient { return EmulatorClient(Client()); } - final transport = Http2ClientPool( - _settings.host ?? 'firestore.googleapis.com', - ); + final host = _settings.host ?? 'firestore.googleapis.com'; + final transport = _HostRoutingClient(host, Http2ClientPool(host), Client()); _transport = transport; final serviceAccountCreds = credential.serviceAccountCredentials; From 7596ec37b5ee3340915cd4d360bfa186f1513107 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 20:15:59 +0100 Subject: [PATCH 09/10] fix(firestore): make the HTTP/2 client multi-host (Http2Client), fixing WIF/OIDC credential negotiation --- .../lib/src/firestore_http_client.dart | 94 ++++++++----------- 1 file changed, 41 insertions(+), 53 deletions(-) diff --git a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart index 29320c56..da1f729b 100644 --- a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart +++ b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart @@ -285,26 +285,40 @@ class ClientPool { } } -/// Sends every request as a stream on a pool of shared HTTP/2 connections -/// to [host] - see [ClientPool]. Pure transport, no auth: wrapped with -/// googleapis_auth's client helpers in [FirestoreHttpClient._createClient] -/// for a refreshing [googleapis_auth.AuthClient]. -class Http2ClientPool extends BaseClient { - Http2ClientPool(this.host, {this.maxStreamsPerConnection = 100}) - : _pool = ClientPool( - () => _handshakeGate.withResource(() => _dial(host)), - maxConcurrentOperations: maxStreamsPerConnection, - destroy: (transport) => transport.finish(), - ); +/// Sends every request as a stream on a pool of shared HTTP/2 connections, +/// dialed per-host - see [ClientPool]. Multi-host so it's safe to use for +/// every request an [googleapis_auth.AuthClient] might send, not just the +/// actual Firestore calls: credential negotiation (OAuth2 token endpoint, +/// WIF/OIDC token exchange) targets different hosts entirely, and each +/// gets its own pool dialed to the right place instead of reusing a +/// connection meant for somewhere else. +/// +/// Pure transport, no auth: wrapped with googleapis_auth's client helpers +/// in [FirestoreHttpClient._createClient] for a refreshing +/// [googleapis_auth.AuthClient]. +class Http2Client extends BaseClient { + Http2Client({this.maxStreamsPerConnection = 100}); - final String host; final int maxStreamsPerConnection; - final ClientPool _pool; + final _pools = >{}; - // Caps concurrent in-flight TCP+TLS handshakes across all pools, + // Caps concurrent in-flight TCP+TLS handshakes across all hosts, // independent of how many total connections any one pool needs. static final _handshakeGate = Pool(50); + // Synchronous (no `await`), so concurrent requests to a new host can't + // race each other into creating two pools for the same host. + ClientPool _poolFor(String host) { + return _pools.putIfAbsent( + host, + () => ClientPool( + () => _handshakeGate.withResource(() => _dial(host)), + maxConcurrentOperations: maxStreamsPerConnection, + destroy: (transport) => transport.finish(), + ), + ); + } + static Future _dial(String host) async { final socket = await SecureSocket.connect( host, @@ -320,52 +334,27 @@ class Http2ClientPool extends BaseClient { return ClientTransportConnection.viaSocket(socket); } - /// The number of connections currently in the pool. - int get connectionCount => _pool.size; + /// The number of connections currently pooled, across all hosts. + int get connectionCount => + _pools.values.fold(0, (total, pool) => total + pool.size); @override - Future send(BaseRequest request) => - _pool.run((transport) => _sendOverHttp2(transport, request)); + Future send(BaseRequest request) => _poolFor( + request.url.host, + ).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 [Http2ClientPool]. - Future terminate() => _pool.terminate(); - - @override - void close() => unawaited(terminate()); -} - -/// Routes requests to [pooledHost] through [pooled]; everything else - -/// credential negotiation, WIF/OIDC token exchange, metadata server calls - -/// falls through to [fallback], since those target hosts the single-host -/// [pooled] transport was never dialed for. -class _HostRoutingClient extends BaseClient { - _HostRoutingClient(this.pooledHost, this.pooled, this.fallback); - - final String pooledHost; - final Http2ClientPool pooled; - final Client fallback; - - @override - Future send(BaseRequest request) { - if (request.url.host == pooledHost) { - return pooled.send(request); + /// this can be awaited by callers who hold a concrete [Http2Client]. + Future terminate() async { + for (final pool in _pools.values) { + await pool.terminate(); } - return fallback.send(request); } @override - void close() { - pooled.close(); - fallback.close(); - } - - Future terminate() async { - await pooled.terminate(); - fallback.close(); - } + void close() => unawaited(terminate()); } /// HTTP client wrapper for Firestore API operations. @@ -447,7 +436,7 @@ class FirestoreHttpClient { // googleapis_auth never closes a caller-supplied baseClient, so // close() must reach this transport itself. - _HostRoutingClient? _transport; + Http2Client? _transport; /// Creates the appropriate HTTP client based on emulator configuration. Future _createClient() async { @@ -457,8 +446,7 @@ class FirestoreHttpClient { return EmulatorClient(Client()); } - final host = _settings.host ?? 'firestore.googleapis.com'; - final transport = _HostRoutingClient(host, Http2ClientPool(host), Client()); + final transport = Http2Client(); _transport = transport; final serviceAccountCreds = credential.serviceAccountCredentials; From dfd56acfde50c1d7aa1d7f251173173f06ada277 Mon Sep 17 00:00:00 2001 From: demolaf Date: Thu, 23 Jul 2026 20:35:21 +0100 Subject: [PATCH 10/10] refactor(firestore): move _sendOverHttp2 into Http2Client, make it and _dial plain instance methods --- .../lib/src/firestore_http_client.dart | 157 +++++++++--------- 1 file changed, 79 insertions(+), 78 deletions(-) diff --git a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart index da1f729b..8bac5b92 100644 --- a/packages/google_cloud_firestore/lib/src/firestore_http_client.dart +++ b/packages/google_cloud_firestore/lib/src/firestore_http_client.dart @@ -81,80 +81,6 @@ class EmulatorClient extends BaseClient implements googleapis_auth.AuthClient { void close() => client.close(); } -/// 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), - 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, - ); -} - class _PooledResource { _PooledResource(this.future); final Future future; @@ -319,7 +245,7 @@ class Http2Client extends BaseClient { ); } - static Future _dial(String host) async { + Future _dial(String host) async { final socket = await SecureSocket.connect( host, 443, @@ -334,14 +260,89 @@ 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. + 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), + 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 all hosts. int get connectionCount => _pools.values.fold(0, (total, pool) => total + pool.size); @override - Future send(BaseRequest request) => _poolFor( - request.url.host, - ).run((transport) => _sendOverHttp2(transport, request)); + Future send(BaseRequest request) => + _poolFor(request.url.host).run( + (transport) => _sendOverHttp2(transport, request), + ); /// Waits for in-flight requests to finish, then closes every connection. ///