fix(firestore): use pooled HTTP/2 connections by default to fix slow/failing concurrent requests

packages/google_cloud_firestore/CHANGELOG.md

@@ -5,6 +5,7 @@
 - Updated `Transaction.delete` and `Transaction.update` type constraints to accept `DocumentReference<Object?>`. (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
 

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<T> {
+  _PooledResource(this.future);
+  final Future<T> 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<T> {
+  ClientPool(
+    Future<T> Function() create, {
+    required this.maxConcurrentOperations,
+    required Future<void> Function(T resource) destroy,
+    this.maxIdleResources = 1,
+  }) : _create = create,
+       _destroy = destroy;
+
+  final Future<T> Function() _create;
+  final Future<void> Function(T resource) _destroy;
+  final int maxConcurrentOperations;
+  final int maxIdleResources;
+
+  final _resources = <_PooledResource<T>>[];
+  var _terminated = false;
+  Completer<void>? _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<R> run<R>(Future<R> 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<T> _acquire() {
+    _PooledResource<T>? 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<T>(_create());
+    _resources.add(resource);
+    return resource;
+  }
+
+  Future<void> _collectIfIdle(_PooledResource<T> 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<void> terminate() async {
+    _terminated = true;
+
+    if (opCount > 0) {
+      _drained = Completer<void>();
+      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 = <String, ClientPool<ClientTransportConnection>>{};
+
+  // 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<ClientTransportConnection> _poolFor(String host) {
+    return _pools.putIfAbsent(
+      host,
+      () => ClientPool<ClientTransportConnection>(
+        () => _handshakeGate.withResource(() => _dial(host)),
+        maxConcurrentOperations: maxStreamsPerConnection,
+        destroy: (transport) => transport.finish(),
+      ),
+    );
+  }
+
+  Future<ClientTransportConnection> _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<StreamedResponse> _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<int>();
+    late final StreamSubscription<StreamMessage> subscription;
+    final bodyController = StreamController<List<int>>(
+      onCancel: () => subscription.cancel(),
+    );
+    final responseHeaders = <String, String>{};
+
+    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<StreamedResponse> 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<void> 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<googleapis_auth.AuthClient> _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<googleapis_auth.AuthClient> _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<void> close() async {
     final client = await _client;
     client.close();
+    await _transport?.terminate();
   }
 }

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: