Track compaction RV in snapshotter and keep last snapshot behind compaction window

@@ -395,8 +395,7 @@ func TestCompactPastLastWrite(t *testing.T) {  	t.Log("Same list after compaction to the listed resourceVersion") 	compactStore(cacher, server.V3Client.Client)(ctx, t, resourceVersionPastLastWrite)-	// TODO: Expect no requests to etcd, same as before compaction.-	require.Len(t, etcdListRequests(t, ctx, cacher, recorder, list), 1)+	require.Empty(t, etcdListRequests(t, ctx, cacher, recorder, list)) }  func TestGetListRecursivePrefix(t *testing.T) {

@@ -174,8 +174,22 @@ func TestStoreSnapshotter(t *testing.T) { 	assert.Equal(t, uint64(100), snapshot.ResourceVersion()) 	assert.Same(t, snap40.tree, snapshot.tree) +	t.Log("Remove snapshot less than 25 keeps snapshot 20 for [25, 30)")+	cache.Compact(25)++	assert.Equal(t, 3, cache.Len())+	assert.False(t, cache.HasSnapshot(24))+	_, err = cache.GetSnapshot(24)+	assert.True(t, errors.IsResourceExpired(err))++	assert.True(t, cache.HasSnapshot(25))+	snapshot, err = cache.GetSnapshot(25)+	require.NoError(t, err)+	assert.Equal(t, uint64(25), snapshot.ResourceVersion())+	assert.Same(t, snap20.tree, snapshot.tree)+ 	t.Log("Remove snapshot less than 30")-	cache.RemoveLess(30)+	cache.Compact(30)  	assert.Equal(t, 2, cache.Len()) 	assert.False(t, cache.HasSnapshot(10))@@ -186,12 +200,30 @@ func TestStoreSnapshotter(t *testing.T) { 	_, err = cache.GetSnapshot(20) 	assert.True(t, errors.IsResourceExpired(err)) +	assert.False(t, cache.HasSnapshot(25))+	_, err = cache.GetSnapshot(25)+	assert.True(t, errors.IsResourceExpired(err))+ 	assert.True(t, cache.HasSnapshot(30)) 	snapshot, err = cache.GetSnapshot(30) 	require.NoError(t, err) 	assert.Equal(t, uint64(30), snapshot.ResourceVersion()) 	assert.Same(t, snap30.tree, snapshot.tree) +	t.Log("Remove snapshot past last write keeps snapshot 40 for [50, 100]")+	cache.Compact(50)++	assert.Equal(t, 1, cache.Len())+	assert.False(t, cache.HasSnapshot(49))+	_, err = cache.GetSnapshot(49)+	assert.True(t, errors.IsResourceExpired(err))++	assert.True(t, cache.HasSnapshot(50))+	snapshot, err = cache.GetSnapshot(50)+	require.NoError(t, err)+	assert.Equal(t, uint64(50), snapshot.ResourceVersion())+	assert.Same(t, snap40.tree, snapshot.tree)+ 	t.Log("Replace resets old RVs and adds the new snapshot") 	cache.Replace(&btreeStore{resourceVersion: 200}) 	assert.Equal(t, 1, cache.Len())

@@ -74,6 +74,7 @@ func newSnapshotter(enabled bool) snapshotter { type snapshotter struct { 	snapshots       *btree.BTree[*btreeStore] 	resourceVersion uint64+	compactRevision uint64 	enabled         bool } @@ -89,6 +90,7 @@ func (s *snapshotter) SetEnabled(enabled bool) { func (s *snapshotter) reset() { 	s.snapshots.Clear(false) 	s.resourceVersion = 0+	s.compactRevision = 0 }  func (s *snapshotter) HasSnapshot(rv uint64) bool {@@ -104,11 +106,11 @@ func (s *snapshotter) OldestResourceVersion() (uint64, bool) { 	if !ok { 		return 0, false 	}-	return oldest.resourceVersion, true+	return max(oldest.resourceVersion, s.compactRevision), true }  func (s *snapshotter) GetSnapshot(rv uint64) (*btreeStore, error) {-	if !s.enabled || s.resourceVersion == 0 {+	if !s.enabled || s.resourceVersion == 0 || rv < s.compactRevision { 		return nil, errors.NewResourceExpired(fmt.Sprintf("too old resource version: %d", rv)) 	} 	if rv > s.resourceVersion {@@ -118,16 +120,12 @@ func (s *snapshotter) GetSnapshot(rv uint64) (*btreeStore, error) { 		klog.ErrorS(nil, "Snapshotter got future resourceVersion that it doesn't properly handle as it depends on caller ensuring consistency", "requestResourceVersion", rv, "currentResourceVersion", s.resourceVersion) 		return nil, storage.NewTooLargeResourceVersionError(rv, s.resourceVersion, 0) 	}-	var result *btreeStore-	s.snapshots.DescendLessOrEqual(&btreeStore{resourceVersion: rv}, func(snap *btreeStore) bool {-		result = snap-		return false-	})-	if result == nil {+	snapshot := s.getSnapshot(rv)+	if snapshot == nil { 		return nil, errors.NewResourceExpired(fmt.Sprintf("too old resource version: %d", rv)) 	} 	return &btreeStore{-		tree:            result.tree,+		tree:            snapshot.tree, 		resourceVersion: rv, 	}, nil }@@ -152,6 +150,9 @@ func (s *snapshotter) Add(snapshot *btreeStore) { 	} 	s.snapshots.ReplaceOrInsert(snapshot) 	s.resourceVersion = max(s.resourceVersion, snapshot.resourceVersion)+	if snapshot.resourceVersion <= s.compactRevision {+		s.compact()+	} }  func (s *snapshotter) UpdateResourceVersion(rv uint64) {@@ -167,22 +168,39 @@ func (s *snapshotter) UpdateResourceVersion(rv uint64) { 	s.resourceVersion = max(s.resourceVersion, rv) } -func (s *snapshotter) RemoveLess(rv uint64) {+func (s *snapshotter) Compact(rv uint64) { 	if !s.enabled { 		return 	}+	s.compactRevision = max(s.compactRevision, rv)+	s.compact()+}++func (s *snapshotter) compact() {+	// Keep the last snapshot at or before compactRevision so requests for+	// RVs in [compactRevision, nextSnapshotRV) can still be served.+	lastSnapshotBeforeCompaction := s.getSnapshot(s.compactRevision)+	if lastSnapshotBeforeCompaction == nil {+		return+	} 	for s.snapshots.Len() > 0 { 		oldest, ok := s.snapshots.Min()-		if !ok {-			break-		}-		if rv <= oldest.resourceVersion {+		if !ok || oldest.resourceVersion >= lastSnapshotBeforeCompaction.resourceVersion { 			break 		} 		s.snapshots.DeleteMin() 	} } +func (s *snapshotter) getSnapshot(rv uint64) *btreeStore {+	var firstSnapBeforeRV *btreeStore+	s.snapshots.DescendLessOrEqual(&btreeStore{resourceVersion: rv}, func(snap *btreeStore) bool {+		firstSnapBeforeRV = snap+		return false+	})+	return firstSnapBeforeRV+}+ func (s *snapshotter) Len() int { 	return s.snapshots.Len() }

@@ -72,7 +72,7 @@ func (w *WatchCacheStorage) oldestExactRV() (uint64, bool) { func (w *WatchCacheStorage) Compact(rev uint64) { 	w.lock.Lock() 	defer w.lock.Unlock()-	w.snapshots.RemoveLess(rev)+	w.snapshots.Compact(rev) }  func (w *WatchCacheStorage) MarkConsistent(consistent bool) {

@@ -207,19 +207,16 @@ func TestWatchCacheStorageOldestExactRVAfterCompact(t *testing.T) { 	t.Log("Compact between writes") 	assert.Equal(t, "rv=15 keys=[10]", exactSnapshot(t, s, 15)) 	s.Compact(15)-	// TODO: Expect "rv=15 keys=[10]", same as before compaction.-	assert.Equal(t, "rv=20 keys=[10 20]", oldestExactSnapshot(t, s))+	assert.Equal(t, "rv=15 keys=[10]", oldestExactSnapshot(t, s))  	t.Log("Compact past last write") 	assert.Equal(t, "rv=22 keys=[10 20]", exactSnapshot(t, s, 22)) 	s.Compact(22)-	// TODO: Expect "rv=22 keys=[10 20]", same as before compaction.-	assert.Equal(t, "none", oldestExactSnapshot(t, s))+	assert.Equal(t, "rv=22 keys=[10 20]", oldestExactSnapshot(t, s))  	t.Log("Write after compact past last write") 	write(40)-	// TODO: Expect "rv=22 keys=[10 20]", a write does not expire older RVs.-	assert.Equal(t, "rv=40 keys=[10 20 40]", oldestExactSnapshot(t, s))+	assert.Equal(t, "rv=22 keys=[10 20]", oldestExactSnapshot(t, s)) }  func oldestExactSnapshot(t *testing.T, s *WatchCacheStorage) string {

@@ -711,7 +711,7 @@ func RunTestPreconditionalDeleteWithOnlySuggestionPass(ctx context.Context, t *t }  func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, compact Compaction, watchCacheEnabled bool, recorder *KubernetesRecorder) {-	initialRV, createdPods, updatedPod, err := seedMultiLevelData(ctx, store)+	initialRV, compactRV, createdPods, updatedPod, err := seedMultiLevelData(ctx, t, store, compact) 	if err != nil { 		t.Fatal(err) 	}@@ -737,7 +737,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 		return nil, fields.Set{"metadata.name": pod.Name, "spec.nodeName": pod.Spec.NodeName}, nil 	} 	// Compact and increase RV to test consistent List.-	compact(ctx, t, createdPods[0].ResourceVersion)+	compact(ctx, t, compactRV) 	currentRV := fmt.Sprintf("%d", continueRV+1)  	tests := []struct {@@ -807,17 +807,17 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 			prefix:      "/pods/first/", 			pred:        storage.Everything, 			expectedOut: []example.Pod{*createdPods[0]},-			rv:          createdPods[0].ResourceVersion,+			rv:          compactRV, 			rvMatch:     metav1.ResourceVersionMatchExact,-			expectRV:    createdPods[0].ResourceVersion,+			expectRV:    compactRV, 			expectCacherRequestsToEtcd: func() []RecordedList { 				if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) { 					return nil 				} 				return []RecordedList{ 					{ 						Key:      "/registry/pods/first/",-						Revision: mustParseResourceVersion(t, createdPods[0].ResourceVersion),+						Revision: mustParseResourceVersion(t, compactRV), 					}, 				} 			},@@ -827,17 +827,17 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 			prefix:      "/pods/second/", 			pred:        storage.Everything, 			expectedOut: []example.Pod{},-			rv:          createdPods[0].ResourceVersion,+			rv:          compactRV, 			rvMatch:     metav1.ResourceVersionMatchExact,-			expectRV:    createdPods[0].ResourceVersion,+			expectRV:    compactRV, 			expectCacherRequestsToEtcd: func() []RecordedList { 				if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) { 					return nil 				} 				return []RecordedList{ 					{ 						Key:      "/registry/pods/second/",-						Revision: mustParseResourceVersion(t, createdPods[0].ResourceVersion),+						Revision: mustParseResourceVersion(t, compactRV), 					}, 				} 			},@@ -1133,17 +1133,17 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 			}, 			expectedOut:    []example.Pod{}, 			expectContinue: false,-			rv:             createdPods[0].ResourceVersion,+			rv:             compactRV, 			rvMatch:        metav1.ResourceVersionMatchExact,-			expectRV:       createdPods[0].ResourceVersion,+			expectRV:       compactRV, 			expectCacherRequestsToEtcd: func() []RecordedList { 				if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) { 					return nil 				} 				return []RecordedList{ 					{ 						Key:      "/registry/pods/second/",-						Revision: mustParseResourceVersion(t, createdPods[0].ResourceVersion),+						Revision: mustParseResourceVersion(t, compactRV), 						Limit:    1, 					}, 				}@@ -1609,21 +1609,54 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 			expectRVTooOld: true, 		}, 		{-			name:        "test List with resource version of first write, match=Exact",+			name:           "test List with compacted resource version of first write, match=Exact",+			prefix:         "/pods/",+			pred:           storage.Everything,+			rv:             createdPods[0].ResourceVersion,+			rvMatch:        metav1.ResourceVersionMatchExact,+			expectRVTooOld: true,+		},+		{+			name:        "test List with compacted resource version between first and second write, match=Exact", 			prefix:      "/pods/", 			pred:        storage.Everything, 			expectedOut: []example.Pod{*createdPods[0]},-			rv:          createdPods[0].ResourceVersion,+			rv:          compactRV,+			rvMatch:     metav1.ResourceVersionMatchExact,+			expectRV:    compactRV,+			expectCacherRequestsToEtcd: func() []RecordedList {+				if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) {+					return nil+				}+				return []RecordedList{+					{+						Key:      "/registry/pods/",+						Revision: mustParseResourceVersion(t, compactRV),+					},+				}+			},+		},+		{+			name:   "test List with compacted resource version between first and second write and limit, match=Exact",+			prefix: "/pods/",+			pred: storage.SelectionPredicate{+				Label: labels.Everything(),+				Field: fields.Everything(),+				Limit: 1,+			},+			expectedOut: []example.Pod{*createdPods[0]},+			rv:          compactRV, 			rvMatch:     metav1.ResourceVersionMatchExact,-			expectRV:    createdPods[0].ResourceVersion,+			expectRV:    compactRV, 			expectCacherRequestsToEtcd: func() []RecordedList { 				if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) { 					return nil 				} 				return []RecordedList{ 					{ 						Key:      "/registry/pods/",-						Revision: 2,+						Revision: mustParseResourceVersion(t, compactRV),+						Limit:    1, 					}, 				} 			},@@ -1643,7 +1676,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 				return []RecordedList{ 					{ 						Key:      "/registry/pods/",-						Revision: 3,+						Revision: mustParseResourceVersion(t, createdPods[1].ResourceVersion), 					}, 				} 			},@@ -1663,7 +1696,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 				return []RecordedList{ 					{ 						Key:      "/registry/pods/",-						Revision: 4,+						Revision: mustParseResourceVersion(t, createdPods[2].ResourceVersion), 					}, 				} 			},@@ -1683,7 +1716,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 				return []RecordedList{ 					{ 						Key:      "/registry/pods/",-						Revision: 5,+						Revision: mustParseResourceVersion(t, createdPods[3].ResourceVersion), 					}, 				} 			},@@ -1703,7 +1736,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 				return []RecordedList{ 					{ 						Key:      "/registry/pods/",-						Revision: 6,+						Revision: mustParseResourceVersion(t, createdPods[4].ResourceVersion), 					}, 				} 			},@@ -1723,7 +1756,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 				return []RecordedList{ 					{ 						Key:      "/registry/pods/",-						Revision: 7,+						Revision: mustParseResourceVersion(t, createdPods[5].ResourceVersion), 					}, 				} 			},@@ -1743,7 +1776,7 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 				return []RecordedList{ 					{ 						Key:      "/registry/pods/",-						Revision: 8,+						Revision: mustParseResourceVersion(t, updatedPod.ResourceVersion), 					}, 				} 			},@@ -2051,24 +2084,47 @@ func RunTestList(ctx context.Context, t *testing.T, store storage.Interface, com 			}, 		}, 		{-			name:   "test List with continue, resource version of first write",+			name:   "test List with continue, compacted resource version of first write",+			prefix: "/pods/",+			pred: storage.SelectionPredicate{+				Label:    labels.Everything(),+				Field:    fields.Everything(),+				Limit:    1,+				Continue: encodeContinueOrDie(createdPods[0].Namespace+"/"+createdPods[0].Name+"\x00", mustParseResourceVersion(t, createdPods[0].ResourceVersion)),+			},+			expectContinueTooOld: true,+			expectCacherRequestsToEtcd: func() []RecordedList {+				if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) {+					return nil+				}+				return []RecordedList{+					{+						Key:      "/registry/pods/",+						Revision: mustParseResourceVersion(t, createdPods[0].ResourceVersion),+						Limit:    4,+					},+				}+			},+		},+		{+			name:   "test List with continue, compacted resource version between first and second write", 			prefix: "/pods/", 			pred: storage.SelectionPredicate{ 				Label:    labels.Everything(), 				Field:    fields.Everything(), 				Limit:    1,-				Continue: encodeContinueOrDie(createdPods[0].Namespace+"/"+createdPods[0].Name+"\x00", int64(mustAtoi(createdPods[0].ResourceVersion))),+				Continue: encodeContinueOrDie(createdPods[0].Namespace+"/"+createdPods[0].Name+"\x00", mustParseResourceVersion(t, compactRV)), 			}, 			expectedOut: []example.Pod{},-			expectRV:    createdPods[0].ResourceVersion,+			expectRV:    compactRV, 			expectCacherRequestsToEtcd: func() []RecordedList { 				if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) { 					return nil 				} 				return []RecordedList{ 					{ 						Key:      "/registry/pods/first/bar\x00",-						Revision: mustParseResourceVersion(t, createdPods[0].ResourceVersion),+						Revision: mustParseResourceVersion(t, compactRV), 						Limit:    1, 					}, 				}@@ -2677,7 +2733,7 @@ func RunTestConsistentList(ctx context.Context, t *testing.T, store storage.Inte  // seedMultiLevelData creates a set of keys with a multi-level structure, returning a resourceVersion // from before any were created along with the full set of objects that were persisted-func seedMultiLevelData(ctx context.Context, store storage.Interface) (initialRV string, created []*example.Pod, updated *example.Pod, err error) {+func seedMultiLevelData(ctx context.Context, t *testing.T, store storage.Interface, compact Compaction) (initialRV, compactRV string, created []*example.Pod, updated *example.Pod, err error) { 	// Setup storage with the following structure: 	//  / 	//   - first/@@ -2732,15 +2788,20 @@ func seedMultiLevelData(ctx context.Context, store storage.Interface) (initialRV 	// we want to figure out the resourceVersion before we create anything 	initialList := &example.PodList{} 	if err := store.GetList(ctx, "/pods/", storage.ListOptions{Predicate: storage.Everything, Recursive: true}, initialList); err != nil {-		return "", nil, nil, fmt.Errorf("failed to determine starting resourceVersion: %w", err)+		return "", "", nil, nil, fmt.Errorf("failed to determine starting resourceVersion: %w", err) 	} 	initialRV = initialList.ResourceVersion  	for i, ps := range preset { 		preset[i].storedObj = &example.Pod{} 		err := store.Create(ctx, ps.key, ps.obj, preset[i].storedObj, 0) 		if err != nil {-			return "", nil, nil, fmt.Errorf("failed to create object: %w", err)+			return "", "", nil, nil, fmt.Errorf("failed to create object: %w", err)+		}+		if i == 0 {+			// Advance revision before creating the second object so there is a non-pod revision strictly between createdPods[0] and createdPods[1].+			compact(ctx, t, initialRV)+			compactRV = fmt.Sprintf("%d", mustParseResourceVersion(t, preset[0].storedObj.ResourceVersion)+1) 		} 	} @@ -2755,21 +2816,21 @@ func seedMultiLevelData(ctx context.Context, store storage.Interface) (initialRV 			pod.Spec.NodeName = "fakeNode" 			return pod, nil, nil 		}, nil); err != nil {-		return "", nil, nil, fmt.Errorf("failed to update object: %w", err)+		return "", "", nil, nil, fmt.Errorf("failed to update object: %w", err) 	}  	// We now delete bazSecond provided it has been created first. We do this to enable 	// testing cases that had an object exist initially and then was deleted and how this 	// would be reflected in responses of different calls. 	storedObj := &example.Pod{} 	if err := store.Delete(ctx, computePodKey(bazSecond), storedObj, nil, storage.ValidateAllObjectFunc, nil, storage.DeleteOptions{}); err != nil {-		return "", nil, nil, fmt.Errorf("failed to delete object: %w", err)+		return "", "", nil, nil, fmt.Errorf("failed to delete object: %w", err) 	}  	for _, item := range preset { 		created = append(created, item.storedObj) 	}-	return initialRV, created, updated, nil+	return initialRV, compactRV, created, updated, nil }  func RunTestGetListNonRecursive(ctx context.Context, t *testing.T, increaseRV IncreaseRVFunc, store storage.Interface) {@@ -5775,7 +5836,9 @@ func RunTestCompactRevision(ctx context.Context, t *testing.T, store storage.Int 		t.Errorf("Unexpected error: %v", err) 	} 	increaseRV(ctx, t)-	expectCompactedRV := currentRV + 1+	increaseRV(ctx, t)+	compactedFutureRV := currentRV + 1+	expectCompactedRV := currentRV + 2 	compact(ctx, t, fmt.Sprintf("%d", expectCompactedRV)) 	if err := store.GetList(ctx, "/pods/", storage.ListOptions{ 		Predicate:            storage.Everything,@@ -5785,18 +5848,26 @@ func RunTestCompactRevision(ctx context.Context, t *testing.T, store storage.Int 	}, listOut); err == nil || !strings.Contains(err.Error(), "The resourceVersion for the provided list is too old") { 		t.Errorf(`Expected "The resourceVersion for the provided list is too old", but got error: %v`, err) 	}-	// Call twice: the first syncs watchCache to expectCompactedRV via bookmark (falling back after waiting),+	// Call twice: the first syncs watchCache to compactedFutureRV via bookmark (falling back after waiting), 	// and the second exercises upfront delegation once watchCache is already fresh. 	for range 2 { 		if err := store.GetList(ctx, "/pods/", storage.ListOptions{ 			Predicate:            storage.Everything, 			Recursive:            true,-			ResourceVersion:      fmt.Sprintf("%d", expectCompactedRV),+			ResourceVersion:      fmt.Sprintf("%d", compactedFutureRV), 			ResourceVersionMatch: metav1.ResourceVersionMatchExact,-		}, listOut); err != nil {-			t.Errorf("Unexpected error listing at compacted RV %d: %v", expectCompactedRV, err)+		}, listOut); err == nil || !strings.Contains(err.Error(), "The resourceVersion for the provided list is too old") {+			t.Errorf(`Expected "The resourceVersion for the provided list is too old", but got error: %v`, err) 		} 	}+	if err := store.GetList(ctx, "/pods/", storage.ListOptions{+		Predicate:            storage.Everything,+		Recursive:            true,+		ResourceVersion:      fmt.Sprintf("%d", expectCompactedRV),+		ResourceVersionMatch: metav1.ResourceVersionMatchExact,+	}, listOut); err != nil {+		t.Errorf("Unexpected error listing at compacted RV %d: %v", expectCompactedRV, err)+	} 	compactedRV = store.CompactRevision() 	if compactedRV != int64(expectCompactedRV) { 		t.Errorf("CompactRevision()=%d, expected: %d", store.CompactRevision(), expectCompactedRV)