Merge remote-tracking branch 'upstream/@grpc/grpc-js@1.8.x' into v1.8…

….x_upmerge

packages/grpc-js-xds/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js-xds",
-  "version": "1.8.1",
+  "version": "1.8.2",
   "description": "Plugin for @grpc/grpc-js. Adds the xds:// URL scheme and associated features.",
   "main": "build/src/index.js",
   "scripts": {

packages/grpc-js-xds/src/xds-stream-state/xds-stream-state.ts

@@ -15,7 +15,7 @@
  *
  */
 
-import { experimental, logVerbosity, StatusObject } from "@grpc/grpc-js";
+import { experimental, logVerbosity, Metadata, status, StatusObject } from "@grpc/grpc-js";
 import { Any__Output } from "../generated/google/protobuf/Any";
 
 const TRACER_NAME = 'xds_client';
@@ -157,47 +157,55 @@ export abstract class BaseXdsStreamState<ResponseType> implements XdsStreamState
     return Array.from(this.subscriptions.keys());
   }
   handleResponses(responses: ResourcePair<ResponseType>[]): HandleResponseResult {
-    const validResponses: ResponseType[] = [];
     let result: HandleResponseResult = {
       accepted: [],
       rejected: [],
       missing: []
     }
+    const allResourceNames = new Set<string>();
     for (const {resource, raw} of responses) {
       const resourceName = this.getResourceName(resource);
+      allResourceNames.add(resourceName);
+      const subscriptionEntry = this.subscriptions.get(resourceName);
       if (this.validateResponse(resource)) {
-        validResponses.push(resource);
         result.accepted.push({
           name: resourceName,
           raw: raw});
+        if (subscriptionEntry) {
+          for (const watcher of subscriptionEntry.watchers) {
+            /* Use process.nextTick to prevent errors from the watcher from
+             * bubbling up through here. */
+            process.nextTick(() => {
+              watcher.onValidUpdate(resource);
+            });
+          }
+          clearTimeout(subscriptionEntry.resourceTimer);
+          subscriptionEntry.cachedResponse = resource;
+          if (subscriptionEntry.deletionIgnored) {
+            experimental.log(logVerbosity.INFO, `Received resource with previously ignored deletion: ${resourceName}`);
+            subscriptionEntry.deletionIgnored = false;
+          }
+        }
       } else {
         this.trace('Validation failed for message ' + JSON.stringify(resource));
         result.rejected.push({
           name: resourceName, 
           raw: raw,
           error: `Validation failed for resource ${resourceName}`
         });
-      }
-    }
-    const allResourceNames = new Set<string>();
-    for (const resource of validResponses) {
-      const resourceName = this.getResourceName(resource);
-      allResourceNames.add(resourceName);
-      const subscriptionEntry = this.subscriptions.get(resourceName);
-      if (subscriptionEntry) {
-        const watchers = subscriptionEntry.watchers;
-        for (const watcher of watchers) {
-          /* Use process.nextTick to prevent errors from the watcher from
-           * bubbling up through here. */
-          process.nextTick(() => {
-            watcher.onValidUpdate(resource);
-          });
-        }
-        clearTimeout(subscriptionEntry.resourceTimer);
-        subscriptionEntry.cachedResponse = resource;
-        if (subscriptionEntry.deletionIgnored) {
-          experimental.log(logVerbosity.INFO, 'Received resource with previously ignored deletion: ' + resourceName);
-          subscriptionEntry.deletionIgnored = false;
+        if (subscriptionEntry) {
+          for (const watcher of subscriptionEntry.watchers) {
+            /* Use process.nextTick to prevent errors from the watcher from
+             * bubbling up through here. */
+            process.nextTick(() => {
+              watcher.onTransientError({
+                code: status.UNAVAILABLE,
+                details: `Validation failed for resource ${resourceName}`,
+                metadata: new Metadata()
+              });
+            });
+          }
+          clearTimeout(subscriptionEntry.resourceTimer);
         }
       }
     }

packages/grpc-js-xds/test/test-nack.ts

@@ -0,0 +1,160 @@
+/*
+ * Copyright 2023 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import * as assert from 'assert';
+import { register } from "../src";
+import { Backend } from "./backend";
+import { XdsTestClient } from "./client";
+import { FakeCluster, FakeRouteGroup } from "./framework";
+import { XdsServer } from "./xds-server";
+
+register();
+
+describe('Validation errors', () => {
+  let xdsServer: XdsServer;
+  let client: XdsTestClient;
+  beforeEach(done => {
+    xdsServer = new XdsServer();
+    xdsServer.startServer(error => {
+      done(error);
+    });
+  });
+  afterEach(() => {
+    client?.close();
+    xdsServer?.shutdownServer();
+  });
+  it('Should continue to use a valid resource after receiving an invalid EDS update', done => {
+    const cluster = new FakeCluster('cluster1', [{backends: [new Backend()], locality: {region: 'region1'}}]);
+    const routeGroup = new FakeRouteGroup('route1', [{cluster: cluster}]);
+    routeGroup.startAllBackends().then(() => {
+      xdsServer.setEdsResource(cluster.getEndpointConfig());
+      xdsServer.setCdsResource(cluster.getClusterConfig());
+      xdsServer.setRdsResource(routeGroup.getRouteConfiguration());
+      xdsServer.setLdsResource(routeGroup.getListener());
+      client = new XdsTestClient('route1', xdsServer);
+      client.startCalls(100);
+      routeGroup.waitForAllBackendsToReceiveTraffic().then(() => {
+        // After backends receive calls, set invalid EDS resource
+        const invalidEdsResource = {cluster_name: cluster.getEndpointConfig().cluster_name, endpoints: [{}]};
+        xdsServer.setEdsResource(invalidEdsResource);
+        let seenNack = false;
+        xdsServer.addResponseListener((typeUrl, responseState) => {
+          if (responseState.state === 'NACKED') {
+            if (seenNack) {
+              return;
+            }
+            seenNack = true;
+            routeGroup.waitForAllBackendsToReceiveTraffic().then(() => {
+              client.stopCalls();
+              done();
+            });
+          }
+        });
+      }, reason => done(reason));
+    }, reason => done(reason));
+  });
+  it('Should continue to use a valid resource after receiving an invalid CDS update', done => {
+    const cluster = new FakeCluster('cluster1', [{backends: [new Backend()], locality: {region: 'region1'}}]);
+    const routeGroup = new FakeRouteGroup('route1', [{cluster: cluster}]);
+    routeGroup.startAllBackends().then(() => {
+      xdsServer.setEdsResource(cluster.getEndpointConfig());
+      xdsServer.setCdsResource(cluster.getClusterConfig());
+      xdsServer.setRdsResource(routeGroup.getRouteConfiguration());
+      xdsServer.setLdsResource(routeGroup.getListener());
+      client = new XdsTestClient('route1', xdsServer);
+      client.startCalls(100);
+      routeGroup.waitForAllBackendsToReceiveTraffic().then(() => {
+        // After backends receive calls, set invalid CDS resource
+        const invalidCdsResource = {name: cluster.getClusterConfig().name};
+        xdsServer.setCdsResource(invalidCdsResource);
+        let seenNack = false;
+        xdsServer.addResponseListener((typeUrl, responseState) => {
+          if (responseState.state === 'NACKED') {
+            if (seenNack) {
+              return;
+            }
+            seenNack = true;
+            routeGroup.waitForAllBackendsToReceiveTraffic().then(() => {
+              client.stopCalls();
+              done();
+            });
+          }
+        });
+      }, reason => done(reason));
+    }, reason => done(reason));
+  });
+  it('Should continue to use a valid resource after receiving an invalid RDS update', done => {
+    const cluster = new FakeCluster('cluster1', [{backends: [new Backend()], locality: {region: 'region1'}}]);
+    const routeGroup = new FakeRouteGroup('route1', [{cluster: cluster}]);
+    routeGroup.startAllBackends().then(() => {
+      xdsServer.setEdsResource(cluster.getEndpointConfig());
+      xdsServer.setCdsResource(cluster.getClusterConfig());
+      xdsServer.setRdsResource(routeGroup.getRouteConfiguration());
+      xdsServer.setLdsResource(routeGroup.getListener());
+      client = new XdsTestClient('route1', xdsServer);
+      client.startCalls(100);
+      routeGroup.waitForAllBackendsToReceiveTraffic().then(() => {
+        // After backends receive calls, set invalid RDS resource
+        const invalidRdsResource = {name: routeGroup.getRouteConfiguration().name, virtual_hosts: [{domains: ['**']}]};
+        xdsServer.setRdsResource(invalidRdsResource);
+        let seenNack = false;
+        xdsServer.addResponseListener((typeUrl, responseState) => {
+          if (responseState.state === 'NACKED') {
+            if (seenNack) {
+              return;
+            }
+            seenNack = true;
+            routeGroup.waitForAllBackendsToReceiveTraffic().then(() => {
+              client.stopCalls();
+              done();
+            });
+          }
+        });
+      }, reason => done(reason));
+    }, reason => done(reason));
+  });
+  it('Should continue to use a valid resource after receiving an invalid LDS update', done => {
+    const cluster = new FakeCluster('cluster1', [{backends: [new Backend()], locality: {region: 'region1'}}]);
+    const routeGroup = new FakeRouteGroup('route1', [{cluster: cluster}]);
+    routeGroup.startAllBackends().then(() => {
+      xdsServer.setEdsResource(cluster.getEndpointConfig());
+      xdsServer.setCdsResource(cluster.getClusterConfig());
+      xdsServer.setRdsResource(routeGroup.getRouteConfiguration());
+      xdsServer.setLdsResource(routeGroup.getListener());
+      client = new XdsTestClient('route1', xdsServer);
+      client.startCalls(100);
+      routeGroup.waitForAllBackendsToReceiveTraffic().then(() => {
+        // After backends receive calls, set invalid LDS resource
+        const invalidLdsResource = {name: routeGroup.getListener().name};
+        xdsServer.setLdsResource(invalidLdsResource);
+        let seenNack = false;
+        xdsServer.addResponseListener((typeUrl, responseState) => {
+          if (responseState.state === 'NACKED') {
+            if (seenNack) {
+              return;
+            }
+            seenNack = true;
+            routeGroup.waitForAllBackendsToReceiveTraffic().then(() => {
+              client.stopCalls();
+              done();
+            });
+          }
+        });
+      }, reason => done(reason));
+    }, reason => done(reason));
+  });
+});

packages/grpc-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js",
-  "version": "1.8.13",
+  "version": "1.8.14",
   "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/load-balancer-pick-first.ts

@@ -322,12 +322,12 @@ export class PickFirstLoadBalancer implements LoadBalancer {
       );
     }
     this.currentPick = subchannel;
-    this.updateState(ConnectivityState.READY, new PickFirstPicker(subchannel));
     subchannel.addConnectivityStateListener(this.pickedSubchannelStateListener);
     subchannel.ref();
     this.channelControlHelper.addChannelzChild(subchannel.getChannelzRef());
     this.resetSubchannelList();
     clearTimeout(this.connectionDelayTimeout);
+    this.updateState(ConnectivityState.READY, new PickFirstPicker(subchannel));
   }
 
   private updateState(newState: ConnectivityState, picker: Picker) {

packages/grpc-js/src/subchannel.ts

@@ -60,7 +60,7 @@ export class Subchannel {
    * state changes. Will be modified by `addConnectivityStateListener` and
    * `removeConnectivityStateListener`
    */
-  private stateListeners: ConnectivityStateListener[] = [];
+  private stateListeners: Set<ConnectivityStateListener> = new Set();
 
   private backoffTimeout: BackoffTimeout;
 
@@ -261,9 +261,7 @@ export class Subchannel {
       default:
         throw new Error(`Invalid state: unknown ConnectivityState ${newState}`);
     }
-    /* We use a shallow copy of the stateListeners array in case a listener
-     * is removed during this iteration */
-    for (const listener of [...this.stateListeners]) {
+    for (const listener of this.stateListeners) {
       listener(this, previousState, newState, this.keepaliveTime);
     }
     return true;
@@ -291,13 +289,15 @@ export class Subchannel {
       if (this.channelzEnabled) {
         this.channelzTrace.addTrace('CT_INFO', 'Shutting down');
       }
-      this.transitionToState(
-        [ConnectivityState.CONNECTING, ConnectivityState.READY],
-        ConnectivityState.IDLE
-      );
       if (this.channelzEnabled) {
         unregisterChannelzRef(this.channelzRef);
       }
+      process.nextTick(() => {
+        this.transitionToState(
+          [ConnectivityState.CONNECTING, ConnectivityState.READY],
+          ConnectivityState.IDLE
+        );
+      });
     }
   }
 
@@ -339,20 +339,22 @@ export class Subchannel {
    * Otherwise, do nothing.
    */
   startConnecting() {
-    /* First, try to transition from IDLE to connecting. If that doesn't happen
-     * because the state is not currently IDLE, check if it is
-     * TRANSIENT_FAILURE, and if so indicate that it should go back to
-     * connecting after the backoff timer ends. Otherwise do nothing */
-    if (
-      !this.transitionToState(
-        [ConnectivityState.IDLE],
-        ConnectivityState.CONNECTING
-      )
-    ) {
-      if (this.connectivityState === ConnectivityState.TRANSIENT_FAILURE) {
-        this.continueConnecting = true;
+    process.nextTick(() => {
+      /* First, try to transition from IDLE to connecting. If that doesn't happen
+      * because the state is not currently IDLE, check if it is
+      * TRANSIENT_FAILURE, and if so indicate that it should go back to
+      * connecting after the backoff timer ends. Otherwise do nothing */
+      if (
+        !this.transitionToState(
+          [ConnectivityState.IDLE],
+          ConnectivityState.CONNECTING
+        )
+      ) {
+        if (this.connectivityState === ConnectivityState.TRANSIENT_FAILURE) {
+          this.continueConnecting = true;
+        }
       }
-    }
+    });
   }
 
   /**
@@ -368,7 +370,7 @@ export class Subchannel {
    * @param listener
    */
   addConnectivityStateListener(listener: ConnectivityStateListener) {
-    this.stateListeners.push(listener);
+    this.stateListeners.add(listener);
   }
 
   /**
@@ -377,21 +379,20 @@ export class Subchannel {
    *     `addConnectivityStateListener`
    */
   removeConnectivityStateListener(listener: ConnectivityStateListener) {
-    const listenerIndex = this.stateListeners.indexOf(listener);
-    if (listenerIndex > -1) {
-      this.stateListeners.splice(listenerIndex, 1);
-    }
+    this.stateListeners.delete(listener);
   }
 
   /**
    * Reset the backoff timeout, and immediately start connecting if in backoff.
    */
   resetBackoff() {
-    this.backoffTimeout.reset();
-    this.transitionToState(
-      [ConnectivityState.TRANSIENT_FAILURE],
-      ConnectivityState.CONNECTING
-    );
+    process.nextTick(() => {
+      this.backoffTimeout.reset();
+      this.transitionToState(
+        [ConnectivityState.TRANSIENT_FAILURE],
+        ConnectivityState.CONNECTING
+      );
+    });
   }
 
   getAddress(): string {