grpc-js: round_robin: always have children reconnect immediately

Author: murgatroid99

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

@@ -15,7 +15,7 @@
  *
  */
 
-import { loadPackageDefinition, sendUnaryData, Server, ServerCredentials, ServerUnaryCall, UntypedServiceImplementation } from "@grpc/grpc-js";
+import { loadPackageDefinition, sendUnaryData, Server, ServerCredentials, ServerOptions, ServerUnaryCall, UntypedServiceImplementation } from "@grpc/grpc-js";
 import { loadSync } from "@grpc/proto-loader";
 import { ProtoGrpcType } from "./generated/echo";
 import { EchoRequest__Output } from "./generated/grpc/testing/EchoRequest";
@@ -43,7 +43,7 @@ export class Backend {
   private receivedCallCount = 0;
   private callListeners: (() => void)[] = [];
   private port: number | null = null;
-  constructor() {
+  constructor(private serverOptions?: ServerOptions) {
   }
   Echo(call: ServerUnaryCall<EchoRequest__Output, EchoResponse>, callback: sendUnaryData<EchoResponse>) {
     // call.request.params is currently ignored
@@ -76,13 +76,12 @@ export class Backend {
     if (this.server) {
       throw new Error("Backend already running");
     }
-    this.server = new Server();
+    this.server = new Server(this.serverOptions);
     this.server.addService(loadedProtos.grpc.testing.EchoTestService.service, this as unknown as UntypedServiceImplementation);
     const boundPort = this.port ?? 0;
     this.server.bindAsync(`localhost:${boundPort}`, ServerCredentials.createInsecure(), (error, port) => {
       if (!error) {
         this.port = port;
-        this.server!.start();
       }
       callback(error, port);
     })

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

@@ -91,4 +91,32 @@ describe('core xDS functionality', () => {
       });
     }, reason => done(reason));
   });
+  it('should handle connections aging out', function(done) {
+    this.timeout(5000);
+    const cluster = new FakeEdsCluster('cluster1', 'endpoint1', [{backends: [new Backend({'grpc.max_connection_age_ms': 1000})], locality:{region: 'region1'}}]);
+    const routeGroup = new FakeRouteGroup('listener1', 'route1', [{cluster: cluster}]);
+    routeGroup.startAllBackends().then(() => {
+      xdsServer.setEdsResource(cluster.getEndpointConfig());
+      xdsServer.setCdsResource(cluster.getClusterConfig());
+      xdsServer.setRdsResource(routeGroup.getRouteConfiguration());
+      xdsServer.setLdsResource(routeGroup.getListener());
+      xdsServer.addResponseListener((typeUrl, responseState) => {
+        if (responseState.state === 'NACKED') {
+          client.stopCalls();
+          assert.fail(`Client NACKED ${typeUrl} resource with message ${responseState.errorMessage}`);
+        }
+      })
+      client = XdsTestClient.createFromServer('listener1', xdsServer);
+      client.sendOneCall(error => {
+        assert.ifError(error);
+        // Make another call after the max_connection_age_ms expires
+        setTimeout(() => {
+          client.sendOneCall(error => {
+            done(error);
+          })
+        }, 1100);
+      });
+    }, reason => done(reason));
+
+  })
 });

packages/grpc-js/package.json

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

@@ -605,6 +605,10 @@ export class LeafLoadBalancer {
     return this.endpoint;
   }
 
+  exitIdle() {
+    this.pickFirstBalancer.exitIdle();
+  }
+
   destroy() {
     this.pickFirstBalancer.destroy();
   }

packages/grpc-js/src/load-balancer-round-robin.ts

@@ -161,6 +161,15 @@ export class RoundRobinLoadBalancer implements LoadBalancer {
     } else {
       this.updateState(ConnectivityState.IDLE, new QueuePicker(this));
     }
+    /* round_robin should keep all children connected, this is how we do that.
+     * We can't do this more efficiently in the individual child's updateState
+     * callback because that doesn't have a reference to which child the state
+     * change is associated with. */
+    for (const child of this.children) {
+      if (child.getConnectivityState() === ConnectivityState.IDLE) {
+        child.exitIdle();
+      }
+    }
   }
 
   private updateState(newState: ConnectivityState, picker: Picker) {