Make applyRemoteEvent idempotent

Author: schmidt-sebastian

packages/firestore/src/local/indexeddb_index_manager.ts

@@ -22,7 +22,6 @@ import { decode, encode } from './encoded_resource_path';
 import { IndexManager } from './index_manager';
 import { IndexedDbPersistence } from './indexeddb_persistence';
 import { DbCollectionParent, DbCollectionParentKey } from './indexeddb_schema';
-import { MemoryCollectionParentIndex } from './memory_index_manager';
 import { PersistenceTransaction } from './persistence';
 import { PersistencePromise } from './persistence_promise';
 import { SimpleDbStore } from './simple_db';
@@ -31,30 +30,17 @@ import { SimpleDbStore } from './simple_db';
  * A persisted implementation of IndexManager.
  */
 export class IndexedDbIndexManager implements IndexManager {
-  /**
-   * An in-memory copy of the index entries we've already written since the SDK
-   * launched. Used to avoid re-writing the same entry repeatedly.
-   *
-   * This is *NOT* a complete cache of what's in persistence and so can never be used to
-   * satisfy reads.
-   */
-  private collectionParentsCache = new MemoryCollectionParentIndex();
-
   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.');
-      const collectionId = collectionPath.lastSegment();
-      const parentPath = collectionPath.popLast();
-      return collectionParentsStore(transaction).put({
-        collectionId,
-        parent: encode(parentPath)
-      });
-    }
-    return PersistencePromise.resolve();
+    const collectionId = collectionPath.lastSegment();
+    const parentPath = collectionPath.popLast();
+    return collectionParentsStore(transaction).put({
+      collectionId,
+      parent: encode(parentPath)
+    });
   }
 
   getCollectionParents(

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

@@ -764,7 +764,10 @@ export class IndexedDbPersistence implements Persistence {
       simpleDbMode,
       ALL_STORES,
       simpleDbTxn => {
-        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';
@@ -91,12 +94,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);
   }
 
   /**
@@ -454,6 +452,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(
@@ -469,6 +471,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));
@@ -492,6 +496,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);