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 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..8bac5b92 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,283 @@ class EmulatorClient extends BaseClient implements googleapis_auth.AuthClient { void close() => client.close(); } +class _PooledResource { + _PooledResource(this.future); + final Future future; + int inFlight = 0; + bool failed = false; +} + +/// 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.'); + } + + 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(); + } +} + +/// 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 int maxStreamsPerConnection; + final _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(), + ), + ); + } + + Future _dial(String host) async { + final socket = await SecureSocket.connect( + host, + 443, + 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), + 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), + ); + + /// 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()); +} + /// HTTP client wrapper for Firestore API operations. /// /// Provides authenticated API access with automatic project ID discovery. @@ -154,24 +435,33 @@ 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. + Http2Client? _transport; + /// Creates the appropriate HTTP client based on emulator configuration. Future _createClient() async { if (_isUsingEmulator) { - // Emulator: Create unauthenticated client. + // 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: Create authenticated client. + final transport = Http2Client(); + _transport = transport; + 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, ); } @@ -204,5 +494,6 @@ class FirestoreHttpClient { Future close() async { final client = await _client; client.close(); + await _transport?.terminate(); } } 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: 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); + }); + }); +}