Adding SnapshotVersion to SharedClientState

packages/firestore/src/core/sync_engine.ts

@@ -505,7 +505,12 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     this.assertSubscribed('rejectListens()');
 
     // PORTING NOTE: Multi-tab only.
-    this.sharedClientState.updateQueryState(targetId, 'rejected', err);
+    this.sharedClientState.updateQueryState(
+      targetId,
+      SnapshotVersion.MIN,
+      'rejected',
+      err
+    );
 
     const limboResolution = this.limboResolutionsByTarget[targetId];
     const limboKey = limboResolution && limboResolution.key;
@@ -552,6 +557,7 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
   // PORTING NOTE: Multi-tab only
   async applyBatchState(
     batchId: BatchId,
+    snapshotVersion: SnapshotVersion,
     batchState: MutationBatchState,
     error?: FirestoreError
   ): Promise<void> {
@@ -570,22 +576,30 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
       return;
     }
 
+    let remoteEvent;
+
     if (batchState === 'pending') {
       // If we are the primary client, we need to send this write to the
       // backend. Secondary clients will ignore these writes since their remote
       // connection is disabled.
       await this.remoteStore.fillWritePipeline();
-    } else if (batchState === 'acknowledged' || batchState === 'rejected') {
+    } else if (batchState === 'acknowledged') {
+      remoteEvent = RemoteEvent.createSynthesizedRemoteEventForSuccessfulWrite(
+        snapshotVersion
+      );
       // NOTE: Both these methods are no-ops for batches that originated from
       // other clients.
-      this.processUserCallback(batchId, error ? error : null);
-
+      this.processUserCallback(batchId, null);
+      this.localStore.removeCachedMutationBatchMetadata(batchId);
+    } else if (batchState === 'rejected') {
+      assert(error !== null, 'Error not set for rejected mutation');
+      this.processUserCallback(batchId, error!);
       this.localStore.removeCachedMutationBatchMetadata(batchId);
     } else {
       fail(`Unknown batchState: ${batchState}`);
     }
 
-    await this.emitNewSnapsAndNotifyLocalStore(documents);
+    await this.emitNewSnapsAndNotifyLocalStore(documents, remoteEvent);
   }
 
   applySuccessfulWrite(
@@ -604,7 +618,13 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     return this.localStore
       .acknowledgeBatch(mutationBatchResult)
       .then(changes => {
-        return this.emitNewSnapsAndNotifyLocalStore(changes);
+        const synthesizedRemoteEvent = RemoteEvent.createSynthesizedRemoteEventForSuccessfulWrite(
+          mutationBatchResult.commitVersion
+        );
+        return this.emitNewSnapsAndNotifyLocalStore(
+          changes,
+          synthesizedRemoteEvent
+        );
       })
       .catch(err => this.ignoreIfPrimaryLeaseLoss(err));
   }
@@ -621,7 +641,12 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     return this.localStore
       .rejectBatch(batchId)
       .then(changes => {
-        this.sharedClientState.updateMutationState(batchId, 'rejected', error);
+        this.sharedClientState.updateMutationState(
+          batchId,
+          SnapshotVersion.MIN,
+          'rejected',
+          error
+        );
         return this.emitNewSnapsAndNotifyLocalStore(changes);
       })
       .catch(err => this.ignoreIfPrimaryLeaseLoss(err));
@@ -761,6 +786,10 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
     const docChangesInAllViews: LocalViewChanges[] = [];
     const queriesProcessed: Array<Promise<void>> = [];
 
+    const snapshotVersion = remoteEvent
+      ? remoteEvent.snapshotVersion
+      : SnapshotVersion.MIN;
+
     this.queryViewsByQuery.forEach((_, queryView) => {
       queriesProcessed.push(
         Promise.resolve()
@@ -792,6 +821,7 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
                 if (this.isPrimary) {
                   this.sharedClientState.updateQueryState(
                     queryView.targetId,
+                    snapshotVersion,
                     viewChange.snapshot.fromCache ? 'not-current' : 'current'
                   );
                 }
@@ -969,6 +999,7 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
   // PORTING NOTE: Multi-tab only
   async applyTargetState(
     targetId: TargetId,
+    snapshotVersion: SnapshotVersion,
     state: QueryTargetState,
     error?: FirestoreError
   ): Promise<void> {
@@ -985,6 +1016,7 @@ export class SyncEngine implements RemoteSyncer, SharedClientStateSyncer {
         case 'not-current': {
           const changes = await this.localStore.getNewDocumentChanges();
           const synthesizedRemoteEvent = RemoteEvent.createSynthesizedRemoteEventForCurrentChange(
+            snapshotVersion,
             targetId,
             state === 'current'
           );

packages/firestore/src/local/local_store.ts

@@ -952,6 +952,7 @@ export class LocalStore {
           // is committed. b/33446471 will remove this reliance.
           this.sharedClientState.updateMutationState(
             batchResult.batch.batchId,
+            SnapshotVersion.MIN, // TODO: Replace with batch version from commit
             'acknowledged'
           );
         });

packages/firestore/src/local/shared_client_state.ts

@@ -34,6 +34,8 @@ import {
 import { AsyncQueue } from '../util/async_queue';
 import { Platform } from '../platform/platform';
 import { TargetIdSet, targetIdSet } from '../model/collections';
+import { SnapshotVersion } from '../core/snapshot_version';
+import { Timestamp } from '../api/timestamp';
 
 const LOG_TAG = 'SharedClientState';
 
@@ -91,6 +93,7 @@ export interface SharedClientState {
    */
   updateMutationState(
     batchId: BatchId,
+    snapshotVersion: SnapshotVersion,
     state: 'acknowledged' | 'rejected',
     error?: FirestoreError
   ): void;
@@ -113,6 +116,7 @@ export interface SharedClientState {
    */
   updateQueryState(
     targetId: TargetId,
+    snapshotVersion: SnapshotVersion,
     state: QueryTargetState,
     error?: FirestoreError
   ): void;
@@ -173,6 +177,7 @@ export interface SharedClientState {
  * encoded as part of the key.
  */
 interface MutationMetadataSchema {
+  snapshotVersion: { seconds: number; nanos: number };
   state: MutationBatchState;
   error?: { code: string; message: string }; // Only set when state === 'rejected'
 }
@@ -186,6 +191,7 @@ export class MutationMetadata {
   constructor(
     readonly user: User,
     readonly batchId: BatchId,
+    readonly snapshotVersion: SnapshotVersion,
     readonly state: MutationBatchState,
     readonly error?: FirestoreError
   ) {
@@ -210,6 +216,7 @@ export class MutationMetadata {
       typeof mutationBatch === 'object' &&
       ['pending', 'acknowledged', 'rejected'].indexOf(mutationBatch.state) !==
         -1 &&
+      isSnapshotVersion(mutationBatch.snapshotVersion) &&
       (mutationBatch.error === undefined ||
         typeof mutationBatch.error === 'object');
 
@@ -231,6 +238,12 @@ export class MutationMetadata {
       return new MutationMetadata(
         user,
         batchId,
+        SnapshotVersion.fromTimestamp(
+          new Timestamp(
+            mutationBatch.snapshotVersion.seconds,
+            mutationBatch.snapshotVersion.nanos
+          )
+        ),
         mutationBatch.state,
         firestoreError
       );
@@ -244,7 +257,12 @@ export class MutationMetadata {
   }
 
   toLocalStorageJSON(): string {
+    const timestamp = this.snapshotVersion.toTimestamp();
     const batchMetadata: MutationMetadataSchema = {
+      snapshotVersion: {
+        seconds: timestamp.seconds,
+        nanos: timestamp.nanoseconds
+      },
       state: this.state
     };
 
@@ -264,6 +282,7 @@ export class MutationMetadata {
  * serialization. The TargetId is omitted as it is encoded as part of the key.
  */
 interface QueryTargetStateSchema {
+  snapshotVersion: { seconds: number; nanos: number };
   state: QueryTargetState;
   error?: { code: string; message: string }; // Only set when state === 'rejected'
 }
@@ -276,6 +295,7 @@ interface QueryTargetStateSchema {
 export class QueryTargetMetadata {
   constructor(
     readonly targetId: TargetId,
+    readonly snapshotVersion: SnapshotVersion,
     readonly state: QueryTargetState,
     readonly error?: FirestoreError
   ) {
@@ -297,6 +317,7 @@ export class QueryTargetMetadata {
 
     let validData =
       typeof targetState === 'object' &&
+      isSnapshotVersion(targetState.snapshotVersion) &&
       ['not-current', 'current', 'rejected'].indexOf(targetState.state) !==
         -1 &&
       (targetState.error === undefined ||
@@ -319,6 +340,12 @@ export class QueryTargetMetadata {
     if (validData) {
       return new QueryTargetMetadata(
         targetId,
+        SnapshotVersion.fromTimestamp(
+          new Timestamp(
+            targetState.snapshotVersion.seconds,
+            targetState.snapshotVersion.nanos
+          )
+        ),
         targetState.state,
         firestoreError
       );
@@ -332,7 +359,12 @@ export class QueryTargetMetadata {
   }
 
   toLocalStorageJSON(): string {
+    const timestamp = this.snapshotVersion.toTimestamp();
     const targetState: QueryTargetStateSchema = {
+      snapshotVersion: {
+        seconds: timestamp.seconds,
+        nanos: timestamp.nanoseconds
+      },
       state: this.state
     };
 
@@ -654,15 +686,16 @@ export class WebStorageSharedClientState implements SharedClientState {
   }
 
   addPendingMutation(batchId: BatchId): void {
-    this.persistMutationState(batchId, 'pending');
+    this.persistMutationState(batchId, SnapshotVersion.MIN, 'pending');
   }
 
   updateMutationState(
     batchId: BatchId,
+    snapshotVersion: SnapshotVersion,
     state: 'acknowledged' | 'rejected',
     error?: FirestoreError
   ): void {
-    this.persistMutationState(batchId, state, error);
+    this.persistMutationState(batchId, snapshotVersion, state, error);
 
     // Once a final mutation result is observed by other clients, they no longer
     // access the mutation's metadata entry. Since LocalStorage replays events
@@ -708,10 +741,11 @@ export class WebStorageSharedClientState implements SharedClientState {
 
   updateQueryState(
     targetId: TargetId,
+    snapshotVersion: SnapshotVersion,
     state: QueryTargetState,
     error?: FirestoreError
   ): void {
-    this.persistQueryTargetState(targetId, state, error);
+    this.persistQueryTargetState(targetId, snapshotVersion, state, error);
   }
 
   handleUserChange(
@@ -836,12 +870,14 @@ export class WebStorageSharedClientState implements SharedClientState {
 
   private persistMutationState(
     batchId: BatchId,
+    snapshotVersion: SnapshotVersion,
     state: MutationBatchState,
     error?: FirestoreError
   ): void {
     const mutationState = new MutationMetadata(
       this.currentUser,
       batchId,
+      snapshotVersion,
       state,
       error
     );
@@ -864,11 +900,17 @@ export class WebStorageSharedClientState implements SharedClientState {
 
   private persistQueryTargetState(
     targetId: TargetId,
+    snapshotVersion: SnapshotVersion,
     state: QueryTargetState,
     error?: FirestoreError
   ): void {
     const targetKey = this.toLocalStorageQueryTargetMetadataKey(targetId);
-    const targetMetadata = new QueryTargetMetadata(targetId, state, error);
+    const targetMetadata = new QueryTargetMetadata(
+      targetId,
+      snapshotVersion,
+      state,
+      error
+    );
     this.setItem(targetKey, targetMetadata.toLocalStorageJSON());
   }
 
@@ -978,6 +1020,7 @@ export class WebStorageSharedClientState implements SharedClientState {
 
     return this.syncEngine.applyBatchState(
       mutationBatch.batchId,
+      mutationBatch.snapshotVersion,
       mutationBatch.state,
       mutationBatch.error
     );
@@ -988,6 +1031,7 @@ export class WebStorageSharedClientState implements SharedClientState {
   ): Promise<void> {
     return this.syncEngine.applyTargetState(
       targetMetadata.targetId,
+      targetMetadata.snapshotVersion,
       targetMetadata.state,
       targetMetadata.error
     );
@@ -1058,6 +1102,7 @@ export class MemorySharedClientState implements SharedClientState {
 
   updateMutationState(
     batchId: BatchId,
+    snapshotVersion: SnapshotVersion,
     state: 'acknowledged' | 'rejected',
     error?: FirestoreError
   ): void {
@@ -1071,6 +1116,7 @@ export class MemorySharedClientState implements SharedClientState {
 
   updateQueryState(
     targetId: TargetId,
+    snapshotVersion: SnapshotVersion,
     state: QueryTargetState,
     error?: FirestoreError
   ): void {
@@ -1112,3 +1158,14 @@ export class MemorySharedClientState implements SharedClientState {
 
   shutdown(): void {}
 }
+
+function isSnapshotVersion(snapshotVersion: {
+  seconds: number;
+  nanos: number;
+}): boolean {
+  return (
+    typeof snapshotVersion === 'object' &&
+    isSafeInteger(snapshotVersion.seconds) &&
+    isSafeInteger(snapshotVersion.nanos)
+  );
+}

packages/firestore/src/local/shared_client_state_syncer.ts

@@ -17,6 +17,7 @@
 import { BatchId, MutationBatchState, TargetId } from '../core/types';
 import { FirestoreError } from '../util/error';
 import { ClientId } from './shared_client_state';
+import { SnapshotVersion } from '../core/snapshot_version';
 
 /** The different states of a watch target. */
 export type QueryTargetState = 'not-current' | 'current' | 'rejected';
@@ -34,13 +35,15 @@ export interface SharedClientStateSyncer {
   /** Applies a mutation state to an existing batch.  */
   applyBatchState(
     batchId: BatchId,
+    snapshotVersion: SnapshotVersion,
     state: MutationBatchState,
     error?: FirestoreError
   ): Promise<void>;
 
   /** Applies a query target change from a different tab. */
   applyTargetState(
     targetId: TargetId,
+    snapshotVersion: SnapshotVersion,
     state: QueryTargetState,
     error?: FirestoreError
   ): Promise<void>;