Merge pull request #2791 from murgatroid99/grpc-js_channel_close_pick_fix

grpc-js: Ensure pending calls end after channel close

Author: murgatroid99

packages/grpc-js/src/internal-channel.ts

@@ -20,7 +20,7 @@ import { ChannelOptions } from './channel-options';
 import { ResolvingLoadBalancer } from './resolving-load-balancer';
 import { SubchannelPool, getSubchannelPool } from './subchannel-pool';
 import { ChannelControlHelper } from './load-balancer';
-import { UnavailablePicker, Picker, QueuePicker } from './picker';
+import { UnavailablePicker, Picker, QueuePicker, PickArgs, PickResult, PickResultType } from './picker';
 import { Metadata } from './metadata';
 import { Status, LogVerbosity, Propagate } from './constants';
 import { FilterStackFactory } from './filter-stack';
@@ -143,6 +143,22 @@ class ChannelSubchannelWrapper
   }
 }
 
+class ShutdownPicker implements Picker {
+  pick(pickArgs: PickArgs): PickResult {
+    return {
+      pickResultType: PickResultType.DROP,
+      status: {
+        code: Status.UNAVAILABLE,
+        details: 'Channel closed before call started',
+        metadata: new Metadata()
+      },
+      subchannel: null,
+      onCallStarted: null,
+      onCallEnded: null
+    }
+  }
+}
+
 export class InternalChannel {
   private readonly resolvingLoadBalancer: ResolvingLoadBalancer;
   private readonly subchannelPool: SubchannelPool;
@@ -536,7 +552,9 @@ export class InternalChannel {
   }
 
   getConfig(method: string, metadata: Metadata): GetConfigResult {
-    this.resolvingLoadBalancer.exitIdle();
+    if (this.connectivityState !== ConnectivityState.SHUTDOWN) {
+      this.resolvingLoadBalancer.exitIdle();
+    }
     if (this.configSelector) {
       return {
         type: 'SUCCESS',
@@ -745,6 +763,15 @@ export class InternalChannel {
   close() {
     this.resolvingLoadBalancer.destroy();
     this.updateState(ConnectivityState.SHUTDOWN);
+    this.currentPicker = new ShutdownPicker();
+    for (const call of this.configSelectionQueue) {
+      call.cancelWithStatus(Status.UNAVAILABLE, 'Channel closed before call started');
+    }
+    this.configSelectionQueue = [];
+    for (const call of this.pickQueue) {
+      call.cancelWithStatus(Status.UNAVAILABLE, 'Channel closed before call started');
+    }
+    this.pickQueue = [];
     clearInterval(this.callRefTimer);
     if (this.idleTimer) {
       clearTimeout(this.idleTimer);

packages/grpc-js/test/test-client.ts

@@ -97,6 +97,21 @@ describe('Client without a server', () => {
       }
     );
   });
+  it('close should force calls to end', done => {
+    client.makeUnaryRequest(
+      '/service/method',
+      x => x,
+      x => x,
+      Buffer.from([]),
+      new grpc.Metadata({waitForReady: true}),
+      (error, value) => {
+        assert(error);
+        assert.strictEqual(error?.code, grpc.status.UNAVAILABLE);
+        done();
+      }
+    );
+    client.close();
+  });
 });
 
 describe('Client with a nonexistent target domain', () => {
@@ -133,4 +148,19 @@ describe('Client with a nonexistent target domain', () => {
       }
     );
   });
+  it('close should force calls to end', done => {
+    client.makeUnaryRequest(
+      '/service/method',
+      x => x,
+      x => x,
+      Buffer.from([]),
+      new grpc.Metadata({waitForReady: true}),
+      (error, value) => {
+        assert(error);
+        assert.strictEqual(error?.code, grpc.status.UNAVAILABLE);
+        done();
+      }
+    );
+    client.close();
+  });
 });