fix: retry rst_stream errors (#1059)

Author: mutianf

src/table.ts

@@ -946,6 +946,19 @@ Please use the format 'prezzy' or '${instance.name}/tables/prezzy'.`);
 
       rowStream = pumpify.obj([requestStream, chunkTransformer, toRowStream]);
 
+      // Retry on "received rst stream" errors
+      const isRstStreamError = (error: ServiceError): boolean => {
+        if (error.code === 13 && error.message) {
+          const error_message = (error.message || '').toLowerCase();
+          return (
+            error.code === 13 &&
+            (error_message.includes('rst_stream') ||
+              error_message.includes('rst stream'))
+          );
+        }
+        return false;
+      };
+
       rowStream
         .on('error', (error: ServiceError) => {
           rowStream.unpipe(userStream);
@@ -959,7 +972,7 @@ Please use the format 'prezzy' or '${instance.name}/tables/prezzy'.`);
           numConsecutiveErrors++;
           if (
             numConsecutiveErrors <= maxRetries &&
-            RETRYABLE_STATUS_CODES.has(error.code)
+            (RETRYABLE_STATUS_CODES.has(error.code) || isRstStreamError(error))
           ) {
             const backOffSettings =
               options.gaxOptions?.retry?.backoffSettings ||

test/table.ts

@@ -1429,6 +1429,28 @@ describe('Bigtable/Table', () => {
           done();
         });
       });
+
+      it('should retry received rst stream errors', done => {
+        const rstStreamError = new Error('Received Rst_stream') as ServiceError;
+        rstStreamError.code = 13;
+        emitters = [
+          ((stream: Duplex) => {
+            stream.emit('error', rstStreamError);
+          }) as {} as EventEmitter,
+          ((stream: Duplex) => {
+            stream.end([{key: 'a'}]);
+          }) as {} as EventEmitter,
+        ];
+
+        const options = {
+          keys: ['a'],
+        };
+
+        callCreateReadStream(options, () => {
+          assert.strictEqual(reqOptsCalls.length, 2);
+          done();
+        });
+      });
     });
   });