apiserver/pkg/storage/cacher/lister_watcher: pass RV for request from the watchlist consistency checker

staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher.go

@@ -31,26 +31,30 @@ import (
 	"k8s.io/apiserver/pkg/storage"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/consistencydetector"
 	"k8s.io/client-go/util/watchlist"
 )
 
 // listerWatcher opaques storage.Interface to expose cache.ListerWatcher.
 type listerWatcher struct {
-	storage                       storage.Interface
-	resourcePrefix                string
-	newListFunc                   func() runtime.Object
-	contextMetadata               metadata.MD
-	unsupportedWatchListSemantics bool
+	storage         storage.Interface
+	resourcePrefix  string
+	newListFunc     func() runtime.Object
+	contextMetadata metadata.MD
+
+	unsupportedWatchListSemantics    bool
+	watchListConsistencyCheckEnabled bool
 }
 
 // NewListerWatcher returns a storage.Interface backed ListerWatcher.
 func NewListerWatcher(storage storage.Interface, resourcePrefix string, newListFunc func() runtime.Object, contextMetadata metadata.MD) cache.ListerWatcher {
 	return &listerWatcher{
-		storage:                       storage,
-		resourcePrefix:                resourcePrefix,
-		newListFunc:                   newListFunc,
-		contextMetadata:               contextMetadata,
-		unsupportedWatchListSemantics: watchlist.DoesClientNotSupportWatchListSemantics(storage),
+		storage:                          storage,
+		resourcePrefix:                   resourcePrefix,
+		newListFunc:                      newListFunc,
+		contextMetadata:                  contextMetadata,
+		unsupportedWatchListSemantics:    watchlist.DoesClientNotSupportWatchListSemantics(storage),
+		watchListConsistencyCheckEnabled: consistencydetector.IsDataConsistencyDetectionForWatchListEnabled(),
 	}
 }
 
@@ -69,6 +73,27 @@ func (lw *listerWatcher) List(options metav1.ListOptions) (runtime.Object, error
 		Predicate:            pred,
 		Recursive:            true,
 	}
+
+	// The ConsistencyChecker built into reflectors for the WatchList feature is responsible
+	// for verifying that the data received from the server (potentially from the watch cache)
+	// is consistent with the data stored in etcd.
+	//
+	// To perform this verification, the checker uses the ResourceVersion obtained from the initial request
+	// and sets the ResourceVersionMatch so that it retrieves exactly the same data directly from etcd.
+	// This allows comparing both data sources and confirming their consistency.
+	//
+	// The code below checks whether the incoming request originates from the ConsistencyChecker.
+	// If so, it allows explicitly setting the ResourceVersion.
+	//
+	// As of Oct 2025, reflector on its own is not setting RVM=Exact.
+	// However, even if that changes in the meantime, we would have to propagate that
+	// down to storage to ensure the correct semantics of the request.
+	watchListEnabled := utilfeature.DefaultFeatureGate.Enabled(features.WatchList)
+	supportedRVM := options.ResourceVersionMatch == metav1.ResourceVersionMatchExact
+	if watchListEnabled && lw.watchListConsistencyCheckEnabled && supportedRVM {
+		storageOpts.ResourceVersion = options.ResourceVersion
+	}
+
 	ctx := context.Background()
 	if lw.contextMetadata != nil {
 		ctx = metadata.NewOutgoingContext(ctx, lw.contextMetadata)

staging/src/k8s.io/apiserver/pkg/storage/cacher/lister_watcher_test.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/apis/example"
 	"k8s.io/apiserver/pkg/features"
+	"k8s.io/apiserver/pkg/storage"
 	storagetesting "k8s.io/apiserver/pkg/storage/testing"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/tools/cache"
@@ -218,3 +219,79 @@ func TestCacherListerWatcherWhenListWatchDisabled(t *testing.T) {
 		t.Fatalf("Expected error %q, but got %q", expectedErrMsg, err.Error())
 	}
 }
+
+func TestListerWatcherListResourceVersionPropagation(t *testing.T) {
+	scenarios := []struct {
+		name                             string
+		options                          metav1.ListOptions
+		watchListEnabled                 bool
+		watchListConsistencyCheckEnabled bool
+		expectedStorageResourceVer       string
+	}{
+		{
+			name: "WatchList FG disabled - RV not propagated",
+			options: metav1.ListOptions{
+				ResourceVersion:      "123",
+				ResourceVersionMatch: metav1.ResourceVersionMatchExact,
+			},
+			watchListEnabled:                 false,
+			watchListConsistencyCheckEnabled: true,
+			expectedStorageResourceVer:       "",
+		},
+		{
+			name: "WatchList consistency check disabled - RV not propagated",
+			options: metav1.ListOptions{
+				ResourceVersion:      "123",
+				ResourceVersionMatch: metav1.ResourceVersionMatchExact,
+			},
+			watchListEnabled:                 true,
+			watchListConsistencyCheckEnabled: false,
+			expectedStorageResourceVer:       "",
+		},
+		{
+			name: "Unsupported RVM - RV not propagated",
+			options: metav1.ListOptions{
+				ResourceVersion:      "123",
+				ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan,
+			},
+			watchListEnabled:                 true,
+			watchListConsistencyCheckEnabled: true,
+			expectedStorageResourceVer:       "",
+		},
+		{
+			name: "all conditions satisfied - RV propagated",
+			options: metav1.ListOptions{
+				ResourceVersion:      "123",
+				ResourceVersionMatch: metav1.ResourceVersionMatchExact,
+			},
+			watchListEnabled:                 true,
+			watchListConsistencyCheckEnabled: true,
+			expectedStorageResourceVer:       "123",
+		},
+	}
+
+	for _, scenario := range scenarios {
+		t.Run(scenario.name, func(t *testing.T) {
+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchList, scenario.watchListEnabled)
+
+			backingStorage := &dummyStorage{}
+			var capturedOpts storage.ListOptions
+			backingStorage.getListFn = func(ctx context.Context, resPrefix string, opts storage.ListOptions, listObj runtime.Object) error {
+				capturedOpts = opts
+				return nil
+			}
+
+			targetInterface := NewListerWatcher(backingStorage, "/pods/", func() runtime.Object { return &example.PodList{} }, nil)
+			target := targetInterface.(*listerWatcher)
+			target.watchListConsistencyCheckEnabled = scenario.watchListConsistencyCheckEnabled
+
+			if _, err := target.List(scenario.options); err != nil {
+				t.Fatalf("List returned error: %v", err)
+			}
+
+			if capturedOpts.ResourceVersion != scenario.expectedStorageResourceVer {
+				t.Fatalf("expected storage ResourceVersion %q, got %q", scenario.expectedStorageResourceVer, capturedOpts.ResourceVersion)
+			}
+		})
+	}
+}