fix: release clients that received a RST_STREAM error (#1380)

* fix: release clients that received a RST_STREAM error

* Fix test

dev/src/pool.ts

@@ -214,13 +214,13 @@ export class ClientPool<T> {
 
     return op(client)
       .catch(async (err: GoogleError) => {
-        await this.release(requestTag, client);
-        if (err.message?.indexOf('RST_STREAM')) {
+        if (err.message?.match(/RST_STREAM/)) {
           // Once a client has seen a RST_STREAM error, the GRPC channel can
           // no longer be used. We mark the client as failed, which ensures that
           // we open a new GRPC channel for the next request.
           this.failedClients.add(client);
         }
+        await this.release(requestTag, client);
         return Promise.reject(err);
       })
       .then(async res => {

dev/test/pool.ts

@@ -214,13 +214,11 @@ describe('Client pool', () => {
     );
     expect(clientPool.size).to.equal(2);
 
-    operationPromises.forEach(deferred => deferred.reject());
+    operationPromises.forEach(deferred => deferred.reject(new Error()));
 
-    return Promise.all(completionPromises.map(p => p.catch(() => {}))).then(
-      () => {
-        expect(clientPool.size).to.equal(0);
-      }
-    );
+    return Promise.all(
+      completionPromises.map(p => p.catch(() => {}))
+    ).then(() => expect(clientPool.size).to.equal(0));
   });
 
   it('garbage collection calls destructor', () => {
@@ -283,6 +281,21 @@ describe('Client pool', () => {
     expect(instanceCount).to.equal(2);
   });
 
+  it('garbage collects after RST_STREAM', async () => {
+    const clientPool = new ClientPool<{}>(1, 1, () => {
+      return {};
+    });
+
+    const op = clientPool.run(REQUEST_TAG, () =>
+      Promise.reject(
+        new GoogleError('13 INTERNAL: Received RST_STREAM with code 2')
+      )
+    );
+    await op.catch(() => {});
+
+    expect(clientPool.size).to.equal(0);
+  });
+
   it('keeps pool of idle clients', async () => {
     const clientPool = new ClientPool<{}>(
       /* concurrentOperationLimit= */ 1,