Implement sequence number migration (#1374)

* Implement sequence number migration

Author: Greg Soltis

packages/firestore/src/local/indexeddb_schema.ts

@@ -23,6 +23,7 @@ import { SnapshotVersion } from '../core/snapshot_version';
 import { BATCHID_UNKNOWN } from '../model/mutation_batch';
 import { encode, EncodedResourcePath } from './encoded_resource_path';
 import { removeMutationBatch } from './indexeddb_mutation_queue';
+import { getHighestListenSequenceNumber } from './indexeddb_query_cache';
 import { dbDocumentSize } from './indexeddb_remote_document_cache';
 import { LocalSerializer } from './local_serializer';
 import { PersistencePromise } from './persistence_promise';
@@ -39,8 +40,10 @@ import { SimpleDbSchemaConverter, SimpleDbTransaction } from './simple_db';
  *    https://github.com/firebase/firebase-ios-sdk/issues/1548
  * 4. Multi-Tab Support.
  * 5. Removal of held write acks.
+ * 6. Create document global for tracking document cache size.
+ * 7. Ensure every cached document has a sentinel row with a sequence number.
  */
-export const SCHEMA_VERSION = 6;
+export const SCHEMA_VERSION = 7;
 
 /** Performs database creation and schema upgrades. */
 export class SchemaConverter implements SimpleDbSchemaConverter {
@@ -115,6 +118,10 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
       });
     }
 
+    if (fromVersion < 7 && toVersion >= 7) {
+      p = p.next(() => this.ensureSequenceNumbers(txn));
+    }
+
     return p;
   }
 
@@ -172,6 +179,53 @@ export class SchemaConverter implements SimpleDbSchemaConverter {
       });
     });
   }
+
+  /**
+   * Ensures that every document in the remote document cache has a corresponding sentinel row
+   * with a sequence number. Missing rows are given the most recently used sequence number.
+   */
+  private ensureSequenceNumbers(
+    txn: SimpleDbTransaction
+  ): PersistencePromise<void> {
+    const documentTargetStore = txn.store<
+      DbTargetDocumentKey,
+      DbTargetDocument
+    >(DbTargetDocument.store);
+    const documentsStore = txn.store<DbRemoteDocumentKey, DbRemoteDocument>(
+      DbRemoteDocument.store
+    );
+
+    return getHighestListenSequenceNumber(txn).next(currentSequenceNumber => {
+      const writeSentinelKey = (
+        path: ResourcePath
+      ): PersistencePromise<void> => {
+        return documentTargetStore.put(
+          new DbTargetDocument(0, encode(path), currentSequenceNumber)
+        );
+      };
+
+      const promises: Array<PersistencePromise<void>> = [];
+      return documentsStore
+        .iterate((key, doc) => {
+          const path = new ResourcePath(key);
+          const docSentinelKey = sentinelKey(path);
+          promises.push(
+            documentTargetStore.get(docSentinelKey).next(maybeSentinel => {
+              if (!maybeSentinel) {
+                return writeSentinelKey(path);
+              } else {
+                return PersistencePromise.resolve();
+              }
+            })
+          );
+        })
+        .next(() => PersistencePromise.waitFor(promises));
+    });
+  }
+}
+
+function sentinelKey(path: ResourcePath): DbTargetDocumentKey {
+  return [0, encode(path)];
 }
 
 // TODO(mikelehen): Get rid of "as any" if/when TypeScript fixes their types.

packages/firestore/test/unit/local/indexeddb_persistence.test.ts

@@ -17,6 +17,7 @@
 import { expect } from 'chai';
 import { PersistenceSettings } from '../../../src/api/database';
 import { SnapshotVersion } from '../../../src/core/snapshot_version';
+import { decode, encode } from '../../../src/local/encoded_resource_path';
 import { IndexedDbPersistence } from '../../../src/local/indexeddb_persistence';
 import {
   DbDocumentMutation,
@@ -32,6 +33,8 @@ import {
   DbRemoteDocumentGlobalKey,
   DbRemoteDocumentKey,
   DbTarget,
+  DbTargetDocument,
+  DbTargetDocumentKey,
   DbTargetGlobal,
   DbTargetGlobalKey,
   DbTargetKey,
@@ -521,6 +524,92 @@ describe('IndexedDbSchema: createOrUpgradeDb', () => {
       });
     });
   });
+
+  it('can upgrade from version 6 to 7', async () => {
+    const oldSequenceNumber = 1;
+    // Set the highest sequence number to this value so that untagged documents
+    // will pick up this value.
+    const newSequenceNumber = 2;
+    await withDb(6, db => {
+      const serializer = TEST_SERIALIZER;
+
+      const sdb = new SimpleDb(db);
+      return sdb.runTransaction('readwrite', V6_STORES, txn => {
+        const targetGlobalStore = txn.store<DbTargetGlobalKey, DbTargetGlobal>(
+          DbTargetGlobal.store
+        );
+        const remoteDocumentStore = txn.store<
+          DbRemoteDocumentKey,
+          DbRemoteDocument
+        >(DbRemoteDocument.store);
+        const targetDocumentStore = txn.store<
+          DbTargetDocumentKey,
+          DbTargetDocument
+        >(DbTargetDocument.store);
+        return targetGlobalStore
+          .get(DbTargetGlobal.key)
+          .next(metadata => {
+            expect(metadata).to.not.be.null;
+            metadata!.highestListenSequenceNumber = newSequenceNumber;
+            return targetGlobalStore.put(DbTargetGlobal.key, metadata!);
+          })
+          .next(() => {
+            // Set up some documents (we only need the keys)
+            // For the odd ones, add sentinel rows.
+            const promises: Array<PersistencePromise<void>> = [];
+            for (let i = 0; i < 10; i++) {
+              const document = doc('docs/doc_' + i, 1, { foo: 'bar' });
+              promises.push(
+                remoteDocumentStore.put(
+                  document.key.path.toArray(),
+                  serializer.toDbRemoteDocument(document)
+                )
+              );
+              if (i % 2 === 1) {
+                promises.push(
+                  targetDocumentStore.put(
+                    new DbTargetDocument(
+                      0,
+                      encode(document.key.path),
+                      oldSequenceNumber
+                    )
+                  )
+                );
+              }
+            }
+            return PersistencePromise.waitFor(promises);
+          });
+      });
+    });
+
+    // Now run the migration and verify
+    await withDb(7, db => {
+      const sdb = new SimpleDb(db);
+      return sdb.runTransaction('readonly', V6_STORES, txn => {
+        const targetDocumentStore = txn.store<
+          DbTargetDocumentKey,
+          DbTargetDocument
+        >(DbTargetDocument.store);
+        const range = IDBKeyRange.bound(
+          [0],
+          [1],
+          /*lowerOpen=*/ false,
+          /*upperOpen=*/ true
+        );
+        return targetDocumentStore.iterate(
+          { range },
+          ([_, path], targetDocument) => {
+            const decoded = decode(path);
+            const lastSegment = decoded.lastSegment();
+            const docNum = +lastSegment.split('_')[1];
+            const expected =
+              docNum % 2 === 1 ? oldSequenceNumber : newSequenceNumber;
+            expect(targetDocument.sequenceNumber).to.equal(expected);
+          }
+        );
+      });
+    });
+  });
 });
 
 describe('IndexedDb: canActAsPrimary', () => {