fix(firestore): large snapshots do not block frame scheduling

packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/FlutterFirebaseFirestorePlugin.java

@@ -989,7 +989,8 @@ public void querySnapshot(
                 includeMetadataChanges,
                 PigeonParser.parsePigeonServerTimestampBehavior(
                     options.getServerTimestampBehavior()),
-                PigeonParser.parseListenSource(source))));
+                PigeonParser.parseListenSource(source),
+                cachedThreadPool)));
   }
 
   @Override

packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotsStreamHandler.java

@@ -19,6 +19,7 @@
 import io.flutter.plugins.firebase.firestore.utils.ExceptionConverter;
 import io.flutter.plugins.firebase.firestore.utils.PigeonParser;
 import java.util.Map;
+import java.util.concurrent.Executor;
 
 public class QuerySnapshotsStreamHandler implements StreamHandler {
 
@@ -29,24 +30,28 @@ public class QuerySnapshotsStreamHandler implements StreamHandler {
   DocumentSnapshot.ServerTimestampBehavior serverTimestampBehavior;
 
   ListenSource source;
+  Executor snapshotExecutor;
 
   public QuerySnapshotsStreamHandler(
       Query query,
       Boolean includeMetadataChanges,
       DocumentSnapshot.ServerTimestampBehavior serverTimestampBehavior,
-      ListenSource source) {
+      ListenSource source,
+      Executor snapshotExecutor) {
     this.query = query;
     this.metadataChanges =
         includeMetadataChanges ? MetadataChanges.INCLUDE : MetadataChanges.EXCLUDE;
     this.serverTimestampBehavior = serverTimestampBehavior;
     this.source = source;
+    this.snapshotExecutor = snapshotExecutor;
   }
 
   @Override
   public void onListen(Object arguments, EventSink events) {
     SnapshotListenOptions.Builder optionsBuilder = new SnapshotListenOptions.Builder();
     optionsBuilder.setMetadataChanges(metadataChanges);
     optionsBuilder.setSource(source);
+    optionsBuilder.setExecutor(snapshotExecutor);
 
     listenerRegistration =
         query.addSnapshotListener(
@@ -63,8 +68,9 @@ public void onListen(Object arguments, EventSink events) {
                 // nested `InternalDocumentSnapshot` / `InternalDocumentChange` /
                 // `InternalSnapshotMetadata` with their proper type codes. Pigeon 26
                 // no longer flattens nested types via `.toList()`.
-                events.success(
-                    PigeonParser.toPigeonQuerySnapshot(querySnapshot, serverTimestampBehavior));
+                Object pigeonSnapshot =
+                    PigeonParser.toPigeonQuerySnapshot(querySnapshot, serverTimestampBehavior);
+                events.success(pigeonSnapshot);
               }
             });
   }

packages/cloud_firestore/cloud_firestore/example/integration_test/query_e2e.dart

@@ -8,6 +8,7 @@ import 'dart:math';
 import 'package:cloud_firestore/cloud_firestore.dart';
 import 'package:firebase_core/firebase_core.dart';
 import 'package:flutter/foundation.dart';
+import 'package:flutter/material.dart';
 import 'package:flutter_test/flutter_test.dart';
 
 void runQueryTests() {
@@ -374,6 +375,88 @@ void runQueryTests() {
         await subscription.cancel();
       });
 
+      testWidgets(
+        'large snapshots do not block frame scheduling',
+        (WidgetTester tester) async {
+          CollectionReference<Map<String, dynamic>> collection =
+              await initializeTest('large-snapshot-listener');
+          const int documentCount = 1000;
+          final String payload = List.filled(1024, 'x').join();
+
+          for (int start = 0; start < documentCount; start += 400) {
+            final WriteBatch batch = firestore.batch();
+            final int end = min(start + 400, documentCount);
+            for (int index = start; index < end; index++) {
+              batch.set(collection.doc('doc-$index'), <String, Object?>{
+                'index': index,
+                'payload': payload,
+              });
+            }
+            await batch.commit();
+          }
+
+          final Completer<void> initialSnapshot = Completer<void>();
+          final Completer<void> receivedUpdates = Completer<void>();
+          var initialSnapshotReceived = false;
+          var updateSnapshots = 0;
+
+          final StreamSubscription<QuerySnapshot<Map<String, dynamic>>>
+              subscription = collection.snapshots().listen((snapshot) {
+            if (!initialSnapshotReceived && snapshot.size == documentCount) {
+              initialSnapshotReceived = true;
+              initialSnapshot.complete();
+              return;
+            }
+
+            if (initialSnapshotReceived && snapshot.docChanges.isNotEmpty) {
+              updateSnapshots++;
+              if (updateSnapshots >= 3 && !receivedUpdates.isCompleted) {
+                receivedUpdates.complete();
+              }
+            }
+          });
+          addTearDown(subscription.cancel);
+
+          await initialSnapshot.future.timeout(const Duration(seconds: 30));
+          await tester.pumpWidget(
+            const Directionality(
+              textDirection: TextDirection.ltr,
+              child: Center(child: CircularProgressIndicator()),
+            ),
+          );
+
+          var updatesDone = false;
+          final updateFuture = Future<void>(() async {
+            for (int index = 0; index < 3; index++) {
+              await collection.doc('doc-0').update(<String, Object?>{
+                'counter': index,
+                'payload': payload,
+              });
+            }
+            await receivedUpdates.future.timeout(const Duration(seconds: 30));
+          }).whenComplete(() {
+            updatesDone = true;
+          });
+
+          final pumpDurations = <Duration>[];
+          while (!updatesDone) {
+            final Stopwatch stopwatch = Stopwatch()..start();
+            await tester.pump(const Duration(milliseconds: 16));
+            stopwatch.stop();
+            pumpDurations.add(stopwatch.elapsed);
+          }
+          await updateFuture;
+
+          expect(pumpDurations, isNotEmpty);
+          final Duration longestPump = pumpDurations.reduce(
+            (current, next) => current > next ? current : next,
+          );
+          expect(longestPump, lessThan(const Duration(milliseconds: 250)));
+        },
+        timeout: const Timeout.factor(10),
+        skip: kIsWeb || defaultTargetPlatform == TargetPlatform.windows,
+      );
+
       test(
         'listeners throws a [FirebaseException] with Query',
         () async {

packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/FLTQuerySnapshotStreamHandler.m

@@ -16,6 +16,7 @@
 
 @interface FLTQuerySnapshotStreamHandler ()
 @property(readwrite, strong) id<FIRListenerRegistration> listenerRegistration;
+@property(nonatomic) dispatch_queue_t snapshotQueue;
 @end
 
 @implementation FLTQuerySnapshotStreamHandler
@@ -32,6 +33,8 @@ - (instancetype)initWithFirestore:(FIRFirestore *)firestore
     _includeMetadataChanges = includeMetadataChanges;
     _serverTimestampBehavior = serverTimestampBehavior;
     _source = source;
+    _snapshotQueue = dispatch_queue_create("io.flutter.plugins.firebase.firestore.query_snapshot",
+                                           DISPATCH_QUEUE_SERIAL);
   }
   return self;
 }
@@ -64,13 +67,15 @@ - (FlutterError *_Nullable)onListenWithArguments:(id _Nullable)arguments
                                           andOptionalNSError:error]);
       });
     } else {
-      dispatch_async(dispatch_get_main_queue(), ^{
+      dispatch_async(self.snapshotQueue, ^{
         // Emit the Pigeon object directly; the Pigeon-aware codec serializes nested
         // `InternalDocumentSnapshot` / `InternalDocumentChange` / `InternalSnapshotMetadata`
         // with their proper type codes. Pigeon 26 no longer flattens nested types
         // via `toList`.
-        events([FirestorePigeonParser toPigeonQuerySnapshot:snapshot
-                                    serverTimestampBehavior:self.serverTimestampBehavior]);
+        InternalQuerySnapshot *pigeonSnapshot =
+            [FirestorePigeonParser toPigeonQuerySnapshot:snapshot
+                                 serverTimestampBehavior:self.serverTimestampBehavior];
+        events(pigeonSnapshot);
       });
     }
   };