Repository: kubernetes/kubernetes
File: staging/src/k8s.io/client-go/tools/cache/reflector.go (Go)

Code before:
				// watch bookmarks, it will ignore this field).
				AllowWatchBookmarks: true,
			}

			w, err = r.listerWatcher.WatchWithContext(ctx, options)
			if err != nil {
				if canRetry := isWatchErrorRetriable(err); canRetry {
					logger.V(4).Info("Watch failed - backing off", "reflector", r.name, "type", r.typeDescription, "err", err)
					select {
					case <-stopCh:
						return nil
					case <-r.backoffManager.Backoff().C():
						continue
					}
				}
				return err
			}
		}

		err = handleWatch(ctx, start, w, r.store, r.expectedType, r.expectedGVK, r.name, r.typeDescription,
			func(rv string, eventReceivedBesidesAdded bool) {
				// We update the resource version in the store only if we have received at least one event that is
				// not an added event, or if the resource version has been set previously. This is because we can
				// encounter 2 scenarios:
				// 1. The watch is started from a resource version specified by the LastSyncResourceVersion field.
				//    In this case, we can update the resource version in the store without worrying about it being
				//    out of order since we will not receive any synthetic added events for resources that may be
				//    out of order.
				// 2. The watch is started when the LastSyncResourceVersion field is empty. In this case, we may not
				//    update the LastSyncResourceVersion until we receive at least one event that is not an added
				//    event, since that is the only way to ensure that the watch has exited the initial list phase.
				rvSynced := r.LastSyncResourceVersion() != ""
				if rvSynced || eventReceivedBesidesAdded {
					r.setLastSyncResourceVersion(rv)
					if rvu, ok := r.store.(ResourceVersionUpdater); ok {
						rvu.UpdateResourceVersion(rv)
					}
				}
			},
			r.clock, resyncerrc)
		// handleWatch always stops the watcher. So we don't need to here.
		// Just set it to nil to trigger a retry on the next loop.
		w = nil
		retry.After(err)
		if err != nil {
			if !errors.Is(err, errorStopRequested) {
				switch {
				case isExpiredError(err):
					// Don't set LastSyncResourceVersionUnavailable - LIST call with ResourceVersion=RV already
					// has a semantic that it returns data at least as fresh as provided RV.
					// So first try to LIST with setting RV to resource version of last observed object.

Review comment (refactor):
tracing this back, the only call path that passes a non-nil watcher to `watch` is the watchList path that already received a bookmark event

if that's the case, I think it's ok to document a non-nil watcher passed here must already be initialized and past any initial synthetic non-rv-order ADDED events

that lets us simplify this to a boolean we track outside the rv-setter closure

```go
// if w is already initialized, it must be past any synthetic non-rv-ordered added events
propagateRVFromStart := true
if w == nil {
			options := metav1.ListOptions{
				ResourceVersion: r.LastSyncResourceVersion(),
				// We want to avoid situations of hanging watchers. Stop any watchers that do not
				// receive any events within the timeout window.
				TimeoutSeconds: &timeoutSeconds,
				// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
				// Reflector doesn't assume bookmarks are returned at all (if the server do not support
				// watch bookmarks, it will ignore this field).
				AllowWatchBookmarks: true,
			}
			if options.ResourceVersion == "" || options.ResourceVersion == "0" {
				// if we're starting the watch at a resource version that will get synthetic ADDED events in non-rv order,
				// wait until we're through that set of events before propagating the RV
				propagateRVFromStart = false
			}
...
			func(rv string, eventReceivedBesidesAdded bool) {
				// We update the resource version in the store only if we have received at least one event that is
				// not an added event, or if the resource version has been set previously. This is because we can
				// encounter 2 scenarios:
				// 1. The watch is started from a resource version specified by the LastSyncResourceVersion field.
				//    In this case, we can update the resource version in the store without worrying about it being
				//    out of order since we will not receive any synthetic added events for resources that may be
				//    out of order.
				// 2. The watch is started when the LastSyncResourceVersion field is empty. In this case, we may not
				//    update the LastSyncResourceVersion until we receive at least one event that is not an added
				//    event, since that is the only way to ensure that the watch has exited the initial list phase.
				if propagateRVFromStart || eventReceivedBesidesAdded {
```

Code after:
				// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
				// Reflector doesn't assume bookmarks are returned at all (if the server do not support
				// watch bookmarks, it will ignore this field).
				AllowWatchBookmarks: true,
			}
			if options.ResourceVersion == "" || options.ResourceVersion == "0" {
				// if we're starting the watch at a resource version that will get synthetic ADDED events in non-rv order,
				// wait until we're through that set of events before propagating the RV
				propagateRVFromStart = false
			}

			w, err = r.listerWatcher.WatchWithContext(ctx, options)
			if err != nil {
				if canRetry := isWatchErrorRetriable(err); canRetry {
					logger.V(4).Info("Watch failed - backing off", "reflector", r.name, "type", r.typeDescription, "err", err)
					select {
					case <-stopCh:
						return nil
					case <-r.backoffManager.Backoff().C():
						continue
					}
				}
				return err
			}
		}

		err = handleWatch(ctx, start, w, r.store, r.expectedType, r.expectedGVK, r.name, r.typeDescription,
			func(rv string, eventReceivedBesidesAdded bool) {
				// We update the resource version in the store only if we have received at least one event that is
				// not an added event, or if the resource version has been set previously. This is because we can
				// encounter 2 scenarios:
				// 1. The watch is started from a resource version specified by the LastSyncResourceVersion field.
				//    In this case, we can update the resource version in the store without worrying about it being
				//    out of order since we will not receive any synthetic added events for resources that may be
				//    out of order.
				// 2. The watch is started when the LastSyncResourceVersion field is empty. In this case, we may not
				//    update the LastSyncResourceVersion until we receive at least one event that is not an added
				//    event, since that is the only way to ensure that the watch has exited the initial list phase.
				if propagateRVFromStart || eventReceivedBesidesAdded {
					r.setLastSyncResourceVersion(rv)
					if rvu, ok := r.store.(ResourceVersionUpdater); ok {
						rvu.UpdateResourceVersion(rv)
					}
				}
			},
			r.clock, resyncerrc)
		// handleWatch always stops the watcher. So we don't need to here.
		// Just set it to nil to trigger a retry on the next loop.
		w = nil
		retry.After(err)
		if err != nil {