easwars review #2

Author: purnesh42H

xds/internal/clients/xdsclient/ads_stream.go

@@ -445,7 +445,9 @@ func (s *adsStreamImpl) sendMessageLocked(stream clients.Stream, names []string,
 		}
 	}
 
-	// Call the event handler to remove unsubscribed cache entries.
+	// Call the event handler to remove unsubscribed cache entries. It is to
+	// ensure the cache entries are deleted even if discovery request fails. In
+	// case of failure when the stream restarts, nonce is reset anyways.
 	s.eventHandler.onRequest(url)
 
 	msg, err := proto.Marshal(req)

xds/internal/clients/xdsclient/authority.go

@@ -293,18 +293,19 @@ func (a *authority) fallbackToServer(xc *xdsChannelWithConfig) bool {
 	// Subscribe to all existing resources from the new management server.
 	for typ, resources := range a.resources {
 		for name, state := range resources {
-			if len(state.watchers) > 0 {
-				if a.logger.V(2) {
-					a.logger.Infof("Resubscribing to resource of type %q and name %q", typ.TypeName, name)
-				}
-				xc.channel.subscribe(typ, name)
-
-				// Add the new channel to the list of xdsChannels from which this
-				// resource has been requested from. Retain the cached resource and
-				// the set of existing watchers (and other metadata fields) in the
-				// resource state.
-				state.xdsChannelConfigs[xc] = true
+			if len(state.watchers) == 0 {
+				continue
+			}
+			if a.logger.V(2) {
+				a.logger.Infof("Resubscribing to resource of type %q and name %q", typ.TypeName, name)
 			}
+			xc.channel.subscribe(typ, name)
+
+			// Add the new channel to the list of xdsChannels from which this
+			// resource has been requested from. Retain the cached resource and
+			// the set of existing watchers (and other metadata fields) in the
+			// resource state.
+			state.xdsChannelConfigs[xc] = true
 		}
 	}
 	return true

xds/internal/clients/xdsclient/channel.go

@@ -147,6 +147,8 @@ func (xc *xdsChannel) close() {
 
 	// Get the resource types that this specific ADS stream was handling
 	// before stopping it.
+	//
+	// TODO: Revisit if we can avoid acquiring the lock of ads (another type).
 	xc.ads.mu.Lock()
 	typesHandledByStream := make([]ResourceType, 0, len(xc.ads.resourceTypeState))
 	for typ := range xc.ads.resourceTypeState {