Schema migration that drops held write acks

packages/firestore/src/core/firestore_client.ts

@@ -293,17 +293,15 @@ export class FirestoreClient {
     // TODO(http://b/33384523): For now we just disable garbage collection
     // when persistence is enabled.
     this.garbageCollector = new NoOpGarbageCollector();
-    const storagePrefix = IndexedDbPersistence.buildStoragePrefix(
-      this.databaseInfo
-    );
+
     // Opt to use proto3 JSON in case the platform doesn't support Uint8Array.
     const serializer = new JsonProtoSerializer(this.databaseInfo.databaseId, {
       useProto3Json: true
     });
 
     return Promise.resolve().then(() => {
       const persistence: IndexedDbPersistence = new IndexedDbPersistence(
-        storagePrefix,
+        this.databaseInfo,
         this.clientId,
         this.platform,
         this.asyncQueue,
@@ -322,6 +320,9 @@ export class FirestoreClient {
         );
       }
 
+      const storagePrefix = IndexedDbPersistence.buildStoragePrefix(
+        this.databaseInfo
+      );
       this.sharedClientState = settings.experimentalTabSynchronization
         ? new WebStorageSharedClientState(
             this.asyncQueue,

packages/firestore/src/local/indexeddb_persistence.ts

@@ -26,12 +26,12 @@ import { IndexedDbQueryCache } from './indexeddb_query_cache';
 import { IndexedDbRemoteDocumentCache } from './indexeddb_remote_document_cache';
 import {
   ALL_STORES,
-  createOrUpgradeDb,
   DbClientMetadataKey,
   DbClientMetadata,
   DbPrimaryClient,
   DbPrimaryClientKey,
-  SCHEMA_VERSION
+  SCHEMA_VERSION,
+  SchemaConverter
 } from './indexeddb_schema';
 import { LocalSerializer } from './local_serializer';
 import { MutationQueue } from './mutation_queue';
@@ -204,15 +204,18 @@ export class IndexedDbPersistence implements Persistence {
   private queryCache: IndexedDbQueryCache;
   private remoteDocumentCache: IndexedDbRemoteDocumentCache;
 
+  private readonly persistenceKey: string;
+
   constructor(
-    private readonly persistenceKey: string,
+    private readonly databaseInfo: DatabaseInfo,
     private readonly clientId: ClientId,
     platform: Platform,
     private readonly queue: AsyncQueue,
     serializer: JsonProtoSerializer,
     synchronizeTabs: boolean
   ) {
-    this.dbName = persistenceKey + IndexedDbPersistence.MAIN_DATABASE;
+    this.persistenceKey = IndexedDbPersistence.buildStoragePrefix(databaseInfo);
+    this.dbName = this.persistenceKey + IndexedDbPersistence.MAIN_DATABASE;
     this.serializer = new LocalSerializer(serializer);
     this.document = platform.document;
     this.window = platform.window;
@@ -241,7 +244,11 @@ export class IndexedDbPersistence implements Persistence {
     assert(!this.started, 'IndexedDbPersistence double-started!');
     assert(this.window !== null, "Expected 'window' to be defined");
 
-    return SimpleDb.openOrCreate(this.dbName, SCHEMA_VERSION, createOrUpgradeDb)
+    return SimpleDb.openOrCreate(
+      this.dbName,
+      SCHEMA_VERSION,
+      new SchemaConverter(this.databaseInfo.databaseId)
+    )
       .then(db => {
         this.simpleDb = db;
       })

packages/firestore/src/local/indexeddb_schema.ts

@@ -21,9 +21,15 @@ import { ResourcePath } from '../model/path';
 import { assert } from '../util/assert';
 
 import { encode, EncodedResourcePath } from './encoded_resource_path';
-import { SimpleDbTransaction } from './simple_db';
+import { SimpleDbSchemaConverter, SimpleDbTransaction } from './simple_db';
 import { PersistencePromise } from './persistence_promise';
 import { SnapshotVersion } from '../core/snapshot_version';
+import { BATCHID_UNKNOWN } from '../model/mutation_batch';
+import { IndexedDbMutationQueue } from './indexeddb_mutation_queue';
+import { LocalSerializer } from './local_serializer';
+import { JsonProtoSerializer } from '../remote/serializer';
+import { IndexedDbTransaction } from './indexeddb_persistence';
+import { DatabaseId } from '../core/database_info';
 
 /**
  * Schema Version for the Web client:
@@ -35,66 +41,130 @@ import { SnapshotVersion } from '../core/snapshot_version';
  *    to limbo resolution. Addresses
  *    https://github.com/firebase/firebase-ios-sdk/issues/1548
  * 4. Multi-Tab Support.
+ * 5. Removal of held write acks (not yet active).
  */
 export const SCHEMA_VERSION = 4;
+// TODO(mrschmidt): As SCHEMA_VERSION becomes 5, uncomment the assert in
+// `createOrUpgrade`.
+
+/** Performs database creation and schema upgrades. */
+export class SchemaConverter implements SimpleDbSchemaConverter {
+  private readonly serializer: LocalSerializer;
+
+  constructor(private databaseId: DatabaseId) {
+    this.serializer = new LocalSerializer(
+      new JsonProtoSerializer(this.databaseId, {
+        useProto3Json: true
+      })
+    );
+  }
 
-/**
- * Performs database creation and schema upgrades.
- *
- * Note that in production, this method is only ever used to upgrade the schema
- * to SCHEMA_VERSION. Different values of toVersion are only used for testing
- * and local feature development.
- */
-export function createOrUpgradeDb(
-  db: IDBDatabase,
-  txn: SimpleDbTransaction,
-  fromVersion: number,
-  toVersion: number
-): PersistencePromise<void> {
-  assert(
-    fromVersion < toVersion && fromVersion >= 0 && toVersion <= SCHEMA_VERSION,
-    'Unexpected schema upgrade from v${fromVersion} to v{toVersion}.'
-  );
+  /**
+   * Performs database creation and schema upgrades.
+   *
+   * Note that in production, this method is only ever used to upgrade the schema
+   * to SCHEMA_VERSION. Different values of toVersion are only used for testing
+   * and local feature development.
+   */
+  createOrUpgrade(
+    db: IDBDatabase,
+    txn: SimpleDbTransaction,
+    fromVersion: number,
+    toVersion: number
+  ): PersistencePromise<void> {
+    // assert(
+    //   fromVersion < toVersion && fromVersion >= 0 && toVersion <= SCHEMA_VERSION,
+    //   `Unexpected schema upgrade from v${fromVersion} to v{toVersion}.`
+    // );
+
+    if (fromVersion < 1 && toVersion >= 1) {
+      createPrimaryClientStore(db);
+      createMutationQueue(db);
+      createQueryCache(db);
+      createRemoteDocumentCache(db);
+    }
 
-  if (fromVersion < 1 && toVersion >= 1) {
-    createPrimaryClientStore(db);
-    createMutationQueue(db);
-    createQueryCache(db);
-    createRemoteDocumentCache(db);
-  }
+    // Migration 2 to populate the targetGlobal object no longer needed since
+    // migration 3 unconditionally clears it.
+
+    let p = PersistencePromise.resolve();
+    if (fromVersion < 3 && toVersion >= 3) {
+      // Brand new clients don't need to drop and recreate--only clients that
+      // potentially have corrupt data.
+      if (fromVersion !== 0) {
+        dropQueryCache(db);
+        createQueryCache(db);
+      }
+      p = p.next(() => writeEmptyTargetGlobalEntry(txn));
+    }
 
-  // Migration 2 to populate the targetGlobal object no longer needed since
-  // migration 3 unconditionally clears it.
+    if (fromVersion < 4 && toVersion >= 4) {
+      if (fromVersion !== 0) {
+        // Schema version 3 uses auto-generated keys to generate globally unique
+        // mutation batch IDs (this was previously ensured internally by the
+        // client). To migrate to the new schema, we have to read all mutations
+        // and write them back out. We preserve the existing batch IDs to guarantee
+        // consistency with other object stores. Any further mutation batch IDs will
+        // be auto-generated.
+        p = p.next(() => upgradeMutationBatchSchemaAndMigrateData(db, txn));
+      }
+
+      p = p.next(() => {
+        createClientMetadataStore(db);
+        createRemoteDocumentChangesStore(db);
+      });
+    }
 
-  let p = PersistencePromise.resolve();
-  if (fromVersion < 3 && toVersion >= 3) {
-    // Brand new clients don't need to drop and recreate--only clients that
-    // potentially have corrupt data.
-    if (fromVersion !== 0) {
-      dropQueryCache(db);
-      createQueryCache(db);
+    if (fromVersion < 5 && toVersion >= 5) {
+      p = p.next(() => this.removeAcknowledgedMutations(txn));
     }
-    p = p.next(() => writeEmptyTargetGlobalEntry(txn));
+
+    return p;
   }
 
-  if (fromVersion < 4 && toVersion >= 4) {
-    if (fromVersion !== 0) {
-      // Schema version 3 uses auto-generated keys to generate globally unique
-      // mutation batch IDs (this was previously ensured internally by the
-      // client). To migrate to the new schema, we have to read all mutations
-      // and write them back out. We preserve the existing batch IDs to guarantee
-      // consistency with other object stores. Any further mutation batch IDs will
-      // be auto-generated.
-      p = p.next(() => upgradeMutationBatchSchemaAndMigrateData(db, txn));
-    }
+  private removeAcknowledgedMutations(
+    txn: SimpleDbTransaction
+  ): PersistencePromise<void> {
+    const queuesStore = txn.store<DbMutationQueueKey, DbMutationQueue>(
+      DbMutationQueue.store
+    );
+    const mutationsStore = txn.store<DbMutationBatchKey, DbMutationBatch>(
+      DbMutationBatch.store
+    );
 
-    p = p.next(() => {
-      createClientMetadataStore(db);
-      createRemoteDocumentChangesStore(db);
+    const indexedDbTransaction = new IndexedDbTransaction(txn);
+    return queuesStore.loadAll().next(queues => {
+      return PersistencePromise.forEach(queues, queue => {
+        const mutationQueue = new IndexedDbMutationQueue(
+          queue.userId,
+          this.serializer
+        );
+        const range = IDBKeyRange.bound(
+          [queue.userId, BATCHID_UNKNOWN],
+          [queue.userId, queue.lastAcknowledgedBatchId]
+        );
+
+        return mutationsStore
+          .loadAll(DbMutationBatch.userMutationsIndex, range)
+          .next(dbBatches => {
+            return PersistencePromise.forEach(dbBatches, dbBatch => {
+              assert(
+                dbBatch.userId === queue.userId,
+                `Cannot process batch ${dbBatch.batchId} from unexpected user`
+              );
+              const batch = this.serializer.fromDbMutationBatch(dbBatch);
+              return mutationQueue.removeMutationBatch(
+                indexedDbTransaction,
+                batch
+              );
+            });
+          })
+          .next(() =>
+            mutationQueue.performConsistencyCheck(indexedDbTransaction)
+          );
+      });
     });
   }
-
-  return p;
 }
 
 // TODO(mikelehen): Get rid of "as any" if/when TypeScript fixes their types.

packages/firestore/src/local/simple_db.ts

@@ -25,6 +25,15 @@ import { Code, FirestoreError } from '../util/error';
 
 const LOG_TAG = 'SimpleDb';
 
+export interface SimpleDbSchemaConverter {
+  createOrUpgrade(
+    db: IDBDatabase,
+    txn: SimpleDbTransaction,
+    fromVersion: number,
+    toVersion: number
+  ): PersistencePromise<void>;
+}
+
 /**
  * Provides a wrapper around IndexedDb with a simplified interface that uses
  * Promise-like return values to chain operations. Real promises cannot be used
@@ -37,12 +46,7 @@ export class SimpleDb {
   static openOrCreate(
     name: string,
     version: number,
-    runUpgrade: (
-      db: IDBDatabase,
-      txn: SimpleDbTransaction,
-      fromVersion: number,
-      toVersion: number
-    ) => PersistencePromise<void>
+    schemaConverter: SimpleDbSchemaConverter
   ): Promise<SimpleDb> {
     assert(
       SimpleDb.isAvailable(),
@@ -87,12 +91,14 @@ export class SimpleDb {
         // we wrap that in a SimpleDbTransaction to allow use of our friendlier
         // API for schema migration operations.
         const txn = new SimpleDbTransaction(request.transaction);
-        runUpgrade(db, txn, event.oldVersion, SCHEMA_VERSION).next(() => {
-          debug(
-            LOG_TAG,
-            'Database upgrade to version ' + SCHEMA_VERSION + ' complete'
-          );
-        });
+        schemaConverter
+          .createOrUpgrade(db, txn, event.oldVersion, SCHEMA_VERSION)
+          .next(() => {
+            debug(
+              LOG_TAG,
+              'Database upgrade to version ' + SCHEMA_VERSION + ' complete'
+            );
+          });
       };
     }).toPromise();
   }

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

@@ -19,6 +19,7 @@ import * as EncodedResourcePath from '../../../src/local/encoded_resource_path';
 import { PersistencePromise } from '../../../src/local/persistence_promise';
 import {
   SimpleDb,
+  SimpleDbSchemaConverter,
   SimpleDbStore,
   SimpleDbTransaction
 } from '../../../src/local/simple_db';
@@ -28,6 +29,18 @@ import { path } from '../../util/helpers';
 let db: SimpleDb;
 const sep = '\u0001\u0001';
 
+class EncodedResourcePathSchemaConverter implements SimpleDbSchemaConverter {
+  createOrUpgrade(
+    db: IDBDatabase,
+    txn: SimpleDbTransaction,
+    fromVersion: number,
+    toVersion: number
+  ): PersistencePromise<void> {
+    db.createObjectStore('test');
+    return PersistencePromise.resolve();
+  }
+}
+
 describe('EncodedResourcePath', () => {
   if (!SimpleDb.isAvailable()) {
     console.warn('No IndexedDB. Skipping EncodedResourcePath tests.');
@@ -39,10 +52,11 @@ describe('EncodedResourcePath', () => {
   beforeEach(() => {
     return SimpleDb.delete(dbName)
       .then(() => {
-        return SimpleDb.openOrCreate(dbName, 1, db => {
-          db.createObjectStore('test');
-          return PersistencePromise.resolve();
-        });
+        return SimpleDb.openOrCreate(
+          dbName,
+          1,
+          new EncodedResourcePathSchemaConverter()
+        );
       })
       .then(simpleDb => {
         db = simpleDb;