Merge pull request #2471 from murgatroid99/grpc-js_channel_idle_timeout

grpc-js: Implement channel idle timeout

Author: murgatroid99

packages/grpc-js/README.md

@@ -63,6 +63,7 @@ Many channel arguments supported in `grpc` are not supported in `@grpc/grpc-js`.
   - `grpc.per_rpc_retry_buffer_size`
   - `grpc.retry_buffer_size`
   - `grpc.service_config_disable_resolution`
+  - `grpc.client_idle_timeout_ms`
   - `grpc-node.max_session_memory`
   - `channelOverride`
   - `channelFactoryOverride`

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

@@ -56,6 +56,7 @@ export interface ChannelOptions {
   'grpc.max_connection_age_grace_ms'?: number;
   'grpc-node.max_session_memory'?: number;
   'grpc.service_config_disable_resolution'?: number;
+  'grpc.client_idle_timeout_ms'?: number;
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   [key: string]: any;
 }
@@ -89,6 +90,7 @@ export const recognizedOptions = {
   'grpc.max_connection_age_grace_ms': true,
   'grpc-node.max_session_memory': true,
   'grpc.service_config_disable_resolution': true,
+  'grpc.client_idle_timeout_ms': true,
 };
 
 export function channelOptionsEqual(

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

@@ -20,7 +20,12 @@ 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, PickResultType } from './picker';
+import {
+  UnavailablePicker,
+  Picker,
+  PickResultType,
+  QueuePicker,
+} from './picker';
 import { Metadata } from './metadata';
 import { Status, LogVerbosity, Propagate } from './constants';
 import { FilterStackFactory } from './filter-stack';
@@ -85,6 +90,11 @@ import {
  */
 const MAX_TIMEOUT_TIME = 2147483647;
 
+const MIN_IDLE_TIMEOUT_MS = 1000;
+
+// 30 minutes
+const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000;
+
 interface ConnectivityStateWatcher {
   currentState: ConnectivityState;
   timer: NodeJS.Timeout | null;
@@ -153,8 +163,8 @@ class ChannelSubchannelWrapper
 }
 
 export class InternalChannel {
-  private resolvingLoadBalancer: ResolvingLoadBalancer;
-  private subchannelPool: SubchannelPool;
+  private readonly resolvingLoadBalancer: ResolvingLoadBalancer;
+  private readonly subchannelPool: SubchannelPool;
   private connectivityState: ConnectivityState = ConnectivityState.IDLE;
   private currentPicker: Picker = new UnavailablePicker();
   /**
@@ -164,17 +174,17 @@ export class InternalChannel {
   private configSelectionQueue: ResolvingCall[] = [];
   private pickQueue: LoadBalancingCall[] = [];
   private connectivityStateWatchers: ConnectivityStateWatcher[] = [];
-  private defaultAuthority: string;
-  private filterStackFactory: FilterStackFactory;
-  private target: GrpcUri;
+  private readonly defaultAuthority: string;
+  private readonly filterStackFactory: FilterStackFactory;
+  private readonly target: GrpcUri;
   /**
    * This timer does not do anything on its own. Its purpose is to hold the
    * event loop open while there are any pending calls for the channel that
    * have not yet been assigned to specific subchannels. In other words,
    * the invariant is that callRefTimer is reffed if and only if pickQueue
    * is non-empty.
    */
-  private callRefTimer: NodeJS.Timer;
+  private readonly callRefTimer: NodeJS.Timer;
   private configSelector: ConfigSelector | null = null;
   /**
    * This is the error from the name resolver if it failed most recently. It
@@ -184,17 +194,22 @@ export class InternalChannel {
    * than TRANSIENT_FAILURE.
    */
   private currentResolutionError: StatusObject | null = null;
-  private retryBufferTracker: MessageBufferTracker;
+  private readonly retryBufferTracker: MessageBufferTracker;
   private keepaliveTime: number;
-  private wrappedSubchannels: Set<ChannelSubchannelWrapper> = new Set();
+  private readonly wrappedSubchannels: Set<ChannelSubchannelWrapper> =
+    new Set();
+
+  private callCount = 0;
+  private idleTimer: NodeJS.Timer | null = null;
+  private readonly idleTimeoutMs: number;
 
   // Channelz info
   private readonly channelzEnabled: boolean = true;
-  private originalTarget: string;
-  private channelzRef: ChannelRef;
-  private channelzTrace: ChannelzTrace;
-  private callTracker = new ChannelzCallTracker();
-  private childrenTracker = new ChannelzChildrenTracker();
+  private readonly originalTarget: string;
+  private readonly channelzRef: ChannelRef;
+  private readonly channelzTrace: ChannelzTrace;
+  private readonly callTracker = new ChannelzCallTracker();
+  private readonly childrenTracker = new ChannelzChildrenTracker();
 
   constructor(
     target: string,
@@ -265,6 +280,10 @@ export class InternalChannel {
         DEFAULT_PER_RPC_RETRY_BUFFER_SIZE_BYTES
     );
     this.keepaliveTime = options['grpc.keepalive_time_ms'] ?? -1;
+    this.idleTimeoutMs = Math.max(
+      options['grpc.client_idle_timeout_ms'] ?? DEFAULT_IDLE_TIMEOUT_MS,
+      MIN_IDLE_TIMEOUT_MS
+    );
     const channelControlHelper: ChannelControlHelper = {
       createSubchannel: (
         subchannelAddress: SubchannelAddress,
@@ -548,6 +567,49 @@ export class InternalChannel {
     this.callRefTimerRef();
   }
 
+  private enterIdle() {
+    this.resolvingLoadBalancer.destroy();
+    this.updateState(ConnectivityState.IDLE);
+    this.currentPicker = new QueuePicker(this.resolvingLoadBalancer);
+  }
+
+  private maybeStartIdleTimer() {
+    if (this.callCount === 0) {
+      this.idleTimer = setTimeout(() => {
+        this.trace(
+          'Idle timer triggered after ' +
+            this.idleTimeoutMs +
+            'ms of inactivity'
+        );
+        this.enterIdle();
+      }, this.idleTimeoutMs);
+      this.idleTimer.unref?.();
+    }
+  }
+
+  private onCallStart() {
+    if (this.channelzEnabled) {
+      this.callTracker.addCallStarted();
+    }
+    this.callCount += 1;
+    if (this.idleTimer) {
+      clearTimeout(this.idleTimer);
+      this.idleTimer = null;
+    }
+  }
+
+  private onCallEnd(status: StatusObject) {
+    if (this.channelzEnabled) {
+      if (status.code === Status.OK) {
+        this.callTracker.addCallSucceeded();
+      } else {
+        this.callTracker.addCallFailed();
+      }
+    }
+    this.callCount -= 1;
+    this.maybeStartIdleTimer();
+  }
+
   createLoadBalancingCall(
     callConfig: CallConfig,
     method: string,
@@ -653,16 +715,10 @@ export class InternalChannel {
       callNumber
     );
 
-    if (this.channelzEnabled) {
-      this.callTracker.addCallStarted();
-      call.addStatusWatcher(status => {
-        if (status.code === Status.OK) {
-          this.callTracker.addCallSucceeded();
-        } else {
-          this.callTracker.addCallFailed();
-        }
-      });
-    }
+    this.onCallStart();
+    call.addStatusWatcher(status => {
+      this.onCallEnd(status);
+    });
     return call;
   }
 
@@ -685,6 +741,7 @@ export class InternalChannel {
     const connectivityState = this.connectivityState;
     if (tryToConnect) {
       this.resolvingLoadBalancer.exitIdle();
+      this.maybeStartIdleTimer();
     }
     return connectivityState;
   }

packages/grpc-js/src/load-balancer-child-handler.ts

@@ -150,6 +150,10 @@ export class ChildLoadBalancerHandler implements LoadBalancer {
     }
   }
   destroy(): void {
+    /* Note: state updates are only propagated from the child balancer if that
+     * object is equal to this.currentChild or this.pendingChild. Since this
+     * function sets both of those to null, no further state updates will
+     * occur after this function returns. */
     if (this.currentChild) {
       this.currentChild.destroy();
       this.currentChild = null;

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

@@ -207,6 +207,9 @@ class DnsResolver implements Resolver {
       this.pendingLookupPromise = dnsLookupPromise(hostname, { all: true });
       this.pendingLookupPromise.then(
         addressList => {
+          if (this.pendingLookupPromise === null) {
+            return;
+          }
           this.pendingLookupPromise = null;
           this.backoff.reset();
           this.backoff.stop();
@@ -248,6 +251,9 @@ class DnsResolver implements Resolver {
           );
         },
         err => {
+          if (this.pendingLookupPromise === null) {
+            return;
+          }
           trace(
             'Resolution error for target ' +
               uriToString(this.target) +
@@ -268,6 +274,9 @@ class DnsResolver implements Resolver {
         this.pendingTxtPromise = resolveTxtPromise(hostname);
         this.pendingTxtPromise.then(
           txtRecord => {
+            if (this.pendingTxtPromise === null) {
+              return;
+            }
             this.pendingTxtPromise = null;
             try {
               this.latestServiceConfig = extractAndSelectServiceConfig(
@@ -348,10 +357,21 @@ class DnsResolver implements Resolver {
     }
   }
 
+  /**
+   * Reset the resolver to the same state it had when it was created. In-flight
+   * DNS requests cannot be cancelled, but they are discarded and their results
+   * will be ignored.
+   */
   destroy() {
     this.continueResolving = false;
+    this.backoff.reset();
     this.backoff.stop();
     this.stopNextResolutionTimer();
+    this.pendingLookupPromise = null;
+    this.pendingTxtPromise = null;
+    this.latestLookupResult = null;
+    this.latestServiceConfig = null;
+    this.latestServiceConfigError = null;
   }
 
   /**

packages/grpc-js/src/resolver.ts

@@ -82,7 +82,10 @@ export interface Resolver {
   updateResolution(): void;
 
   /**
-   * Destroy the resolver. Should be called when the owning channel shuts down.
+   * Discard all resources owned by the resolver. A later call to
+   * `updateResolution` should reinitialize those resources.  No
+   * `ResolverListener` callbacks should be called after `destroy` is called
+   * until `updateResolution` is called again.
    */
   destroy(): void;
 }

packages/grpc-js/src/resolving-load-balancer.ts

@@ -94,9 +94,9 @@ export class ResolvingLoadBalancer implements LoadBalancer {
   /**
    * The resolver class constructed for the target address.
    */
-  private innerResolver: Resolver;
+  private readonly innerResolver: Resolver;
 
-  private childLoadBalancer: ChildLoadBalancerHandler;
+  private readonly childLoadBalancer: ChildLoadBalancerHandler;
   private latestChildState: ConnectivityState = ConnectivityState.IDLE;
   private latestChildPicker: Picker = new QueuePicker(this);
   /**
@@ -324,7 +324,13 @@ export class ResolvingLoadBalancer implements LoadBalancer {
   destroy() {
     this.childLoadBalancer.destroy();
     this.innerResolver.destroy();
-    this.updateState(ConnectivityState.SHUTDOWN, new UnavailablePicker());
+    this.backoffTimeout.reset();
+    this.backoffTimeout.stop();
+    this.latestChildState = ConnectivityState.IDLE;
+    this.latestChildPicker = new QueuePicker(this);
+    this.currentState = ConnectivityState.IDLE;
+    this.previousServiceConfig = null;
+    this.continueResolving = false;
   }
 
   getTypeName() {