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: []*xdsChannelWithConfig{xdsChannel},
			}
			resources[resourceName] = state
			xdsChannel.xc.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.
		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() {}) })
		}
		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", 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
			// there when the watch was registered.
			resources := a.resources[rType]
			state := resources[resourceName]

			// Delete this particular watcher from the list of watchers, so that its
			// callback will not be invoked in the future.

Review comment (bug):
nit: Is the reason for the failure always that the authority is closed? If so, I think mentioning this would make the log message more helpful.

Could this late watch request indicate a synchronization issue in which the authority is closed before the users of the authority?

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: []*xdsChannelWithConfig{xdsChannel},
			}
			resources[resourceName] = state
			xdsChannel.xc.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.
		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() {}) })
		}
		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
			// there when the watch was registered.
			resources := a.resources[rType]
			state := resources[resourceName]

			// Delete this particular watcher from the list of watchers, so that its
			// callback will not be invoked in the future.