Make applyRemoteEvent idempotent

packages/firestore/src/local/indexeddb_index_manager.ts

@@ -40,15 +40,28 @@ export class IndexedDbIndexManager implements IndexManager {
    */
   private collectionParentsCache = new MemoryCollectionParentIndex();
 
+  /**
+   * Adds a new entry to the collection parent index.
+   *
+   * Repeated calls for the same collectionPath should be avoided within a
+   * transaction as IndexedDbIndexManager only caches writes once a transaction
+   * has been committed.
+   */
   addToCollectionParentIndex(
     transaction: PersistenceTransaction,
     collectionPath: ResourcePath
   ): PersistencePromise<void> {
     assert(collectionPath.length % 2 === 1, 'Expected a collection path.');
-    if (this.collectionParentsCache.add(collectionPath)) {
-      assert(collectionPath.length >= 1, 'Invalid collection path.');
+    if (!this.collectionParentsCache.has(collectionPath)) {
       const collectionId = collectionPath.lastSegment();
       const parentPath = collectionPath.popLast();
+
+      transaction.addOnCommittedListener(() => {
+        // Add the collection to the in memory cache only if the transaction was
+        // successfully committed.
+        this.collectionParentsCache.add(collectionPath);
+      });
+
       return collectionParentsStore(transaction).put({
         collectionId,
         parent: encode(parentPath)

packages/firestore/src/local/indexeddb_mutation_queue.ts

@@ -181,23 +181,28 @@ export class IndexedDbMutationQueue implements MutationQueue {
       this.documentKeysByBatchId[batchId] = batch.keys();
 
       const promises: Array<PersistencePromise<void>> = [];
+      let collectionParents = new SortedSet<ResourcePath>((l, r) =>
+        primitiveComparator(l.canonicalString(), r.canonicalString())
+      );
       for (const mutation of mutations) {
         const indexKey = DbDocumentMutation.key(
           this.userId,
           mutation.key.path,
           batchId
         );
+        collectionParents = collectionParents.add(mutation.key.path.popLast());
         promises.push(mutationStore.put(dbBatch));
         promises.push(
           documentStore.put(indexKey, DbDocumentMutation.PLACEHOLDER)
         );
+      }
+
+      collectionParents.forEach(parent => {
         promises.push(
-          this.indexManager.addToCollectionParentIndex(
-            transaction,
-            mutation.key.path.popLast()
-          )
+          this.indexManager.addToCollectionParentIndex(transaction, parent)
         );
-      }
+      });
+
       return PersistencePromise.waitFor(promises).next(() => batch);
     });
   }

packages/firestore/src/local/indexeddb_persistence.ts

@@ -760,7 +760,10 @@ export class IndexedDbPersistence implements Persistence {
           this.listenSequence.next()
         );
 
-        if (mode === 'readwrite-primary') {
+        if (
+          mode === 'readwrite-primary' ||
+          mode === 'readwrite-primary-idempotent'
+        ) {
           // While we merely verify that we have (or can acquire) the lease
           // immediately, we wait to extend the primary lease until after
           // executing transactionOperation(). This ensures that even if the

packages/firestore/src/local/indexeddb_query_cache.ts

@@ -164,7 +164,7 @@ export class IndexedDbQueryCache implements QueryCache {
         const queryData = this.serializer.fromDbTarget(value);
         if (
           queryData.sequenceNumber <= upperBound &&
-          activeTargetIds[queryData.targetId] === undefined
+          activeTargetIds.get(queryData.targetId) === null
         ) {
           count++;
           promises.push(this.removeQueryData(txn, queryData));

packages/firestore/src/local/indexeddb_remote_document_cache.ts

@@ -29,7 +29,10 @@ import {
 } from '../model/collections';
 import { Document, MaybeDocument, NoDocument } from '../model/document';
 import { DocumentKey } from '../model/document_key';
+import { ResourcePath } from '../model/path';
+import { primitiveComparator } from '../util/misc';
 import { SortedMap } from '../util/sorted_map';
+import { SortedSet } from '../util/sorted_set';
 
 import { SnapshotVersion } from '../core/snapshot_version';
 import { assert, fail } from '../util/assert';
@@ -71,12 +74,7 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
     doc: DbRemoteDocument
   ): PersistencePromise<void> {
     const documentStore = remoteDocumentsStore(transaction);
-    return documentStore.put(dbKey(key), doc).next(() => {
-      this.indexManager.addToCollectionParentIndex(
-        transaction,
-        key.path.popLast()
-      );
-    });
+    return documentStore.put(dbKey(key), doc);
   }
 
   /**
@@ -449,6 +447,10 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
 
       let sizeDelta = 0;
 
+      let collectionParents = new SortedSet<ResourcePath>((l, r) =>
+        primitiveComparator(l.canonicalString(), r.canonicalString())
+      );
+
       this.changes.forEach((key, maybeDocument) => {
         const previousSize = this.documentSizes.get(key);
         assert(
@@ -464,6 +466,8 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
             maybeDocument,
             this.readTime
           );
+          collectionParents = collectionParents.add(key.path.popLast());
+
           const size = dbDocumentSize(doc);
           sizeDelta += size - previousSize!;
           promises.push(this.documentCache.addEntry(transaction, key, doc));
@@ -487,6 +491,15 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
         }
       });
 
+      collectionParents.forEach(parent => {
+        promises.push(
+          this.documentCache.indexManager.addToCollectionParentIndex(
+            transaction,
+            parent
+          )
+        );
+      });
+
       promises.push(this.documentCache.updateMetadata(transaction, sizeDelta));
 
       return PersistencePromise.waitFor(promises);