Merge pull request #2369 from murgatroid99/grpc-js_pick_first_fix

grpc-js: Fix bugs in pick first LB policy and channel subchannel wrapper

Author: murgatroid99

packages/grpc-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js",
-  "version": "1.8.9",
+  "version": "1.8.10",
   "description": "gRPC Library for Node - pure JS implementation",
   "homepage": "https://grpc.io/",
   "repository": "https://github.com/grpc/grpc-node/tree/master/packages/grpc-js",

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

@@ -86,16 +86,14 @@ const DEFAULT_RETRY_BUFFER_SIZE_BYTES = 1<<24; // 16 MB
 const DEFAULT_PER_RPC_RETRY_BUFFER_SIZE_BYTES = 1<<20; // 1 MB
 
 class ChannelSubchannelWrapper extends BaseSubchannelWrapper implements SubchannelInterface {
-  private stateListeners: ConnectivityStateListener[] = [];
   private refCount = 0;
+  private subchannelStateListener: ConnectivityStateListener;
   constructor(childSubchannel: SubchannelInterface, private channel: InternalChannel) {
     super(childSubchannel);
-    childSubchannel.addConnectivityStateListener((subchannel, previousState, newState, keepaliveTime) => {
+    this.subchannelStateListener = (subchannel, previousState, newState, keepaliveTime) => {
       channel.throttleKeepalive(keepaliveTime);
-      for (const listener of this.stateListeners) {
-        listener(this, previousState, newState, keepaliveTime);
-      }
-    });
+    };
+    childSubchannel.addConnectivityStateListener(this.subchannelStateListener);
   }
 
   ref(): void {
@@ -107,6 +105,7 @@ class ChannelSubchannelWrapper extends BaseSubchannelWrapper implements Subchann
     this.child.unref();
     this.refCount -= 1;
     if (this.refCount <= 0) {
+      this.child.removeConnectivityStateListener(this.subchannelStateListener);
       this.channel.removeWrappedSubchannel(this);
     }
   }

packages/grpc-js/src/load-balancer-pick-first.ts

@@ -33,6 +33,7 @@ import {
 } from './picker';
 import {
   SubchannelAddress,
+  subchannelAddressEqual,
   subchannelAddressToString,
 } from './subchannel-address';
 import * as logging from './logging';
@@ -168,7 +169,7 @@ export class PickFirstLoadBalancer implements LoadBalancer {
        * connecting to the next one instead of waiting for the connection
        * delay timer. */
       if (
-        subchannel === this.subchannels[this.currentSubchannelIndex] &&
+        subchannel.getRealSubchannel() === this.subchannels[this.currentSubchannelIndex].getRealSubchannel() &&
         newState === ConnectivityState.TRANSIENT_FAILURE
       ) {
         this.startNextSubchannelConnecting();
@@ -420,7 +421,7 @@ export class PickFirstLoadBalancer implements LoadBalancer {
     if (
       this.subchannels.length === 0 ||
       !this.latestAddressList.every(
-        (value, index) => addressList[index] === value
+        (value, index) => subchannelAddressEqual(addressList[index], value)
       )
     ) {
       this.latestAddressList = addressList;