Repository: grpc/grpc-go
File: xds/internal/xdsclient/authority.go (Go)

Code before:
			if a.logger.V(2) {
				a.logger.Infof("First watch for type %q, resource name %q", rType.TypeName(), resourceName)
			}
			state = &resourceState{
				watchers:          make(map[xdsresource.ResourceWatcher]bool),
				md:                xdsresource.UpdateMetadata{Status: xdsresource.ServiceStatusRequested},
				xdsChannelConfigs: map[*xdsChannelWithConfig]bool{xdsChannel: true},
			}
			resources[resourceName] = state
			xdsChannel.channel.subscribe(rType, resourceName)
		}
		// Always add the new watcher to the set of watchers.
		state.watchers[watcher] = true

		// If we have a cached copy of the resource, notify the new watcher
		// immediately.
		if state.cache != nil {
			if a.logger.V(2) {
				a.logger.Infof("Resource type %q with resource name %q found in cache: %s", rType.TypeName(), resourceName, state.cache.ToJSON())
			}
			resource := state.cache
			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnUpdate(resource, func() {}) })
		}
		// If last update was NACK'd, notify the new watcher of error
		// immediately as well.
		if state.md.Status == xdsresource.ServiceStatusNACKed && state.md.ErrState != nil {
			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnError(state.md.ErrState.Err, func() {}) })
		}
		cleanup = a.unwatchResource(rType, resourceName, watcher)
	}, func() {
		if a.logger.V(2) {
			a.logger.Infof("Failed to schedule a watch for type %q, resource name %q, because the xDS client is closed", rType.TypeName(), resourceName)
		}
		close(done)
	})
	<-done
	return cleanup
}

func (a *authority) unwatchResource(rType xdsresource.Type, resourceName string, watcher xdsresource.ResourceWatcher) func() {
	return grpcsync.OnceFunc(func() {
		done := make(chan struct{})
		a.xdsClientSerializer.ScheduleOr(func(context.Context) {
			defer close(done)

			if a.logger.V(2) {
				a.logger.Infof("Canceling a watch for type %q, resource name %q", rType.TypeName(), resourceName)
			}

			// Lookup the resource type from the resource cache. The entry is
			// guaranteed to be present, since *we* were the ones who added it in

Review comment (suggestion):
Is it not enough if we check for `state.md.Status == xdsresource.ServiceStatusNACKed` alone? There is a single place where the status is set to `xdsresource.ServiceStatusNACKed`, and in there, we always set the error field as well.

Code after:
			if a.logger.V(2) {
				a.logger.Infof("First watch for type %q, resource name %q", rType.TypeName(), resourceName)
			}
			state = &resourceState{
				watchers:          make(map[xdsresource.ResourceWatcher]bool),
				md:                xdsresource.UpdateMetadata{Status: xdsresource.ServiceStatusRequested},
				xdsChannelConfigs: map[*xdsChannelWithConfig]bool{xdsChannel: true},
			}
			resources[resourceName] = state
			xdsChannel.channel.subscribe(rType, resourceName)
		}
		// Always add the new watcher to the set of watchers.
		state.watchers[watcher] = true

		// If we have a cached copy of the resource, notify the new watcher
		// immediately.
		if state.cache != nil {
			if a.logger.V(2) {
				a.logger.Infof("Resource type %q with resource name %q found in cache: %s", rType.TypeName(), resourceName, state.cache.ToJSON())
			}
			resource := state.cache
			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnUpdate(resource, func() {}) })
		}
		// If last update was NACK'd, notify the new watcher of error
		// immediately as well.
		if state.md.Status == xdsresource.ServiceStatusNACKed {
			if a.logger.V(2) {
				a.logger.Infof("Resource type %q with resource name %q was NACKed: %s", rType.TypeName(), resourceName, state.cache.ToJSON())
			}
			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnError(state.md.ErrState.Err, func() {}) })
		}
		// If the metadata field is updated to indicate that the management
		// server does not have this resource, notify the new watcher.
		if state.md.Status == xdsresource.ServiceStatusNotExist {
			a.watcherCallbackSerializer.TrySchedule(func(context.Context) { watcher.OnResourceDoesNotExist(func() {}) })
		}
		cleanup = a.unwatchResource(rType, resourceName, watcher)
	}, func() {
		if a.logger.V(2) {
			a.logger.Infof("Failed to schedule a watch for type %q, resource name %q, because the xDS client is closed", rType.TypeName(), resourceName)
		}
		close(done)
	})
	<-done
	return cleanup
}

func (a *authority) unwatchResource(rType xdsresource.Type, resourceName string, watcher xdsresource.ResourceWatcher) func() {
	return grpcsync.OnceFunc(func() {
		done := make(chan struct{})
		a.xdsClientSerializer.ScheduleOr(func(context.Context) {