fix: ensure that during resumption of a scan, rows that have not been observed by the caller are re-requested

src/index.ts

@@ -819,7 +819,11 @@ export class Bigtable {
     };
 
     if (isStreamMode) {
-      stream = streamEvents(new PassThrough({objectMode: true}));
+      stream = streamEvents(
+        new PassThrough({
+          objectMode: true,
+        })
+      );
       stream.abort = () => {
         if (gaxStream && gaxStream.cancel) {
           gaxStream.cancel();

src/table.ts

@@ -726,7 +726,7 @@
     let filter: {} | null;
     const rowsLimit = options.limit || 0;
     const hasLimit = rowsLimit !== 0;
-    let rowsRead = 0;
+
     let numConsecutiveErrors = 0;
     let numRequestsMade = 0;
     let retryTimer: NodeJS.Timeout | null;
@@ -749,14 +749,22 @@
     let rowStream: Duplex;
 
     let userCanceled = false;
+    // The key of the last row that was emitted by the per attempt pipeline
+    // Note: this must be updated from the operation level userStream to avoid referencing buffered rows that will be
+    // discarded in the per attempt subpipeline (rowStream)
+    let lastRowKey = '';
+    let rowsRead = 0;
     const userStream = new PassThrough({
       objectMode: true,
       readableHighWaterMark: 0,
+      writableHighWaterMark: 0,
       transform(row, _encoding, callback) {
         if (userCanceled) {
           callback();
           return;
         }
+        lastRowKey = row.id;
+        rowsRead++;
         callback(null, row);
       },
     });
@@ -796,9 +804,10 @@
       // cancelled the stream in the middle of a retry
       retryTimer = null;
 
-      const lastRowKey = chunkTransformer ? chunkTransformer.lastRowKey : '';
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
-      chunkTransformer = new ChunkTransformer({decode: options.decode} as any);
+      chunkTransformer = new ChunkTransformer({
+        decode: options.decode,
+      } as any);
 
       const reqOpts = {
         tableName: this.name,
@@ -810,7 +819,7 @@
         // Handling retries in this client. Specify the retry options to
         // make sure nothing is retried in retry-request.
         noResponseRetries: 0,
         shouldRetryFn: (_: any) => {
           return false;
         },
       };
@@ -918,7 +927,6 @@
           ) {
             return next();
           }
-          rowsRead++;
           const row = this.row(rowData.key);
           row.data = rowData.data;
           next(null, row);
@@ -969,7 +977,7 @@
             userStream.emit('error', error);
           }
         })
         .on('data', _ => {
           // Reset error count after a successful read so the backoff
           // time won't keep increasing when as stream had multiple errors
           numConsecutiveErrors = 0;
@@ -1592,7 +1600,7 @@
         // Handling retries in this client. Specify the retry options to
         // make sure nothing is retried in retry-request.
         noResponseRetries: 0,
         shouldRetryFn: (_: any) => {
           return false;
         },
       };

src/util/mock-servers/mock-server.ts

@@ -37,7 +37,6 @@ export class MockServer {
       `localhost:${this.port}`,
       grpc.ServerCredentials.createInsecure(),
       () => {
-        server.start();
         callback ? callback(portString) : undefined;
       }
     );

test/readrows.ts

@@ -23,6 +23,7 @@ import {BigtableClientMockService} from '../src/util/mock-servers/service-implem
 import {MockService} from '../src/util/mock-servers/mock-service';
 import {debugLog, readRowsImpl} from './utils/readRowsImpl';
 import {UntypedHandleCall} from '@grpc/grpc-js';
+import {readRowsImpl2} from './utils/readRowsImpl2';
 
 describe('Bigtable/ReadRows', () => {
   let server: MockServer;
@@ -317,6 +318,52 @@ describe('Bigtable/ReadRows', () => {
     });
   });
 
+  it('should return row data in the right order', done => {
+    // 1000 rows must be enough to reproduce issues with losing the data and to create backpressure
+    const keyFrom = undefined;
+    const keyTo = undefined;
+    // the server will error after sending this chunk (not row)
+    const errorAfterChunkNo = 100;
+    const dataResults = [];
+
+    service.setService({
+      ReadRows: readRowsImpl2(keyFrom, keyTo, errorAfterChunkNo) as any,
+    });
+    const sleep = (ms: any) => {
+      return new Promise(resolve => setTimeout(resolve, ms));
+    };
+    (async () => {
+      try {
+        // Setup
+        // const bigtableClient = new Bigtable();
+        // const instance = bigtableClient.instance(INSTANCE_ID);
+        // const table = await prepareTable({instance: instance, N: 150});
+
+        // console.log('read rows');
+        const stream = table.createReadStream({
+          start: '00000000',
+          end: '00000150',
+        });
+
+        for await (const row of stream) {
+          dataResults.push(row.id);
+          console.log('logging row');
+          console.log(row.id, row.data);
+          await sleep(50);
+        }
+        const expectedResults = Array.from(Array(150).keys())
+          .map(i => '00000000' + i.toString())
+          .map(i => i.slice(-8));
+        assert.deepStrictEqual(dataResults, expectedResults);
+        console.log('No more data in the stream');
+        done();
+      } catch (error) {
+        console.error('Something went wrong:', error);
+        done(error);
+      }
+    })();
+  });
+
   after(async () => {
     server.shutdown(() => {});
   });

test/table.ts

@@ -1262,7 +1262,9 @@ describe('Bigtable/Table', () => {
           // eslint-disable-next-line @typescript-eslint/no-explicit-any
           ((stream: any) => {
             stream.push([{key: 'a'}]);
-            stream.emit('error', makeRetryableError());
+            setImmediate(() => {
+              stream.emit('error', makeRetryableError());
+            });
           }) as {} as EventEmitter,
           ((stream: Writable) => {
             stream.end();
@@ -1285,7 +1287,9 @@ describe('Bigtable/Table', () => {
         emitters = [
           ((stream: Duplex) => {
             stream.push([{key: 'a'}]);
-            stream.emit('error', makeRetryableError());
+            setImmediate(() => {
+              stream.emit('error', makeRetryableError());
+            });
           }) as {} as EventEmitter,
           ((stream: Writable) => {
             stream.end();
@@ -1310,7 +1314,9 @@ describe('Bigtable/Table', () => {
           ((stream: Duplex) => {
             stream.push([{key: 'a'}]);
             stream.push([{key: 'b'}]);
-            stream.emit('error', makeRetryableError());
+            setImmediate(() => {
+              stream.emit('error', makeRetryableError());
+            });
           }) as {} as EventEmitter,
           ((stream: Duplex) => {
             stream.push([{key: 'c'}]);
@@ -1343,7 +1349,9 @@ describe('Bigtable/Table', () => {
         emitters = [
           ((stream: Duplex) => {
             stream.push([{key: 'a'}]);
-            stream.emit('error', makeRetryableError());
+            setImmediate(() => {
+              stream.emit('error', makeRetryableError());
+            });
           }) as {} as EventEmitter,
           ((stream: Duplex) => {
             stream.end([{key: 'c'}]);
@@ -1362,7 +1370,9 @@ describe('Bigtable/Table', () => {
           ((stream: Duplex) => {
             stream.push([{key: 'a'}]);
             stream.push([{key: 'b'}]);
-            stream.emit('error', makeRetryableError());
+            setImmediate(() => {
+              stream.emit('error', makeRetryableError());
+            });
           }) as {} as EventEmitter,
         ];
 
@@ -1381,7 +1391,9 @@ describe('Bigtable/Table', () => {
         emitters = [
           ((stream: Duplex) => {
             stream.push([{key: 'a'}]);
-            stream.emit('error', makeRetryableError());
+            setImmediate(() => {
+              stream.emit('error', makeRetryableError());
+            });
           }) as {} as EventEmitter,
         ];
 
@@ -1395,7 +1407,9 @@ describe('Bigtable/Table', () => {
         emitters = [
           ((stream: Duplex) => {
             stream.push([{key: 'c'}]);
-            stream.emit('error', makeRetryableError());
+            setImmediate(() => {
+              stream.emit('error', makeRetryableError());
+            });
           }) as {} as EventEmitter,
         ];
 
@@ -1418,7 +1432,9 @@ describe('Bigtable/Table', () => {
           ((stream: Duplex) => {
             stream.push([{key: 'a1'}]);
             stream.push([{key: 'd'}]);
-            stream.emit('error', makeRetryableError());
+            setImmediate(() => {
+              stream.emit('error', makeRetryableError());
+            });
           }) as {} as EventEmitter,
         ];