Watchcache background resync

@@ -2294,6 +2294,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate 		{Version: version.MustParse("1.37"), Default: true, PreRelease: featuregate.GA, LockToDefault: true}, 	}, +	genericfeatures.WatchCacheStallResume: {+		{Version: version.MustParse("1.38"), Default: false, PreRelease: featuregate.Alpha},+	},+ 	genericfeatures.WatchFromStorageWithoutPrevKV: { 		{Version: version.MustParse("1.38"), Default: true, PreRelease: featuregate.Beta}, 	},@@ -2775,6 +2779,8 @@ var defaultKubernetesFeatureGateDependencies = map[featuregate.Feature][]feature  	genericfeatures.WatchCacheInitializationPostStartHook: {}, +	genericfeatures.WatchCacheStallResume: {},+ 	genericfeatures.WatchFromStorageWithoutPrevKV: {},  	genericfeatures.WatchList: {},

@@ -331,6 +331,12 @@ const ( 	// Enables post-start-hook for storage readiness 	WatchCacheInitializationPostStartHook featuregate.Feature = "WatchCacheInitializationPostStartHook" +	// owner: @serathius+	// kep: https://kep.k8s.io/6268+	//+	// Enables watch cache to pause and resume stalled watchers instead of closing them.+	WatchCacheStallResume featuregate.Feature = "WatchCacheStallResume"+ 	// owner: @shadowofs 	// 	// Enables watch cache ingestion without fetching previous values from etcd.@@ -552,6 +558,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate 		{Version: version.MustParse("1.37"), Default: true, PreRelease: featuregate.GA, LockToDefault: true}, 	}, +	WatchCacheStallResume: {+		{Version: version.MustParse("1.38"), Default: false, PreRelease: featuregate.Alpha},+	},+ 	WatchFromStorageWithoutPrevKV: { 		{Version: version.MustParse("1.38"), Default: true, PreRelease: featuregate.Beta}, 	},

@@ -77,6 +77,10 @@ const ( 	// defaultBookmarkFrequency defines how frequently watch bookmarks should be send 	// in addition to sending a bookmark right before watch deadline. 	defaultBookmarkFrequency = time.Minute++	// watcherResyncPeriod defines how frequently unsynchronized watchers are resynced+	// with the watch cache.+	watcherResyncPeriod = 100 * time.Millisecond )  // Config contains the configuration for a given Cache.@@ -333,7 +337,7 @@ type Cacher struct { 	stopCh   chan struct{} 	stopWg   sync.WaitGroup -	clock clock.Clock+	clock clock.WithTicker 	// timer is used to avoid unnecessary allocations in underlying watchers. 	timer *time.Timer @@ -343,6 +347,8 @@ type Cacher struct { 	// watchersBuffer is a list of watchers potentially interested in currently 	// dispatched event. 	watchersBuffer []*cacheWatcher+	// unsyncedWatchers is a map that need to be synchronized before sending events.+	unsyncedWatchers watchersMap 	// blockedWatchers is a list of watchers whose buffer is currently full. 	blockedWatchers []*cacheWatcher 	// watchersToStop is a list of watchers that were supposed to be stopped@@ -429,6 +435,7 @@ func NewCacherFromConfig(config Config) (*Cacher, error) { 		stopCh:                        stopCh, 		clock:                         config.Clock, 		timer:                         time.NewTimer(time.Duration(0)),+		unsyncedWatchers:              make(watchersMap), 		bookmarkWatchers:              newTimeBucketWatchers(config.Clock, defaultBookmarkFrequency), 		watcherMetrics:                metrics.NewWatcherMetricsObservers(config.GroupResource), 		disableConsistentGetFromCache: config.DisableConsistentGetFromCache,@@ -485,6 +492,9 @@ func NewCacherFromConfig(config Config) (*Cacher, error) { 	}  	go cacher.dispatchEvents()+	if utilfeature.DefaultFeatureGate.Enabled(features.WatchCacheStallResume) {+		go cacher.startBackgroundResync()+	} 	go progressRequester.Run(stopCh) 	go watchCache.freshness.Run(stopCh) @@ -525,6 +535,62 @@ func (c *Cacher) startCaching(stopChannel <-chan struct{}) { 	} } +func (c *Cacher) startBackgroundResync() {+	c.stopWg.Add(1)+	ticker := c.clock.NewTicker(watcherResyncPeriod)+	go func() {+		defer c.stopWg.Done()+		defer ticker.Stop()+		for {+			select {+			case <-c.stopCh:+				return+			case <-ticker.C():+				c.resyncWatchers()+			}+		}+	}()+}++func (c *Cacher) resyncWatchers() {+	c.RLock()+	hasUnsynced := len(c.unsyncedWatchers) > 0+	c.RUnlock()+	if !hasUnsynced {+		return+	}++	c.watchCache.RLock()+	defer c.watchCache.RUnlock()++	c.Lock()+	defer c.Unlock()++	if !c.Ready() {+		return+	}++	for _, watcher := range c.unsyncedWatchers {+		c.unsyncedWatchers.deleteWatcher(watcher)+		if watcher.stopped {+			continue+		}+		// TODO: Once stalled watchers are moved back to unsyncedWatchers,+		// compute and send only catchup events after the last dispatched+		// resourceVersion instead of re-initializing the watcher interval.+		cacheInterval, err := c.initWatcherIntervalLocked(watcher)+		if err != nil {+			watcher.input <- watcherInput{err: err}+			// Drain input so watcher.done stays open until process() delivers+			// the queued watch.Error event; non-draining stop closes watcher.done+			// immediately, causing sendWatchError to drop the error.+			c.stopWatcherLocked(watcher, true)+			continue+		}+		c.registerWatcherLocked(watcher, cacheInterval)+	}+}+ type namespacedName struct { 	namespace string 	name      string@@ -655,9 +721,12 @@ func (c *Cacher) watch(ctx context.Context, key string, requestedWatchRV uint64, 		watcher.requiredResourceVersion = requiredResourceVersion 	} -	cacheInterval, err := c.initWatcherIntervalLocked(watcher)-	if err != nil {-		return nil, err+	var cacheInterval *watchCacheInterval+	if !utilfeature.DefaultFeatureGate.Enabled(features.WatchCacheStallResume) {+		cacheInterval, err = c.initWatcherIntervalLocked(watcher)+		if err != nil {+			return nil, err+		} 	}  	c.Lock()@@ -681,7 +750,12 @@ func (c *Cacher) watch(ctx context.Context, key string, requestedWatchRV uint64, 			originalForget(drainWatcher) 		} 	}-	c.registerWatcherLocked(watcher, cacheInterval)+	if utilfeature.DefaultFeatureGate.Enabled(features.WatchCacheStallResume) {+		c.unsyncedWatchers.addWatcher(watcher)+	} else {+		c.registerWatcherLocked(watcher, cacheInterval)+	}+	c.watcherIdx++ 	return watcher, nil } @@ -1269,6 +1343,9 @@ func (c *Cacher) terminateAllWatchers() { 	c.watchers.terminateAll(c.groupResource, func(watcher *cacheWatcher) { 		c.stopWatcherLocked(watcher, false) 	})+	c.unsyncedWatchers.terminateAll(func(watcher *cacheWatcher) {+		c.stopWatcherLocked(watcher, false)+	}) }  // watcherToStop is a stop deferred to the end of a dispatch. It keeps its own@@ -1335,6 +1412,7 @@ func forgetWatcher(c *Cacher, w *cacheWatcher) func(bool) { 		// simultaneous Stop() and terminateAllWatchers(), but it is safe to call stopLocked() 		// on a watcher multiple times. 		c.watchers.deleteWatcher(w)+		c.unsyncedWatchers.deleteWatcher(w) 		c.stopWatcherLocked(w, drainWatcher) 	} }

@@ -1057,7 +1057,15 @@ func TestWatchCacheBypass(t *testing.T) { }  func TestEmptyWatchEventCache(t *testing.T) {-	t.Parallel()+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			testEmptyWatchEventCache(t)+		})+	}+}++func testEmptyWatchEventCache(t *testing.T) { 	server, etcdStorage := cachertesting.NewEtcdTestStorage(t, etcd3testing.PathPrefix()) 	defer server.Terminate(t) @@ -1342,6 +1350,15 @@ func TestCacherDontAcceptRequestsStopped(t *testing.T) { }  func TestCacherDontMissEventsOnReinitialization(t *testing.T) {+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			testCacherDontMissEventsOnReinitialization(t)+		})+	}+}++func testCacherDontMissEventsOnReinitialization(t *testing.T) { 	makePod := func(i int) *example.Pod { 		return &example.Pod{ 			ObjectMeta: metav1.ObjectMeta{@@ -1620,6 +1637,15 @@ func TestCacherSendBookmarkEvents(t *testing.T) { }  func TestInitialEventsEndBookmark(t *testing.T) {+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			testInitialEventsEndBookmark(t)+		})+	}+}++func testInitialEventsEndBookmark(t *testing.T) { 	for _, listFromCacheSnapshot := range []bool{true, false} { 		t.Run(fmt.Sprintf("ListFromCacheSnapshot=%v", listFromCacheSnapshot), func(t *testing.T) { 			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchList, true)@@ -1719,7 +1745,15 @@ func TestInitialEventsEndBookmark(t *testing.T) { }  func TestCacherSendsMultipleWatchBookmarks(t *testing.T) {-	t.Parallel()+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			testCacherSendsMultipleWatchBookmarks(t)+		})+	}+}++func testCacherSendsMultipleWatchBookmarks(t *testing.T) { 	backingStorage := &cachertesting.MockStorage{} 	cacher, _, err := newTestCacher(backingStorage) 	if err != nil {@@ -1790,6 +1824,15 @@ func TestCacherSendsMultipleWatchBookmarks(t *testing.T) { }  func TestDispatchingBookmarkEventsWithConcurrentStop(t *testing.T) {+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			testDispatchingBookmarkEventsWithConcurrentStop(t)+		})+	}+}++func testDispatchingBookmarkEventsWithConcurrentStop(t *testing.T) { 	backingStorage := &cachertesting.MockStorage{} 	cacher, _, err := newTestCacher(backingStorage) 	if err != nil {@@ -1865,7 +1908,15 @@ func TestDispatchingBookmarkEventsWithConcurrentStop(t *testing.T) { }  func TestBookmarksOnResourceVersionUpdates(t *testing.T) {-	t.Parallel()+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			testBookmarksOnResourceVersionUpdates(t)+		})+	}+}++func testBookmarksOnResourceVersionUpdates(t *testing.T) { 	backingStorage := &cachertesting.MockStorage{} 	cacher, _, err := newTestCacher(backingStorage) 	if err != nil {@@ -2018,7 +2069,15 @@ func TestStartingResourceVersion(t *testing.T) { }  func TestDispatchEventWillNotBeBlockedByTimedOutWatcher(t *testing.T) {-	t.Parallel()+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			testDispatchEventWillNotBeBlockedByTimedOutWatcher(t)+		})+	}+}++func testDispatchEventWillNotBeBlockedByTimedOutWatcher(t *testing.T) { 	backingStorage := &cachertesting.MockStorage{} 	cacher, _, err := newTestCacher(backingStorage) 	if err != nil {@@ -2161,6 +2220,15 @@ func verifyEvents(t *testing.T, w watch.Interface, events []watch.Event, strictO }  func TestCachingDeleteEvents(t *testing.T) {+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			testCachingDeleteEvents(t)+		})+	}+}++func testCachingDeleteEvents(t *testing.T) { 	backingStorage := &cachertesting.MockStorage{} 	cacher, _, err := newTestCacher(backingStorage) 	if err != nil {@@ -2330,8 +2398,13 @@ func testCachingObjects(t *testing.T, watchersCount int) { }  func TestCachingObjects(t *testing.T) {-	t.Run("single watcher", func(t *testing.T) { testCachingObjects(t, 1) })-	t.Run("many watcher", func(t *testing.T) { testCachingObjects(t, 3) })+	for _, watchCacheStallResume := range []bool{true, false} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+			t.Run("single watcher", func(t *testing.T) { testCachingObjects(t, 1) })+			t.Run("many watcher", func(t *testing.T) { testCachingObjects(t, 3) })+		})+	} }  func TestCacheIntervalInvalidationStopsWatch(t *testing.T) {

@@ -574,11 +574,16 @@ func TestCorrectness(t *testing.T) { 	for _, consistentGetFromCache := range []bool{true, false} { 		t.Run(fmt.Sprintf("ConsistentGetFromCache=%v", consistentGetFromCache), func(t *testing.T) { 			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ConsistentGetFromCache, consistentGetFromCache)-			cachertesting.ForceRequestWatchProgressSupport(t)-			ctx, cacher, server, terminate := setup.SetupWithEtcdServer(t)-			t.Cleanup(terminate)-			keyFunc := func(obj runtime.Object) (string, error) { return storage.NamespaceKeyFunc("/pods/", obj) }-			correctness.RunTestCorrectness(ctx, t, cacher, etcd3testing.PathPrefix(), keyFunc, setup.CompactStore(cacher, server.V3Client.Client))+			for _, watchCacheStallResume := range []bool{true, false} {+				t.Run(fmt.Sprintf("WatchCacheStallResume=%v", watchCacheStallResume), func(t *testing.T) {+					featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, watchCacheStallResume)+					cachertesting.ForceRequestWatchProgressSupport(t)+					ctx, cacher, server, terminate := setup.SetupWithEtcdServer(t)+					t.Cleanup(terminate)+					keyFunc := func(obj runtime.Object) (string, error) { return storage.NamespaceKeyFunc("/pods/", obj) }+					correctness.RunTestCorrectness(ctx, t, cacher, etcd3testing.PathPrefix(), keyFunc, setup.CompactStore(cacher, server.V3Client.Client))+				})+			} 		}) 	} }

@@ -32,15 +32,23 @@ import ( )  func TestWatch(t *testing.T) {-	for _, watchWithoutPrevKV := range []bool{true, false} {-		t.Run(fmt.Sprintf("WatchWithoutPrevKV=%v", watchWithoutPrevKV), func(t *testing.T) {-			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchFromStorageWithoutPrevKV, watchWithoutPrevKV)-			testWatch(t)+	for _, tc := range []struct {+		watchCacheStallResume bool+		watchWithoutPrevKV    bool+	}{+		{watchCacheStallResume: false, watchWithoutPrevKV: true},+		{watchCacheStallResume: false, watchWithoutPrevKV: false},+		{watchCacheStallResume: true, watchWithoutPrevKV: true},+	} {+		t.Run(fmt.Sprintf("WatchCacheStallResume=%v/WatchWithoutPrevKV=%v", tc.watchCacheStallResume, tc.watchWithoutPrevKV), func(t *testing.T) {+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchCacheStallResume, tc.watchCacheStallResume)+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchFromStorageWithoutPrevKV, tc.watchWithoutPrevKV)+			testWatch(t, tc.watchCacheStallResume) 		}) 	} } -func testWatch(t *testing.T) {+func testWatch(t *testing.T, watchCacheStallResume bool) { 	t.Run("Watch", func(t *testing.T) { 		t.Parallel() 		ctx, cacher, terminate := setup.Setup(t)@@ -146,7 +154,7 @@ func testWatch(t *testing.T) { 			fakeClock.Step(cacher.DefaultBookmarkFrequency + time.Second) 			time.Sleep(50 * time.Millisecond) 		}-		storagetesting.RunTestWatchResync(ctx, t, store, true, cacher.CompactWatch(delegator, server.V3Client.Client), resync, triggerBookmarks)+		storagetesting.RunTestWatchResync(ctx, t, store, true, watchCacheStallResume, cacher.CompactWatch(delegator, server.V3Client.Client), resync, triggerBookmarks) 	}) 	// These tests override WatchList and must finish before the parallel subtests start. 	t.Run("WatchSemantics", func(t *testing.T) {

@@ -174,7 +174,7 @@ func TestWatch(t *testing.T) { 	}) 	t.Run("WatchResync", func(t *testing.T) { 		ctx, store, client := testSetup(t)-		storagetesting.RunTestWatchResync(ctx, t, store, false, compactStorage(store, client.Client), nil, nil)+		storagetesting.RunTestWatchResync(ctx, t, store, false, false, compactStorage(store, client.Client), nil, nil) 	}) } 

@@ -235,7 +235,7 @@ func TestStorageWatchListMatchSingle(t *testing.T) {  func TestStorageWatchResync(t *testing.T) { 	s := newTestStorage()-	storagetesting.RunTestWatchResync(t.Context(), t, s, false, compactFunc(s), nil, nil)+	storagetesting.RunTestWatchResync(t.Context(), t, s, false, false, compactFunc(s), nil, nil) }  func TestStorageCorrectness(t *testing.T) {

@@ -2447,20 +2447,20 @@ func RunWatchErrorIsBlockingFurtherEvents(ctx context.Context, t *testing.T, sto 	wg.Wait() } -func RunTestWatchResync(ctx context.Context, t *testing.T, store storage.Interface, isCacher bool, compact Compaction, resync, triggerBookmarks func()) {+func RunTestWatchResync(ctx context.Context, t *testing.T, store storage.Interface, isCacher, syncBackground bool, compact Compaction, resync, triggerBookmarks func()) { 	for i := range 3 { 		t.Run(fmt.Sprintf("compact=%d", i), func(t *testing.T) {-			testWatchResync(ctx, t, store, isCacher, compact, resync, triggerBookmarks, i, false)+			testWatchResync(ctx, t, store, isCacher, syncBackground, compact, resync, triggerBookmarks, i, false) 		}) 	} 	for i := range 2 { 		t.Run(fmt.Sprintf("stall,compact=%d", i), func(t *testing.T) {-			testWatchResync(ctx, t, store, isCacher, compact, resync, triggerBookmarks, i, true)+			testWatchResync(ctx, t, store, isCacher, syncBackground, compact, resync, triggerBookmarks, i, true) 		}) 	} } -func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, isCacher bool, compact Compaction, resync, triggerBookmarks func(), compactIdx int, stall bool) {+func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, isCacher, syncBackground bool, compact Compaction, resync, triggerBookmarks func(), compactIdx int, stall bool) { 	type testRV string 	const ( 		rvUnset        testRV = ""@@ -2522,6 +2522,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 		expect              *WatchExpectation 		expectEtcd          *WatchExpectation 		expectCacher        *WatchExpectation+		expectCacherResync  *WatchExpectation 	}{ 		{ 			name:                "allowWatchBookmarks=true, sendInitialEvents=true, RV=unset",@@ -2532,6 +2533,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			podsAfterResync:     []*example.Pod{pod3}, 			expectEtcd:          ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3)}), 			expectCacher:        ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3), bookmarkWithPod3RV}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{added(pod1), added(pod2), initialBookmarkWithPod2RV, added(pod3), bookmarkWithPod3RV}), 		}, 		{ 			name:                "allowWatchBookmarks=true, sendInitialEvents=false, RV=unset",@@ -2576,6 +2578,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			podsAfterResync:     []*example.Pod{pod3}, 			expectEtcd:          ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3)}), 			expectCacher:        ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3), bookmarkWithPod3RV}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{added(pod1), added(pod2), initialBookmarkWithPod2RV, added(pod3), bookmarkWithPod3RV}), 		}, 		{ 			name:                "allowWatchBookmarks=true, sendInitialEvents=false, RV=0",@@ -2624,6 +2627,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			podsAfterResync:     []*example.Pod{pod3}, 			expectEtcd:          ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3)}), 			expectCacher:        ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3), bookmarkWithPod3RV}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{added(pod1), added(pod2), initialBookmarkWithPod2RV, added(pod3), bookmarkWithPod3RV}), 		}, 		{ 			name:                "allowWatchBookmarks=true, sendInitialEvents=false, RV=1",@@ -2674,6 +2678,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			podsAfterResync:     []*example.Pod{pod3}, 			expectEtcd:          ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3)}), 			expectCacher:        ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3), bookmarkWithPod3RV}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{added(pod1), added(pod2), initialBookmarkWithPod2RV, added(pod3), bookmarkWithPod3RV}), 		}, 		{ 			name:                "allowWatchBookmarks=true, sendInitialEvents=false, RV=beforeTest",@@ -2722,6 +2727,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			podsAfterResync:     []*example.Pod{pod3}, 			expectEtcd:          ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3)}), 			expectCacher:        ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3), bookmarkWithPod3RV}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{added(pod1), added(pod2), initialBookmarkWithPod2RV, added(pod3), bookmarkWithPod3RV}), 		}, 		{ 			name:                "allowWatchBookmarks=true, sendInitialEvents=false, RV=beforeWatch",@@ -2824,6 +2830,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			compactBeforeResync: pod2, 			podsAfterResync:     []*example.Pod{pod3}, 			expect:              ExpectWatchEventsExact([]watch.Event{added(pod2), added(pod3)}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{expiredCacherError}), 		}, 		{ 			name:                "compacted before resync, sendInitialEvents=false, RV=unset",@@ -2833,6 +2840,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			compactBeforeResync: pod2, 			podsAfterResync:     []*example.Pod{pod3}, 			expect:              ExpectWatchEventsExact([]watch.Event{added(pod2), added(pod3)}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{expiredCacherError}), 		}, 		{ 			name:                "compacted before resync, sendInitialEvents=false, RV=0",@@ -2843,6 +2851,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			compactBeforeResync: pod2, 			podsAfterResync:     []*example.Pod{pod3}, 			expect:              ExpectWatchEventsExact([]watch.Event{added(pod2), added(pod3)}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{expiredCacherError}), 		}, 		{ 			name:                "compacted before resync, legacy, RV=unset",@@ -2871,6 +2880,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			podsAfterResync:     []*example.Pod{pod3}, 			expectEtcd:          ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3)}), 			expectCacher:        ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2), added(pod3), bookmarkWithPod3RV}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{added(pod1), added(pod2), initialBookmarkWithPod2RV, added(pod3), bookmarkWithPod3RV}), 		}, 		{ 			name:                "label selector, allowWatchBookmarks=true, sendInitialEvents=true, RV=unset",@@ -2879,6 +2889,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			pred:                storage.SelectionPredicate{Label: labels.SelectorFromSet(labels.Set{"foo": "bar"})}, 			expectEtcd:          ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod3)}), 			expectCacher:        ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod3), bookmarkWithPod3RV}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod2RV, added(pod3), bookmarkWithPod3RV}), 		}, 		{ 			name:                "indexed field selector, allowWatchBookmarks=true, sendInitialEvents=true, RV=unset",@@ -2887,6 +2898,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			pred:                storage.SelectionPredicate{Field: fields.ParseSelectorOrDie("spec.nodeName=node-1"), IndexFields: []string{"spec.nodeName"}}, 			expectEtcd:          ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod3)}), 			expectCacher:        ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod3), bookmarkWithPod3RV}),+			expectCacherResync:  ExpectWatchEventsExact([]watch.Event{added(pod1), initialBookmarkWithPod2RV, added(pod3), bookmarkWithPod3RV}), 		}, 		{ 			name:                "indexed field selector, allowWatchBookmarks=true, sendInitialEvents=true, RV=beforeResync",@@ -2910,6 +2922,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			stall:               true, 			expectEtcd:          ExpectWatchEventsExact(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2)}, stallEvents, []watch.Event{added(pod3)})), 			expectCacher:        ExpectWatchEventsPrefixAndClose(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2)}, stallEvents), stallMin, stallMax),+			expectCacherResync:  ExpectWatchEventsPrefixAndClose(slices.Concat([]watch.Event{added(pod1), added(pod2), initialBookmarkWithPod2RV}, stallEvents), stallMin, stallMax), 		}, 		{ 			name:                "stalled, allowWatchBookmarks=true, sendInitialEvents=false, RV=beforeWatch",@@ -2946,6 +2959,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			stall:               true, 			expectEtcd:          ExpectWatchEventsExact(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod1RV}, stallEvents, []watch.Event{added(pod3)})), 			expectCacher:        ExpectWatchEventsPrefixAndClose(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod1RV}, stallEvents), stallMin, stallMax),+			expectCacherResync:  ExpectWatchEventsPrefixAndClose(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod2RV}, stallEvents), stallMin, stallMax), 		}, 		{ 			name:                "stalled, indexed field selector, allowWatchBookmarks=true, sendInitialEvents=true, RV=unset",@@ -2955,6 +2969,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			stall:               true, 			expectEtcd:          ExpectWatchEventsExact(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod1RV}, stallEvents, []watch.Event{added(pod3)})), 			expectCacher:        ExpectWatchEventsPrefixAndClose(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod1RV}, stallEvents), stallMin, stallMax),+			expectCacherResync:  ExpectWatchEventsPrefixAndClose(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod2RV}, stallEvents), stallMin, stallMax), 		}, 		{ 			name:                "compacted after stall, allowWatchBookmarks=true, sendInitialEvents=false, RV=beforeWatch",@@ -2974,6 +2989,7 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			compactAfterResync:  stallCompacted, 			expectEtcd:          ExpectWatchEventsExact(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2)}, stallEvents, []watch.Event{added(pod3)})), 			expectCacher:        ExpectWatchEventsPrefixAndClose(slices.Concat([]watch.Event{added(pod1), initialBookmarkWithPod1RV, added(pod2)}, stallEvents), stallMin, stallMax),+			expectCacherResync:  ExpectWatchEventsPrefixAndClose(slices.Concat([]watch.Event{added(pod1), added(pod2), initialBookmarkWithPod2RV}, stallEvents), stallMin, stallMax), 		}, 		{ 			name:                "compacted after stall, indexed field selector, allowWatchBookmarks=true, sendInitialEvents=false, RV=beforeTest",@@ -3030,6 +3046,9 @@ func testWatchResync(ctx context.Context, t *testing.T, store storage.Interface, 			if isCacher && tt.expectCacher != nil { 				expectedOut = tt.expectCacher 			}+			if isCacher && syncBackground && tt.expectCacherResync != nil {+				expectedOut = tt.expectCacherResync+			} 			if !isCacher && tt.expectEtcd != nil { 				expectedOut = tt.expectEtcd 			}

@@ -234,6 +234,7 @@ | VolumeBindMountOptions |  |  | 1.37– |  |  |  |  | [code](https://cs.k8s.io/?q=%5CbVolumeBindMountOptions%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbVolumeBindMountOptions%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | VolumeLimitScaling | :ballot_box_with_check: 1.37+ |  | 1.35–1.36 | 1.37– |  |  |  | [code](https://cs.k8s.io/?q=%5CbVolumeLimitScaling%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbVolumeLimitScaling%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | WatchCacheInitializationPostStartHook | :ballot_box_with_check: 1.36+ | :closed_lock_with_key: 1.37+ |  | 1.31–1.36 | 1.37– |  |  | [code](https://cs.k8s.io/?q=%5CbWatchCacheInitializationPostStartHook%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbWatchCacheInitializationPostStartHook%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |+| WatchCacheStallResume |  |  | 1.38– |  |  |  |  | [code](https://cs.k8s.io/?q=%5CbWatchCacheStallResume%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbWatchCacheStallResume%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | WatchFromStorageWithoutPrevKV | :ballot_box_with_check: 1.38+ |  |  | 1.38– |  |  |  | [code](https://cs.k8s.io/?q=%5CbWatchFromStorageWithoutPrevKV%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbWatchFromStorageWithoutPrevKV%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | WatchList | :ballot_box_with_check: 1.32+ |  | 1.27–1.31 | 1.32– |  |  |  | [code](https://cs.k8s.io/?q=%5CbWatchList%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbWatchList%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | WatchListClient | :ballot_box_with_check: 1.35+ |  |  | 1.30– |  |  |  | [code](https://cs.k8s.io/?q=%5CbWatchListClient%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbWatchListClient%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |

@@ -2141,6 +2141,12 @@     lockToDefault: true     preRelease: GA     version: "1.37"+- name: WatchCacheStallResume+  versionedSpecs:+  - default: false+    lockToDefault: false+    preRelease: Alpha+    version: "1.38" - name: WatchFromStorageWithoutPrevKV   versionedSpecs:   - default: true