diff --git a/packages/grpc-js-xds/src/xds-client.ts b/packages/grpc-js-xds/src/xds-client.ts index 785c11943..e3c8947af 100644 --- a/packages/grpc-js-xds/src/xds-client.ts +++ b/packages/grpc-js-xds/src/xds-client.ts @@ -150,6 +150,7 @@ class ResourceTimer { if (!resourceState) { return; } + resourceState.cachedResource = null; resourceState.meta.clientStatus = 'DOES_NOT_EXIST'; for (const watcher of resourceState.watchers) { watcher.onResourceDoesNotExist(); @@ -372,6 +373,7 @@ class AdsCallState { experimental.log(logVerbosity.ERROR, 'Ignoring nonexistent resource ' + xdsResourceNameToString({authority, key}, result.type!.getTypeUrl())); resourceState.deletionIgnored = true; } else { + resourceState.cachedResource = null; resourceState.meta.clientStatus = 'DOES_NOT_EXIST'; process.nextTick(() => { for (const watcher of resourceState.watchers) { @@ -404,6 +406,13 @@ class AdsCallState { this.trace( 'ADS stream ended. code=' + streamStatus.code + ' details= ' + streamStatus.details ); + for (const typeState of this.typeStates.values()) { + for (const authorityMap of typeState.subscribedResources.values()) { + for (const timer of authorityMap.values()) { + timer.maybeCancelTimer(); + } + } + } if (streamStatus.code !== status.OK && !this.receivedAnyResponse) { for (const watcher of this.allWatchers()) { watcher.onError(streamStatus); @@ -458,6 +467,7 @@ class AdsCallState { if (!authorityMap) { return; } + authorityMap.get(name.key)?.maybeCancelTimer(); authorityMap.delete(name.key); if (authorityMap.size === 0) { typeState.subscribedResources.delete(name.authority); @@ -937,6 +947,9 @@ class XdsSingleServerClient { const metadata = new Metadata({waitForReady: true}); const call = this.adsClient.StreamAggregatedResources(metadata); this.adsCallState = new AdsCallState(this, call, this.xdsClient.adsNode!); + if (this.adsClient.getChannel().getConnectivityState(false) === connectivityState.READY) { + this.adsCallState.markStreamStarted(); + } this.adsBackoff.runOnce(); } diff --git a/packages/grpc-js-xds/test/test-core.ts b/packages/grpc-js-xds/test/test-core.ts index 3eef81dd9..523a28327 100644 --- a/packages/grpc-js-xds/test/test-core.ts +++ b/packages/grpc-js-xds/test/test-core.ts @@ -197,5 +197,47 @@ describe('core xDS functionality', () => { xdsServer.setRdsResource(routeGroup2.getRouteConfiguration()); await cluster2.waitForAllBackendsToReceiveTraffic(); client.stopCalls(); - }) + }); + it('should recover when a deleted LDS resource is restored with identical content', async () => { + const [backend] = await createBackends(1); + const serverRoute = new FakeServerRoute(backend.getPort(), 'serverRoute'); + xdsServer.setRdsResource(serverRoute.getRouteConfiguration()); + xdsServer.setLdsResource(serverRoute.getListener()); + xdsServer.addResponseListener((typeUrl, responseState) => { + if (responseState.state === 'NACKED') { + client?.stopCalls(); + assert.fail(`Client NACKED ${typeUrl} resource with message ${responseState.errorMessage}`); + } + }); + const cluster = new FakeEdsCluster('cluster1', 'endpoint1', [{backends: [backend], locality: {region: 'region1'}}]); + const routeGroup = new FakeRouteGroup('listener1', 'route1', [{cluster: cluster}]); + await routeGroup.startAllBackends(xdsServer); + xdsServer.setEdsResource(cluster.getEndpointConfig()); + xdsServer.setCdsResource(cluster.getClusterConfig()); + xdsServer.setRdsResource(routeGroup.getRouteConfiguration()); + xdsServer.setLdsResource(routeGroup.getListener()); + client = XdsTestClient.createFromServer('listener1', xdsServer); + client.startCalls(100); + await routeGroup.waitForAllBackendsToReceiveTraffic(); + client.stopCalls(); + + // Trigger transient LDS deletion (control plane flap) + xdsServer.unsetLdsResource('listener1'); + + // Verify call fails while resource is deleted + await new Promise((resolve) => { + client.sendOneCall((error) => { + assert(error, 'Expected RPC to fail after LDS deletion'); + resolve(); + }); + }); + + // Restore identical LDS resource + xdsServer.setLdsResource(routeGroup.getListener()); + + // Verify client recovers and traffic flows normally + client.startCalls(100); + await routeGroup.waitForAllBackendsToReceiveTraffic(); + client.stopCalls(); + }); });