Skip to content

IndexedDB recovery for waitForPendingWrites() #3038

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
May 18, 2020
Merged

IndexedDB recovery for waitForPendingWrites() #3038

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
0852c5d
IndexedDB recovery for waitForPendingWrites()
schmidt-sebastian May 8, 2020
8f47d7a
Lint
schmidt-sebastian May 8, 2020
a0eb1df
Merge branch 'master' into mrschmidt/pendingwrites
schmidt-sebastian May 13, 2020
d68d8d1
Review
schmidt-sebastian May 13, 2020
001cf9c
Fix test
schmidt-sebastian May 14, 2020
ae2fa31
Merge branch 'master' into mrschmidt/pendingwrites
schmidt-sebastian May 14, 2020
8bc241f
Fix tests after merge
schmidt-sebastian May 14, 2020
d09a3a3
Merge branch 'master' into mrschmidt/pendingwrites
schmidt-sebastian May 15, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 34 additions & 35 deletions packages/firestore/src/core/event_manager.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import { OnlineState } from './types';
import { ChangeType, DocumentViewChange, ViewSnapshot } from './view_snapshot';
import { logError } from '../util/log';
import { Code, FirestoreError } from '../util/error';
import { executeWithIndexedDbRecovery } from '../util/async_queue';

const LOG_TAG = 'EventManager';

Expand Down Expand Up @@ -62,47 +63,45 @@ export class EventManager implements SyncEngineListener {
this.syncEngine.subscribe(this);
}

async listen(listener: QueryListener): Promise<void> {
const query = listener.query;
let firstListen = false;
listen(listener: QueryListener): Promise<void> {
return executeWithIndexedDbRecovery(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can't say this pattern is really different from try/catch, except it's hiding that.

What do you think of the more direct:

try {
  const query = listener.query;
  //
} catch (e) {
  const msg = `Initialization of query '${query}' failed: ${e}`;	
  logError(LOG_TAG, msg);
  throwIfUnrecoverable(e);
  listener.onError(new FirestoreError(Code.UNAVAILABLE, msg));
}

As I see it:

  • It's not really more code than what you have here
  • It uses async/await try/cache directly and looks more natural/synchronous-like
  • On the flip side, there aren't a series of callbacks that you have to remember their meaning.
  • There isn't an extra log message about "Internal operation failed"
  • It still abstracts the handling of the IndexedDbTransactionError type

Building on this, you could actually cut the verbosity of each call down by making a checkRecoverable that takes the message, logs it, and then throws if unrecoverable or returns the FirestoreError if it is?

The catch blocks would look like:

const wrapped = checkRecoverable(e, `Initialization of query '${query}' failed: ${e}`);
listener.onError(new FirestoreError(Code.UNAVAILABLE, msg));

The implementation would look like:

checkRecoverable(e: Error, message: string): FirestoreError {
  logError(LOG_TAG, msg);
  if (e.name == ....) {
    return new FirestoreError(unavailable, message);
  } else {
    throw e;
  }
}

WDYT?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I tried really hard to somehow push the try/catch into my helper, but it might have been a bit too mush. I updated the PR to use a helper that returns a FirestoreError if the error is an IndexedDB error, otherwise it throws.

async () => {
const query = listener.query;
let firstListen = false;

let queryInfo = this.queries.get(query);
if (!queryInfo) {
firstListen = true;
queryInfo = new QueryListenersInfo();
}
let queryInfo = this.queries.get(query);
if (!queryInfo) {
firstListen = true;
queryInfo = new QueryListenersInfo();
}

if (firstListen) {
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;
if (firstListen) {
queryInfo.viewSnap = await this.syncEngine.listen(query);
}
return;
}
}

this.queries.set(query, queryInfo);
queryInfo.listeners.push(listener);
this.queries.set(query, queryInfo);
queryInfo.listeners.push(listener);

// Run global snapshot listeners if a consistent snapshot has been emitted.
const raisedEvent = listener.applyOnlineStateChange(this.onlineState);
debugAssert(
!raisedEvent,
"applyOnlineStateChange() shouldn't raise an event for brand-new listeners."
);

if (queryInfo.viewSnap) {
const raisedEvent = listener.onViewSnapshot(queryInfo.viewSnap);
if (raisedEvent) {
this.raiseSnapshotsInSyncEvent();
// Run global snapshot listeners if a consistent snapshot has been emitted.
const raisedEvent = listener.applyOnlineStateChange(this.onlineState);
debugAssert(
!raisedEvent,
"applyOnlineStateChange() shouldn't raise an event for brand-new listeners."
);

if (queryInfo.viewSnap) {
const raisedEvent = listener.onViewSnapshot(queryInfo.viewSnap);
if (raisedEvent) {
this.raiseSnapshotsInSyncEvent();
}
}
},
e => {
const msg = `Initialization of query '${listener.query}' failed: ${e}`;
logError(LOG_TAG, msg);
listener.onError(new FirestoreError(Code.UNAVAILABLE, msg));
}
}
);
}

async unlisten(listener: QueryListener): Promise<void> {
Expand Down
64 changes: 33 additions & 31 deletions packages/firestore/src/core/sync_engine.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -73,7 +72,7 @@ import {
ViewDocumentChanges
} from './view';
import { ViewSnapshot } from './view_snapshot';
import { AsyncQueue } from '../util/async_queue';
import { AsyncQueue, executeWithIndexedDbRecovery } from '../util/async_queue';
import { TransactionRunner } from './transaction_runner';

const LOG_TAG = 'SyncEngine';
Expand Down Expand Up @@ -350,30 +349,25 @@ export class SyncEngine implements RemoteSyncer {
* userCallback is resolved once the write was acked/rejected by the
* backend (or failed locally for any other reason).
*/
async write(batch: Mutation[], userCallback: Deferred<void>): Promise<void> {
write(batch: Mutation[], userCallback: Deferred<void>): Promise<void> {
this.assertSubscribed('write()');

let result: LocalWriteResult;
try {
result = await this.localStore.localWrite(batch);
} catch (e) {
if (e.name === 'IndexedDbTransactionError') {
return executeWithIndexedDbRecovery(
async () => {
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();
},
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.
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;
const msg = `Failed to persist write: ${e}`;
logError(LOG_TAG, msg);
userCallback.reject(new FirestoreError(Code.UNAVAILABLE, msg));
}
}

this.sharedClientState.addPendingMutation(result.batchId);
this.addMutationCallback(result.batchId, userCallback);
await this.emitNewSnapsAndNotifyLocalStore(result.changes);
await this.remoteStore.fillWritePipeline();
);
}

/**
Expand Down Expand Up @@ -584,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;
}

const callbacks = this.pendingWritesCallbacks.get(highestBatchId) || [];
callbacks.push(callback);
this.pendingWritesCallbacks.set(highestBatchId, callbacks);
return executeWithIndexedDbRecovery(
async () => {
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);
},
e => {
const msg = `Initialization of waitForPendingWrites() operation failed: ${e}`;
logError(LOG_TAG, msg);
callback.reject(new FirestoreError(Code.UNAVAILABLE, msg));
}
);
}

/**
Expand Down
30 changes: 12 additions & 18 deletions packages/firestore/src/local/lru_garbage_collector.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,11 @@
import { ListenSequence } from '../core/listen_sequence';
import { ListenSequenceNumber, TargetId } from '../core/types';
import { debugAssert } from '../util/assert';
import { AsyncQueue, TimerId } from '../util/async_queue';
import {
AsyncQueue,
TimerId,
executeWithIndexedDbRecovery
} from '../util/async_queue';
import { getLogLevel, logDebug, LogLevel } from '../util/log';
import { primitiveComparator } from '../util/misc';
import { CancelablePromise } from '../util/promise';
Expand All @@ -32,8 +36,6 @@ import {
import { PersistencePromise } from './persistence_promise';
import { TargetData } from './target_data';

const LOG_TAG = 'LruGarbageCollector';

/**
* Persistence layers intending to use LRU Garbage collection should have reference delegates that
* implement this interface. This interface defines the operations that the LRU garbage collector
Expand Down Expand Up @@ -267,23 +269,15 @@ export class LruScheduler implements GarbageCollectionScheduler {
this.gcTask = this.asyncQueue.enqueueAfterDelay(
TimerId.LruGarbageCollection,
delay,
async () => {
() => {
this.gcTask = null;
this.hasRun = true;
try {
await localStore.collectGarbage(this.garbageCollector);
} catch (e) {
if (e.name === 'IndexedDbTransactionError') {
logDebug(
LOG_TAG,
'Ignoring IndexedDB error during garbage collection: ',
e
);
} else {
await ignoreIfPrimaryLeaseLoss(e);
}
}
await this.scheduleGC(localStore);
return executeWithIndexedDbRecovery(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In particular, this change has fairly significantly harmed this method:

  • Why is the extra .then(() => {}) required?
  • Going back to then/catch from async/await is a bummer.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is now reverted, but there are some arguments for using Promise chains here:

  • In ES5 code, these are shorter since the function doesn't have to be wrapped in a generator. We haven't really optimized for this in other cases and the overhead might be outweighed by readability.
  • The Promise version doesn't re-schedule the GC task if we have lost the lease. As pointed out in the previous PR, this is also not really an issue though since we stop the scheduler during the next lease refresh.

() => localStore.collectGarbage(this.garbageCollector).then(() => {}),
/* recoveryHandler= */ () => {}
)
.then(() => this.scheduleGC(localStore))
.catch(ignoreIfPrimaryLeaseLoss);
}
);
}
Expand Down
20 changes: 20 additions & 0 deletions packages/firestore/src/util/async_queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 { IndexedDbTransactionError } from '../local/simple_db';

const LOG_TAG = 'AsyncQueue';

Expand Down Expand Up @@ -501,3 +502,22 @@ export class AsyncQueue {
this.delayedOperations.splice(index, 1);
}
}

/**
* Runs the provided `op`. If `op` fails with an `IndexedDbTransactionError`,
* calls `recoveryHandler` and returns a resolved Promise. If `op` is successful
* or fails with another type of error, returns op's result.
*/
export function executeWithIndexedDbRecovery<T>(
op: () => Promise<void>,
recoveryHandler: (e: IndexedDbTransactionError) => void
): Promise<void> {
return op().catch(e => {
logDebug(LOG_TAG, 'Internal operation failed: ', e);
if (e.name === 'IndexedDbTransactionError') {
recoveryHandler(e);
} else {
throw e;
}
});
}