fix: properly handle asynchronous read from stream #1284

alexander-fenster · 2023-05-19T22:15:07Z

Fixes #607.

~~Note: I'm currently comparing this pull request to #1282, I will change the base to main after #1282 is merged. This way it's easier to review.~~

Several changes here:

Remove the logic in table.ts that would skip the unprocessed records if a ChunkTransformer has _destroyed flag set. There are two separate cases when the ChunkTransformer gets destroyed:
1. if it's a network problem or anything else that forced gRPC to emit an error event. In such case, it always makes sense to process all the rows we have received before giving up.
2. if it's a user cancellation (by calling stream.end() from the user code). In this case we need to stop emitting new rows.

~~So let's introduce a new flag, _canceled, and ignore the incoming rows only if the stream is canceled from .end().~~ Let's have the new flag, userCanceled, and use it in userStream transform method to stop emitting new rows when the user don't need them, but we'll still keep emitting all rows if there was no user cancellation. This change fixes #607.

When a network error happens, we need to update the lastRowKey only if the row processing has completed (the commit bit was set in the last chunk). Until then, don't update lastRowKey. The current code makes recovery skip one record - the one that was incomplete. This was actually a test problem, there is no problem here.
Minor change to make modern TypeScript compiler happy when comparing chunk.timestampMicros to zero. chunk.timestampMicros can theoretically be an instance of Long, so let's handle this accordingly. It was showing as an error in my editor and it makes sense so I'm fixing it in the safest way possible. In the default case, the || expression will short circuit in the typeof chunk.timestampMicros === 'number' branch.

…into stream-tests

alexander-fenster · 2023-05-20T00:08:32Z

Note: the system and samples tests won't run here until I rebase this PR to be against the main branch, but I ran them locally and they pass.

Also note: I cherry-picked a commit from #1285 to avoid an unrelated Windows test failure.

igorbernstein2 · 2023-05-22T14:21:08Z

src/chunktransformer.ts

    this.reset();
  }

+  get canceled() {


It seems like we expose ChunkTransformer as part of our public api. Which I think is incorrect to begin with. To minimize further leakage of implementation details, can we mark this method as internal?

igorbernstein2 · 2023-05-22T14:22:59Z

src/chunktransformer.ts

+      (typeof chunk.timestampMicros === 'number' &&
+        chunk.timestampMicros! > 0) ||
+      // if it's an instance of Long
+      (typeof chunk.timestampMicros === 'object' &&
+        'compare' in chunk.timestampMicros &&
+        typeof chunk.timestampMicros.compare === 'function' &&
+        chunk.timestampMicros.compare(0) === 1);


This is a bit strange, chunk is a protobuf, so shouldnt it have a stable type?

This is the specifics of protobufjs handling 64 bit integers. In JavaScript, all numbers are 64-bit IEEE 754 floats, so they can reliably represent up to 2⁵³ – 1. So protobufjs accepts and may emit an object of type Long (from https://www.npmjs.com/package/long). As far as I understand in this case it's always number in practice, but in my clean workspace VSCode shows error here, so should be safe to fix.

please leave a comment in the code explaining this

igorbernstein2 · 2023-05-22T14:28:16Z

src/table.ts

        transform: (rowData, _, next) => {
          if (
-            chunkTransformer._destroyed ||
+            chunkTransformer.canceled ||


It seems like there are only 2 things that care about the cancelled flag:

the closure for userstream.end()

the closure for toRowStream

ChunkTransformer seems like an innocent bystander here. Why not move the userCancelled flag as a local var in createReadStream(), whose scope is shared by the 2 closure that care about the flag?

Alternatively, maybe create a subclass for userStream that holds the state

Moving the flag closer to the consumer makes sense to me, I'll try to do it. I only put it here to replace the _destroyed logic, but you are right, it will make more sense downstream.

igorbernstein2 · 2023-05-22T14:37:05Z

src/chunktransformer.ts

      }
-      if (this._destroyed) {
-        return;
-      }
-    }
-    if (data.lastScannedRowKey && data.lastScannedRowKey.length > 0) {
-      this.lastRowKey = Mutation.convertFromBytes(
-        data.lastScannedRowKey as Bytes,
-        {
-          userOptions: this.options,
-        }
-      );
    }


I think this breaks request resumption logic in makeNewRequest

const lastRowKey = chunkTransformer ? chunkTransformer.lastRowKey : '';

lastRowKey is set when committing, it should not be set here because the row is still incomplete at this moment.

This is my fault, I forgot to mention one other detail of ReadRows protocol, there are 2 ways that a readrows resumption request can be built:

based on the last received row key of a committed row chunk. ( This one you know about)

As a heartbeat sent via an empty ReadRowsResponse. This is will be used when a caller has very restrictive filters that cause the scan to omit entire tablets. The protocol allows for the server to emit a heartbeat with the last scanned (as opposed to returned) row key

This is not currently enable on the serverside, but is specified by the protocol

igorbernstein2 · 2023-05-22T15:39:34Z

Also I think we still have a problem of end() on the userStream being delayed. I think we can deal with it in a separate PR but I think that in the case where the consumer is slower than the grpc producer (ie. the user transformer calls the callback in a setTimeout()). Then rows can get buffered in the userStream. When the end user calls cancels the stream early via an end(), the rows buffered in the userStream will still be emitted.

I'm not sure how to deal with this. We would need the userStream to signal the stream end on read() when cancelled()

igorbernstein2 · 2023-05-23T21:19:41Z

src/chunktransformer.ts

+      (typeof chunk.timestampMicros === 'number' &&
+        chunk.timestampMicros! > 0) ||
+      // if it's an instance of Long
+      (typeof chunk.timestampMicros === 'object' &&
+        'compare' in chunk.timestampMicros &&
+        typeof chunk.timestampMicros.compare === 'function' &&
+        chunk.timestampMicros.compare(0) === 1);


please leave a comment in the code explaining this

igorbernstein2 · 2023-05-23T21:24:13Z

src/table.ts

+    });
+    const originalEnd = userStream.end.bind(userStream);
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    userStream.end = (chunk?: any, encoding?: any, cb?: () => void) => {


please add a comment explaining that this is needed to fulfill a pre-existing contract with the caller, where the caller can prematurely cancel the stream by calling end. And we need to be able to disambiguate between a user cancellation and the a normal end. So we capture the originalEnd to be used for normal termination and the overwritten end is meant for end users

igorbernstein2 · 2023-05-23T21:28:04Z

src/table.ts

      rowStream?.unpipe(userStream);
+      rowStream.removeListener('end', originalEnd);


can you add 2 helper functions rowStreamPipe & rowStreamUnpipe and add comments that we need to do special handling of the end event to re-direct the handler to the originalEnd

igorbernstein2 · 2023-05-23T21:28:27Z

src/table.ts

          rowStream.unpipe(userStream);
+          rowStream.removeListener('end', originalEnd);


please delegate to a helper

danieljbruce

Looks good. A couple of questions/comments about specific details.

danieljbruce · 2023-05-24T17:28:28Z

src/table.ts

+          callback();
+          return;
+        }
+        callback(null, row);


Do we add this logic here to prevent the user from receiving rows when they make it past the race condition in toRowStream? If so then do we need the userCanceled || check in toRowStream at all?

Yes. The reason we also have the condition in toRowStream.transform is just because it does not make much sense to pipe more data through toRowStream after the user explicitly asked us to stop.

danieljbruce · 2023-05-24T17:38:55Z

src/table.ts

+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    userStream.end = (chunk?: any, encoding?: any, cb?: () => void) => {
+      rowStreamUnpipe(rowStream, userStream);


Does this prevent the data from being passed on from rowStream to userStream? It looks like we are using both userCanceled and unpipe to stop passing data along. Should we also unpipe the data being passed from the chunk transformer to toRowStream?

We don't really care about the data passed from chunkTransformer to toRowStream because this is all inside rowStream. We only need to stop that data from flowing into the user stream. Since it's an end of the user stream, it's not really important if more or less data has flown inside the rowStream pipeline.

alexander-fenster and others added 7 commits May 19, 2023 02:55

test: test ReadRows logic with local gRPC server

318aeff

test: PR feedback

2c229de

test: fix race condition in initialization

1924249

Merge branch 'main' into stream-tests

99813aa

test: PR feedback, renaming a variable for readability

3353666

Merge branch 'stream-tests' of github.com:googleapis/nodejs-bigtable …

3f2385b

…into stream-tests

fix: properly handle asynchronous read from stream

bf950d0

alexander-fenster requested review from a team as code owners May 19, 2023 22:15

product-auto-label bot added size: m Pull request size is medium. api: bigtable Issues related to the googleapis/nodejs-bigtable API. labels May 19, 2023

alexander-fenster added the kokoro:run Add this label to force Kokoro to re-run the tests. label May 19, 2023

test: skip failing Windows test

fbd43a0

alexander-fenster force-pushed the fix-streams branch from e7f4249 to fbd43a0 Compare May 20, 2023 00:15

test: increase timeout on Windows

aaff430

alexander-fenster requested review from danieljbruce, galz10, igorbernstein2 and leahecole May 20, 2023 02:23

igorbernstein2 requested changes May 22, 2023

View reviewed changes

alexander-fenster added 3 commits May 22, 2023 17:59

fix: PR feedback

1e52970

test: only set lastScannedRowKey for completed rows

b02c587

fix: bring back the lastScannedRowKey logic

d944d58

Base automatically changed from stream-tests to main May 22, 2023 23:15

alexander-fenster added 2 commits May 22, 2023 23:49

Merge branch 'main' into fix-streams

69077d9

test: pick latest changes from main branch

56d4364

product-auto-label bot added size: s Pull request size is small. and removed size: m Pull request size is medium. labels May 22, 2023

yoshi-kokoro removed the kokoro:run Add this label to force Kokoro to re-run the tests. label May 22, 2023

fix: add transform method to userStream, handle cancellation in it

71b67a1

product-auto-label bot added size: m Pull request size is medium. and removed size: s Pull request size is small. labels May 23, 2023

alexander-fenster requested a review from igorbernstein2 May 23, 2023 20:50

igorbernstein2 requested changes May 23, 2023

View reviewed changes

fix: PR feedback

4099b2b

danieljbruce reviewed May 24, 2023

View reviewed changes

Merge branch 'main' into fix-streams

e54f607

danieljbruce mentioned this pull request May 29, 2023

Trying readable #1293

Closed

igorbernstein2 approved these changes May 30, 2023

View reviewed changes

Merge branch 'main' into fix-streams

63453b9

danieljbruce approved these changes May 30, 2023

View reviewed changes

alexander-fenster merged commit 55d86ba into main May 30, 2023

alexander-fenster deleted the fix-streams branch May 30, 2023 18:08

release-please bot mentioned this pull request May 30, 2023

chore(main): release 4.6.1 #1295

Merged

		rowStream?.unpipe(userStream);
		rowStream.removeListener('end', originalEnd);

		rowStream.unpipe(userStream);
		rowStream.removeListener('end', originalEnd);

fix: properly handle asynchronous read from stream #1284

fix: properly handle asynchronous read from stream #1284

Uh oh!

Conversation

alexander-fenster commented May 19, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

alexander-fenster commented May 20, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexander-fenster May 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

igorbernstein2 commented May 22, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danieljbruce left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

alexander-fenster commented May 19, 2023 •

edited

Loading

alexander-fenster May 22, 2023 •

edited

Loading