Schema migration that drops held write acks

Author: schmidt-sebastian

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

@@ -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,12 @@ 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.databaseInfo.databaseId,
+      this.dbName,
+      SCHEMA_VERSION,
+      createOrUpgradeDb
+    )
       .then(db => {
         this.simpleDb = db;
       })

packages/firestore/src/local/indexeddb_schema.ts

@@ -24,6 +24,12 @@ import { encode, EncodedResourcePath } from './encoded_resource_path';
 import { 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,8 +41,11 @@ 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
+// `createOrUpgradeDb`.
 
 /**
  * Performs database creation and schema upgrades.
@@ -47,14 +56,15 @@ export const SCHEMA_VERSION = 4;
  */
 export function createOrUpgradeDb(
   db: IDBDatabase,
+  databaseId: DatabaseId,
   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}.'
-  );
+  // assert(
+  //   fromVersion < toVersion && fromVersion >= 0 && toVersion <= SCHEMA_VERSION,
+  //   `Unexpected schema upgrade from v${fromVersion} to v{toVersion}.`
+  // );
 
   if (fromVersion < 1 && toVersion >= 1) {
     createPrimaryClientStore(db);
@@ -94,6 +104,10 @@ export function createOrUpgradeDb(
     });
   }
 
+  if (fromVersion < 5 && toVersion >= 5) {
+    p = p.next(() => removeAcknowledgedMutations(db, databaseId, txn));
+  }
+
   return p;
 }
 
@@ -295,6 +309,62 @@ function upgradeMutationBatchSchemaAndMigrateData(
   });
 }
 
+function removeAcknowledgedMutations(
+  db: IDBDatabase,
+  databaseId: DatabaseId,
+  txn: SimpleDbTransaction
+): PersistencePromise<void> {
+  const queuesStore = txn.store<DbMutationQueueKey, DbMutationQueue>(
+    DbMutationQueue.store
+  );
+  const mutationsStore = txn.store<DbMutationBatchKey, DbMutationBatch>(
+    DbMutationBatch.store
+  );
+  const serializer = new LocalSerializer(
+    new JsonProtoSerializer(databaseId, {
+      useProto3Json: true
+    })
+  );
+
+  const indexedDbTransaction = new IndexedDbTransaction(txn);
+  return queuesStore.loadAll().next(queues => {
+    let p = PersistencePromise.resolve();
+    for (const queue of queues) {
+      p = p.next(() => {
+        const mutationQueue = new IndexedDbMutationQueue(
+          queue.userId,
+          serializer
+        );
+        const range = IDBKeyRange.bound(
+          [queue.userId, BATCHID_UNKNOWN],
+          [queue.userId, queue.lastAcknowledgedBatchId]
+        );
+
+        return mutationsStore
+          .loadAll(DbMutationBatch.userMutationsIndex, range)
+          .next(dbBatches => {
+            let removeP = PersistencePromise.resolve();
+            for (const dbBatch of dbBatches) {
+              assert(
+                dbBatch.userId === queue.userId,
+                `Cannot process batch ${dbBatch.batchId} from unexpected user`
+              );
+              const batch = serializer.fromDbMutationBatch(dbBatch);
+              removeP = removeP.next(() =>
+                mutationQueue.removeMutationBatch(indexedDbTransaction, batch)
+              );
+            }
+            return removeP;
+          })
+          .next(() =>
+            mutationQueue.performConsistencyCheck(indexedDbTransaction)
+          );
+      });
+    }
+    return p;
+  });
+}
+
 /**
  * An object to be stored in the 'documentMutations' store in IndexedDb.
  *

packages/firestore/src/local/simple_db.ts

@@ -22,6 +22,7 @@ import { SCHEMA_VERSION } from './indexeddb_schema';
 import { AnyJs } from '../util/misc';
 import { Deferred } from '../util/promise';
 import { Code, FirestoreError } from '../util/error';
+import { DatabaseId } from '../core/database_info';
 
 const LOG_TAG = 'SimpleDb';
 
@@ -35,10 +36,12 @@ const LOG_TAG = 'SimpleDb';
 export class SimpleDb {
   /** Opens the specified database, creating or upgrading it if necessary. */
   static openOrCreate(
+    databaseId: DatabaseId,
     name: string,
     version: number,
     runUpgrade: (
       db: IDBDatabase,
+      databaseId: DatabaseId,
       txn: SimpleDbTransaction,
       fromVersion: number,
       toVersion: number
@@ -87,12 +90,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'
-          );
-        });
+        runUpgrade(db, databaseId, 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

@@ -24,6 +24,7 @@ import {
 } from '../../../src/local/simple_db';
 import { ResourcePath } from '../../../src/model/path';
 import { path } from '../../util/helpers';
+import { INDEXEDDB_TEST_DATABASE_ID } from './persistence_test_helpers';
 
 let db: SimpleDb;
 const sep = '\u0001\u0001';
@@ -39,10 +40,15 @@ describe('EncodedResourcePath', () => {
   beforeEach(() => {
     return SimpleDb.delete(dbName)
       .then(() => {
-        return SimpleDb.openOrCreate(dbName, 1, db => {
-          db.createObjectStore('test');
-          return PersistencePromise.resolve();
-        });
+        return SimpleDb.openOrCreate(
+          INDEXEDDB_TEST_DATABASE_ID,
+          dbName,
+          1,
+          db => {
+            db.createObjectStore('test');
+            return PersistencePromise.resolve();
+          }
+        );
       })
       .then(simpleDb => {
         db = simpleDb;