feat: add onWriteError() and onWriteResult() handlers to BulkWriter #1315

thebrianchen · 2020-09-30T17:54:33Z

Adds global onWriteError() error and onWriteResult() success handlers to BulkWriter instances.
Adds BulkWriter.retry(operation: BulkWriterOperation) method to retry failed operations.
All failed operations throw a BulkWriterError that contain an underlying BulkWriterOperation.

codecov · 2020-09-30T19:48:46Z

Codecov Report

Merging #1315 into master will decrease coverage by 0.01%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##           master    #1315      +/-   ##
==========================================
- Coverage   98.56%   98.55%   -0.02%     
==========================================
  Files          32       32              
  Lines       19115    19274     +159     
  Branches     1373     1386      +13     
==========================================
+ Hits        18841    18995     +154     
- Misses        271      276       +5     
  Partials        3        3

Impacted Files	Coverage Δ
dev/src/bulk-writer.ts	`99.80% <100.00%> (+0.04%)`	⬆️
dev/src/util.ts	`98.50% <100.00%> (+0.11%)`	⬆️
dev/src/write-batch.ts	`100.00% <100.00%> (ø)`
dev/src/backoff.ts	`98.29% <0.00%> (-1.71%)`	⬇️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update a18ab50...fff6a4e. Read the comment docs.

schmidt-sebastian

This looks good. Before you address the feedback, we should see if we want to incorporate @bcoe's API suggestion.

dev/src/write-batch.ts

dev/src/bulk-writer.ts

schmidt-sebastian · 2020-10-02T22:39:26Z

dev/src/bulk-writer.ts

+    const pendingOp = bulkCommitBatch.getLastOp();
+    resultPromise.catch(err => {
+      const operation = new BulkWriterOperation(
+        documentRef,
+        'create',
+        pendingOp
+      );
+      this.errorFn(err, operation);
+    });


Ideally, we would only attach our own Promise if errorFn is defined, but that would require users to register an error function before any other API call. I think that we could re-throw the original exception here though if the error callback is not defined. This can be a simple change if you change the default error function to just throw.

I don't fully follow along here. If we change the default function to throw, that would cause an UnhandledPromiseRejection to surface again. Isn't the point of attaching a .catch() to the existing promise to prevent the UnhandledPromiseRejection from being thrown in the first place?

Something like:

const defaultCatch = (e) => throw e; set() { this.op().catch(() => defaultCatch()) } onError(errorFn) { defaultCatch = errorFn; }

This should avoid the "uncaught Promise rejection" handling only when onError is called.

dev/src/bulk-writer.ts

dev/test/bulk-writer.ts

types/firestore.d.ts

…e into bc/bulk-error

tilgovi · 2020-10-13T16:10:47Z

dev/src/bulk-writer.ts

@@ -449,10 +574,10 @@ export class BulkWriter {
   * delete.
   * @param {Timestamp=} precondition.lastUpdateTime If set, enforces that the
   * document was last updated at lastUpdateTime. Fails the batch if the
-   * document doesn't exist or was last updated at a different time.
+   * document doesn't exist or was last updatedra at a different time.


typo slipped in here

tilgovi · 2020-10-13T16:14:20Z

dev/src/bulk-writer.ts

+    if (this.batchQueue.length > 0) {
+      await this._flush();
+      await this.closedDeferred.promise;
+    }


I don't think I see how it's possible for this to be the case after awaiting the flush, above, because it's not possible to add retries after this.closeCalled = true, right?

If it is expected, does this need to be in a loop?

I now see that verifyNotClosed only throws if fully closed when called from retry. In which case, I think my second question still stands. Do we know that the await this._flush() won't cause more retries? Do we either need to do this in a loop or at least await another flush after this.closeDeferred.promise resolves?

I think we can use Promise chaining here as proposed in the diff above. Instead of requiring void as the return type of onWriteError, we can require the user to call retry and to pass the result of retry to use via the error handler. The signature would then be:

onWriteError((err:BulkWriterError) => Promise<WriteResult>)

Once we do that, we can chain the result of this error callback and block on the retry attempt in the flush call.

If we don't do this, we have to somehow ensure that all error callbacks have fired once we process the results. This would be a bit tricky and may not port to Java or other languages:

error.reject(...); setImmediate(() => { // This only works in Node, but due to the way that ticks are processed in the VM, we can use a `setImmediate` // to ensure that our code only runs once all Promise chains are processed. });

That would be elegant, I think. The ultimate result of the original write would become the result of the retry. It would just be important to document well so that callers don't forget to return the result of retry() from their error handler or the retry promise would end up unhandled and the original write would resolve with undefined.

schmidt-sebastian · 2020-10-13T19:18:40Z

dev/src/bulk-writer.ts

@@ -40,6 +40,9 @@ import {
  validateInteger,
  validateOptional,
 } from './validate';
+// TODO: How do I get rid of this?
+// eslint-disable-next-line no-undef


https://gist.github.com/schmidt-sebastian/ebb63b96c048f3c43187edac7523f9c4

The idea here is that you can use GrpcStatus in the typings file and the Gax status in code. They are interchangeable.

Please address.

I think this is a necessary evil unfortunately.

schmidt-sebastian · 2020-10-13T19:21:59Z

dev/src/bulk-writer.ts

+    readonly operation: BulkWriterOperation,
+    readonly code: GrpcStatus,
+    readonly message: string


Do these show up in JSDoc? Should we add comments?

schmidt-sebastian · 2020-10-13T19:23:52Z

dev/src/bulk-writer.ts

  /**
   * Whether this BulkWriter instance is closed. Once closed, it cannot be
   * opened again.
   */
-  private closed = false;
+  private isClosed = false;


We may want to align this with the status codes in BulkWriterBatch and use an enum (either a new enum or a specific enum for BulkWriter (open/closing/closed)).

schmidt-sebastian · 2020-10-13T20:04:04Z

dev/src/bulk-writer.ts


+  /** @hideconstructor */
  constructor(
    private readonly firestore: Firestore,
    options?: firestore.BulkWriterOptions
  ) {
    this.firestore._incrementBulkWritersCount();
    this.errorFn = () => {};


By default, you probably want this error function to throw. You still want "Unhandled Promise Rejections" when there is no error handler and no other callback attached.

schmidt-sebastian · 2020-10-13T20:06:29Z

dev/src/bulk-writer.ts

+      .catch(err => {
+        this.errorFn(err);
+      });


I think you want to return the Promise that has the error handler attached. That way, the default error handler can re-throw and any user specified error handler can prevent Unhandled Promise Rejections.

types/firestore.d.ts

schmidt-sebastian · 2020-10-13T20:33:04Z

dev/src/bulk-writer.ts

+    return this._flush();
+  }
+
+  private async _flush(): Promise<void> {
    const trackedBatches = this.batchQueue;
    const writePromises = trackedBatches.map(batch => batch.awaitBulkCommit());
    this.sendReadyBatches();


Random thought: Should the already build-in retry use the new Public APIs? This would simplify some of the code a bit (https://gist.github.com/schmidt-sebastian/302c19baa2eda89b23435da41afb077f) and it would allow us to use consistent backoff for all retries.

schmidt-sebastian · 2020-10-13T21:01:43Z

dev/src/bulk-writer.ts

+    if (this.batchQueue.length > 0) {
+      await this._flush();
+      await this.closedDeferred.promise;
+    }


I think we can use Promise chaining here as proposed in the diff above. Instead of requiring void as the return type of onWriteError, we can require the user to call retry and to pass the result of retry to use via the error handler. The signature would then be:

onWriteError((err:BulkWriterError) => Promise<WriteResult>)

Once we do that, we can chain the result of this error callback and block on the retry attempt in the flush call.

If we don't do this, we have to somehow ensure that all error callbacks have fired once we process the results. This would be a bit tricky and may not port to Java or other languages:

error.reject(...); setImmediate(() => { // This only works in Node, but due to the way that ticks are processed in the VM, we can use a `setImmediate` // to ensure that our code only runs once all Promise chains are processed. });

schmidt-sebastian · 2020-10-13T21:06:23Z

dev/src/bulk-writer.ts

+  private verifyNotClosed(checkIsClosed = false): void {
+    if (!checkIsClosed && this.closeCalled) {
+      throw new Error('BulkWriter has already been closed.');
+    }
+
+    if (checkIsClosed && this.isClosed) {
      throw new Error('BulkWriter has already been closed.');
    }
  }


Should these be two different functions?

types/firestore.d.ts

thebrianchen · 2020-10-15T04:45:32Z

@tilgovi Quick update -- Still working on it in a separate branch #1330.

At this point, I have the auto-retry functionality working. onWriteError() now takes a callback with the signature (error: BulkWriterError): boolean where returning true automatically retries, and returning false throws. Aside from a lot of code cleanup to do, the last bug that is in the implementation is that close() resolves a tick before the last retry operation resolves. Hoping to fix that tomorrow.

Remove usage of BulkWriterOperation

google-cla · 2020-10-28T19:47:22Z

All (the pull request submitter and all commit authors) CLAs are signed, but one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project. Please have them confirm that by leaving a comment that contains only @googlebot I consent. in this pull request.

Note to project maintainer: There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent. In those cases, you can manually confirm consent of the commit author(s), and set the cla label to yes (if enabled on your project).

ℹ️ Googlers: Go here for more info.

schmidt-sebastian

Whee. Very nice. Some nits left but it looks like we are basically good to go

schmidt-sebastian · 2020-10-29T16:54:39Z

dev/src/bulk-writer.ts

@@ -40,6 +40,9 @@ import {
  validateInteger,
  validateOptional,
 } from './validate';
+// TODO: How do I get rid of this?
+// eslint-disable-next-line no-undef


Please address.

schmidt-sebastian · 2020-10-29T16:55:41Z

dev/src/bulk-writer.ts

 import {
  invalidArgumentMessage,
  validateInteger,
  validateOptional,
 } from './validate';
+// TODO(chenbrian): Figure some way to get rid of this.
+// eslint-disable-next-line no-undef
+import GrpcStatus = FirebaseFirestore.GrpcStatus;


Please move this line below all other import. This will help us pretend that we care about import statement order.

schmidt-sebastian · 2020-10-29T16:57:06Z

dev/src/bulk-writer.ts

    assert(
      this.state === BatchState.OPEN,
      'Batch should be OPEN when adding writes'
    );
    const deferred = new Deferred<BatchWriteResult>();
    this.pendingOps.push({
      writeBatchIndex: this.opCount,
-      key: documentRef.path,
      deferred: deferred,
    });


Could pendingOps be an array of writeBatchIndex to deferred (and not a list of these elements)?

schmidt-sebastian · 2020-10-29T17:21:05Z

dev/src/bulk-writer.ts

+  /**
+   * Schedules and runs the provided operation.
+   */
+  private async _executeWrite(
+    documentRef: firestore.DocumentReference,
+    operationType: 'create' | 'set' | 'update' | 'delete',
+    operationFn: (bulkCommitBatch: BulkCommitBatch) => Promise<WriteResult>
+  ): Promise<WriteResult> {
+    // A deferred promise that resolves when operationFn completes.
+    const operationCompletedDeferred = new Deferred<void>();
+    this._pendingOps.add(operationCompletedDeferred.promise);
+    try {
+      for (let failedAttempts = 0; ; ++failedAttempts) {
+        const batchQueue =
+          failedAttempts > 0 ? this._retryBatchQueue : this._batchQueue;
+        const bulkCommitBatch = this.getEligibleBatch(batchQueue);
+
+        // Send ready batches if this is the first attempt. Subsequent retry
+        // batches are scheduled after the initial batch returns.
+        if (failedAttempts === 0) {
+          this.sendReadyBatches(batchQueue);
+        }
+
+        try {
+          const operationResult = await operationFn(bulkCommitBatch);
+          this._successFn(documentRef, operationResult);
+          return operationResult;
+        } catch (error) {
+          const bulkWriterError = new BulkWriterError(
+            error.code,
+            error.message,
+            documentRef,
+            operationType,
+            failedAttempts
+          );
+          const shouldRetry = this._errorFn(bulkWriterError);
+          logger(
+            'BulkWriter.errorFn',
+            null,
+            'Running error callback on error code:',
+            error.code,
+            ', shouldRetry:',
+            shouldRetry
+          );
+          if (!shouldRetry) {
+            throw bulkWriterError;
+          }
+        }
+      }
+    } finally {
+      operationCompletedDeferred.resolve();
+      this._pendingOps.delete(operationCompletedDeferred.promise);
+    }
+  }
+


My last effort cleanup attempt:

Suggested change

/**

* Schedules and runs the provided operation.

*/

private async _executeWrite(

documentRef: firestore.DocumentReference,

operationType: 'create' | 'set' | 'update' | 'delete',

operationFn: (bulkCommitBatch: BulkCommitBatch) => Promise<WriteResult>

): Promise<WriteResult> {

// A deferred promise that resolves when operationFn completes.

const operationCompletedDeferred = new Deferred<void>();

this._pendingOps.add(operationCompletedDeferred.promise);

try {

for (let failedAttempts = 0; ; ++failedAttempts) {

const batchQueue =

failedAttempts > 0 ? this._retryBatchQueue : this._batchQueue;

const bulkCommitBatch = this.getEligibleBatch(batchQueue);

// Send ready batches if this is the first attempt. Subsequent retry

// batches are scheduled after the initial batch returns.

if (failedAttempts === 0) {

this.sendReadyBatches(batchQueue);

}

try {

const operationResult = await operationFn(bulkCommitBatch);

this._successFn(documentRef, operationResult);

return operationResult;

} catch (error) {

const bulkWriterError = new BulkWriterError(

error.code,

error.message,

documentRef,

operationType,

failedAttempts

);

const shouldRetry = this._errorFn(bulkWriterError);

logger(

'BulkWriter.errorFn',

null,

'Running error callback on error code:',

error.code,

', shouldRetry:',

shouldRetry

);

if (!shouldRetry) {

throw bulkWriterError;

}

}

}

} finally {

operationCompletedDeferred.resolve();

this._pendingOps.delete(operationCompletedDeferred.promise);

}

}

/**

* Schedules and runs the provided operation.

*/

private async _executeWrite(

documentRef: firestore.DocumentReference,

operationType: 'create' | 'set' | 'update' | 'delete',

operationFn: (bulkCommitBatch: BulkCommitBatch) => Promise<WriteResult>

): Promise<WriteResult> {

let result : Promise<WriteResult> | undefined;

// A deferred promise that resolves when operationFn completes.

const operationCompletedDeferred = new Deferred<void>();

this._pendingOps.add(operationCompletedDeferred.promise);

let batchQueue = this._batchQueue;

this.sendReadyBatches(batchQueue);

for (let failedAttempts = 0; result != undefined; ++failedAttempts) {

const bulkCommitBatch = this.getEligibleBatch(batchQueue);

try {

const writeResult = await operationFn(bulkCommitBatch);

this._successFn(documentRef, writeResult);

result = Promise.resolve(writeResult);

} catch (error) {

const bulkWriterError = new BulkWriterError(

error.code,

error.message,

documentRef,

operationType,

failedAttempts

);

const shouldRetry = this._errorFn(bulkWriterError);

logger(

'BulkWriter.errorFn',

null,

'Running error callback on error code:',

error.code,

', shouldRetry:',

shouldRetry

);

if (shouldRetry) {

batchQueue = this._retryBatchQueue;

} else {

result = Promise.reject(bulkWriterError);

}

}

}

result.then(() => {

operationCompletedDeferred.resolve();

this._pendingOps.delete(operationCompletedDeferred.promise);

});

return result!;

}

Let me know what you think.

post discussion: The main issue with this implementation is that if the user error callback throws an error, the deferred operation will never be resolved. The double try/catch blocks, while ugly, serve to ensure that the deferred operation is always removed.

dev/system-test/firestore.ts

dev/test/bulk-writer.ts

Co-authored-by: Sebastian Schmidt <mrschmidt@google.com>

schmidt-sebastian

Whee!

feat: add onError handler to BulkWriter

09bc422

thebrianchen requested review from a team as code owners September 30, 2020 17:54

thebrianchen self-assigned this Sep 30, 2020

google-cla bot added the cla: yes This human has signed the Contributor License Agreement. label Sep 30, 2020

Brian Chen added 3 commits September 30, 2020 14:42

add additional test

8721708

Merge branch 'master' into bc/bulk-error

a35e9ad

rename retry to addOperation

d748c20

thebrianchen requested a review from schmidt-sebastian September 30, 2020 19:47

thebrianchen assigned schmidt-sebastian and unassigned thebrianchen Sep 30, 2020

product-auto-label bot added the api: firestore Issues related to the googleapis/nodejs-firestore API. label Oct 1, 2020

schmidt-sebastian reviewed Oct 2, 2020

View reviewed changes

schmidt-sebastian assigned thebrianchen and unassigned schmidt-sebastian Oct 2, 2020

Brian Chen and others added 7 commits October 7, 2020 16:53

feat: add onError handler to BulkWriter

7c292c4

add additional test

2183baa

rename retry to addOperation

0a26962

Merge branch 'master' into bc/bulk-error

e704188

Merge branch 'bc/bulk-error' of github.com:googleapis/nodejs-firestor…

dbba557

…e into bc/bulk-error

resolve comments

6ed9bda

Add BulkWriterError

2bfcb0b

thebrianchen changed the title ~~feat: add onError handler to BulkWriter~~ feat: add onWriteError() and onWriteResult() handlers to BulkWriter Oct 13, 2020

tilgovi reviewed Oct 13, 2020

View reviewed changes

schmidt-sebastian reviewed Oct 13, 2020

View reviewed changes

Brian Chen added 2 commits October 21, 2020 14:02

update comments

d630ec1

Remove usage of BulkWriterOperation

massive cleanup + proper flush() resolution

be8bc9b

Merge branch 'master' into bc/bulk-error

085836f

google-cla bot added cla: no This human has *not* signed the Contributor License Agreement. and removed cla: yes This human has signed the Contributor License Agreement. labels Oct 28, 2020

Update branch to match debug branch

166952b

thebrianchen force-pushed the bc/bulk-error branch from 9b4fd13 to 166952b Compare October 28, 2020 20:34

google-cla bot added cla: yes This human has signed the Contributor License Agreement. and removed cla: no This human has *not* signed the Contributor License Agreement. labels Oct 28, 2020

thebrianchen requested a review from schmidt-sebastian October 29, 2020 16:16

thebrianchen assigned schmidt-sebastian and unassigned thebrianchen Oct 29, 2020

schmidt-sebastian reviewed Oct 29, 2020

View reviewed changes

schmidt-sebastian removed their assignment Oct 29, 2020

Brian Chen and others added 3 commits October 29, 2020 11:42

Apply suggestions from code review

f600f37

Co-authored-by: Sebastian Schmidt <mrschmidt@google.com>

resolve comments

6a5fb8f

Merge branch 'master' into bc/bulk-error

e30e6d4

thebrianchen requested a review from schmidt-sebastian October 29, 2020 23:13

thebrianchen assigned schmidt-sebastian Oct 29, 2020

schmidt-sebastian approved these changes Oct 29, 2020

View reviewed changes

schmidt-sebastian assigned thebrianchen Oct 29, 2020

Merge branch 'master' into bc/bulk-error

fff6a4e

thebrianchen merged commit a173f4d into master Nov 3, 2020

thebrianchen deleted the bc/bulk-error branch November 3, 2020 17:23

thebrianchen mentioned this pull request Nov 9, 2020

feat: add error and success handlers to BulkWriter googleapis/java-firestore#458

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add onWriteError() and onWriteResult() handlers to BulkWriter #1315

feat: add onWriteError() and onWriteResult() handlers to BulkWriter #1315

thebrianchen commented Sep 30, 2020 •

edited

Loading

codecov bot commented Sep 30, 2020 •

edited

Loading

schmidt-sebastian left a comment

schmidt-sebastian Oct 2, 2020

thebrianchen Oct 8, 2020

schmidt-sebastian Oct 13, 2020

tilgovi Oct 13, 2020

tilgovi Oct 13, 2020

tilgovi Oct 13, 2020

schmidt-sebastian Oct 13, 2020

tilgovi Oct 13, 2020

schmidt-sebastian Oct 13, 2020

schmidt-sebastian Oct 29, 2020

thebrianchen Oct 29, 2020

schmidt-sebastian Oct 13, 2020

schmidt-sebastian Oct 13, 2020

schmidt-sebastian Oct 13, 2020

schmidt-sebastian Oct 13, 2020

schmidt-sebastian Oct 13, 2020

schmidt-sebastian Oct 13, 2020

schmidt-sebastian Oct 13, 2020

thebrianchen commented Oct 15, 2020

google-cla bot commented Oct 28, 2020

schmidt-sebastian left a comment

schmidt-sebastian Oct 29, 2020

schmidt-sebastian Oct 29, 2020

thebrianchen Oct 29, 2020

schmidt-sebastian Oct 29, 2020

thebrianchen Oct 29, 2020

schmidt-sebastian Oct 29, 2020 •

edited

Loading

thebrianchen Oct 29, 2020

schmidt-sebastian left a comment

feat: add onWriteError() and onWriteResult() handlers to BulkWriter #1315

feat: add onWriteError() and onWriteResult() handlers to BulkWriter #1315

Conversation

thebrianchen commented Sep 30, 2020 • edited Loading

codecov bot commented Sep 30, 2020 • edited Loading

Codecov Report

schmidt-sebastian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

thebrianchen commented Oct 15, 2020

google-cla bot commented Oct 28, 2020

schmidt-sebastian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

schmidt-sebastian Oct 29, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

schmidt-sebastian left a comment

Choose a reason for hiding this comment

thebrianchen commented Sep 30, 2020 •

edited

Loading

codecov bot commented Sep 30, 2020 •

edited

Loading

schmidt-sebastian Oct 29, 2020 •

edited

Loading