fix(cloud_firestore): deliver query snapshots before snapshotsInSync

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

@@ -57,12 +57,17 @@
 import io.flutter.plugins.firebase.firestore.utils.PipelineParser;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class FlutterFirebaseFirestorePlugin
@@ -90,6 +95,7 @@ public class FlutterFirebaseFirestorePlugin
   private final Map<String, EventChannel> eventChannels = new HashMap<>();
   private final Map<String, StreamHandler> streamHandlers = new HashMap<>();
   private final Map<String, OnTransactionResultListener> transactionHandlers = new HashMap<>();
+  private final Map<String, ExecutorService> snapshotDeliveryExecutors = new ConcurrentHashMap<>();
 
   // Used in the decoder to know which ServerTimestampBehavior to use
   public static final Map<Integer, DocumentSnapshot.ServerTimestampBehavior>
@@ -116,15 +122,30 @@ protected static void setCachedFirebaseFirestoreInstanceForKey(
   protected static FirebaseFirestore getFirestoreInstanceByNameAndDatabaseUrl(
       String appName, String databaseURL) {
     synchronized (firestoreInstanceCache) {
-      for (Map.Entry<FirebaseFirestore, FlutterFirebaseFirestoreExtension> entry :
-          firestoreInstanceCache.entrySet()) {
-        if (entry.getValue().getInstance().getApp().getName().equals(appName)
+      Iterator<Map.Entry<FirebaseFirestore, FlutterFirebaseFirestoreExtension>> iterator =
+          firestoreInstanceCache.entrySet().iterator();
+      FirebaseFirestore match = null;
+      while (iterator.hasNext()) {
+        Map.Entry<FirebaseFirestore, FlutterFirebaseFirestoreExtension> entry = iterator.next();
+        // getName() throws after delete(), and the SDK has no public check.
+        final String cachedAppName;
+        try {
+          cachedAppName = entry.getValue().getInstance().getApp().getName();
+        } catch (IllegalStateException exception) {
+          if ("FirebaseApp was deleted".equals(exception.getMessage())) {
+            iterator.remove();
+            continue;
+          }
+          throw exception;
+        }
+        if (match == null
+            && cachedAppName.equals(appName)
             && entry.getValue().getDatabaseURL().equals(databaseURL)) {
-          return entry.getKey();
+          match = entry.getKey();
         }
       }
+      return match;
     }
-    return null;
   }
 
   private static void destroyCachedFirebaseFirestoreInstanceForKey(FirebaseFirestore firestore) {
@@ -360,6 +381,15 @@ public static FirebaseFirestore getFirestoreFromPigeon(
     }
   }
 
+  private ExecutorService snapshotDeliveryExecutor(
+      GeneratedAndroidFirebaseFirestore.FirestorePigeonFirebaseApp app) {
+    String key = app.getAppName() + "|" + app.getDatabaseURL();
+    return snapshotDeliveryExecutors.computeIfAbsent(
+        key,
+        ignored ->
+            new ThreadPoolExecutor(0, 1, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>()));
+  }
+
   @Override
   public void loadBundle(
       @NonNull GeneratedAndroidFirebaseFirestore.FirestorePigeonFirebaseApp app,
@@ -563,7 +593,8 @@ public void snapshotsInSyncSetup(
 
     result.success(
         registerEventChannel(
-            METHOD_CHANNEL_NAME + "/snapshotsInSync", new SnapshotsInSyncStreamHandler(firestore)));
+            METHOD_CHANNEL_NAME + "/snapshotsInSync",
+            new SnapshotsInSyncStreamHandler(firestore, snapshotDeliveryExecutor(app))));
   }
 
   @Override
@@ -1019,7 +1050,7 @@ public void querySnapshot(
                 PigeonParser.parsePigeonServerTimestampBehavior(
                     options.getServerTimestampBehavior()),
                 PigeonParser.parseListenSource(source),
-                cachedThreadPool)));
+                snapshotDeliveryExecutor(app))));
   }
 
   @Override

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

@@ -6,25 +6,31 @@
 
 package io.flutter.plugins.firebase.firestore.streamhandler;
 
+import android.os.Handler;
+import android.os.Looper;
 import com.google.firebase.firestore.FirebaseFirestore;
 import com.google.firebase.firestore.ListenerRegistration;
 import io.flutter.plugin.common.EventChannel.EventSink;
 import io.flutter.plugin.common.EventChannel.StreamHandler;
+import java.util.concurrent.Executor;
 
 public class SnapshotsInSyncStreamHandler implements StreamHandler {
 
   ListenerRegistration listenerRegistration;
   FirebaseFirestore firestore;
+  private final Executor snapshotExecutor;
+  private final Handler mainHandler = new Handler(Looper.getMainLooper());
 
-  public SnapshotsInSyncStreamHandler(FirebaseFirestore firestore) {
+  public SnapshotsInSyncStreamHandler(FirebaseFirestore firestore, Executor snapshotExecutor) {
     this.firestore = firestore;
+    this.snapshotExecutor = snapshotExecutor;
   }
 
   @Override
   public void onListen(Object arguments, EventSink events) {
-    Runnable snapshotsInSyncRunnable = () -> events.success(null);
-
-    listenerRegistration = firestore.addSnapshotsInSyncListener(snapshotsInSyncRunnable);
+    listenerRegistration =
+        firestore.addSnapshotsInSyncListener(
+            snapshotExecutor, () -> mainHandler.post(() -> events.success(null)));
   }
 
   @Override

packages/cloud_firestore/cloud_firestore/example/integration_test/instance_e2e.dart

@@ -62,6 +62,109 @@ void runInstanceTests() {
       await snapshots.cancel();
     }, skip: kIsWeb);
 
+    test(
+      'snapshotsInSync() follows a remote query snapshot',
+      () async {
+        final writerApp = await Firebase.initializeApp(
+          name: 'insync-writer-${DateTime.now().microsecondsSinceEpoch}',
+          options: Firebase.app().options,
+        );
+        final writer = FirebaseFirestore.instanceFor(app: writerApp);
+        writer.settings = const Settings(persistenceEnabled: false);
+        writer.useFirestoreEmulator('localhost', 8080);
+
+        final collection = firestore
+            .collection('flutter-tests')
+            .doc('insync-${DateTime.now().microsecondsSinceEpoch}')
+            .collection('items');
+        const docId = 'shared';
+        final events = <String>[];
+
+        final querySub = collection.snapshots().listen((snapshot) {
+          String? value;
+          for (final document in snapshot.docs) {
+            if (document.id == docId) {
+              value = '${document.data()['n']}';
+            }
+          }
+          events.add('query:$value:${snapshot.metadata.hasPendingWrites}');
+        });
+        final syncSub = firestore.snapshotsInSync().listen((_) {
+          events.add('sync');
+        });
+
+        Future<void> waitUntil(bool Function() ready) async {
+          final deadline = DateTime.now().add(const Duration(seconds: 5));
+          while (!ready()) {
+            expect(
+              DateTime.now().isBefore(deadline),
+              isTrue,
+              reason: '$events',
+            );
+            await Future<void>.delayed(const Duration(milliseconds: 20));
+          }
+        }
+
+        Future<void> settle() async {
+          final deadline = DateTime.now().add(const Duration(seconds: 2));
+          var count = events.length;
+          while (DateTime.now().isBefore(deadline)) {
+            await Future<void>.delayed(const Duration(milliseconds: 80));
+            if (events.length == count) {
+              return;
+            }
+            count = events.length;
+          }
+        }
+
+        await waitUntil(
+          () =>
+              events.contains('sync') &&
+              events.any((event) => event.startsWith('query:')),
+        );
+        await settle();
+        events.clear();
+
+        for (var i = 1; i <= 24; i++) {
+          final marker = events.length;
+          await writer.collection(collection.path).doc(docId).set({'n': i});
+          await waitUntil(
+            () => events.sublist(marker).contains('query:$i:false'),
+          );
+          await waitUntil(() => events.sublist(marker).contains('sync'));
+          final slice = events.sublist(marker);
+          final queryAt = slice.indexOf('query:$i:false');
+          final syncAt = slice.indexOf('sync');
+          expect(queryAt, lessThan(syncAt), reason: 'write $i: $slice');
+          await settle();
+        }
+
+        await querySub.cancel();
+        await syncSub.cancel();
+        await writerApp.delete();
+
+        // This database is not cached, so the lookup has to walk every
+        // entry, including the deleted writer's.
+        final uncached = FirebaseFirestore.instanceFor(
+          app: Firebase.app(),
+          databaseId: 'insync-after-delete',
+        );
+        uncached.settings = const Settings(persistenceEnabled: false);
+        uncached.useFirestoreEmulator('localhost', 8080);
+        await uncached.doc('flutter-tests/insync-after-delete').set({
+          'ok': true,
+        });
+        await firestore.doc('flutter-tests/insync-after-delete').set({
+          'ok': true,
+        });
+      },
+      timeout: const Timeout(Duration(minutes: 2)),
+      skip:
+          defaultTargetPlatform != TargetPlatform.iOS &&
+          defaultTargetPlatform != TargetPlatform.macOS &&
+          defaultTargetPlatform != TargetPlatform.android,
+    );
+
     test('enableNetwork()', () async {
       // Write some data while online
       await firestore.enableNetwork();

packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/FLTFirebaseFirestorePlugin.swift

@@ -35,6 +35,8 @@ public class FLTFirebaseFirestorePlugin: NSObject, FlutterPlugin, FLTFirebasePlu
   private var transactionHandlers: [String: TransactionStreamHandler] = [:]
   private let transactionLock = NSLock()
   private let listenersLock = NSLock()
+  private var snapshotDeliveryQueues: [String: DispatchQueue] = [:]
+  private let snapshotDeliveryQueuesLock = NSLock()
 
   static let serverTimestampMap = NSCache<NSNumber, NSString>()
 
@@ -153,6 +155,20 @@ public class FLTFirebaseFirestorePlugin: NSObject, FlutterPlugin, FLTFirebasePlu
     return identifier
   }
 
+  private func snapshotDeliveryQueue(appName: String, databaseURL: String) -> DispatchQueue {
+    let key = "\(appName)|\(databaseURL)"
+    snapshotDeliveryQueuesLock.lock()
+    defer { snapshotDeliveryQueuesLock.unlock() }
+    if let queue = snapshotDeliveryQueues[key] {
+      return queue
+    }
+    let queue = DispatchQueue(
+      label: "io.flutter.plugins.firebase.firestore.snapshot_delivery.\(key)"
+    )
+    snapshotDeliveryQueues[key] = queue
+    return queue
+  }
+
   private func firestore(from pigeonApp: FirestorePigeonFirebaseApp) -> Firestore {
     objc_sync_enter(self)
     defer { objc_sync_exit(self) }
@@ -353,7 +369,10 @@ public class FLTFirebaseFirestorePlugin: NSObject, FlutterPlugin, FLTFirebasePlu
     let firestore = firestore(from: app)
     let identifier = registerEventChannel(
       prefix: kFLTFirebaseFirestoreSnapshotsInSyncEventChannelName,
-      streamHandler: SnapshotsInSyncStreamHandler(firestore: firestore)
+      streamHandler: SnapshotsInSyncStreamHandler(
+        firestore: firestore,
+        snapshotQueue: snapshotDeliveryQueue(appName: app.appName, databaseURL: app.databaseURL)
+      )
     )
     completion(.success(identifier))
   }
@@ -732,7 +751,8 @@ public class FLTFirebaseFirestorePlugin: NSObject, FlutterPlugin, FLTFirebasePlu
         serverTimestampBehavior: PigeonParser.parseServerTimestampBehavior(
           options.serverTimestampBehavior
         ),
-        source: PigeonParser.parseListenSource(source)
+        source: PigeonParser.parseListenSource(source),
+        snapshotQueue: snapshotDeliveryQueue(appName: app.appName, databaseURL: app.databaseURL)
       )
     )
     completion(.success(identifier))

packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/QuerySnapshotStreamHandler.swift

@@ -24,22 +24,22 @@ final class QuerySnapshotStreamHandler: NSObject, FlutterStreamHandler {
   private let serverTimestampBehavior: FirebaseFirestore.ServerTimestampBehavior
   private let source: FirebaseFirestore.ListenSource
   private var listenerRegistration: ListenerRegistration?
-  private let snapshotQueue = DispatchQueue(
-    label: "io.flutter.plugins.firebase.firestore.query_snapshot"
-  )
+  private let snapshotQueue: DispatchQueue
 
   init(
     firestore: Firestore,
     query: Query?,
     includeMetadataChanges: Bool,
     serverTimestampBehavior: FirebaseFirestore.ServerTimestampBehavior,
-    source: FirebaseFirestore.ListenSource
+    source: FirebaseFirestore.ListenSource,
+    snapshotQueue: DispatchQueue
   ) {
     self.firestore = firestore
     self.query = query
     self.includeMetadataChanges = includeMetadataChanges
     self.serverTimestampBehavior = serverTimestampBehavior
     self.source = source
+    self.snapshotQueue = snapshotQueue
   }
 
   func onListen(withArguments arguments: Any?, eventSink events: @escaping FlutterEventSink)

packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/SnapshotsInSyncStreamHandler.swift

@@ -13,18 +13,22 @@ import Foundation
 
 final class SnapshotsInSyncStreamHandler: NSObject, FlutterStreamHandler {
   private let firestore: Firestore
+  private let snapshotQueue: DispatchQueue
   private var listenerRegistration: ListenerRegistration?
 
-  init(firestore: Firestore) {
+  init(firestore: Firestore, snapshotQueue: DispatchQueue) {
     self.firestore = firestore
+    self.snapshotQueue = snapshotQueue
   }
 
   func onListen(withArguments arguments: Any?, eventSink events: @escaping FlutterEventSink)
     -> FlutterError?
   {
-    listenerRegistration = firestore.addSnapshotsInSyncListener {
-      DispatchQueue.main.async {
-        events(nil)
+    listenerRegistration = firestore.addSnapshotsInSyncListener { [snapshotQueue] in
+      snapshotQueue.async {
+        DispatchQueue.main.async {
+          events(nil)
+        }
       }
     }
     return nil