test: reproduce data duplication error on node 14 (#1458)

* Create the same test as before with setImmediate * Rename parameter in sleep function to ticks * Remove only * Remove TODO * Linting fix * Add a comment to describe the sleep function * Move the comment about 150 rows to the right place * Add comment about backpressure * increase the timeout of the test in question * Remove only * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Do not rely on the removed service anymore * linter fix * linting fix --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: danieljbruce

src/mutation.ts

@@ -129,7 +129,8 @@ export class Mutation {
  bytes: Buffer | string,
  options?: ConvertFromBytesOptions
  ): Buffer | Value | string {
- const buf = bytes instanceof Buffer ? bytes : Buffer.from(bytes, 'base64');
+ const buf =
+ bytes instanceof Buffer ? bytes : Buffer.from(bytes as string, 'base64');
  if (options && options.isPossibleNumber && buf.length === 8) {
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  const num = Long.fromBytes(buf as any).toNumber();

test/readrows.ts

@@ -402,6 +402,64 @@ describe('Bigtable/ReadRows', () => {
  }
  })();
  });
+ it('should return row data in the right order with a predictable sleep function', function (done) {
+ this.timeout(600000);
+ const keyFrom = undefined;
+ const keyTo = undefined;
+ // the server will error after sending this chunk (not row)
+ const errorAfterChunkNo = 100;
+ const dataResults = [];
+
+ // keyTo and keyFrom are not provided so they will be determined from
+ // the request that is passed in.
+ service.setService({
+ ReadRows: ReadRowsImpl.createService({
+ errorAfterChunkNo: 100, // the server will error after sending this chunk (not row)
+ valueSize: 1,
+ chunkSize: 1,
+ chunksPerResponse: 1,
+ debugLog,
+ }) as ServerImplementationInterface,
+ });
+ const sleep = (ticks: number) => {
+ // Adds an event to the end of the event loop `ticks` times
+ // This creates a predictable delay using the event loop and
+ // allows the streams to create a predictable amount of back pressure.
+ return new Promise(resolve => {
+ const nextEventLoop = () => {
+ if (ticks > 0) {
+ ticks = ticks - 1;
+ setImmediate(nextEventLoop);
+ } else {
+ resolve(ticks);
+ }
+ };
+ nextEventLoop();
+ });
+ };
+ (async () => {
+ try {
+ // 150 rows must be enough to reproduce issues with losing the data and to create backpressure
+ const stream = table.createReadStream({
+ start: '00000000',
+ end: '00000150',
+ });
+
+ for await (const row of stream) {
+ dataResults.push(row.id);
+ // sleep parameter needs to be high enough to produce backpressure.
+ await sleep(4000);
+ }
+ const expectedResults = Array.from(Array(150).keys())
+ .map(i => '00000000' + i.toString())
+ .map(i => i.slice(-8));
+ assert.deepStrictEqual(dataResults, expectedResults);
+ done();
+ } catch (error) {
+ done(error);
+ }
+ })();
+ });
 
  after(async () => {
  server.shutdown(() => {});