IndexedDB recovery for waitForPendingWrites() (#3038)

Author: schmidt-sebastian

packages/firestore/src/core/event_manager.ts

@@ -22,10 +22,7 @@ import { Query } from './query';
 import { SyncEngine, SyncEngineListener } from './sync_engine';
 import { OnlineState } from './types';
 import { ChangeType, DocumentViewChange, ViewSnapshot } from './view_snapshot';
-import { logError } from '../util/log';
-import { Code, FirestoreError } from '../util/error';
-
-const LOG_TAG = 'EventManager';
+import { wrapInUserErrorIfRecoverable } from '../util/async_queue';
 
 /**
  * Holds the listeners and the last received ViewSnapshot for a query being
@@ -76,13 +73,11 @@ export class EventManager implements SyncEngineListener {
       try {
         queryInfo.viewSnap = await this.syncEngine.listen(query);
       } catch (e) {
-        const msg = `Initialization of query '${query}' failed: ${e}`;
-        logError(LOG_TAG, msg);
-        if (e.name === 'IndexedDbTransactionError') {
-          listener.onError(new FirestoreError(Code.UNAVAILABLE, msg));
-        } else {
-          throw e;
-        }
+        const firestoreError = wrapInUserErrorIfRecoverable(
+          e,
+          `Initialization of query '${listener.query}' failed`
+        );
+        listener.onError(firestoreError);
         return;
       }
     }

packages/firestore/src/core/sync_engine.ts

@@ -19,7 +19,6 @@ import { User } from '../auth/user';
 import {
   ignoreIfPrimaryLeaseLoss,
   LocalStore,
-  LocalWriteResult,
   MultiTabLocalStore
 } from '../local/local_store';
 import { LocalViewChanges } from '../local/local_view_changes';
@@ -39,7 +38,7 @@ import { RemoteStore } from '../remote/remote_store';
 import { RemoteSyncer } from '../remote/remote_syncer';
 import { debugAssert, fail, hardAssert } from '../util/assert';
 import { Code, FirestoreError } from '../util/error';
-import { logDebug, logError } from '../util/log';
+import { logDebug } from '../util/log';
 import { primitiveComparator } from '../util/misc';
 import { ObjectMap } from '../util/obj_map';
 import { Deferred } from '../util/promise';
@@ -73,7 +72,7 @@ import {
   ViewDocumentChanges
 } from './view';
 import { ViewSnapshot } from './view_snapshot';
-import { AsyncQueue } from '../util/async_queue';
+import { AsyncQueue, wrapInUserErrorIfRecoverable } from '../util/async_queue';
 import { TransactionRunner } from './transaction_runner';
 
 const LOG_TAG = 'SyncEngine';
@@ -353,27 +352,18 @@ export class SyncEngine implements RemoteSyncer {
   async write(batch: Mutation[], userCallback: Deferred<void>): Promise<void> {
     this.assertSubscribed('write()');
 
-    let result: LocalWriteResult;
     try {
-      result = await this.localStore.localWrite(batch);
+      const result = await this.localStore.localWrite(batch);
+      this.sharedClientState.addPendingMutation(result.batchId);
+      this.addMutationCallback(result.batchId, userCallback);
+      await this.emitNewSnapsAndNotifyLocalStore(result.changes);
+      await this.remoteStore.fillWritePipeline();
     } catch (e) {
-      if (e.name === 'IndexedDbTransactionError') {
-        // If we can't persist the mutation, we reject the user callback and
-        // don't send the mutation. The user can then retry the write.
-        logError(LOG_TAG, 'Dropping write that cannot be persisted: ' + e);
-        userCallback.reject(
-          new FirestoreError(Code.UNAVAILABLE, 'Failed to persist write: ' + e)
-        );
-        return;
-      } else {
-        throw e;
-      }
+      // If we can't persist the mutation, we reject the user callback and
+      // don't send the mutation. The user can then retry the write.
+      const error = wrapInUserErrorIfRecoverable(e, `Failed to persist write`);
+      userCallback.reject(error);
     }
-
-    this.sharedClientState.addPendingMutation(result.batchId);
-    this.addMutationCallback(result.batchId, userCallback);
-    await this.emitNewSnapsAndNotifyLocalStore(result.changes);
-    await this.remoteStore.fillWritePipeline();
   }
 
   /**
@@ -588,16 +578,24 @@ export class SyncEngine implements RemoteSyncer {
       );
     }
 
-    const highestBatchId = await this.localStore.getHighestUnacknowledgedBatchId();
-    if (highestBatchId === BATCHID_UNKNOWN) {
-      // Trigger the callback right away if there is no pending writes at the moment.
-      callback.resolve();
-      return;
-    }
+    try {
+      const highestBatchId = await this.localStore.getHighestUnacknowledgedBatchId();
+      if (highestBatchId === BATCHID_UNKNOWN) {
+        // Trigger the callback right away if there is no pending writes at the moment.
+        callback.resolve();
+        return;
+      }
 
-    const callbacks = this.pendingWritesCallbacks.get(highestBatchId) || [];
-    callbacks.push(callback);
-    this.pendingWritesCallbacks.set(highestBatchId, callbacks);
+      const callbacks = this.pendingWritesCallbacks.get(highestBatchId) || [];
+      callbacks.push(callback);
+      this.pendingWritesCallbacks.set(highestBatchId, callbacks);
+    } catch (e) {
+      const firestoreError = wrapInUserErrorIfRecoverable(
+        e,
+        'Initialization of waitForPendingWrites() operation failed'
+      );
+      callback.reject(firestoreError);
+    }
   }
 
   /**

packages/firestore/src/local/indexeddb_persistence.ts

@@ -72,7 +72,12 @@ import {
 import { PersistencePromise } from './persistence_promise';
 import { ClientId } from './shared_client_state';
 import { TargetData } from './target_data';
-import { SimpleDb, SimpleDbStore, SimpleDbTransaction } from './simple_db';
+import {
+  isIndexedDbTransactionError,
+  SimpleDb,
+  SimpleDbStore,
+  SimpleDbTransaction
+} from './simple_db';
 
 const LOG_TAG = 'IndexedDbPersistence';
 
@@ -287,7 +292,7 @@ export class IndexedDbPersistence implements Persistence {
         // having the persistence lock), so it's the first thing we should do.
         return this.updateClientMetadataAndTryBecomePrimary();
       })
-      .then(e => {
+      .then(() => {
         if (!this.isPrimary && !this.allowTabSynchronization) {
           // Fail `start()` if `synchronizeTabs` is disabled and we cannot
           // obtain the primary lease.
@@ -423,7 +428,7 @@ export class IndexedDbPersistence implements Persistence {
     )
       .catch(e => {
         if (!this.allowTabSynchronization) {
-          if (e.name === 'IndexedDbTransactionError') {
+          if (isIndexedDbTransactionError(e)) {
             logDebug(LOG_TAG, 'Failed to extend owner lease: ', e);
             // Proceed with the existing state. Any subsequent access to
             // IndexedDB will verify the lease.

packages/firestore/src/local/local_store.ts

@@ -66,6 +66,7 @@ import { IndexedDbMutationQueue } from './indexeddb_mutation_queue';
 import { IndexedDbRemoteDocumentCache } from './indexeddb_remote_document_cache';
 import { IndexedDbTargetCache } from './indexeddb_target_cache';
 import { extractFieldMask } from '../model/object_value';
+import { isIndexedDbTransactionError } from './simple_db';
 
 const LOG_TAG = 'LocalStore';
 
@@ -727,7 +728,7 @@ export class LocalStore {
         }
       );
     } catch (e) {
-      if (e.name === 'IndexedDbTransactionError') {
+      if (isIndexedDbTransactionError(e)) {
         // If `notifyLocalViewChanges` fails, we did not advance the sequence
         // number for the documents that were included in this transaction.
         // This might trigger them to be deleted earlier than they otherwise

packages/firestore/src/local/lru_garbage_collector.ts

@@ -31,6 +31,7 @@ import {
 } from './persistence';
 import { PersistencePromise } from './persistence_promise';
 import { TargetData } from './target_data';
+import { isIndexedDbTransactionError } from './simple_db';
 
 const LOG_TAG = 'LruGarbageCollector';
 
@@ -273,7 +274,7 @@ export class LruScheduler implements GarbageCollectionScheduler {
         try {
           await localStore.collectGarbage(this.garbageCollector);
         } catch (e) {
-          if (e.name === 'IndexedDbTransactionError') {
+          if (isIndexedDbTransactionError(e)) {
             logDebug(
               LOG_TAG,
               'Ignoring IndexedDB error during garbage collection: ',

packages/firestore/src/local/simple_db.ts

@@ -418,6 +418,13 @@ export class IndexedDbTransactionError extends FirestoreError {
   }
 }
 
+/** Verifies whether `e` is an IndexedDbTransactionError. */
+export function isIndexedDbTransactionError(e: Error): boolean {
+  // Use name equality, as instanceof checks on errors don't work with errors
+  // that wrap other errors.
+  return e.name === 'IndexedDbTransactionError';
+}
+
 /**
  * Wraps an IDBTransaction and exposes a store() method to get a handle to a
  * specific object store.

packages/firestore/src/remote/remote_store.ts

@@ -54,6 +54,7 @@ import {
   WatchTargetChangeState
 } from './watch_change';
 import { ByteString } from '../util/byte_string';
+import { isIndexedDbTransactionError } from '../local/simple_db';
 
 const LOG_TAG = 'RemoteStore';
 
@@ -448,7 +449,7 @@ export class RemoteStore implements TargetMetadataProvider {
    * `enqueueRetryable()`.
    */
   private async disableNetworkUntilRecovery(e: FirestoreError): Promise<void> {
-    if (e.name === 'IndexedDbTransactionError') {
+    if (isIndexedDbTransactionError(e)) {
       debugAssert(
         !this.indexedDbFailed,
         'Unexpected network event when IndexedDB was marked failed.'

packages/firestore/src/util/async_queue.ts

@@ -21,6 +21,7 @@ import { logDebug, logError } from './log';
 import { CancelablePromise, Deferred } from './promise';
 import { ExponentialBackoff } from '../remote/backoff';
 import { PlatformSupport } from '../platform/platform';
+import { isIndexedDbTransactionError } from '../local/simple_db';
 
 const LOG_TAG = 'AsyncQueue';
 
@@ -338,7 +339,7 @@ export class AsyncQueue {
           deferred.resolve();
           this.backoff.reset();
         } catch (e) {
-          if (e.name === 'IndexedDbTransactionError') {
+          if (isIndexedDbTransactionError(e)) {
             logDebug(LOG_TAG, 'Operation failed with retryable error: ' + e);
             this.backoff.backoffAndRun(retryingOp);
           } else {
@@ -501,3 +502,19 @@ export class AsyncQueue {
     this.delayedOperations.splice(index, 1);
   }
 }
+
+/**
+ * Returns a FirestoreError that can be surfaced to the user if the provided
+ * error is an IndexedDbTransactionError. Re-throws the error otherwise.
+ */
+export function wrapInUserErrorIfRecoverable(
+  e: Error,
+  msg: string
+): FirestoreError {
+  logError(LOG_TAG, `${msg}: ${e}`);
+  if (isIndexedDbTransactionError(e)) {
+    return new FirestoreError(Code.UNAVAILABLE, `${msg}: ${e}`);
+  } else {
+    throw e;
+  }
+}