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

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<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, entry.value),
+  ], endStream: bodyBytes.isEmpty);
+
+  if (bodyBytes.isNotEmpty) stream.sendData(bodyBytes, endStream: true);
+
+  final statusCompleter = Completer<int>();
+  final bodyController = StreamController<List<int>>();
+  final responseHeaders = <String, String>{};
+
+  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<ClientTransportConnection> 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<ClientTransportConnection> _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<StreamedResponse> 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,34 @@ class FirestoreHttpClient {
   /// 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 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());
     }
 
-    // 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,
     );
   }
 

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: