Adding authoritative updates back

Author: schmidt-sebastian

packages/firestore/src/local/indexeddb_schema.ts

@@ -450,7 +450,7 @@ export class DbRemoteDocument {
      * not known, but yet we no the document exist (e.g. it had a successful
      * update applied to it)
      */
-    public unknownDocument: DbUnknownDocument | null,
+    public unknownDocument: DbUnknownDocument | null | undefined,
     /**
      * Set to an instance of a DbNoDocument if it is known that no document
      * exists.
@@ -467,7 +467,7 @@ export class DbRemoteDocument {
      * documents are potentially inconsistent with the document's version on
      * the backend.
      */
-    public hasCommittedMutations?: boolean
+    public hasCommittedMutations: boolean | undefined
   ) {}
 }
 

packages/firestore/src/local/local_store.ts

@@ -50,6 +50,7 @@ import { ReferenceSet } from './reference_set';
 import { RemoteDocumentCache } from './remote_document_cache';
 import { RemoteDocumentChangeBuffer } from './remote_document_change_buffer';
 import { ClientId, SharedClientState } from './shared_client_state';
+import { Document } from '../model/document';
 
 const LOG_TAG = 'LocalStore';
 
@@ -498,13 +499,30 @@ export class LocalStore {
     const documentBuffer = new RemoteDocumentChangeBuffer(this.remoteDocuments);
     return this.persistence.runTransaction('Apply remote event', true, txn => {
       const promises = [] as Array<PersistencePromise<void>>;
+      let authoritativeUpdates = documentKeySet();
       objUtils.forEachNumber(
         remoteEvent.targetChanges,
         (targetId: TargetId, change: TargetChange) => {
           // Do not ref/unref unassigned targetIds - it may lead to leaks.
           let queryData = this.queryDataByTarget[targetId];
           if (!queryData) return;
 
+          // When a global snapshot contains updates (either add or modify) we
+          // can completely trust these updates as authoritative and blindly
+          // apply them to our cache (as a defensive measure to promote
+          // self-healing in the unfortunate case that our cache is ever somehow
+          // corrupted / out-of-sync).
+          //
+          // If the document is only updated while removing it from a target
+          // then watch isn't obligated to send the absolute latest version: it
+          // can send the first version that caused the document not to match.
+          change.addedDocuments.forEach(key => {
+            authoritativeUpdates = authoritativeUpdates.add(key);
+          });
+          change.modifiedDocuments.forEach(key => {
+            authoritativeUpdates = authoritativeUpdates.add(key);
+          });
+
           promises.push(
             this.queryCache
               .removeMatchingKeys(txn, change.removedDocuments, targetId)
@@ -542,9 +560,16 @@ export class LocalStore {
         changedDocKeys = changedDocKeys.add(key);
         promises.push(
           documentBuffer.getEntry(txn, key).next(existingDoc => {
+            // If a document update isn't authoritative, make sure we don't
+            // apply an old document version to the remote cache. We make an
+            // exception for SnapshotVersion.MIN which can happen for
+            // manufactured events (e.g. in the case of a limbo document
+            // resolution failing).
             if (
               existingDoc == null ||
               doc.version.isEqual(SnapshotVersion.MIN) ||
+              (authoritativeUpdates.has(doc.key) &&
+                !existingDoc.hasPendingWrites()) ||
               doc.version.compareTo(existingDoc.version) >= 0
             ) {
               documentBuffer.addEntry(doc);

packages/firestore/src/model/document.ts

@@ -38,6 +38,8 @@ export abstract class MaybeDocument {
     return DocumentKey.comparator(d1.key, d2.key);
   }
 
+  abstract hasPendingWrites(): boolean;
+
   abstract isEqual(other: MaybeDocument | null | undefined): boolean;
 }
 
@@ -92,6 +94,10 @@ export class Document extends MaybeDocument {
     );
   }
 
+  hasPendingWrites(): boolean {
+    return this.hasLocalMutations || this.hasCommittedMutations;
+  }
+
   static compareByField(field: FieldPath, d1: Document, d2: Document): number {
     const v1 = d1.field(field);
     const v2 = d2.field(field);
@@ -117,6 +123,10 @@ export class NoDocument extends MaybeDocument {
     return `NoDocument(${this.key}, ${this.version})`;
   }
 
+  hasPendingWrites(): boolean {
+    return false;
+  }
+
   isEqual(other: MaybeDocument | null | undefined): boolean {
     return (
       other instanceof NoDocument &&
@@ -139,6 +149,10 @@ export class UnknownDocument extends MaybeDocument {
     return `UnknownDocument(${this.key}, ${this.version})`;
   }
 
+  hasPendingWrites(): boolean {
+    return true;
+  }
+
   isEqual(other: MaybeDocument | null | undefined): boolean {
     return (
       other instanceof UnknownDocument &&

packages/firestore/src/model/mutation.ts

@@ -499,6 +499,7 @@ export class TransformMutation extends Mutation {
       maybeDoc,
       mutationResult.transformResults!
     );
+
     const version = mutationResult.version;
     const newData = this.transformObject(doc.data, transformResults);
     return new Document(this.key, version, newData, {

packages/firestore/test/unit/specs/listen_spec.test.ts

@@ -369,6 +369,69 @@ describeSpec('Listens:', [], () => {
     );
   });
 
+  specTest('Waits until Watch catches up to local deletes ', [], () => {
+    const query1 = Query.atPath(path('collection'));
+    const docAv1 = doc('collection/a', 1000, { v: '1' });
+    const docAv2 = doc('collection/a', 2000, { v: '2' });
+    const docAv3 = doc('collection/a', 4000, { v: '3' });
+    const docAv5 = doc('collection/a', 5000, { v: '5' });
+
+    return (
+      spec()
+        // Disable GC so the cache persists across listens.
+        .withGCEnabled(false)
+        .userListens(query1)
+        .watchAcksFull(query1, 1000, docAv1)
+        .expectEvents(query1, { added: [docAv1] })
+        .userDeletes('collection/a')
+        .expectEvents(query1, { removed: [docAv1] })
+        .watchSends({ affects: [query1] }, docAv2)
+        .watchSnapshots(2000)
+        // The write stream acks our delete at version 4000.
+        .writeAcks('collection/a', 4000)
+        .watchSends({ affects: [query1] }, docAv3)
+        .watchSnapshots(3000)
+        // Watch now sends us a document past our deleted version.
+        .watchSends({ affects: [query1] }, docAv5)
+        .watchSnapshots(5000)
+        .expectEvents(query1, { added: [docAv5] })
+    );
+  });
+
+  specTest('Deleted documents in cache are fixed', [], () => {
+    const allQuery = Query.atPath(path('collection'));
+    const setupQuery = allQuery.addFilter(filter('key', '==', 'a'));
+
+    const docAv1 = doc('collection/a', 1000, { key: 'a' });
+    const docDeleted = deletedDoc('collection/a', 2000);
+
+    return (
+      spec()
+        // Presuppose an initial state where the remote document cache has a
+        // broken synthesized delete at a timestamp later than the true version
+        // of the document. This requires both adding and later removing the
+        // document in order to force the watch change aggregator to propagate
+        // the deletion.
+        .withGCEnabled(false)
+        .userListens(setupQuery)
+        .watchAcksFull(setupQuery, 1000, docAv1)
+        .expectEvents(setupQuery, { added: [docAv1], fromCache: false })
+        .watchSends({ removed: [setupQuery] }, docDeleted)
+        .watchSnapshots(2000, [setupQuery], 'resume-token-2000')
+        .watchSnapshots(2000)
+        .expectEvents(setupQuery, { removed: [docAv1], fromCache: false })
+        .userUnlistens(setupQuery)
+        .watchRemoves(setupQuery)
+
+        // Now when the client listens expect the cached NoDocument to be
+        // discarded because the global snapshot version exceeds what came
+        // before.
+        .userListens(allQuery)
+        .watchAcksFull(allQuery, 3000, docAv1)
+        .expectEvents(allQuery, { added: [docAv1], fromCache: false })
+    );
+  });
+
   specTest('Listens are reestablished after network disconnect', [], () => {
     const expectRequestCount = requestCounts =>
       requestCounts.addTarget + requestCounts.removeTarget;

packages/firestore/test/unit/specs/write_spec.test.ts

@@ -882,40 +882,6 @@ describeSpec('Writes:', [], () => {
         .expectEvents(query, {
           metadata: [docA]
         })
-    );
-  });
-
-  specTest('Writes are held during primary failover', ['multi-client'], () => {
-    const query1 = Query.atPath(path('collection'));
-    const query2 = Query.atPath(path('collection/doc'));
-    const docV1 = doc(
-      'collection/doc',
-      0,
-      { v: 1 },
-      { hasLocalMutations: true }
-    );
-    const docV1Committed = doc(
-      'collection/doc',
-      2000,
-      { v: 1 },
-      { hasCommittedMutations: true }
-    );
-    const docV1Acknowledged = doc('collection/doc', 2000, { v: 1 });
-    return (
-      client(0)
-        .userListens(query1)
-        .userSets('collection/doc', { v: 1 })
-        .expectEvents(query1, {
-          hasPendingWrites: true,
-          added: [docV1],
-          fromCache: true
-        })
-        .watchAcksFull(query1, 1000)
-        .expectEvents(query1, {
-          hasPendingWrites: true
-        })
-        .writeAcks('collection/doc', 2000)
-        // Start a new client. DocV1 still has pending writes.
         .client(1)
         .expectUserCallbacks({
           acknowledged: ['collection/a']