Add sequence number to QueryData (#1194)

Greg Soltis · web-flow · commit dbc8d70966c4 · 2018-09-05T16:03:15.000-07:00
* Add sequence numbers to QueryData

* [AUTOMATED]: Prettier Code Styling

* Dont used separate db prefix

* [AUTOMATED]: Prettier Code Styling
diff --git a/packages/firestore/src/core/sync_engine.ts b/packages/firestore/src/core/sync_engine.ts
@@ -69,6 +69,7 @@ import { ClientId, SharedClientState } from '../local/shared_client_state';
 import { SortedSet } from '../util/sorted_set';
 import * as objUtils from '../util/obj';
 import { isPrimaryLeaseLostError } from '../local/indexeddb_persistence';
+import { ListenSequence } from './listen_sequence';
 
 const LOG_TAG = 'SyncEngine';
 
@@ -742,7 +743,12 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
       const query = Query.atPath(key.path);
       this.limboResolutionsByTarget[limboTargetId] = new LimboResolution(key);
       this.remoteStore.listen(
-        new QueryData(query, limboTargetId, QueryPurpose.LimboResolution)
+        new QueryData(
+          query,
+          limboTargetId,
+          QueryPurpose.LimboResolution,
+          ListenSequence.INVALID
+        )
       );
       this.limboTargetsByKey = this.limboTargetsByKey.insert(
         key,
diff --git a/packages/firestore/src/local/indexeddb_query_cache.ts b/packages/firestore/src/local/indexeddb_query_cache.ts
@@ -86,6 +86,14 @@ export class IndexedDbQueryCache implements QueryCache {
     });
   }
 
+  getHighestSequenceNumber(
+    transaction: PersistenceTransaction
+  ): PersistencePromise<ListenSequenceNumber> {
+    return getHighestListenSequenceNumber(
+      (transaction as IndexedDbTransaction).simpleDbTransaction
+    );
+  }
+
   setTargetsMetadata(
     transaction: PersistenceTransaction,
     highestListenSequenceNumber: number,
@@ -96,6 +104,9 @@ export class IndexedDbQueryCache implements QueryCache {
       if (lastRemoteSnapshotVersion) {
         metadata.lastRemoteSnapshotVersion = lastRemoteSnapshotVersion.toTimestamp();
       }
+      if (highestListenSequenceNumber > metadata.highestListenSequenceNumber) {
+        metadata.highestListenSequenceNumber = highestListenSequenceNumber;
+      }
       return this.saveMetadata(transaction, metadata);
     });
   }
@@ -165,13 +176,17 @@ export class IndexedDbQueryCache implements QueryCache {
     queryData: QueryData,
     metadata: DbTargetGlobal
   ): boolean {
+    let updated = false;
     if (queryData.targetId > metadata.highestTargetId) {
       metadata.highestTargetId = queryData.targetId;
-      return true;
+      updated = true;
     }
 
-    // TODO(GC): add sequence number check
-    return false;
+    if (queryData.sequenceNumber > metadata.highestListenSequenceNumber) {
+      metadata.highestListenSequenceNumber = queryData.sequenceNumber;
+      updated = true;
+    }
+    return updated;
   }
 
   getQueryCount(
diff --git a/packages/firestore/src/local/local_serializer.ts b/packages/firestore/src/local/local_serializer.ts
@@ -175,6 +175,7 @@ export class LocalSerializer {
       query,
       dbTarget.targetId,
       QueryPurpose.Listen,
+      dbTarget.lastListenSequenceNumber,
       version,
       dbTarget.resumeToken
     );
@@ -216,7 +217,7 @@ export class LocalSerializer {
       queryData.query.canonicalId(),
       dbTimestamp,
       resumeToken,
-      0,
+      queryData.sequenceNumber,
       queryProto
     );
   }
diff --git a/packages/firestore/src/local/local_store.ts b/packages/firestore/src/local/local_store.ts
@@ -459,7 +459,8 @@ export class LocalStore {
             const oldQueryData = queryData;
             queryData = queryData.copy({
               resumeToken,
-              snapshotVersion: remoteEvent.snapshotVersion
+              snapshotVersion: remoteEvent.snapshotVersion,
+              sequenceNumber: txn.currentSequenceNumber
             });
             this.queryDataByTarget[targetId] = queryData;
 
@@ -527,7 +528,7 @@ export class LocalStore {
             );
             return this.queryCache.setTargetsMetadata(
               txn,
-              /*highestSequenceNumber=*/ 0,
+              txn.currentSequenceNumber,
               remoteVersion
             );
           });
@@ -653,7 +654,12 @@ export class LocalStore {
             return PersistencePromise.resolve();
           } else {
             return this.queryCache.allocateTargetId(txn).next(targetId => {
-              queryData = new QueryData(query, targetId, QueryPurpose.Listen);
+              queryData = new QueryData(
+                query,
+                targetId,
+                QueryPurpose.Listen,
+                txn.currentSequenceNumber
+              );
               return this.queryCache.addQueryData(txn, queryData);
             });
           }
diff --git a/packages/firestore/src/local/memory_query_cache.ts b/packages/firestore/src/local/memory_query_cache.ts
@@ -16,7 +16,7 @@
 
 import { Query } from '../core/query';
 import { SnapshotVersion } from '../core/snapshot_version';
-import { TargetId } from '../core/types';
+import { ListenSequenceNumber, TargetId } from '../core/types';
 import { DocumentKeySet } from '../model/collections';
 import { DocumentKey } from '../model/document_key';
 import { ObjectMap } from '../util/obj_map';
@@ -40,6 +40,8 @@ export class MemoryQueryCache implements QueryCache {
   private lastRemoteSnapshotVersion = SnapshotVersion.MIN;
   /** The highest numbered target ID encountered. */
   private highestTargetId: TargetId = 0;
+  /** The highest sequence number encountered. */
+  private highestSequenceNumber: ListenSequenceNumber = 0;
   /**
    * A ordered bidirectional mapping between documents and the remote target
    * IDs.
@@ -56,6 +58,12 @@ export class MemoryQueryCache implements QueryCache {
     return PersistencePromise.resolve(this.lastRemoteSnapshotVersion);
   }
 
+  getHighestSequenceNumber(
+    transaction: PersistenceTransaction
+  ): PersistencePromise<ListenSequenceNumber> {
+    return PersistencePromise.resolve(this.highestSequenceNumber);
+  }
+
   allocateTargetId(
     transaction: PersistenceTransaction
   ): PersistencePromise<TargetId> {
@@ -72,6 +80,9 @@ export class MemoryQueryCache implements QueryCache {
     if (lastRemoteSnapshotVersion) {
       this.lastRemoteSnapshotVersion = lastRemoteSnapshotVersion;
     }
+    if (highestListenSequenceNumber > this.highestSequenceNumber) {
+      this.highestSequenceNumber = highestListenSequenceNumber;
+    }
     return PersistencePromise.resolve();
   }
 
@@ -81,7 +92,9 @@ export class MemoryQueryCache implements QueryCache {
     if (targetId > this.highestTargetId) {
       this.highestTargetId = targetId;
     }
-    // TODO(GC): track sequence number
+    if (queryData.sequenceNumber > this.highestSequenceNumber) {
+      this.highestSequenceNumber = queryData.sequenceNumber;
+    }
   }
 
   addQueryData(
diff --git a/packages/firestore/src/local/query_cache.ts b/packages/firestore/src/local/query_cache.ts
@@ -16,7 +16,7 @@
 
 import { Query } from '../core/query';
 import { SnapshotVersion } from '../core/snapshot_version';
-import { TargetId } from '../core/types';
+import { ListenSequenceNumber, TargetId } from '../core/types';
 import { DocumentKeySet } from '../model/collections';
 
 import { GarbageSource } from './garbage_source';
@@ -45,6 +45,14 @@ export interface QueryCache extends GarbageSource {
     transaction: PersistenceTransaction
   ): PersistencePromise<SnapshotVersion>;
 
+  /**
+   * @return The highest sequence number observed, including any that might be
+   *         persisted on-disk.
+   */
+  getHighestSequenceNumber(
+    transaction: PersistenceTransaction
+  ): PersistencePromise<ListenSequenceNumber>;
+
   /**
    * Set the highest listen sequence number and optionally updates the
    * snapshot version of the last consistent snapshot received from the backend
diff --git a/packages/firestore/src/local/query_data.ts b/packages/firestore/src/local/query_data.ts
@@ -16,7 +16,7 @@
 
 import { Query } from '../core/query';
 import { SnapshotVersion } from '../core/snapshot_version';
-import { ProtoByteString, TargetId } from '../core/types';
+import { ListenSequenceNumber, ProtoByteString, TargetId } from '../core/types';
 import { emptyByteString } from '../platform/platform';
 
 /** An enumeration of the different purposes we have for queries. */
@@ -47,6 +47,8 @@ export class QueryData {
     readonly targetId: TargetId,
     /** The purpose of the query. */
     readonly purpose: QueryPurpose,
+    /** The sequence number of the last transaction during which this query data was modified */
+    readonly sequenceNumber: ListenSequenceNumber,
     /** The latest snapshot version seen for this target. */
     readonly snapshotVersion: SnapshotVersion = SnapshotVersion.MIN,
     /**
@@ -65,11 +67,15 @@ export class QueryData {
   copy(overwrite: {
     resumeToken?: ProtoByteString;
     snapshotVersion?: SnapshotVersion;
+    sequenceNumber?: ListenSequenceNumber;
   }): QueryData {
     return new QueryData(
       this.query,
       this.targetId,
       this.purpose,
+      overwrite.sequenceNumber === undefined
+        ? this.sequenceNumber
+        : overwrite.sequenceNumber,
       overwrite.snapshotVersion === undefined
         ? this.snapshotVersion
         : overwrite.snapshotVersion,
@@ -83,6 +89,7 @@ export class QueryData {
     return (
       this.targetId === other.targetId &&
       this.purpose === other.purpose &&
+      this.sequenceNumber === other.sequenceNumber &&
       this.snapshotVersion.isEqual(other.snapshotVersion) &&
       this.resumeToken === other.resumeToken &&
       this.query.isEqual(other.query)
diff --git a/packages/firestore/src/remote/remote_store.ts b/packages/firestore/src/remote/remote_store.ts
@@ -442,7 +442,8 @@ export class RemoteStore implements TargetMetadataProvider {
       const requestQueryData = new QueryData(
         queryData.query,
         targetId,
-        QueryPurpose.ExistenceFilterMismatch
+        QueryPurpose.ExistenceFilterMismatch,
+        queryData.sequenceNumber
       );
       this.sendWatchRequest(requestQueryData);
     });
diff --git a/packages/firestore/test/unit/local/query_cache.test.ts b/packages/firestore/test/unit/local/query_cache.test.ts
@@ -34,6 +34,7 @@ import {
 import * as persistenceHelpers from './persistence_test_helpers';
 import { TestGarbageCollector } from './test_garbage_collector';
 import { TestQueryCache } from './test_query_cache';
+import { Timestamp } from '../../../src/api/timestamp';
 
 describe('MemoryQueryCache', () => {
   genericQueryCacheTests(persistenceHelpers.testMemoryPersistence);
@@ -51,6 +52,44 @@ describe('IndexedDbQueryCache', () => {
   });
 
   genericQueryCacheTests(() => persistencePromise);
+
+  it('persists metadata across restarts', async () => {
+    const db1 = await persistencePromise;
+
+    const queryCache1 = new TestQueryCache(db1, db1.getQueryCache());
+    expect(await queryCache1.getHighestSequenceNumber()).to.equal(0);
+
+    const originalSequenceNumber = 1234;
+    const targetId = 5;
+    const snapshotVersion = SnapshotVersion.fromTimestamp(new Timestamp(1, 2));
+    const query = Query.atPath(path('rooms'));
+    await queryCache1.addQueryData(
+      new QueryData(
+        query,
+        targetId,
+        QueryPurpose.Listen,
+        originalSequenceNumber,
+        snapshotVersion
+      )
+    );
+    // Snapshot version needs to be set separately
+    await queryCache1.setTargetsMetadata(
+      originalSequenceNumber,
+      snapshotVersion
+    );
+    await db1.shutdown(/* deleteData= */ false);
+
+    const db2 = await persistenceHelpers.testIndexedDbPersistence({
+      dontPurgeData: true
+    });
+    const queryCache2 = new TestQueryCache(db2, db2.getQueryCache());
+    expect(await queryCache2.getHighestSequenceNumber()).to.equal(
+      originalSequenceNumber
+    );
+    const actualSnapshotVersion = await queryCache2.getLastRemoteSnapshotVersion();
+    expect(snapshotVersion.isEqual(actualSnapshotVersion)).to.be.true;
+    await db2.shutdown(/* deleteData= */ true);
+  });
 });
 
 /**
@@ -70,6 +109,7 @@ function genericQueryCacheTests(
    * Creates a new QueryData object from the the given parameters, synthesizing
    * a resume token from the snapshot version.
    */
+  let previousSequenceNumber = 0;
   function testQueryData(
     query: Query,
     targetId: TargetId,
@@ -84,6 +124,7 @@ function genericQueryCacheTests(
       query,
       targetId,
       QueryPurpose.Listen,
+      ++previousSequenceNumber,
       snapshotVersion,
       resumeToken
     );
@@ -96,7 +137,9 @@ function genericQueryCacheTests(
   });
 
   afterEach(async () => {
-    await persistence.shutdown(/* deleteData= */ true);
+    if (persistence.started) {
+      await persistence.shutdown(/* deleteData= */ true);
+    }
   });
 
   it('returns null for query not in cache', () => {
@@ -344,4 +387,25 @@ function genericQueryCacheTests(
         );
       });
   });
+
+  it('sets the highest sequence number', async () => {
+    const query1 = new QueryData(QUERY_ROOMS, 1, QueryPurpose.Listen, 10);
+    await cache.addQueryData(query1);
+    const query2 = new QueryData(QUERY_HALLS, 2, QueryPurpose.Listen, 20);
+    await cache.addQueryData(query2);
+    expect(await cache.getHighestSequenceNumber()).to.equal(20);
+
+    // Sequence numbers can never come down
+    await cache.removeQueryData(query2);
+    expect(await cache.getHighestSequenceNumber()).to.equal(20);
+
+    const query3 = new QueryData(QUERY_GARAGES, 3, QueryPurpose.Listen, 100);
+    await cache.addQueryData(query3);
+    expect(await cache.getHighestSequenceNumber()).to.equal(100);
+
+    await cache.removeQueryData(query1);
+    expect(await cache.getHighestSequenceNumber()).to.equal(100);
+    await cache.removeQueryData(query3);
+    expect(await cache.getHighestSequenceNumber()).to.equal(100);
+  });
 }
diff --git a/packages/firestore/test/unit/local/test_query_cache.ts b/packages/firestore/test/unit/local/test_query_cache.ts
@@ -16,7 +16,7 @@
 
 import { Query } from '../../../src/core/query';
 import { SnapshotVersion } from '../../../src/core/snapshot_version';
-import { TargetId } from '../../../src/core/types';
+import { ListenSequenceNumber, TargetId } from '../../../src/core/types';
 import { Persistence } from '../../../src/local/persistence';
 import { QueryCache } from '../../../src/local/query_cache';
 import { QueryData } from '../../../src/local/query_data';
@@ -70,6 +70,16 @@ export class TestQueryCache {
     );
   }
 
+  getHighestSequenceNumber(): Promise<ListenSequenceNumber> {
+    return this.persistence.runTransaction(
+      'getHighestSequenceNumber',
+      false,
+      txn => {
+        return this.cache.getHighestSequenceNumber(txn);
+      }
+    );
+  }
+
   allocateTargetId(): Promise<TargetId> {
     return this.persistence.runTransaction('allocateTargetId', false, txn => {
       return this.cache.allocateTargetId(txn);
@@ -125,7 +135,7 @@ export class TestQueryCache {
   }
 
   setTargetsMetadata(
-    highestListenSequenceNumber: number,
+    highestListenSequenceNumber: ListenSequenceNumber,
     lastRemoteSnapshotVersion?: SnapshotVersion
   ): Promise<void> {
     return this.persistence.runTransaction('setTargetsMetadata', true, txn =>
diff --git a/packages/firestore/test/unit/local/web_storage_shared_client_state.test.ts b/packages/firestore/test/unit/local/web_storage_shared_client_state.test.ts
diff --git a/packages/firestore/test/unit/remote/node/serializer.test.ts b/packages/firestore/test/unit/remote/node/serializer.test.ts
diff --git a/packages/firestore/test/unit/specs/spec_test_runner.ts b/packages/firestore/test/unit/specs/spec_test_runner.ts
diff --git a/packages/firestore/test/util/helpers.ts b/packages/firestore/test/util/helpers.ts
