Add sequence numbers and sync them (#1172)

* Add sequence numbers and sync them

* Lint and comments

* FirestoreClient and ListenSequence review feedback

* Make webStorage non-optional

* Return a promise of removeddocs from removeMutationBatch

* [AUTOMATED]: Prettier Code Styling

* Rename sequence number parsing function

* [AUTOMATED]: Prettier Code Styling

* IndexedDb startup refactoring to throw proper errors early

* Renamed tests

* Refactor Syncer interface for sequence numbers to conform to other examples

* [AUTOMATED]: Prettier Code Styling

* Shorten the comment

* Drop 'only'

* Add comment re multiClientParams

* SharedClientState existing doesn't imply multiple clients

* Remove start() from Persistence interface (#1179)

* WIP on switching to static constructors

* Remove start() from Persistence interface, use static helpers for IndexedDbPersistence construction

* [AUTOMATED]: Prettier Code Styling

* Mark start() private

* Remove unused import

* Use explicit type, switch async to Promise.resolve

* Export the type

* Condense a few lines

* Use persistence key to scope sequence numbers to a project

* [AUTOMATED]: Prettier Code Styling

* Fix test to use new sequence number key

* updateSequenceNumber -> writeSequenceNumber

* Add comments to interface, switch to assert for sequenceNumberHandler

* [AUTOMATED]: Prettier Code Styling

Author: Greg Soltis

packages/firestore/src/core/firestore_client.ts

@@ -301,17 +301,7 @@ export class FirestoreClient {
       useProto3Json: true
     });
 
-    return Promise.resolve().then(() => {
-      const persistence: IndexedDbPersistence = new IndexedDbPersistence(
-        storagePrefix,
-        this.clientId,
-        this.platform,
-        this.asyncQueue,
-        serializer,
-        settings.experimentalTabSynchronization
-      );
-      this.persistence = persistence;
-
+    return Promise.resolve().then(async () => {
       if (
         settings.experimentalTabSynchronization &&
         !WebStorageSharedClientState.isAvailable(this.platform)
@@ -322,16 +312,32 @@ export class FirestoreClient {
         );
       }
 
-      this.sharedClientState = settings.experimentalTabSynchronization
-        ? new WebStorageSharedClientState(
-            this.asyncQueue,
-            this.platform,
-            storagePrefix,
-            this.clientId,
-            user
-          )
-        : new MemorySharedClientState();
-      return persistence.start();
+      if (settings.experimentalTabSynchronization) {
+        this.sharedClientState = new WebStorageSharedClientState(
+          this.asyncQueue,
+          this.platform,
+          storagePrefix,
+          this.clientId,
+          user
+        );
+        this.persistence = await IndexedDbPersistence.createMultiClientIndexedDbPersistence(
+          storagePrefix,
+          this.clientId,
+          this.platform,
+          this.asyncQueue,
+          serializer,
+          { sequenceNumberSyncer: this.sharedClientState }
+        );
+      } else {
+        this.sharedClientState = new MemorySharedClientState();
+        this.persistence = await IndexedDbPersistence.createIndexedDbPersistence(
+          storagePrefix,
+          this.clientId,
+          this.platform,
+          this.asyncQueue,
+          serializer
+        );
+      }
     });
   }
 
@@ -344,7 +350,7 @@ export class FirestoreClient {
     this.garbageCollector = new EagerGarbageCollector();
     this.persistence = new MemoryPersistence(this.clientId);
     this.sharedClientState = new MemorySharedClientState();
-    return this.persistence.start();
+    return Promise.resolve();
   }
 
   /**

packages/firestore/src/core/listen_sequence.ts

@@ -0,0 +1,71 @@
+/**
+ * Copyright 2018 Google Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ListenSequenceNumber } from './types';
+
+/**
+ * `SequenceNumberSyncer` defines the methods required to keep multiple instances of a `ListenSequence` in sync.
+ */
+export interface SequenceNumberSyncer {
+  // Notify the syncer that a new sequence number has been used.
+  writeSequenceNumber(sequenceNumber: ListenSequenceNumber): void;
+  // Setting this property allows the syncer to notify when a sequence number has been used, and
+  // and lets the ListenSequence adjust its internal previous value accordingly.
+  sequenceNumberHandler:
+    | ((sequenceNumber: ListenSequenceNumber) => void)
+    | null;
+}
+
+/**
+ * `ListenSequence` is a monotonic sequence. It is initialized with a minimum value to
+ * exceed. All subsequent calls to next will return increasing values. If provided with a
+ * `SequenceNumberSyncer`, it will additionally bump its next value when told of a new value, as well as write out
+ * sequence numbers that it produces via `next()`.
+ */
+export class ListenSequence {
+  static readonly INVALID: ListenSequenceNumber = -1;
+
+  private writeNewSequenceNumber?: (
+    newSequenceNumber: ListenSequenceNumber
+  ) => void;
+
+  constructor(
+    private previousValue: ListenSequenceNumber,
+    sequenceNumberSyncer?: SequenceNumberSyncer
+  ) {
+    if (sequenceNumberSyncer) {
+      sequenceNumberSyncer.sequenceNumberHandler = sequenceNumber =>
+        this.setPreviousValue(sequenceNumber);
+      this.writeNewSequenceNumber = sequenceNumber =>
+        sequenceNumberSyncer.writeSequenceNumber(sequenceNumber);
+    }
+  }
+
+  private setPreviousValue(
+    externalPreviousValue: ListenSequenceNumber
+  ): ListenSequenceNumber {
+    this.previousValue = Math.max(externalPreviousValue, this.previousValue);
+    return this.previousValue;
+  }
+
+  next(): ListenSequenceNumber {
+    const nextValue = ++this.previousValue;
+    if (this.writeNewSequenceNumber) {
+      this.writeNewSequenceNumber(nextValue);
+    }
+    return nextValue;
+  }
+}

packages/firestore/src/core/types.ts

@@ -26,6 +26,8 @@ export type BatchId = number;
  */
 export type TargetId = number;
 
+export type ListenSequenceNumber = number;
+
 // TODO(b/35918695): In GRPC / node, tokens are Uint8Array. In WebChannel,
 // they're strings. We should probably (de-)serialize to a common internal type.
 export type ProtoByteString = Uint8Array | string;

packages/firestore/src/local/indexeddb_mutation_queue.ts

@@ -41,8 +41,11 @@ import { LocalSerializer } from './local_serializer';
 import { MutationQueue } from './mutation_queue';
 import { PersistenceTransaction } from './persistence';
 import { PersistencePromise } from './persistence_promise';
-import { SimpleDbStore } from './simple_db';
-import { IndexedDbPersistence } from './indexeddb_persistence';
+import { SimpleDbStore, SimpleDbTransaction } from './simple_db';
+import {
+  IndexedDbPersistence,
+  IndexedDbTransaction
+} from './indexeddb_persistence';
 
 /** A mutation queue for a specific user, backed by IndexedDB. */
 export class IndexedDbMutationQueue implements MutationQueue {
@@ -479,41 +482,18 @@ export class IndexedDbMutationQueue implements MutationQueue {
     transaction: PersistenceTransaction,
     batch: MutationBatch
   ): PersistencePromise<void> {
-    const mutationStore = mutationsStore(transaction);
-    const indexTxn = documentMutationsStore(transaction);
-    const promises: Array<PersistencePromise<void>> = [];
-
-    const range = IDBKeyRange.only(batch.batchId);
-    let numDeleted = 0;
-    const removePromise = mutationStore.iterate(
-      { range },
-      (key, value, control) => {
-        numDeleted++;
-        return control.delete();
-      }
-    );
-    promises.push(
-      removePromise.next(() => {
-        assert(
-          numDeleted === 1,
-          'Dangling document-mutation reference found: Missing batch ' +
-            batch.batchId
-        );
-      })
-    );
-    for (const mutation of batch.mutations) {
-      const indexKey = DbDocumentMutation.key(
-        this.userId,
-        mutation.key.path,
-        batch.batchId
-      );
-      promises.push(indexTxn.delete(indexKey));
+    return removeMutationBatch(
+      (transaction as IndexedDbTransaction).simpleDbTransaction,
+      this.userId,
+      batch
+    ).next(removedDocuments => {
+      this.removeCachedMutationKeys(batch.batchId);
       if (this.garbageCollector !== null) {
-        this.garbageCollector.addPotentialGarbageKey(mutation.key);
+        for (const key of removedDocuments) {
+          this.garbageCollector.addPotentialGarbageKey(key);
+        }
       }
-    }
-    this.removeCachedMutationKeys(batch.batchId);
-    return PersistencePromise.waitFor(promises);
+    });
   }
 
   removeCachedMutationKeys(batchId: BatchId): void {
@@ -598,6 +578,54 @@ export class IndexedDbMutationQueue implements MutationQueue {
   }
 }
 
+/**
+ * Delete a mutation batch and the associated document mutations.
+ * @return A PersistencePromise of the document mutations that were removed.
+ */
+export function removeMutationBatch(
+  txn: SimpleDbTransaction,
+  userId: string,
+  batch: MutationBatch
+): PersistencePromise<DocumentKey[]> {
+  const mutationStore = txn.store<DbMutationBatchKey, DbMutationBatch>(
+    DbMutationBatch.store
+  );
+  const indexTxn = txn.store<DbDocumentMutationKey, DbDocumentMutation>(
+    DbDocumentMutation.store
+  );
+  const promises: Array<PersistencePromise<void>> = [];
+
+  const range = IDBKeyRange.only(batch.batchId);
+  let numDeleted = 0;
+  const removePromise = mutationStore.iterate(
+    { range },
+    (key, value, control) => {
+      numDeleted++;
+      return control.delete();
+    }
+  );
+  promises.push(
+    removePromise.next(() => {
+      assert(
+        numDeleted === 1,
+        'Dangling document-mutation reference found: Missing batch ' +
+          batch.batchId
+      );
+    })
+  );
+  const removedDocuments: DocumentKey[] = [];
+  for (const mutation of batch.mutations) {
+    const indexKey = DbDocumentMutation.key(
+      userId,
+      mutation.key.path,
+      batch.batchId
+    );
+    promises.push(indexTxn.delete(indexKey));
+    removedDocuments.push(mutation.key);
+  }
+  return PersistencePromise.waitFor(promises).next(() => removedDocuments);
+}
+
 function convertStreamToken(token: ProtoByteString): string {
   if (token instanceof Uint8Array) {
     // TODO(b/78771403): Convert tokens to strings during deserialization