Merge pull request #2078 from grpc/@grpc/grpc-js@1.6.x

Upmerge 1.6.x into master

packages/grpc-js-xds/README.md

@@ -27,4 +27,5 @@ const client = new MyServiceClient('xds:///example.com:123');
  - [xDS Timeouts](https://github.com/grpc/proposal/blob/master/A31-xds-timeout-support-and-config-selector.md)
  - [xDS Circuit Breaking](https://github.com/grpc/proposal/blob/master/A32-xds-circuit-breaking.md)
  - [xDS Client-Side Fault Injection](https://github.com/grpc/proposal/blob/master/A33-Fault-Injection.md)
- - [Client Status Discovery Service](https://github.com/grpc/proposal/blob/master/A40-csds-support.md)
+ - [Client Status Discovery Service](https://github.com/grpc/proposal/blob/master/A40-csds-support.md)
+ - [Outlier Detection](https://github.com/grpc/proposal/blob/master/A50-xds-outlier-detection.md) (experimental, disabled by default, enabled by setting the environment variable `GRPC_EXPERIMENTAL_ENABLE_OUTLIER_DETECTION=true`)

packages/grpc-js-xds/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js-xds",
-  "version": "1.5.2",
+  "version": "1.6.1",
   "description": "Plugin for @grpc/grpc-js. Adds the xds:// URL scheme and associated features.",
   "main": "build/src/index.js",
   "scripts": {
@@ -47,7 +47,7 @@
     "re2-wasm": "^1.0.1"
   },
   "peerDependencies": {
-    "@grpc/grpc-js": "~1.5.0"
+    "@grpc/grpc-js": "~1.6.0"
   },
   "engines": {
     "node": ">=10.10.0"

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

@@ -688,7 +688,6 @@ export class XdsClient {
   ack(serviceKind: AdsServiceKind) {
     /* An ack is the best indication of a successful interaction between the
      * client and the server, so we can reset the backoff timer here. */
-    this.adsBackoff.stop();
     this.adsBackoff.reset();
 
     this.updateNames(serviceKind);

packages/grpc-js/README.md

@@ -58,6 +58,7 @@ Many channel arguments supported in `grpc` are not supported in `@grpc/grpc-js`.
   - `grpc.enable_http_proxy`
   - `grpc.default_compression_algorithm`
   - `grpc.enable_channelz`
+  - `grpc.dns_min_time_between_resolutions_ms`
   - `grpc-node.max_session_memory`
   - `channelOverride`
   - `channelFactoryOverride`

packages/grpc-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js",
-  "version": "1.5.10",
+  "version": "1.6.2",
   "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

@@ -37,14 +37,47 @@ export interface BackoffOptions {
 }
 
 export class BackoffTimeout {
-  private initialDelay: number = INITIAL_BACKOFF_MS;
-  private multiplier: number = BACKOFF_MULTIPLIER;
-  private maxDelay: number = MAX_BACKOFF_MS;
-  private jitter: number = BACKOFF_JITTER;
+  /**
+   * The delay time at the start, and after each reset.
+   */
+  private readonly initialDelay: number = INITIAL_BACKOFF_MS;
+  /**
+   * The exponential backoff multiplier.
+   */
+  private readonly multiplier: number = BACKOFF_MULTIPLIER;
+  /**
+   * The maximum delay time
+   */
+  private readonly maxDelay: number = MAX_BACKOFF_MS;
+  /**
+   * The maximum fraction by which the delay time can randomly vary after
+   * applying the multiplier.
+   */
+  private readonly jitter: number = BACKOFF_JITTER;
+  /**
+   * The delay time for the next time the timer runs.
+   */
   private nextDelay: number;
+  /**
+   * The handle of the underlying timer. If running is false, this value refers
+   * to an object representing a timer that has ended, but it can still be
+   * interacted with without error.
+   */
   private timerId: NodeJS.Timer;
+  /**
+   * Indicates whether the timer is currently running.
+   */
   private running = false;
+  /**
+   * Indicates whether the timer should keep the Node process running if no
+   * other async operation is doing so.
+   */
   private hasRef = true;
+  /**
+   * The time that the currently running timer was started. Only valid if
+   * running is true.
+   */
+  private startTime: Date = new Date();
 
   constructor(private callback: () => void, options?: BackoffOptions) {
     if (options) {
@@ -66,18 +99,23 @@ export class BackoffTimeout {
     clearTimeout(this.timerId);
   }
 
+  private runTimer(delay: number) {
+    this.timerId = setTimeout(() => {
+      this.callback();
+      this.running = false;
+    }, delay);
+    if (!this.hasRef) {
+      this.timerId.unref?.();
+    }
+  }
+
   /**
    * Call the callback after the current amount of delay time
    */
   runOnce() {
     this.running = true;
-    this.timerId = setTimeout(() => {
-      this.callback();
-      this.running = false;
-    }, this.nextDelay);
-    if (!this.hasRef) {
-      this.timerId.unref?.();
-    }
+    this.startTime = new Date();
+    this.runTimer(this.nextDelay);
     const nextBackoff = Math.min(
       this.nextDelay * this.multiplier,
       this.maxDelay
@@ -97,21 +135,44 @@ export class BackoffTimeout {
   }
 
   /**
-   * Reset the delay time to its initial value.
+   * Reset the delay time to its initial value. If the timer is still running,
+   * retroactively apply that reset to the current timer.
    */
   reset() {
     this.nextDelay = this.initialDelay;
+    if (this.running) {
+      const now = new Date();
+      const newEndTime = this.startTime;
+      newEndTime.setMilliseconds(newEndTime.getMilliseconds() + this.nextDelay);
+      clearTimeout(this.timerId);
+      if (now < newEndTime) {
+        this.runTimer(newEndTime.getTime() - now.getTime());
+      } else {
+        this.running = false;
+      }
+    }
   }
 
+  /**
+   * Check whether the timer is currently running.
+   */
   isRunning() {
     return this.running;
   }
 
+  /**
+   * Set that while the timer is running, it should keep the Node process
+   * running.
+   */
   ref() {
     this.hasRef = true;
     this.timerId.ref?.();
   }
 
+  /**
+   * Set that while the timer is running, it should not keep the Node process
+   * running.
+   */
   unref() {
     this.hasRef = false;
     this.timerId.unref?.();

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

@@ -839,7 +839,16 @@ export class Http2CallStream implements Call {
       message,
       flags: context.flags,
     };
-    const cb: WriteCallback = context.callback ?? (() => {});
+    const cb: WriteCallback = (error?: Error | null) => {
+      let code: Status = Status.UNAVAILABLE;
+      if ((error as NodeJS.ErrnoException)?.code === 'ERR_STREAM_WRITE_AFTER_END') {
+        code = Status.INTERNAL;
+      }
+      if (error) {
+        this.cancelWithStatus(code, `Write error: ${error.message}`);
+      }
+      context.callback?.();
+    };
     this.isWriteFilterPending = true;
     this.filterStack.sendMessage(Promise.resolve(writeObj)).then((message) => {
       this.isWriteFilterPending = false;

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

@@ -43,6 +43,7 @@ export interface ChannelOptions {
   'grpc.http_connect_creds'?: string;
   'grpc.default_compression_algorithm'?: CompressionAlgorithms;
   'grpc.enable_channelz'?: number;
+  'grpc.dns_min_time_between_resolutions_ms'?: number;
   'grpc-node.max_session_memory'?: number;
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   [key: string]: any;
@@ -69,6 +70,7 @@ export const recognizedOptions = {
   'grpc.max_receive_message_length': true,
   'grpc.enable_http_proxy': true,
   'grpc.enable_channelz': true,
+  'grpc.dns_min_time_between_resolutions_ms': true,
   'grpc-node.max_session_memory': true,
 };
 

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

@@ -45,6 +45,8 @@ function trace(text: string): void {
  */
 const DEFAULT_PORT = 443;
 
+const DEFAULT_MIN_TIME_BETWEEN_RESOLUTIONS_MS = 30_000;
+
 const resolveTxtPromise = util.promisify(dns.resolveTxt);
 const dnsLookupPromise = util.promisify(dns.lookup);
 
@@ -79,6 +81,12 @@ class DnsResolver implements Resolver {
   private readonly ipResult: SubchannelAddress[] | null;
   private readonly dnsHostname: string | null;
   private readonly port: number | null;
+  /**
+   * Minimum time between resolutions, measured as the time between starting
+   * successive resolution requests. Only applies to successful resolutions.
+   * Failures are handled by the backoff timer.
+   */
+  private readonly minTimeBetweenResolutionsMs: number;
   private pendingLookupPromise: Promise<dns.LookupAddress[]> | null = null;
   private pendingTxtPromise: Promise<string[][]> | null = null;
   private latestLookupResult: TcpSubchannelAddress[] | null = null;
@@ -88,6 +96,8 @@ class DnsResolver implements Resolver {
   private defaultResolutionError: StatusObject;
   private backoff: BackoffTimeout;
   private continueResolving = false;
+  private nextResolutionTimer: NodeJS.Timer;
+  private isNextResolutionTimerRunning = false;
   constructor(
     private target: GrpcUri,
     private listener: ResolverListener,
@@ -134,6 +144,10 @@ class DnsResolver implements Resolver {
       }
     }, backoffOptions);
     this.backoff.unref();
+
+    this.minTimeBetweenResolutionsMs = channelOptions['grpc.dns_min_time_between_resolutions_ms'] ?? DEFAULT_MIN_TIME_BETWEEN_RESOLUTIONS_MS;
+    this.nextResolutionTimer = setTimeout(() => {}, 0);
+    clearTimeout(this.nextResolutionTimer);
   }
 
   /**
@@ -183,6 +197,7 @@ class DnsResolver implements Resolver {
         (addressList) => {
           this.pendingLookupPromise = null;
           this.backoff.reset();
+          this.backoff.stop();
           const ip4Addresses: dns.LookupAddress[] = addressList.filter(
             (addr) => addr.family === 4
           );
@@ -229,6 +244,7 @@ class DnsResolver implements Resolver {
               (err as Error).message
           );
           this.pendingLookupPromise = null;
+          this.stopNextResolutionTimer();
           this.listener.onError(this.defaultResolutionError);
         }
       );
@@ -282,17 +298,34 @@ class DnsResolver implements Resolver {
     }
   }
 
+  private startNextResolutionTimer() {
+    this.nextResolutionTimer = setTimeout(() => {
+      this.stopNextResolutionTimer();
+      if (this.continueResolving) {
+        this.startResolutionWithBackoff();
+      }
+    }, this.minTimeBetweenResolutionsMs).unref?.();
+    this.isNextResolutionTimerRunning = true;
+  }
+
+  private stopNextResolutionTimer() {
+    clearTimeout(this.nextResolutionTimer);
+    this.isNextResolutionTimerRunning = false;
+  }
+
   private startResolutionWithBackoff() {
       this.startResolution();
       this.backoff.runOnce();
+      this.startNextResolutionTimer();
   }
 
   updateResolution() {
     /* If there is a pending lookup, just let it finish. Otherwise, if the
-     * backoff timer is running, do another lookup when it ends, and if not,
-     * do another lookup immeidately. */
+     * nextResolutionTimer or backoff timer is running, set the
+     * continueResolving flag to resolve when whichever of those timers
+     * fires. Otherwise, start resolving immediately. */
     if (this.pendingLookupPromise === null) {
-      if (this.backoff.isRunning()) {
+      if (this.isNextResolutionTimerRunning || this.backoff.isRunning()) {
         this.continueResolving = true;
       } else {
         this.startResolutionWithBackoff();
@@ -301,9 +334,9 @@ class DnsResolver implements Resolver {
   }
 
   destroy() {
-    /* Do nothing. There is not a practical way to cancel in-flight DNS
-     * requests, and after this function is called we can expect that
-     * updateResolution will not be called again. */
+    this.continueResolving = false;
+    this.backoff.stop();
+    this.stopNextResolutionTimer();
   }
 
   /**