Merge pull request #3007 from murgatroid99/grpc-js_1.13_upmerge

Merge v1.13.x into master

Author: murgatroid99

packages/grpc-js-xds/interop/test-client.Dockerfile

@@ -42,7 +42,7 @@ COPY --from=build /node/src/grpc-node/packages/grpc-js ./packages/grpc-js/
 COPY --from=build /node/src/grpc-node/packages/grpc-js-xds ./packages/grpc-js-xds/
 
 ENV GRPC_VERBOSITY="DEBUG"
-ENV GRPC_TRACE=xds_client,xds_resolver,xds_cluster_manager,cds_balancer,xds_cluster_resolver,xds_cluster_impl,priority,weighted_target,round_robin,resolving_load_balancer,subchannel,keepalive,dns_resolver,fault_injection,http_filter,csds,outlier_detection,server,server_call,ring_hash,transport,certificate_provider,xds_channel_credentials
+ENV GRPC_TRACE=xds_client,xds_resolver,xds_cluster_manager,cds_balancer,xds_cluster_resolver,xds_cluster_impl,priority,weighted_target,round_robin,resolving_load_balancer,subchannel,keepalive,dns_resolver,fault_injection,http_filter,csds,outlier_detection,server,server_call,ring_hash,transport,certificate_provider,xds_channel_credentials,backoff
 ENV NODE_XDS_INTEROP_VERBOSITY=1
 
 ENTRYPOINT [ "/nodejs/bin/node", "/node/src/grpc-node/packages/grpc-js-xds/build/interop/xds-interop-client" ]

packages/grpc-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js",
-  "version": "1.13.0",
+  "version": "1.13.4",
   "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/backoff-timeout.ts

@@ -15,6 +15,11 @@
  *
  */
 
+import { LogVerbosity } from './constants';
+import * as logging from './logging';
+
+const TRACER_NAME = 'backoff';
+
 const INITIAL_BACKOFF_MS = 1000;
 const BACKOFF_MULTIPLIER = 1.6;
 const MAX_BACKOFF_MS = 120000;
@@ -84,7 +89,12 @@ export class BackoffTimeout {
    */
   private endTime: Date = new Date();
 
+  private id: number;
+
+  private static nextId = 0;
+
   constructor(private callback: () => void, options?: BackoffOptions) {
+    this.id = BackoffTimeout.getNextId();
     if (options) {
       if (options.initialDelay) {
         this.initialDelay = options.initialDelay;
@@ -99,20 +109,31 @@ export class BackoffTimeout {
         this.maxDelay = options.maxDelay;
       }
     }
+    this.trace('constructed initialDelay=' + this.initialDelay + ' multiplier=' + this.multiplier + ' jitter=' + this.jitter + ' maxDelay=' + this.maxDelay);
     this.nextDelay = this.initialDelay;
     this.timerId = setTimeout(() => {}, 0);
     clearTimeout(this.timerId);
   }
 
+  private static getNextId() {
+    return this.nextId++;
+  }
+
+  private trace(text: string) {
+    logging.trace(LogVerbosity.DEBUG, TRACER_NAME, '{' + this.id + '} ' + text);
+  }
+
   private runTimer(delay: number) {
+    this.trace('runTimer(delay=' + delay + ')');
     this.endTime = this.startTime;
     this.endTime.setMilliseconds(
       this.endTime.getMilliseconds() + delay
     );
     clearTimeout(this.timerId);
     this.timerId = setTimeout(() => {
-      this.callback();
+      this.trace('timer fired');
       this.running = false;
+      this.callback();
     }, delay);
     if (!this.hasRef) {
       this.timerId.unref?.();
@@ -123,6 +144,7 @@ export class BackoffTimeout {
    * Call the callback after the current amount of delay time
    */
   runOnce() {
+    this.trace('runOnce()');
     this.running = true;
     this.startTime = new Date();
     this.runTimer(this.nextDelay);
@@ -140,6 +162,7 @@ export class BackoffTimeout {
    * again.
    */
   stop() {
+    this.trace('stop()');
     clearTimeout(this.timerId);
     this.running = false;
   }
@@ -149,6 +172,7 @@ export class BackoffTimeout {
    * retroactively apply that reset to the current timer.
    */
   reset() {
+    this.trace('reset() running=' + this.running);
     this.nextDelay = this.initialDelay;
     if (this.running) {
       const now = new Date();

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

@@ -206,6 +206,18 @@ function getConnectionOptions(secureContext: SecureContext, verifyOptions: Verif
   const connectionOptions: ConnectionOptions = {
     secureContext: secureContext
   };
+  let realTarget: GrpcUri = channelTarget;
+  if ('grpc.http_connect_target' in options) {
+    const parsedTarget = parseUri(options['grpc.http_connect_target']!);
+    if (parsedTarget) {
+      realTarget = parsedTarget;
+    }
+  }
+  const targetPath = getDefaultAuthority(realTarget);
+  const hostPort = splitHostPort(targetPath);
+  const remoteHost = hostPort?.host ?? targetPath;
+  connectionOptions.host = remoteHost;
+
   if (verifyOptions.checkServerIdentity) {
     connectionOptions.checkServerIdentity = verifyOptions.checkServerIdentity;
   }
@@ -225,36 +237,11 @@ function getConnectionOptions(secureContext: SecureContext, verifyOptions: Verif
     };
     connectionOptions.servername = sslTargetNameOverride;
   } else {
-    if ('grpc.http_connect_target' in options) {
-      /* This is more or less how servername will be set in createSession
-       * if a connection is successfully established through the proxy.
-       * If the proxy is not used, these connectionOptions are discarded
-       * anyway */
-      const targetPath = getDefaultAuthority(
-        parseUri(options['grpc.http_connect_target'] as string) ?? {
-          path: 'localhost',
-        }
-      );
-      const hostPort = splitHostPort(targetPath);
-      connectionOptions.servername = hostPort?.host ?? targetPath;
-    }
+    connectionOptions.servername = remoteHost;
   }
   if (options['grpc-node.tls_enable_trace']) {
     connectionOptions.enableTrace = true;
   }
-
-  let realTarget: GrpcUri = channelTarget;
-  if ('grpc.http_connect_target' in options) {
-    const parsedTarget = parseUri(options['grpc.http_connect_target']!);
-    if (parsedTarget) {
-      realTarget = parsedTarget;
-    }
-  }
-  const targetPath = getDefaultAuthority(realTarget);
-  const hostPort = splitHostPort(targetPath);
-  const remoteHost = hostPort?.host ?? targetPath;
-  connectionOptions.host = remoteHost;
-  connectionOptions.servername = remoteHost;
   return connectionOptions;
 }
 
@@ -268,7 +255,7 @@ class SecureConnectorImpl implements SecureConnector {
     };
     return new Promise<SecureConnectResult>((resolve, reject) => {
       const tlsSocket = tlsConnect(tlsConnectOptions, () => {
-        if (!tlsSocket.authorized) {
+        if ((this.connectionOptions.rejectUnauthorized ?? true) && !tlsSocket.authorized) {
           reject(tlsSocket.authorizationError);
           return;
         }
@@ -364,7 +351,7 @@ class CertificateProviderChannelCredentialsImpl extends ChannelCredentials {
         const tlsSocket = tlsConnect(tlsConnectOptions, () => {
           tlsSocket.removeListener('close', closeCallback);
           tlsSocket.removeListener('error', errorCallback);
-          if (!tlsSocket.authorized) {
+          if ((this.parent.verifyOptions.rejectUnauthorized ?? true) && !tlsSocket.authorized) {
             reject(tlsSocket.authorizationError);
             return;
           }

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

@@ -297,16 +297,16 @@ export class InternalChannel {
     /* The global boolean parameter to getSubchannelPool has the inverse meaning to what
      * the grpc.use_local_subchannel_pool channel option means. */
     this.subchannelPool = getSubchannelPool(
-      (options['grpc.use_local_subchannel_pool'] ?? 0) === 0
+      (this.options['grpc.use_local_subchannel_pool'] ?? 0) === 0
     );
     this.retryBufferTracker = new MessageBufferTracker(
-      options['grpc.retry_buffer_size'] ?? DEFAULT_RETRY_BUFFER_SIZE_BYTES,
-      options['grpc.per_rpc_retry_buffer_size'] ??
+      this.options['grpc.retry_buffer_size'] ?? DEFAULT_RETRY_BUFFER_SIZE_BYTES,
+      this.options['grpc.per_rpc_retry_buffer_size'] ??
         DEFAULT_PER_RPC_RETRY_BUFFER_SIZE_BYTES
     );
-    this.keepaliveTime = options['grpc.keepalive_time_ms'] ?? -1;
+    this.keepaliveTime = this.options['grpc.keepalive_time_ms'] ?? -1;
     this.idleTimeoutMs = Math.max(
-      options['grpc.client_idle_timeout_ms'] ?? DEFAULT_IDLE_TIMEOUT_MS,
+      this.options['grpc.client_idle_timeout_ms'] ?? DEFAULT_IDLE_TIMEOUT_MS,
       MIN_IDLE_TIMEOUT_MS
     );
     const channelControlHelper: ChannelControlHelper = {
@@ -372,7 +372,7 @@ export class InternalChannel {
     this.resolvingLoadBalancer = new ResolvingLoadBalancer(
       this.target,
       channelControlHelper,
-      options,
+      this.options,
       (serviceConfig, configSelector) => {
         if (serviceConfig.retryThrottling) {
           RETRY_THROTTLER_MAP.set(

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

@@ -58,15 +58,15 @@ export class RetryThrottler {
   }
 
   addCallSucceeded() {
-    this.tokens = Math.max(this.tokens + this.tokenRatio, this.maxTokens);
+    this.tokens = Math.min(this.tokens + this.tokenRatio, this.maxTokens);
   }
 
   addCallFailed() {
-    this.tokens = Math.min(this.tokens - 1, 0);
+    this.tokens = Math.max(this.tokens - 1, 0);
   }
 
   canRetryCall() {
-    return this.tokens > this.maxTokens / 2;
+    return this.tokens > (this.maxTokens / 2);
   }
 }
 
@@ -225,7 +225,10 @@ export class RetryingCall implements Call, DeadlineInfoProvider {
     const maxAttemptsLimit =
       channel.getOptions()['grpc-node.retry_max_attempts_limit'] ??
       DEFAULT_MAX_ATTEMPTS_LIMIT;
-    if (callConfig.methodConfig.retryPolicy) {
+    if (channel.getOptions()['grpc.enable_retries'] === 0) {
+      this.state = 'NO_RETRY';
+      this.maxAttempts = 1;
+    } else if (callConfig.methodConfig.retryPolicy) {
       this.state = 'RETRY';
       const retryPolicy = callConfig.methodConfig.retryPolicy;
       this.nextRetryBackoffSec = this.initialRetryBackoffSec = Number(
@@ -241,9 +244,6 @@ export class RetryingCall implements Call, DeadlineInfoProvider {
         callConfig.methodConfig.hedgingPolicy.maxAttempts,
         maxAttemptsLimit
       );
-    } else if (channel.getOptions()['grpc.enable_retries'] === 0) {
-      this.state = 'NO_RETRY';
-      this.maxAttempts = 1;
     } else {
       this.state = 'TRANSPARENT_ONLY';
       this.maxAttempts = 1;
@@ -483,6 +483,9 @@ export class RetryingCall implements Call, DeadlineInfoProvider {
         callback(true);
         this.attempts += 1;
         this.startNewAttempt();
+      } else {
+        this.trace('Retry attempt denied by throttling policy');
+        callback(false);
       }
     }, retryDelayMs);
   }

packages/grpc-js/src/transport.ts

@@ -712,19 +712,19 @@ export class Http2SubchannelConnector implements SubchannelConnector {
           reject(`${errorMessage} (${new Date().toISOString()})`);
         }
       };
-      const session = http2.connect(`${scheme}://${targetPath}`, {
+      const sessionOptions: http2.ClientSessionOptions = {
         createConnection: (authority, option) => {
           return secureConnectResult.socket;
         },
         settings: {
           initialWindowSize:
             options['grpc-node.flow_control_window'] ??
-            http2.getDefaultSettings().initialWindowSize ?? 65535,
+            http2.getDefaultSettings?.()?.initialWindowSize ?? 65535,
         }
-      });
-
+      };
+      const session = http2.connect(`${scheme}://${targetPath}`, sessionOptions);
       // Prepare window size configuration for remoteSettings handler
-      const defaultWin = http2.getDefaultSettings().initialWindowSize ?? 65535; // 65 535 B
+      const defaultWin = http2.getDefaultSettings?.()?.initialWindowSize ?? 65535; // 65 535 B
       const connWin = options[
         'grpc-node.flow_control_window'
       ] as number | undefined;
@@ -745,7 +745,7 @@ export class Http2SubchannelConnector implements SubchannelConnector {
             if (delta > 0) (session as any).incrementWindowSize(delta);
           }
         }
-        
+
         session.removeAllListeners();
         secureConnectResult.socket.removeListener('close', closeHandler);
         secureConnectResult.socket.removeListener('error', errorHandler);
@@ -799,6 +799,7 @@ export class Http2SubchannelConnector implements SubchannelConnector {
       await secureConnector.waitForReady();
       this.trace(addressString + ' secureConnector is ready');
       tcpConnection = await this.tcpConnect(address, options);
+      tcpConnection.setNoDelay();
       this.trace(addressString + ' Established TCP connection');
       secureConnectResult = await secureConnector.connect(tcpConnection);
       this.trace(addressString + ' Established secure connection');

packages/grpc-js/test/test-channel-credentials.ts

@@ -228,6 +228,15 @@ describe('ChannelCredentials usage', () => {
       done();
     });
   })
+  it('Should accept self-signed certs with acceptUnauthorized', done => {
+    const client = new echoService(`localhost:${portNum}`, grpc.credentials.createSsl(null, null, null, {rejectUnauthorized: false}));
+    client.echo({ value: 'test value', value2: 3 }, (error: ServiceError | null, response: any) => {
+      client.close();
+      assert.ifError(error);
+      assert.deepStrictEqual(response, { value: 'test value', value2: 3 });
+      done();
+    });
+  });
 });
 
 describe('Channel credentials mtls', () => {

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

@@ -175,6 +175,10 @@ describe('Retries', () => {
             },
           },
         ],
+        retryThrottling: {
+          maxTokens: 1000,
+          tokenRatio: 0.1,
+        },
       };
       client = new EchoService(
         `localhost:${port}`,