Merge

Author: schmidt-sebastian

packages/firestore/src/core/firestore_client.ts

@@ -400,6 +400,8 @@ export class FirestoreClient {
         // TODO(index-free): Use IndexFreeQueryEngine/IndexedQueryEngine as appropriate.
         const queryEngine = new SimpleQueryEngine();
         this.localStore = new LocalStore(this.persistence, queryEngine, user);
+        await this.localStore.start();
+
         if (maybeLruGc) {
           // We're running LRU Garbage collection. Set up the scheduler.
           this.lruScheduler = new LruScheduler(

packages/firestore/src/local/indexeddb_persistence.ts

@@ -317,15 +317,12 @@ export class IndexedDbPersistence implements Persistence {
 
         this.scheduleClientMetadataAndPrimaryLeaseRefreshes();
 
-        return this.startRemoteDocumentCache();
-      })
-      .then(() =>
-        this.simpleDb.runTransaction(
+        return this.simpleDb.runTransaction(
           'readonly-idempotent',
           [DbTargetGlobal.store],
           txn => getHighestListenSequenceNumber(txn)
-        )
-      )
+        );
+      })
       .then(highestListenSequenceNumber => {
         this.listenSequence = new ListenSequence(
           highestListenSequenceNumber,
@@ -341,12 +338,6 @@ export class IndexedDbPersistence implements Persistence {
       });
   }
 
-  private startRemoteDocumentCache(): Promise<void> {
-    return this.simpleDb.runTransaction('readonly', ALL_STORES, txn =>
-      this.remoteDocumentCache.start(txn)
-    );
-  }
-
   setPrimaryStateListener(
     primaryStateListener: PrimaryStateListener
   ): Promise<void> {
@@ -769,10 +760,14 @@ export class IndexedDbPersistence implements Persistence {
           this.listenSequence.next()
         );
 
+<<<<<<< HEAD
         if (
           mode === 'readwrite-primary' ||
           mode === 'readwrite-primary-idempotent'
         ) {
+=======
+        if (mode === 'readwrite-primary') {
+>>>>>>> mrschmidt/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_remote_document_cache.ts

@@ -49,18 +49,10 @@ import { PersistenceTransaction } from './persistence';
 import { PersistencePromise } from './persistence_promise';
 import { RemoteDocumentCache } from './remote_document_cache';
 import { RemoteDocumentChangeBuffer } from './remote_document_change_buffer';
-import {
-  IterateOptions,
-  SimpleDb,
-  SimpleDbStore,
-  SimpleDbTransaction
-} from './simple_db';
+import { IterateOptions, SimpleDbStore } from './simple_db';
 import { ObjectMap } from '../util/obj_map';
 
 export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
-  /** The read time of the last entry consumed by `getNewDocumentChanges()`. */
-  private lastProcessedReadTime = SnapshotVersion.MIN;
-
   /**
    * @param {LocalSerializer} serializer The document serializer.
    * @param {IndexManager} indexManager The query indexes that need to be maintained.
@@ -70,18 +62,6 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
     private readonly indexManager: IndexManager
   ) {}
 
-  /**
-   * Starts up the remote document cache.
-   *
-   * Reads the ID of the last  document change from the documentChanges store.
-   * Existing changes will not be returned as part of
-   * `getNewDocumentChanges()`.
-   */
-  // PORTING NOTE: This is only used for multi-tab synchronization.
-  start(transaction: SimpleDbTransaction): PersistencePromise<void> {
-    return this.synchronizeLastProcessedReadTime(transaction);
-  }
-
   /**
    * Adds the supplied entries to the cache.
    *
@@ -311,14 +291,21 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
       .next(() => results);
   }
 
+  /**
+   * Returns the set of documents that have been updated since the specified read
+   * time.
+   */
+  // PORTING NOTE: This is only used for multi-tab synchronization.
   getNewDocumentChanges(
-    transaction: PersistenceTransaction
-  ): PersistencePromise<MaybeDocumentMap> {
+    transaction: PersistenceTransaction,
+    sinceReadTime: SnapshotVersion
+  ): PersistencePromise<{
+    changedDocs: MaybeDocumentMap;
+    readTime: SnapshotVersion;
+  }> {
     let changedDocs = maybeDocumentMap();
 
-    const lastReadTime = this.serializer.toDbTimestampKey(
-      this.lastProcessedReadTime
-    );
+    let lastReadTime = this.serializer.toDbTimestampKey(sinceReadTime);
 
     const documentsStore = remoteDocumentsStore(transaction);
     const range = IDBKeyRange.lowerBound(lastReadTime, true);
@@ -330,40 +317,48 @@ export class IndexedDbRemoteDocumentCache implements RemoteDocumentCache {
           // the documents directly since we want to keep sentinel deletes.
           const doc = this.serializer.fromDbRemoteDocument(dbRemoteDoc);
           changedDocs = changedDocs.insert(doc.key, doc);
-          this.lastProcessedReadTime = this.serializer.fromDbTimestampKey(
-            dbRemoteDoc.readTime!
-          );
+          lastReadTime = dbRemoteDoc.readTime!;
         }
       )
-      .next(() => changedDocs);
+      .next(() => {
+        return {
+          changedDocs,
+          readTime: this.serializer.fromDbTimestampKey(lastReadTime)
+        };
+      });
   }
 
   /**
-   * Sets the last processed read time to the maximum read time of the backing
-   * object store, allowing calls to getNewDocumentChanges() to return subsequent
-   * changes.
+   * Returns the last document that has changed, as well as the read time of the
+   * last change. If no document has changed, returns SnapshotVersion.MIN.
    */
-  private synchronizeLastProcessedReadTime(
-    transaction: SimpleDbTransaction
-  ): PersistencePromise<void> {
-    const documentsStore = SimpleDb.getStore<
-      DbRemoteDocumentKey,
-      DbRemoteDocument
-    >(transaction, DbRemoteDocument.store);
-
-    // If there are no existing entries, we set `lastProcessedReadTime` to 0.
-    this.lastProcessedReadTime = SnapshotVersion.forDeletedDoc();
-    return documentsStore.iterate(
-      { index: DbRemoteDocument.readTimeIndex, reverse: true },
-      (key, value, control) => {
-        if (value.readTime) {
-          this.lastProcessedReadTime = this.serializer.fromDbTimestampKey(
-            value.readTime
-          );
+  // PORTING NOTE: This is only used for multi-tab synchronization.
+  getLastDocumentChange(
+    transaction: PersistenceTransaction
+  ): PersistencePromise<{
+    changedDoc: MaybeDocument | undefined;
+    readTime: SnapshotVersion;
+  }> {
+    const documentsStore = remoteDocumentsStore(transaction);
+
+    // If there are no existing entries, we return SnapshotVersion.MIN.
+    let readTime = SnapshotVersion.MIN;
+    let changedDoc: MaybeDocument | undefined;
+
+    return documentsStore
+      .iterate(
+        { index: DbRemoteDocument.readTimeIndex, reverse: true },
+        (key, dbRemoteDoc, control) => {
+          changedDoc = this.serializer.fromDbRemoteDocument(dbRemoteDoc);
+          if (dbRemoteDoc.readTime) {
+            readTime = this.serializer.fromDbTimestampKey(dbRemoteDoc.readTime);
+          }
+          control.done();
         }
-        control.done();
-      }
-    );
+      )
+      .next(() => {
+        return { changedDoc, readTime };
+      });
   }
 
   newChangeBuffer(options?: {

packages/firestore/src/local/local_store.ts

@@ -46,6 +46,7 @@ import { SortedMap } from '../util/sorted_map';
 import { LocalDocumentsView } from './local_documents_view';
 import { LocalViewChanges } from './local_view_changes';
 import { LruGarbageCollector, LruResults } from './lru_garbage_collector';
+import { IndexedDbRemoteDocumentCache } from './indexeddb_remote_document_cache';
 import { MutationQueue } from './mutation_queue';
 import { Persistence, PersistenceTransaction } from './persistence';
 import { PersistencePromise } from './persistence_promise';
@@ -177,6 +178,13 @@ export class LocalStore {
     q.canonicalId()
   );
 
+  /**
+   * The read time of the last entry processed by `getNewDocumentChanges()`.
+   *
+   * PORTING NOTE: This is only used for multi-tab synchronization.
+   */
+  private lastDocumentChangeReadTime = SnapshotVersion.MIN;
+
   constructor(
     /** Manages our in-memory or durable persistence. */
     private persistence: Persistence,
@@ -201,6 +209,11 @@ export class LocalStore {
     this.queryEngine.setLocalDocumentsView(this.localDocuments);
   }
 
+  /** Starts the LocalStore. */
+  start(): Promise<void> {
+    return this.synchronizeLastDocumentChangeReadTime();
+  }
+
   /**
    * Tells the LocalStore that the currently authenticated user has changed.
    *
@@ -209,10 +222,13 @@ export class LocalStore {
    */
   // PORTING NOTE: Android and iOS only return the documents affected by the
   // change.
-  handleUserChange(user: User): Promise<UserChangeResult> {
-    return this.persistence.runTransaction(
+  async handleUserChange(user: User): Promise<UserChangeResult> {
+    let newMutationQueue = this.mutationQueue;
+    let newLocalDocuments = this.localDocuments;
+
+    const result = await this.persistence.runTransaction(
       'Handle user change',
-      'readonly',
+      'readonly-idempotent',
       txn => {
         // Swap out the mutation queue, grabbing the pending mutation batches
         // before and after.
@@ -222,17 +238,16 @@ export class LocalStore {
           .next(promisedOldBatches => {
             oldBatches = promisedOldBatches;
 
-            this.mutationQueue = this.persistence.getMutationQueue(user);
+            newMutationQueue = this.persistence.getMutationQueue(user);
 
             // Recreate our LocalDocumentsView using the new
             // MutationQueue.
-            this.localDocuments = new LocalDocumentsView(
+            newLocalDocuments = new LocalDocumentsView(
               this.remoteDocuments,
-              this.mutationQueue,
+              newMutationQueue,
               this.persistence.getIndexManager()
             );
-            this.queryEngine.setLocalDocumentsView(this.localDocuments);
-            return this.mutationQueue.getAllMutationBatches(txn);
+            return newMutationQueue.getAllMutationBatches(txn);
           })
           .next(newBatches => {
             const removedBatchIds: BatchId[] = [];
@@ -257,7 +272,7 @@ export class LocalStore {
 
             // Return the set of all (potentially) changed documents and the list
             // of mutation batch IDs that were affected by change.
-            return this.localDocuments
+            return newLocalDocuments
               .getDocuments(txn, changedKeys)
               .next(affectedDocuments => {
                 return {
@@ -269,7 +284,14 @@ export class LocalStore {
           });
       }
     );
+
+    this.mutationQueue = newMutationQueue;
+    this.localDocuments = newLocalDocuments;
+    this.queryEngine.setLocalDocumentsView(this.localDocuments);
+
+    return result;
   }
+
   /* Accept locally generated Mutations and commit them to storage. */
   localWrite(mutations: Mutation[]): Promise<LocalWriteResult> {
     const localWriteTime = Timestamp.now();
@@ -1037,14 +1059,45 @@ export class LocalStore {
     }
   }
 
+  /**
+   * Returns the set of documents that have been updated since the last call.
+   * If this is the first call, returns the set of changes since client
+   * initialization. Further invocations will return document changes since
+   * the point of rejection.
+   */
   // PORTING NOTE: Multi-tab only.
   getNewDocumentChanges(): Promise<MaybeDocumentMap> {
-    return this.persistence.runTransaction(
-      'Get new document changes',
-      'readonly',
-      txn => {
-        return this.remoteDocuments.getNewDocumentChanges(txn);
-      }
-    );
+    return this.persistence
+      .runTransaction('Get new document changes', 'readonly-idempotent', txn =>
+        this.remoteDocuments.getNewDocumentChanges(
+          txn,
+          this.lastDocumentChangeReadTime
+        )
+      )
+      .then(({ changedDocs, readTime }) => {
+        this.lastDocumentChangeReadTime = readTime;
+        return changedDocs;
+      });
+  }
+
+  /**
+   * Reads the newest document change from persistence and forwards the internal
+   * synchronization marker so that calls to `getNewDocumentChanges()`
+   * only return changes that happened after client initialization.
+   */
+  // PORTING NOTE: Multi-tab only.
+  async synchronizeLastDocumentChangeReadTime(): Promise<void> {
+    if (this.remoteDocuments instanceof IndexedDbRemoteDocumentCache) {
+      const remoteDocumentCache = this.remoteDocuments;
+      return this.persistence
+        .runTransaction(
+          'Synchronize last document change read time',
+          'readonly-idempotent',
+          txn => remoteDocumentCache.getLastDocumentChange(txn)
+        )
+        .then(({ readTime }) => {
+          this.lastDocumentChangeReadTime = readTime;
+        });
+    }
   }
 }