Allow remote updates from watch to heal a cache with synthesized deletes in it

packages/firestore/src/local/local_store.ts

@@ -425,13 +425,30 @@ export class LocalStore {
     const documentBuffer = new RemoteDocumentChangeBuffer(this.remoteDocuments);
     return this.persistence.runTransaction('Apply remote event', 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.targetIds[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)
@@ -463,13 +480,15 @@ export class LocalStore {
         changedDocKeys = changedDocKeys.add(key);
         promises.push(
           documentBuffer.getEntry(txn, key).next(existingDoc => {
-            // Make sure we don't apply an old document version to the remote
-            // cache, though 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 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) ||
               doc.version.compareTo(existingDoc.version) >= 0
             ) {
               documentBuffer.addEntry(doc);

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

@@ -271,7 +271,10 @@ describeSpec('Listens:', [], () => {
         // (the document falls out) and send us a snapshot that's ahead of
         // docAv3 (which is already in our cache).
         .userListens(visibleQuery, 'resume-token-1000')
-        .watchAcksFull(visibleQuery, 5000, docAv2)
+        .watchAcks(visibleQuery)
+        .watchSends({ removed: [visibleQuery] }, docAv2)
+        .watchCurrents(visibleQuery, 'resume-token-5000')
+        .watchSnapshots(5000)
         .expectEvents(visibleQuery, { fromCache: false })
         .userUnlistens(visibleQuery)
         .watchRemoves(visibleQuery)
@@ -283,6 +286,86 @@ describeSpec('Listens:', [], () => {
     );
   });
 
+  specTest('Individual (deleted) documents cannot revert', [], () => {
+    const allQuery = Query.atPath(path('collection'));
+    const visibleQuery = Query.atPath(path('collection')).addFilter(
+      filter('visible', '==', true)
+    );
+    const docAv1 = doc('collection/a', 1000, { visible: true, v: 'v1000' });
+    const docAv2 = doc('collection/a', 2000, { visible: false, v: 'v2000' });
+    const docAv3 = deletedDoc('collection/a', 3000);
+
+    return (
+      spec()
+        // Disable GC so the cache persists across listens.
+        .withGCEnabled(false)
+        .userListens(visibleQuery)
+        .watchAcksFull(visibleQuery, 1000, docAv1)
+        .expectEvents(visibleQuery, { added: [docAv1] })
+        .userUnlistens(visibleQuery)
+        .watchRemoves(visibleQuery)
+        .userListens(allQuery)
+        .expectEvents(allQuery, { added: [docAv1], fromCache: true })
+        .watchAcks(allQuery)
+        .watchSends({ removed: [allQuery] }, docAv3)
+        .watchCurrents(allQuery, 'resume-token-4000')
+        .watchSnapshots(4000)
+        .expectEvents(allQuery, { removed: [docAv1], fromCache: false })
+        .userUnlistens(allQuery)
+        .watchRemoves(allQuery)
+        // Supposing we sent a resume token for visibleQuery, watch could catch
+        // us up to docAV2 since that's the last relevant change to the query
+        // (the document falls out) and send us a snapshot that's ahead of
+        // docAv3 (which is already in our cache).
+        .userListens(visibleQuery, 'resume-token-1000')
+        .watchAcks(visibleQuery)
+        .watchSends({ removed: [visibleQuery] }, docAv2)
+        .watchCurrents(visibleQuery, 'resume-token-5000')
+        .watchSnapshots(5000)
+        .expectEvents(visibleQuery, { fromCache: false })
+        .userUnlistens(visibleQuery)
+        .watchRemoves(visibleQuery)
+        // Listen to allQuery again and make sure we still get no docs.
+        .userListens(allQuery, 'resume-token-4000')
+        .watchAcksFull(allQuery, 6000)
+        .expectEvents(allQuery, { fromCache: false })
+    );
+  });
+
+  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;