Merge branch 'master' into grpc-js-xds_interop_server

Author: murgatroid99

Diff for: doc/environment_variables.md

@@ -11,8 +11,10 @@ can be set.
   checked in order, and the first one that has a value is used.
 
 * no_grpc_proxy, no_proxy
-  A comma separated list of hostnames to connect to without using a proxy even
-  if a proxy is set. These variables are checked in order, and the first one
+  A comma separated list of hostnames, IP addresses,
+  or CIDR blocks to connect to without using a proxy even
+  if a proxy is set, for example: no_proxy=example.com,192.168.0.1,192.168.0.0/16.
+  These variables are checked in order, and the first one
   that has a value is used.
 
 * GRPC_SSL_CIPHER_SUITES
@@ -66,4 +68,4 @@ can be set.
 * GRPC_NODE_USE_ALTERNATIVE_RESOLVER
   Allows changing dns resolve behavior and parse DNS server authority as described in https://github.com/grpc/grpc/blob/master/doc/naming.md
   - true - use alternative resolver
-  - false - use default resolver (default)
+  - false - use default resolver (default)

Diff for: examples/retry/README.md

@@ -0,0 +1,57 @@
+# Retry
+
+This example shows how to enable and configure retry on gRPC clients.
+
+## Documentation
+
+[gRFC for client-side retry support](https://github.com/grpc/proposal/blob/master/A6-client-retries.md)
+
+## Try it
+
+This example includes a service implementation that fails requests three times with status code Unavailable, then passes the fourth. The client is configured to make four retry attempts when receiving an Unavailable status code.
+
+First start the server:
+
+```
+node server.js
+```
+
+Then run the client:
+
+```
+node client.js
+```
+
+## Usage
+
+### Define your retry policy
+
+Retry is configured via the service config, which can be provided by the name resolver, or as a channel option (described below). In the below example, we set the retry policy for the "grpc.example.echo.Echo" method.
+
+```js
+const serviceConfig = {
+  loadBalancingConfig: [],
+  methodConfig: [
+    {
+      name: [
+        {
+          service: 'grpc.examples.echo.Echo',
+        },
+      ],
+      retryPolicy: {
+        maxAttempts: 4,
+        initialBackoff: '0.01s',
+        maxBackoff: '0.01s',
+        backoffMultiplier: 1.0,
+        retryableStatusCodes: ['UNAVAILABLE'],
+      },
+    },
+  ],
+};
+```
+
+### Providing the retry policy as a channel option
+
+```js
+const client = new Echo('localhost:50052', grpc.credentials.createInsecure(), { 'grpc.service_config': JSON.stringify(serviceConfig) });
+```

Diff for: examples/retry/client.js

@@ -0,0 +1,75 @@
+/*
+ *
+ * Copyright 2025 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.
+ *
+ */
+
+const grpc = require('@grpc/grpc-js');
+const protoLoader = require('@grpc/proto-loader');
+const parseArgs = require('minimist');
+
+const PROTO_PATH = __dirname + '/../protos/echo.proto';
+
+const packageDefinition = protoLoader.loadSync(
+  PROTO_PATH,
+  {keepCase: true,
+   longs: String,
+   enums: String,
+   defaults: true,
+   oneofs: true
+  });
+const echoProto = grpc.loadPackageDefinition(packageDefinition).grpc.examples.echo;
+
+const serviceConfig = {
+  loadBalancingConfig: [],
+  methodConfig: [
+    {
+      name: [
+        {
+          service: 'grpc.examples.echo.Echo',
+        },
+      ],
+      retryPolicy: {
+        maxAttempts: 4,
+        initialBackoff: '0.01s',
+        maxBackoff: '0.01s',
+        backoffMultiplier: 1.0,
+        retryableStatusCodes: ['UNAVAILABLE'],
+      },
+    },
+  ],
+};
+
+function main() {
+  let argv = parseArgs(process.argv.slice(2), {
+    string: 'target',
+    default: {target: 'localhost:50052'}
+  });
+
+	// Set up a connection to the server with service config and create the channel.
+	// However, the recommended approach is to fetch the retry configuration
+	// (which is part of the service config) from the name resolver rather than
+	// defining it on the client side.
+  const client = new echoProto.Echo('localhost:50052', grpc.credentials.createInsecure(), { 'grpc.service_config': JSON.stringify(serviceConfig) });
+  client.unaryEcho({message: 'Try and Success'}, (error, value) => {
+    if (error) {
+      console.log(`Unexpected error from UnaryEcho: ${error}`);
+      return;
+    }
+    console.log(`RPC response: ${JSON.stringify(value)}`);
+  });
+}
+
+main();

Diff for: examples/retry/server.js

@@ -0,0 +1,73 @@
+/*
+ *
+ * Copyright 2025 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.
+ *
+ */
+
+const grpc = require('@grpc/grpc-js');
+const protoLoader = require('@grpc/proto-loader');
+const parseArgs = require('minimist');
+
+const PROTO_PATH = __dirname + '/../protos/echo.proto';
+
+const packageDefinition = protoLoader.loadSync(
+  PROTO_PATH,
+  {keepCase: true,
+   longs: String,
+   enums: String,
+   defaults: true,
+   oneofs: true
+  });
+const echoProto = grpc.loadPackageDefinition(packageDefinition).grpc.examples.echo;
+
+const SUCCEED_EVERY = 4
+let callCount = 0;
+
+/* This method will succeed every SUCCEED_EVERY calls, and fail all others with status code
+ * UNAVAILABLE. */
+function unaryEcho(call, callback) {
+  callCount++;
+  if (callCount % SUCCEED_EVERY === 0) {
+    console.log(`Request succeeded count: ${callCount}`);
+    callback(null, call.request);
+  } else {
+    console.log(`Request failed count: ${callCount}`);
+    callback({
+      code: grpc.status.UNAVAILABLE,
+      details: 'Request failed by policy'
+    });
+  }
+}
+
+const serviceImplementation = {
+  unaryEcho
+};
+
+function main() {
+  const argv = parseArgs(process.argv.slice(2), {
+    string: 'port',
+    default: {port: '50052'}
+  });
+  const server = new grpc.Server();
+  server.addService(echoProto.Echo.service, serviceImplementation);
+  server.bindAsync(`0.0.0.0:${argv.port}`, grpc.ServerCredentials.createInsecure(), (err, port) => {
+    if (err != null) {
+      return console.error(err);
+    }
+    console.log(`gRPC listening on ${port}`)
+  });
+}
+
+main();

Diff for: packages/grpc-js-xds/src/resolver-xds.ts

@@ -83,6 +83,30 @@ const RETRY_CODES: {[key: string]: status} = {
 export const XDS_CONFIG_KEY = `${experimental.SUBCHANNEL_ARGS_EXCLUDE_KEY_PREFIX}.xds_config`;
 export const XDS_CLIENT_KEY = 'grpc.internal.xds_client';
 
+/**
+ * Tracks a dynamic subscription to a cluster that is currently or previously
+ * referenced in a RouteConfiguration.
+ */
+class ClusterRef {
+  private refCount = 0;
+  constructor(private unsubscribe: () => void) {}
+
+  ref() {
+    this.refCount += 1;
+  }
+
+  unref() {
+    this.refCount -= 1;
+    if (this.refCount <= 0) {
+      this.unsubscribe();
+    }
+  }
+
+  hasRef() {
+    return this.refCount > 0;
+  }
+}
+
 class XdsResolver implements Resolver {
 
   private listenerResourceName: string | null = null;
@@ -93,6 +117,7 @@ class XdsResolver implements Resolver {
 
   private xdsConfigWatcher: XdsConfigWatcher;
   private xdsDependencyManager: XdsDependencyManager | null = null;
+  private clusterRefs: Map<string, ClusterRef> = new Map();
 
   constructor(
     private target: GrpcUri,
@@ -123,11 +148,20 @@ class XdsResolver implements Resolver {
     }
   }
 
+  private pruneUnusedClusters() {
+    for (const [cluster, clusterRef] of this.clusterRefs) {
+      if (!clusterRef.hasRef()) {
+        this.clusterRefs.delete(cluster);
+      }
+    }
+  }
+
   private async handleXdsConfig(xdsConfig: XdsConfig) {
     /* We need to load the xxhash API before this function finishes, because
      * it is invoked in the config selector, which can be called immediately
      * after this function returns. */
     await loadXxhashApi();
+    this.pruneUnusedClusters();
     const httpConnectionManager = decodeSingleResource(HTTP_CONNECTION_MANGER_TYPE_URL, xdsConfig.listener.api_listener!.api_listener!.value);
     const configDefaultTimeout = httpConnectionManager.common_http_protocol_options?.idle_timeout;
     let defaultTimeout: Duration | undefined = undefined;
@@ -312,44 +346,60 @@ class XdsResolver implements Resolver {
       const routeMatcher = getPredicateForMatcher(route.match!);
       matchList.push({matcher: routeMatcher, action: routeAction});
     }
-    const configSelector: ConfigSelector = (methodName, metadata, channelId) => {
-      for (const {matcher, action} of matchList) {
-        if (matcher.apply(methodName, metadata)) {
-          const clusterResult = action.getCluster();
-          const unrefCluster = this.xdsDependencyManager!.addClusterSubscription(clusterResult.name);
-          const onCommitted = () => {
-            unrefCluster();
-          }
-          let hash: string;
-          if (EXPERIMENTAL_RING_HASH) {
-            hash = `${action.getHash(metadata, channelId)}`;
-          } else {
-            hash = '';
+    for (const cluster of allConfigClusters) {
+      let clusterRef = this.clusterRefs.get(cluster);
+      if (!clusterRef) {
+        clusterRef = new ClusterRef(this.xdsDependencyManager!.addClusterSubscription(cluster));
+        this.clusterRefs.set(cluster, clusterRef);
+      }
+      clusterRef.ref();
+    }
+    const configSelector: ConfigSelector = {
+      invoke: (methodName, metadata, channelId) => {
+        for (const {matcher, action} of matchList) {
+          if (matcher.apply(methodName, metadata)) {
+            const clusterResult = action.getCluster();
+            const clusterRef = this.clusterRefs.get(clusterResult.name)!;
+            clusterRef.ref();
+            const onCommitted = () => {
+              clusterRef.unref();
+            }
+            let hash: string;
+            if (EXPERIMENTAL_RING_HASH) {
+              hash = `${action.getHash(metadata, channelId)}`;
+            } else {
+              hash = '';
+            }
+            return {
+              methodConfig: clusterResult.methodConfig,
+              onCommitted: onCommitted,
+              pickInformation: {cluster: clusterResult.name, hash: hash},
+              status: status.OK,
+              dynamicFilterFactories: clusterResult.dynamicFilterFactories
+            };
           }
-          return {
-            methodConfig: clusterResult.methodConfig,
-            onCommitted: onCommitted,
-            pickInformation: {cluster: clusterResult.name, hash: hash},
-            status: status.OK,
-            dynamicFilterFactories: clusterResult.dynamicFilterFactories
-          };
+        }
+        return {
+          methodConfig: {name: []},
+          // These fields won't be used here, but they're set because of some TypeScript weirdness
+          pickInformation: {cluster: '', hash: ''},
+          status: status.UNAVAILABLE,
+          dynamicFilterFactories: []
+        };
+      },
+      unref: () => {
+        for (const cluster of allConfigClusters) {
+          this.clusterRefs.get(cluster)?.unref();
         }
       }
-      return {
-        methodConfig: {name: []},
-        // These fields won't be used here, but they're set because of some TypeScript weirdness
-        pickInformation: {cluster: '', hash: ''},
-        status: status.UNAVAILABLE,
-        dynamicFilterFactories: []
-      };
-    };
+    }
     trace('Created ConfigSelector with configuration:');
     for (const {matcher, action} of matchList) {
       trace(matcher.toString());
       trace('=> ' + action.toString());
     }
     const clusterConfigMap: {[key: string]: {child_policy: LoadBalancingConfig[]}} = {};
-    for (const clusterName of allConfigClusters) {
+    for (const clusterName of this.clusterRefs.keys()) {
       clusterConfigMap[clusterName] = {child_policy: [{cds: {cluster: clusterName}}]};
     }
     const lbPolicyConfig = {xds_cluster_manager: {children: clusterConfigMap}};