Merge branch 'master' into grpc-js-xds_server_creds

packages/grpc-js-xds/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js-xds",
-  "version": "1.11.0",
+  "version": "1.12.0",
   "description": "Plugin for @grpc/grpc-js. Adds the xds:// URL scheme and associated features.",
   "main": "build/src/index.js",
   "scripts": {
@@ -53,7 +53,7 @@
     "xxhash-wasm": "^1.0.2"
   },
   "peerDependencies": {
-    "@grpc/grpc-js": "~1.11.0"
+    "@grpc/grpc-js": "~1.12.0"
   },
   "engines": {
     "node": ">=10.10.0"

packages/grpc-js-xds/src/environment.ts

@@ -25,4 +25,4 @@ export const EXPERIMENTAL_FEDERATION = (process.env.GRPC_EXPERIMENTAL_XDS_FEDERA
 export const EXPERIMENTAL_CUSTOM_LB_CONFIG = (process.env.GRPC_EXPERIMENTAL_XDS_CUSTOM_LB_CONFIG ?? 'true') === 'true';
 export const EXPERIMENTAL_RING_HASH = (process.env.GRPC_XDS_EXPERIMENTAL_ENABLE_RING_HASH ?? 'true') === 'true';
 export const EXPERIMENTAL_PICK_FIRST = (process.env.GRPC_EXPERIMENTAL_PICKFIRST_LB_CONFIG ?? 'false') === 'true';
-export const EXPERIMENTAL_DUALSTACK_ENDPOINTS = (process.env.GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS ?? 'false') === 'true';
+export const EXPERIMENTAL_DUALSTACK_ENDPOINTS = (process.env.GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS ?? 'true') === 'true';

packages/grpc-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js",
-  "version": "1.11.0",
+  "version": "1.12.1",
   "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/channel-credentials.ts

@@ -54,6 +54,7 @@ export interface VerifyOptions {
    * has been performed on the peer certificate.
    */
   checkServerIdentity?: CheckServerIdentityCallback;
+  rejectUnauthorized?: boolean;
 }
 
 /**
@@ -207,6 +208,11 @@ class SecureChannelCredentialsImpl extends ChannelCredentials {
       this.connectionOptions.checkServerIdentity =
         verifyOptions.checkServerIdentity;
     }
+
+    if (verifyOptions?.rejectUnauthorized !== undefined) {
+      this.connectionOptions.rejectUnauthorized =
+        verifyOptions.rejectUnauthorized;
+    }
   }
 
   compose(callCredentials: CallCredentials): ChannelCredentials {

packages/grpc-js/src/duration.ts

@@ -43,6 +43,6 @@ export function parseDuration(value: string): Duration | null {
   }
   return {
     seconds: Number.parseInt(match[1], 10),
-    nanos: Number.parseInt(match[2].padEnd(9, '0'), 10)
+    nanos: match[2] ? Number.parseInt(match[2].padEnd(9, '0'), 10) : 0
   };
 }

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

@@ -31,7 +31,7 @@ import {
   getDefaultAuthority,
   mapUriDefaultScheme,
 } from './resolver';
-import { trace } from './logging';
+import { trace, isTracerEnabled } from './logging';
 import { SubchannelAddress } from './subchannel-address';
 import { mapProxyName } from './http_proxy';
 import { GrpcUri, parseUri, uriToString } from './uri-parser';
@@ -426,15 +426,17 @@ export class InternalChannel {
         JSON.stringify(options, undefined, 2)
     );
     const error = new Error();
-    trace(
-      LogVerbosity.DEBUG,
-      'channel_stacktrace',
-      '(' +
-        this.channelzRef.id +
-        ') ' +
-        'Channel constructed \n' +
-        error.stack?.substring(error.stack.indexOf('\n') + 1)
-    );
+    if (isTracerEnabled('channel_stacktrace')){
+      trace(
+        LogVerbosity.DEBUG,
+        'channel_stacktrace',
+        '(' +
+          this.channelzRef.id +
+          ') ' +
+          'Channel constructed \n' +
+          error.stack?.substring(error.stack.indexOf('\n') + 1)
+      );
+    }
     this.lastActivityTimestamp = new Date();
   }
 

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

@@ -214,8 +214,6 @@ export class PickFirstLoadBalancer implements LoadBalancer {
    */
   private connectionDelayTimeout: NodeJS.Timeout;
 
-  private triedAllSubchannels = false;
-
   /**
    * The LB policy enters sticky TRANSIENT_FAILURE mode when all
    * subchannels have failed to connect at least once, and it stays in that
@@ -226,12 +224,6 @@ export class PickFirstLoadBalancer implements LoadBalancer {
 
   private reportHealthStatus: boolean;
 
-  /**
-   * Indicates whether we called channelControlHelper.requestReresolution since
-   * the last call to updateAddressList
-   */
-  private requestedResolutionSinceLastUpdate = false;
-
   /**
    * The most recent error reported by any subchannel as it transitioned to
    * TRANSIENT_FAILURE.
@@ -261,6 +253,10 @@ export class PickFirstLoadBalancer implements LoadBalancer {
     return this.children.every(child => child.hasReportedTransientFailure);
   }
 
+  private resetChildrenReportedTF() {
+    this.children.every(child => child.hasReportedTransientFailure = false);
+  }
+
   private calculateAndReportNewState() {
     if (this.currentPick) {
       if (this.reportHealthStatus && !this.currentPick.isHealthy()) {
@@ -293,23 +289,15 @@ export class PickFirstLoadBalancer implements LoadBalancer {
   }
 
   private requestReresolution() {
-    this.requestedResolutionSinceLastUpdate = true;
     this.channelControlHelper.requestReresolution();
   }
 
   private maybeEnterStickyTransientFailureMode() {
     if (!this.allChildrenHaveReportedTF()) {
       return;
     }
-    if (!this.requestedResolutionSinceLastUpdate) {
-      /* Each time we get an update we reset each subchannel's
-       * hasReportedTransientFailure flag, so the next time we get to this
-       * point after that, each subchannel has reported TRANSIENT_FAILURE
-       * at least once since then. That is the trigger for requesting
-       * reresolution, whether or not the LB policy is already in sticky TF
-       * mode. */
-      this.requestReresolution();
-    }
+    this.requestReresolution();
+    this.resetChildrenReportedTF();
     if (this.stickyTransientFailureMode) {
       this.calculateAndReportNewState();
       return;
@@ -323,21 +311,16 @@ export class PickFirstLoadBalancer implements LoadBalancer {
 
   private removeCurrentPick() {
     if (this.currentPick !== null) {
-      /* Unref can cause a state change, which can cause a change in the value
-       * of this.currentPick, so we hold a local reference to make sure that
-       * does not impact this function. */
-      const currentPick = this.currentPick;
-      this.currentPick = null;
-      currentPick.unref();
-      currentPick.removeConnectivityStateListener(this.subchannelStateListener);
+      this.currentPick.removeConnectivityStateListener(this.subchannelStateListener);
       this.channelControlHelper.removeChannelzChild(
-        currentPick.getChannelzRef()
+        this.currentPick.getChannelzRef()
       );
-      if (this.reportHealthStatus) {
-        currentPick.removeHealthStateWatcher(
-          this.pickedSubchannelHealthListener
-        );
-      }
+      this.currentPick.removeHealthStateWatcher(
+        this.pickedSubchannelHealthListener
+      );
+      // Unref last, to avoid triggering listeners
+      this.currentPick.unref();
+      this.currentPick = null;
     }
   }
 
@@ -377,9 +360,6 @@ export class PickFirstLoadBalancer implements LoadBalancer {
 
   private startNextSubchannelConnecting(startIndex: number) {
     clearTimeout(this.connectionDelayTimeout);
-    if (this.triedAllSubchannels) {
-      return;
-    }
     for (const [index, child] of this.children.entries()) {
       if (index >= startIndex) {
         const subchannelState = child.subchannel.getConnectivityState();
@@ -392,7 +372,6 @@ export class PickFirstLoadBalancer implements LoadBalancer {
         }
       }
     }
-    this.triedAllSubchannels = true;
     this.maybeEnterStickyTransientFailureMode();
   }
 
@@ -421,20 +400,25 @@ export class PickFirstLoadBalancer implements LoadBalancer {
     this.connectionDelayTimeout.unref?.();
   }
 
+  /**
+   * Declare that the specified subchannel should be used to make requests.
+   * This functions the same independent of whether subchannel is a member of
+   * this.children and whether it is equal to this.currentPick.
+   * Prerequisite: subchannel.getConnectivityState() === READY.
+   * @param subchannel
+   */
   private pickSubchannel(subchannel: SubchannelInterface) {
-    if (this.currentPick && subchannel.realSubchannelEquals(this.currentPick)) {
-      return;
-    }
     trace('Pick subchannel with address ' + subchannel.getAddress());
     this.stickyTransientFailureMode = false;
-    this.removeCurrentPick();
-    this.currentPick = subchannel;
+    /* Ref before removeCurrentPick and resetSubchannelList to avoid the
+     * refcount dropping to 0 during this process. */
     subchannel.ref();
-    if (this.reportHealthStatus) {
-      subchannel.addHealthStateWatcher(this.pickedSubchannelHealthListener);
-    }
     this.channelControlHelper.addChannelzChild(subchannel.getChannelzRef());
+    this.removeCurrentPick();
     this.resetSubchannelList();
+    subchannel.addConnectivityStateListener(this.subchannelStateListener);
+    subchannel.addHealthStateWatcher(this.pickedSubchannelHealthListener);
+    this.currentPick = subchannel;
     clearTimeout(this.connectionDelayTimeout);
     this.calculateAndReportNewState();
   }
@@ -451,20 +435,11 @@ export class PickFirstLoadBalancer implements LoadBalancer {
 
   private resetSubchannelList() {
     for (const child of this.children) {
-      if (
-        !(
-          this.currentPick &&
-          child.subchannel.realSubchannelEquals(this.currentPick)
-        )
-      ) {
-        /* The connectivity state listener is the same whether the subchannel
-         * is in the list of children or it is the currentPick, so if it is in
-         * both, removing it here would cause problems. In particular, that
-         * always happens immediately after the subchannel is picked. */
-        child.subchannel.removeConnectivityStateListener(
-          this.subchannelStateListener
-        );
-      }
+      /* Always remoev the connectivity state listener. If the subchannel is
+         getting picked, it will be re-added then. */
+      child.subchannel.removeConnectivityStateListener(
+        this.subchannelStateListener
+      );
       /* Refs are counted independently for the children list and the
        * currentPick, so we call unref whether or not the child is the
        * currentPick. Channelz child references are also refcounted, so
@@ -476,20 +451,16 @@ export class PickFirstLoadBalancer implements LoadBalancer {
     }
     this.currentSubchannelIndex = 0;
     this.children = [];
-    this.triedAllSubchannels = false;
-    this.requestedResolutionSinceLastUpdate = false;
   }
 
   private connectToAddressList(addressList: SubchannelAddress[]) {
+    trace('connectToAddressList([' + addressList.map(address => subchannelAddressToString(address)) + '])');
     const newChildrenList = addressList.map(address => ({
       subchannel: this.channelControlHelper.createSubchannel(address, {}, null),
       hasReportedTransientFailure: false,
     }));
-    trace('connectToAddressList([' + addressList.map(address => subchannelAddressToString(address)) + '])');
     for (const { subchannel } of newChildrenList) {
       if (subchannel.getConnectivityState() === ConnectivityState.READY) {
-        this.channelControlHelper.addChannelzChild(subchannel.getChannelzRef());
-        subchannel.addConnectivityStateListener(this.subchannelStateListener);
         this.pickSubchannel(subchannel);
         return;
       }

packages/grpc-js/src/load-balancer-round-robin.ts

@@ -112,6 +112,13 @@ export class RoundRobinLoadBalancer implements LoadBalancer {
       channelControlHelper,
       {
         updateState: (connectivityState, picker) => {
+          /* Ensure that name resolution is requested again after active
+           * connections are dropped. This is more aggressive than necessary to
+           * accomplish that, so we are counting on resolvers to have
+           * reasonable rate limits. */
+          if (this.currentState === ConnectivityState.READY && connectivityState !== ConnectivityState.READY) {
+            this.channelControlHelper.requestReresolution();
+          }
           this.calculateAndUpdateState();
         },
       }

packages/grpc-js/src/resolver-dns.ts

@@ -20,7 +20,7 @@ import {
   registerResolver,
   registerDefaultScheme,
 } from './resolver';
-import { promises as dns } from 'node:dns';
+import { promises as dns } from 'dns';
 import { extractAndSelectServiceConfig, ServiceConfig } from './service-config';
 import { Status } from './constants';
 import { StatusObject } from './call-interface';

packages/grpc-js/src/retrying-call.ts

@@ -398,7 +398,7 @@ export class RetryingCall implements Call, DeadlineInfoProvider {
     return list.some(
       value =>
         value === code ||
-        value.toString().toLowerCase() === Status[code].toLowerCase()
+        value.toString().toLowerCase() === Status[code]?.toLowerCase()
     );
   }
 

packages/grpc-js/src/transport.ts

@@ -223,9 +223,8 @@ class Http2Transport implements Transport {
     );
 
     session.once('error', error => {
-      /* Do nothing here. Any error should also trigger a close event, which is
-       * where we want to handle that.  */
       this.trace('connection closed with error ' + (error as Error).message);
+      this.handleDisconnect();
     });
 
     if (logging.isTracerEnabled(TRACER_NAME)) {
@@ -383,6 +382,9 @@ class Http2Transport implements Transport {
    * Handle connection drops, but not GOAWAYs.
    */
   private handleDisconnect() {
+    if (this.disconnectHandled) {
+      return;
+    }
     this.clearKeepaliveTimeout();
     this.reportDisconnectToOwner(false);
     /* Give calls an event loop cycle to finish naturally before reporting the
@@ -773,6 +775,7 @@ export class Http2SubchannelConnector implements SubchannelConnector {
       );
       this.session = session;
       let errorMessage = 'Failed to connect';
+      let reportedError = false;
       session.unref();
       session.once('connect', () => {
         session.removeAllListeners();
@@ -783,12 +786,19 @@ export class Http2SubchannelConnector implements SubchannelConnector {
         this.session = null;
         // Leave time for error event to happen before rejecting
         setImmediate(() => {
-          reject(`${errorMessage} (${new Date().toISOString()})`);
+          if (!reportedError) {
+            reportedError = true;
+            reject(`${errorMessage} (${new Date().toISOString()})`);
+          }
         });
       });
       session.once('error', error => {
         errorMessage = (error as Error).message;
         this.trace('connection failed with error ' + errorMessage);
+        if (!reportedError) {
+          reportedError = true;
+          reject(`${errorMessage} (${new Date().toISOString()})`);
+        }
       });
     });
   }