Use ListOptions for request arguments but panic on unsupported combinations

@@ -23,6 +23,7 @@ import (  	"github.com/stretchr/testify/require" +	apierrors "k8s.io/apimachinery/pkg/api/errors" 	"k8s.io/apimachinery/pkg/api/meta" 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" 	"k8s.io/apimachinery/pkg/runtime"@@ -59,6 +60,7 @@ func correctnessTestSteps() []testStep { 	pod4Key := mustGetKey(pod4) 	pod5Key := mustGetKey(pod5) 	wrongRV := "99"+	_, invalidRVErr := storage.APIObjectVersioner{}.ParseResourceVersion("abc") 	pod2RV := "3" 	pod3RV8 := "8" 	pod3RV9 := "9"@@ -1111,6 +1113,146 @@ func correctnessTestSteps() []testStep { 				{Object: nil, Err: nil}, 			}, 		},+		{+			Name: "49. Create pod1 with ResourceVersion set returns ErrResourceVersionSetOnCreate",+			Request: Request{+				Op:     OpCreate,+				Key:    pod1Key,+				Create: CreateRequest{Object: withRV(pod1, "5")},+			},+			CorrectResponse: Response{Err: storage.ErrResourceVersionSetOnCreate},+			InvalidResponses: []Response{+				{Object: withRV(pod1, "19")},+				{Object: nil, Err: nil},+			},+		},+		{+			Name: "50. Create with empty key returns empty key error",+			Request: Request{+				Op:     OpCreate,+				Key:    "",+				Create: CreateRequest{Object: pod1},+			},+			CorrectResponse: Response{Err: fmt.Errorf("empty key: %q", "")},+			InvalidResponses: []Response{+				{Object: withRV(pod1, "19")},+				{Object: nil, Err: nil},+			},+		},+		{+			Name: "51. Get with key escaping the prefix returns invalid key error",+			Request: Request{+				Op:  OpGet,+				Key: "/pods/../secrets/s1",+			},+			CorrectResponse: Response{Err: fmt.Errorf("invalid key: %q", "/pods/../secrets/s1")},+			InvalidResponses: []Response{+				{Object: nil, Err: storage.NewKeyNotFoundError("/pods/../secrets/s1", 0)},+				{Object: nil, Err: nil},+			},+		},+		{+			Name: "52. Update with key containing . returns invalid key error",+			Request: Request{+				Op:  OpUpdate,+				Key: "/pods/./ns1/pod4",+				Update: UpdateRequest{+					UpdateFunc: storage.SimpleUpdate(func(obj runtime.Object) (runtime.Object, error) { return obj, nil }),+				},+			},+			CorrectResponse: Response{Err: fmt.Errorf("invalid key: %q", "/pods/./ns1/pod4")},+			InvalidResponses: []Response{+				{Object: nil, Err: storage.NewKeyNotFoundError("/pods/./ns1/pod4", 18)},+				{Object: nil, Err: nil},+			},+		},+		{+			Name: "53. Delete with key / returns empty key error",+			Request: Request{+				Op:  OpDelete,+				Key: "/",+			},+			CorrectResponse: Response{Err: fmt.Errorf("empty key: %q", "/")},+			InvalidResponses: []Response{+				{Object: nil, Err: storage.NewKeyNotFoundError("/", 18)},+				{Object: nil, Err: nil},+			},+		},+		{+			Name: "54. List with key ending with .. returns invalid key error",+			Request: Request{+				Op:   OpList,+				Key:  "/pods/..",+				List: listRecursive,+			},+			CorrectResponse: Response{Err: fmt.Errorf("invalid key: %q", "/pods/..")},+			InvalidResponses: []Response{+				{Object: newTestPodList("18", withLabel(pod4, "17", "version", "v2"))},+				{Object: nil, Err: nil},+			},+		},+		{+			Name: "55. List with unparsable ResourceVersion returns bad request",+			Request: Request{+				Op:  OpList,+				Key: "/pods/",+				List: ListRequest{Options: storage.ListOptions{+					ResourceVersion: "abc",+					Recursive:       true,+					Predicate:       storage.Everything,+				}},+			},+			CorrectResponse: Response{Err: apierrors.NewBadRequest(fmt.Sprintf("invalid resource version: %v", invalidRVErr))},+			InvalidResponses: []Response{+				{Object: newTestPodList("18", withLabel(pod4, "17", "version", "v2"))},+				{Object: nil, Err: nil},+			},+		},+		{+			Name: "56. List with unknown ResourceVersionMatch returns error",+			Request: Request{+				Op:  OpList,+				Key: "/pods/",+				List: ListRequest{Options: storage.ListOptions{+					ResourceVersion:      "15",+					ResourceVersionMatch: "Newest",+					Recursive:            true,+					Predicate:            storage.Everything,+				}},+			},+			CorrectResponse: Response{Err: fmt.Errorf("unknown ResourceVersionMatch value: %v", "Newest")},+			InvalidResponses: []Response{+				{Object: newTestPodList("18", withLabel(pod4, "17", "version", "v2"))},+				{Object: nil, Err: nil},+			},+		},+	}+}++type watchTestCase struct {+	Name        string+	Request     WatchRequest+	ExpectError error+}++func watchTestCasesInvalid() []watchTestCase {+	_, invalidRVErr := storage.APIObjectVersioner{}.ParseResourceVersion("abc")+	return []watchTestCase{+		{+			Name:        "Watch with empty key returns empty key error",+			Request:     WatchRequest{Key: "", Options: storage.ListOptions{ResourceVersion: "1", Predicate: storage.Everything, Recursive: true}},+			ExpectError: fmt.Errorf("empty key: %q", ""),+		},+		{+			Name:        "Watch with key escaping the prefix returns invalid key error",+			Request:     WatchRequest{Key: "/pods/../secrets", Options: storage.ListOptions{ResourceVersion: "1", Predicate: storage.Everything, Recursive: true}},+			ExpectError: fmt.Errorf("invalid key: %q", "/pods/../secrets"),+		},+		{+			Name:        "Watch with unparsable ResourceVersion returns invalid error",+			Request:     WatchRequest{Key: "/pods/", Options: storage.ListOptions{ResourceVersion: "abc", Predicate: storage.Everything, Recursive: true}},+			ExpectError: invalidRVErr,+		}, 	} } @@ -1121,8 +1263,8 @@ func RunTestCorrectness(ctx context.Context, t *testing.T, store storage.Interfa 	initialState := NewEmptyModel(storagePrefix, func() runtime.Object { return &example.Pod{} }, func() runtime.Object { return &example.PodList{} }, versioner) 	model := initialState.Clone() -	watchRequest := WatchRequest{ResourceVersion: "1"}-	watcher, err := store.Watch(ctx, "/pods/", storage.ListOptions{ResourceVersion: watchRequest.ResourceVersion, Predicate: storage.Everything, Recursive: true})+	watchRequest := WatchRequest{Key: "/pods/", Options: storage.ListOptions{ResourceVersion: "1", Predicate: storage.Everything, Recursive: true}}+	watcher, err := store.Watch(ctx, watchRequest.Key, watchRequest.Options) 	require.NoError(t, err) 	defer watcher.Stop() @@ -1171,6 +1313,14 @@ func RunTestCorrectness(ctx context.Context, t *testing.T, store storage.Interfa 	gotEvents := collectEventsTillRV(t, watcher, versioner, model.ResourceVersion) 	validator := NewWatchValidator(versioner, replay, keyFunc) 	require.NoError(t, validator.ValidateWatch(watchRequest, WatchResponse{Events: gotEvents}))++	for _, step := range watchTestCasesInvalid() {+		w, err := store.Watch(ctx, step.Request.Key, step.Request.Options)+		if err == nil {+			w.Stop()+		}+		require.NoError(t, validator.ValidateWatch(step.Request, WatchResponse{Err: err}), "step %s", step.Name)+	} }  func collectEventsTillRV(t *testing.T, watcher watch.Interface, versioner storage.Versioner, targetRV uint64) []watch.Event {

@@ -117,6 +117,30 @@ func (s *Model) Step(input Request, output Response) (ok bool, next *Model, chan 	case OpGet: 		expected = s.get(input.Key, input.Get.Options) 	case OpList:+		if err := checkKey(input.Key, input.List.Options.Recursive); err != nil {+			expected = Response{Err: err}+			break+		}+		if _, _, err := storage.ValidateListOptions("", s.Versioner, input.List.Options); err != nil {+			expected = Response{Err: err}+			break+		}+		if input.List.Options.Predicate.Label == nil || input.List.Options.Predicate.Field == nil {+			// etcd3 and the cacher call methods on both selectors.+			panic("nil label or field selector is not supported, use storage.Everything to match everything")+		}+		if input.List.Options.Predicate.Limit != 0 || input.List.Options.Predicate.Continue != "" {+			panic("pagination (limit, continue) is not supported")+		}+		if !input.List.Options.Predicate.Empty() && input.List.Options.Predicate.GetAttrs == nil {+			panic("selectors without GetAttrs are not supported")+		}+		if input.List.Options.RecordTimestamps {+			panic("recordTimestamps is not supported, it wraps objects in storage-internal types")+		}+		if !input.List.Options.Predicate.Empty() {+			panic("label and field selectors are not supported, the model doesn't filter lists")+		} 		// Model only validates consistent reads, for stale reads we just validate RV and contents are validated later during replay. 		if input.List.Options.ResourceVersion != "" { 			return s.validateListRV(input.List.Options, output), s, nil@@ -125,6 +149,8 @@ func (s *Model) Step(input Request, output Response) (ok bool, next *Model, chan 	case OpUpdate: 		next = s.Clone() 		expected, change = next.update(context.Background(), input.Key, input.Update.IgnoreNotFound, input.Update.Preconditions, input.Update.UpdateFunc, input.Update.CachedExistingObject)+	default:+		panic(fmt.Sprintf("unknown operation %q", input.Op)) 	} 	if !reflect.DeepEqual(expected, output) { 		return false, s, nil@@ -179,6 +205,12 @@ func (s *Model) validateListRV(opts storage.ListOptions, output Response) bool { }  func (s *Model) update(ctx context.Context, key string, ignoreNotFound bool, preconditions *storage.Preconditions, tryUpdate storage.UpdateFunc, cachedExistingObject runtime.Object) (Response, *Change) {+	if err := checkKey(key, false); err != nil {+		return Response{Err: err}, nil+	}+	if tryUpdate == nil {+		panic("tryUpdate must be set")+	} 	stored, exists := s.Items[key] 	var currentObj runtime.Object 	var currentRV uint64@@ -204,10 +236,6 @@ func (s *Model) update(ctx context.Context, key string, ignoreNotFound bool, pre 		return Response{Object: nil, Err: err}, nil 	} -	if tryUpdate == nil {-		return Response{Object: nil, Err: fmt.Errorf("tryUpdate function must not be nil")}, nil-	}- 	updated, _, err := tryUpdate(currentObj, storage.ResponseMeta{ResourceVersion: currentRV}) 	if err != nil { 		return Response{Object: nil, Err: err}, nil@@ -237,11 +265,16 @@ func (s *Model) update(ctx context.Context, key string, ignoreNotFound bool, pre 	} 	accessor.SetResourceVersion(strconv.FormatUint(s.ResourceVersion, 10)) 	s.Items[key] = copied-	// With ignoreNotFound on a missing key, stored is nil and this is a create.-	return Response{Object: copied, Err: nil}, &Change{ResourceVersion: s.ResourceVersion, Object: copied, PrevObject: stored}+	return Response{Object: copied, Err: nil}, &Change{Key: key, ResourceVersion: s.ResourceVersion, Object: copied, PrevObject: stored} }  func (s *Model) create(key string, obj runtime.Object) (Response, *Change) {+	if err := checkKey(key, false); err != nil {+		return Response{Err: err}, nil+	}+	if rv, err := s.Versioner.ObjectResourceVersion(obj); err == nil && rv != 0 {+		return Response{Err: storage.ErrResourceVersionSetOnCreate}, nil+	} 	if _, exists := s.Items[key]; exists { 		return Response{Object: nil, Err: storage.NewKeyExistsError(s.Prefix+key, 0)}, nil 	}@@ -253,10 +286,16 @@ func (s *Model) create(key string, obj runtime.Object) (Response, *Change) { 	} 	accessor.SetResourceVersion(strconv.FormatUint(s.ResourceVersion, 10)) 	s.Items[key] = copied-	return Response{Object: copied, Err: nil}, &Change{ResourceVersion: s.ResourceVersion, Object: copied}+	return Response{Object: copied, Err: nil}, &Change{Key: key, ResourceVersion: s.ResourceVersion, Object: copied} }  func (s *Model) get(key string, opts storage.GetOptions) Response {+	if err := checkKey(key, false); err != nil {+		return Response{Err: err}+	}+	if opts.ResourceVersion != "" {+		panic("get with resourceVersion is not supported, the model only serves the latest state")+	} 	stored, exists := s.Items[key] 	if !exists { 		if opts.IgnoreNotFound {@@ -268,10 +307,9 @@ func (s *Model) get(key string, opts storage.GetOptions) Response { }  func (s *Model) list(key string, opts storage.ListOptions) Response {-	prefix := strings.TrimSuffix(key, "/") + "/" 	var items []runtime.Object 	for _, k := range slices.Sorted(maps.Keys(s.Items)) {-		if (opts.Recursive && strings.HasPrefix(k, prefix)) || (!opts.Recursive && k == key) {+		if keyInScope(key, opts.Recursive, k) { 			items = append(items, s.Items[k].DeepCopyObject()) 		} 	}@@ -285,7 +323,24 @@ func (s *Model) list(key string, opts storage.ListOptions) Response { 	return Response{Object: list, Err: nil} } +// keyInScope reports whether k is selected by a list or watch on key.+func keyInScope(key string, recursive bool, k string) bool {+	if !recursive {+		return k == key+	}+	// Like storage.PrepareKey, "/a" must not select "/ab".+	return strings.HasPrefix(k, strings.TrimSuffix(key, "/")+"/")+}++func checkKey(key string, recursive bool) error {+	_, err := storage.PrepareKey("", key, recursive)+	return err+}+ func (s *Model) delete(ctx context.Context, key string, preconditions *storage.Preconditions, validateDeletion storage.ValidateObjectFunc) (Response, *Change) {+	if err := checkKey(key, false); err != nil {+		return Response{Err: err}, nil+	} 	stored, exists := s.Items[key] 	if !exists { 		return Response{Object: nil, Err: storage.NewKeyNotFoundError(s.Prefix+key, int64(s.ResourceVersion))}, nil@@ -306,7 +361,7 @@ func (s *Model) delete(ctx context.Context, key string, preconditions *storage.P 	} 	accessor.SetResourceVersion(strconv.FormatUint(s.ResourceVersion, 10)) 	delete(s.Items, key)-	return Response{Object: deletedObj, Err: nil}, &Change{ResourceVersion: s.ResourceVersion, PrevObject: stored}+	return Response{Object: deletedObj, Err: nil}, &Change{Key: key, ResourceVersion: s.ResourceVersion, PrevObject: stored} }  func (s *Model) Describe() string {

@@ -65,3 +65,16 @@ func TestCorrectness(t *testing.T) { 	} 	require.Equal(t, expectEvents, gotEvents) }++// TestOptionsInvalid checks the model expects the same error storage returns+// for watches it rejects. Invalid requests are covered by correctnessTestSteps.+func TestOptionsInvalid(t *testing.T) {+	versioner := storage.APIObjectVersioner{}+	validator := NewWatchValidator(versioner, &Replay{versioner: versioner}, getKey)+	for _, step := range watchTestCasesInvalid() {+		t.Run(step.Name, func(t *testing.T) {+			require.NoError(t, validator.ValidateWatch(step.Request, WatchResponse{Err: step.ExpectError}))+			require.Error(t, validator.ValidateWatch(step.Request, WatchResponse{}))+		})+	}+}

@@ -94,7 +94,10 @@ func (r *Replay) Events(request WatchRequest, rvRange *ResourceVersionRange) ([] 		if change.ResourceVersion < rvRange.Min || change.ResourceVersion >= rvRange.Max { 			continue 		}-		watchEvent, err := change.toWatchEvent(r.versioner, request.Predicate)+		if !keyInScope(request.Key, request.Options.Recursive, change.Key) {+			continue+		}+		watchEvent, err := change.toWatchEvent(r.versioner, request.Options.Predicate) 		if err != nil { 			return nil, err 		}

@@ -161,16 +161,16 @@ type Response struct { // create and Object is nil for a delete. Like etcd3 and the cacher, deciding // what a watcher with a predicate receives requires both objects. type Change struct {+	Key             string 	ResourceVersion uint64 	Object          runtime.Object 	PrevObject      runtime.Object } -// WatchRequest contains parameters for a watch stream.+// WatchRequest contains parameters for a watch stream, exactly as passed to storage.Interface.Watch. type WatchRequest struct {-	ResourceVersion string-	// Predicate filters events. The zero value matches everything.-	Predicate storage.SelectionPredicate+	Key     string+	Options storage.ListOptions }  // WatchResponse contains the events and any terminal error received from a watch stream.

@@ -44,6 +44,12 @@ func NewWatchValidator(versioner storage.Versioner, replay *Replay, keyFunc func }  func (v WatchValidator) ValidateWatch(request WatchRequest, response WatchResponse) error {+	if err := v.checkWatch(request); err != nil {+		if !reflect.DeepEqual(err, response.Err) {+			return fmt.Errorf("watch %+v: expected error %v, got %v", request, err, response.Err)+		}+		return nil+	} 	if response.Err != nil { 		return fmt.Errorf("watch %+v: unexpected error: %w", request, response.Err) 	}@@ -59,6 +65,35 @@ func (v WatchValidator) ValidateWatch(request WatchRequest, response WatchRespon 	return nil } +// checkWatch returns the error storage returns for an invalid watch and panics+// on watches the model can't reproduce.+func (v WatchValidator) checkWatch(request WatchRequest) error {+	opts := request.Options+	if err := checkKey(request.Key, opts.Recursive); err != nil {+		return err+	}+	if _, err := v.versioner.ParseResourceVersion(opts.ResourceVersion); err != nil {+		return err+	}+	if opts.Predicate.Label == nil || opts.Predicate.Field == nil {+		// etcd3 and the cacher call methods on both selectors.+		panic("nil label or field selector is not supported, use storage.Everything to match everything")+	}+	if opts.Predicate.Limit != 0 || opts.Predicate.Continue != "" {+		panic("pagination (limit, continue) is not supported")+	}+	if !opts.Predicate.Empty() && opts.Predicate.GetAttrs == nil {+		panic("selectors without GetAttrs are not supported")+	}+	if opts.RecordTimestamps {+		panic("recordTimestamps is not supported, it wraps objects in storage-internal types")+	}+	if opts.SendInitialEvents != nil && *opts.SendInitialEvents {+		panic("initial events are not supported, set sendInitialEvents=false for resourceVersion \"\" or \"0\"")+	}+	return nil+}+ func validateMaybeLastEventError(events []watch.Event) error { 	for i, ev := range events { 		if ev.Type != watch.Error {@@ -212,8 +247,8 @@ func watchRevisionRange(versioner storage.Versioner, request WatchRequest, event 			maxRV = max(maxRV, rv+1) 		} 	}-	if request.ResourceVersion != "0" && request.ResourceVersion != "" {-		requestedRV, err := versioner.ParseResourceVersion(request.ResourceVersion)+	if request.Options.ResourceVersion != "0" && request.Options.ResourceVersion != "" {+		requestedRV, err := versioner.ParseResourceVersion(request.Options.ResourceVersion) 		if err != nil { 			return nil, err 		}

@@ -45,14 +45,16 @@ func TestValidateWatch(t *testing.T) { 		pod1RV7       = watch.Event{Type: watch.Modified, Object: dropLabel(pod1, "7", "color")} 		deletePod1RV8 = watch.Event{Type: watch.Deleted, Object: withRV(pod1, "8")} 	)+	pod1Key := mustGetKey(pod1)+	pod2Key := mustGetKey(pod2) 	history := []Change{-		{ResourceVersion: 2, Object: addPod1RV2.Object},-		{ResourceVersion: 3, Object: addPod2RV3.Object},-		{ResourceVersion: 4, Object: bluePod1RV4.Object, PrevObject: addPod1RV2.Object},-		{ResourceVersion: 5, PrevObject: addPod2RV3.Object},-		{ResourceVersion: 6, Object: redPod1RV6.Object, PrevObject: bluePod1RV4.Object},-		{ResourceVersion: 7, Object: pod1RV7.Object, PrevObject: redPod1RV6.Object},-		{ResourceVersion: 8, PrevObject: pod1RV7.Object},+		{Key: pod1Key, ResourceVersion: 2, Object: addPod1RV2.Object},+		{Key: pod2Key, ResourceVersion: 3, Object: addPod2RV3.Object},+		{Key: pod1Key, ResourceVersion: 4, Object: bluePod1RV4.Object, PrevObject: addPod1RV2.Object},+		{Key: pod2Key, ResourceVersion: 5, PrevObject: addPod2RV3.Object},+		{Key: pod1Key, ResourceVersion: 6, Object: redPod1RV6.Object, PrevObject: bluePod1RV4.Object},+		{Key: pod1Key, ResourceVersion: 7, Object: pod1RV7.Object, PrevObject: redPod1RV6.Object},+		{Key: pod1Key, ResourceVersion: 8, PrevObject: pod1RV7.Object}, 	} 	// Events for object transitions between selectors. 	var (@@ -81,6 +83,16 @@ func TestValidateWatch(t *testing.T) { 		Field:    fields.OneTermEqualSelector("metadata.name", "pod2"), 		GetAttrs: storage.DefaultNamespaceScopedAttr, 	}+	// watchRequest mirrors the watches the integration traffic opens.+	watchRequest := func(rv string, pred storage.SelectionPredicate) WatchRequest {+		opts := storage.ListOptions{ResourceVersion: rv, Predicate: pred, Recursive: true}+		if rv == "" || rv == "0" {+			// Otherwise storage starts with initial events, which aren't supported.+			opts.SendInitialEvents = new(false)+			opts.ResourceVersionMatch = metav1.ResourceVersionMatchNotOlderThan+		}+		return WatchRequest{Key: "/pods/", Options: opts}+	}  	tests := []struct { 		name        string@@ -91,82 +103,82 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "whole history", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events: []watch.Event{addPod1RV2, addPod2RV3, bluePod1RV4, deletePod2RV5, redPod1RV6, pod1RV7, deletePod1RV8}, 		}, 		{ 			name: "partial history from beginning", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything), 			}, 			events: []watch.Event{addPod2RV3, bluePod1RV4, deletePod2RV5, redPod1RV6, pod1RV7, deletePod1RV8}, 		}, 		{ 			name: "partial history from the end", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events: []watch.Event{addPod1RV2, addPod2RV3, bluePod1RV4}, 		}, 		{ 			name: "watch opened on exact resource version", 			requests: []WatchRequest{-				{ResourceVersion: "3"},+				watchRequest("3", storage.Everything), 			}, 			events: []watch.Event{bluePod1RV4, deletePod2RV5}, 		}, 		{ 			name: "missing event in the middle", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events:      []watch.Event{addPod2RV3 /*bluePod1RV4,*/, deletePod2RV5}, 			expectError: true, 		}, 		{ 			name: "duplicate event", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events:      []watch.Event{addPod1RV2, addPod2RV3, addPod2RV3, bluePod1RV4}, 			expectError: true, 		}, 		{ 			name: "events out of order of different type", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events:      []watch.Event{addPod1RV2, addPod2RV3, deletePod2RV5, bluePod1RV4}, 			expectError: true, 		}, 		{ 			name: "events out of order of the same type", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events:      []watch.Event{addPod2RV3, addPod1RV2, bluePod1RV4, deletePod2RV5}, 			expectError: true, 		}, 		{ 			name: "event the history never produced", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events: []watch.Event{ 				addPod1RV2,@@ -178,9 +190,9 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "wrong event type", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events: []watch.Event{ 				addPod1RV2,@@ -192,7 +204,7 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "wrong object in event", 			requests: []WatchRequest{-				{ResourceVersion: "1"},+				watchRequest("1", storage.Everything), 			}, 			events: []watch.Event{ 				addPod1RV2,@@ -204,25 +216,25 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "event at the requested resource version", 			requests: []WatchRequest{-				{ResourceVersion: "3"},+				watchRequest("3", storage.Everything), 			}, 			events:      []watch.Event{addPod2RV3, bluePod1RV4, deletePod2RV5}, 			expectError: true, 		}, 		{ 			name: "bookmark after every event events", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events: []watch.Event{newBookmark("1"), addPod1RV2, newBookmark("2"), addPod2RV3, newBookmark("3"), bluePod1RV4, newBookmark("4"), deletePod2RV5, newBookmark("5")}, 		}, 		{ 			name: "bookmark after the event", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything), 			}, 			events: []watch.Event{ 				addPod2RV3,@@ -232,8 +244,8 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "bookmark ahead of the event", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything), 			}, 			events: []watch.Event{ 				newBookmark("3"),@@ -244,9 +256,9 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "bookmark on fresh watch", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "2"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("2", storage.Everything), 			}, 			events: []watch.Event{ 				newBookmark("3"),@@ -255,9 +267,9 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "bookmark on fresh watch", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "2"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("2", storage.Everything), 			}, 			events: []watch.Event{ 				addPod2RV3,@@ -268,106 +280,106 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "label selector color=blue", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isBlue},-				{ResourceVersion: "0", Predicate: isBlue},-				{ResourceVersion: "1", Predicate: isBlue},-				{ResourceVersion: "2", Predicate: isBlue},-				{ResourceVersion: "3", Predicate: isBlue},+				watchRequest("", isBlue),+				watchRequest("0", isBlue),+				watchRequest("1", isBlue),+				watchRequest("2", isBlue),+				watchRequest("3", isBlue), 			}, 			events: []watch.Event{addBluePod1RV4, deleteBluePod1RV6}, 		}, 		{ 			name: "label selector opened on exact resource version", 			requests: []WatchRequest{-				{ResourceVersion: "4", Predicate: isBlue},-				{ResourceVersion: "5", Predicate: isBlue},+				watchRequest("4", isBlue),+				watchRequest("5", isBlue), 			}, 			events: []watch.Event{deleteBluePod1RV6}, 		}, 		{ 			name: "label selector color=red", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isRed},-				{ResourceVersion: "0", Predicate: isRed},-				{ResourceVersion: "1", Predicate: isRed},-				{ResourceVersion: "2", Predicate: isRed},-				{ResourceVersion: "3", Predicate: isRed},-				{ResourceVersion: "4", Predicate: isRed},-				{ResourceVersion: "5", Predicate: isRed},+				watchRequest("", isRed),+				watchRequest("0", isRed),+				watchRequest("1", isRed),+				watchRequest("2", isRed),+				watchRequest("3", isRed),+				watchRequest("4", isRed),+				watchRequest("5", isRed), 			}, 			events: []watch.Event{addRedPod1RV6, deleteRedPod1RV7}, 		}, 		{ 			name: "field selector name=pod1", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isPod1},-				{ResourceVersion: "0", Predicate: isPod1},-				{ResourceVersion: "1", Predicate: isPod1},+				watchRequest("", isPod1),+				watchRequest("0", isPod1),+				watchRequest("1", isPod1), 			}, 			events: []watch.Event{addPod1RV2, bluePod1RV4, redPod1RV6, pod1RV7, deletePod1RV8}, 		}, 		{ 			name: "field selector name=pod2", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isPod2},-				{ResourceVersion: "0", Predicate: isPod2},-				{ResourceVersion: "1", Predicate: isPod2},-				{ResourceVersion: "2", Predicate: isPod2},+				watchRequest("", isPod2),+				watchRequest("0", isPod2),+				watchRequest("1", isPod2),+				watchRequest("2", isPod2), 			}, 			events: []watch.Event{addPod2RV3, deletePod2RV5}, 		}, 		{ 			name: "event not matching the selector", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isBlue},-				{ResourceVersion: "0", Predicate: isBlue},-				{ResourceVersion: "1", Predicate: isBlue},+				watchRequest("", isBlue),+				watchRequest("0", isBlue),+				watchRequest("1", isBlue), 			}, 			events:      []watch.Event{addPod1RV2, addBluePod1RV4, deleteBluePod1RV6}, 			expectError: true, 		}, 		{ 			name: "missing move into the selector", 			requests: []WatchRequest{-				{ResourceVersion: "1", Predicate: isBlue},-				{ResourceVersion: "2", Predicate: isBlue},-				{ResourceVersion: "3", Predicate: isBlue},+				watchRequest("1", isBlue),+				watchRequest("2", isBlue),+				watchRequest("3", isBlue), 			}, 			events:      []watch.Event{ /*addBluePod1RV4,*/ deleteBluePod1RV6}, 			expectError: true, 		}, 		{ 			name: "move into the selector as modified instead of add", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isBlue},-				{ResourceVersion: "0", Predicate: isBlue},-				{ResourceVersion: "1", Predicate: isBlue},-				{ResourceVersion: "2", Predicate: isBlue},-				{ResourceVersion: "3", Predicate: isBlue},+				watchRequest("", isBlue),+				watchRequest("0", isBlue),+				watchRequest("1", isBlue),+				watchRequest("2", isBlue),+				watchRequest("3", isBlue), 			}, 			events:      []watch.Event{bluePod1RV4, deleteBluePod1RV6}, 			expectError: true, 		}, 		{ 			name: "move out of the selector as modified instead of delete", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isBlue},-				{ResourceVersion: "0", Predicate: isBlue},-				{ResourceVersion: "1", Predicate: isBlue},-				{ResourceVersion: "2", Predicate: isBlue},-				{ResourceVersion: "3", Predicate: isBlue},+				watchRequest("", isBlue),+				watchRequest("0", isBlue),+				watchRequest("1", isBlue),+				watchRequest("2", isBlue),+				watchRequest("3", isBlue), 			}, 			events:      []watch.Event{addBluePod1RV4, redPod1RV6}, 			expectError: true, 		}, 		{ 			name: "move out of the selector with update object instead of previous object", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isBlue},-				{ResourceVersion: "0", Predicate: isBlue},-				{ResourceVersion: "1", Predicate: isBlue},-				{ResourceVersion: "2", Predicate: isBlue},-				{ResourceVersion: "3", Predicate: isBlue},+				watchRequest("", isBlue),+				watchRequest("0", isBlue),+				watchRequest("1", isBlue),+				watchRequest("2", isBlue),+				watchRequest("3", isBlue), 			}, 			events: []watch.Event{ 				addBluePod1RV4,@@ -378,11 +390,11 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "move out of the selector with update object instead of previous object", 			requests: []WatchRequest{-				{ResourceVersion: "", Predicate: isBlue},-				{ResourceVersion: "0", Predicate: isBlue},-				{ResourceVersion: "1", Predicate: isBlue},-				{ResourceVersion: "2", Predicate: isBlue},-				{ResourceVersion: "3", Predicate: isBlue},+				watchRequest("", isBlue),+				watchRequest("0", isBlue),+				watchRequest("1", isBlue),+				watchRequest("2", isBlue),+				watchRequest("3", isBlue), 			}, 			events: []watch.Event{ 				addBluePod1RV4,@@ -393,52 +405,83 @@ func TestValidateWatch(t *testing.T) { 		{ 			name: "watch ending with error event", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events: []watch.Event{addPod1RV2, addPod2RV3, bluePod1RV4, newErrorEvent()}, 		}, 		{ 			name: "watch with only error event", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},-				{ResourceVersion: "3"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything),+				watchRequest("3", storage.Everything), 			}, 			events: []watch.Event{newErrorEvent()}, 		}, 		{ 			name: "event after error event", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events:      []watch.Event{addPod1RV2, newErrorEvent(), addPod2RV3}, 			expectError: true, 		}, 		{ 			name: "missing event before error event", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events:      []watch.Event{addPod1RV2 /*addPod2RV3,*/, bluePod1RV4, newErrorEvent()}, 			expectError: true, 		}, 		{ 			name: "error event with non-status object", 			requests: []WatchRequest{-				{ResourceVersion: ""},-				{ResourceVersion: "0"},-				{ResourceVersion: "1"},+				watchRequest("", storage.Everything),+				watchRequest("0", storage.Everything),+				watchRequest("1", storage.Everything), 			}, 			events:      []watch.Event{addPod1RV2, {Type: watch.Error, Object: withRV(pod2, "3")}}, 			expectError: true, 		},+		{+			name: "watch on a single key",+			requests: []WatchRequest{+				{Key: pod1Key, Options: storage.ListOptions{ResourceVersion: "1", Predicate: storage.Everything}},+			},+			events: []watch.Event{addPod1RV2, bluePod1RV4, redPod1RV6, pod1RV7, deletePod1RV8},+		},+		{+			name: "watch on a single key with event for another key",+			requests: []WatchRequest{+				{Key: pod1Key, Options: storage.ListOptions{ResourceVersion: "1", Predicate: storage.Everything}},+			},+			events:      []watch.Event{addPod1RV2, addPod2RV3, bluePod1RV4},+			expectError: true,+		},+		{+			name: "watch on a namespace",+			requests: []WatchRequest{+				{Key: "/pods/ns1", Options: storage.ListOptions{ResourceVersion: "1", Predicate: storage.Everything, Recursive: true}},+				{Key: "/pods/ns1/", Options: storage.ListOptions{ResourceVersion: "1", Predicate: storage.Everything, Recursive: true}},+			},+			events: []watch.Event{addPod1RV2, addPod2RV3, bluePod1RV4, deletePod2RV5, redPod1RV6, pod1RV7, deletePod1RV8},+		},+		{+			name: "watch on another namespace with event from outside of it",+			requests: []WatchRequest{+				{Key: "/pods/ns", Options: storage.ListOptions{ResourceVersion: "1", Predicate: storage.Everything, Recursive: true}},+			},+			events:      []watch.Event{addPod1RV2},+			expectError: true,+		}, 	}  	versioner := storage.APIObjectVersioner{}

@@ -480,41 +480,43 @@ func runTraffic(ctx context.Context, store storage.Interface, request *correctne }  func randomWatchRequest(ctx context.Context, store storage.Interface, distribution []ChoiceWeight[WatchRequestType]) correctness.WatchRequest {-	var offset int64+	var rv string 	switch selected := PickRandom(distribution); selected { 	case RVEmpty:-		return correctness.WatchRequest{ResourceVersion: ""}+		rv = "" 	case RVZero:-		return correctness.WatchRequest{ResourceVersion: "0"}+		rv = "0" 	case RVOne:-		return correctness.WatchRequest{ResourceVersion: "1"}+		rv = "1" 	case RVCurrent:-		offset = 0+		rv = relativeRV(ctx, store, 0) 	case RVPast:-		offset = -int64(1 + rand.Intn(10))+		rv = relativeRV(ctx, store, -int64(1+rand.Intn(10))) 	case RVFuture:-		offset = int64(1 + rand.Intn(10))+		rv = relativeRV(ctx, store, int64(1+rand.Intn(10))) 	default: 		panic(fmt.Sprintf("%v: unknown watch request type", selected)) 	}+	opts := storage.ListOptions{ResourceVersion: rv, Predicate: storage.Everything, Recursive: true}+	if rv == "" || rv == "0" {+		// Otherwise storage starts with synthetic ADDED events for existing+		// objects. API validation requires the match with sendInitialEvents.+		opts.SendInitialEvents = new(false)+		opts.ResourceVersionMatch = metav1.ResourceVersionMatchNotOlderThan+	}+	return correctness.WatchRequest{Key: "/pods/", Options: opts}+}++func relativeRV(ctx context.Context, store storage.Interface, offset int64) string { 	currentRV, err := store.GetCurrentResourceVersion(ctx) 	if err != nil { 		panic(err) 	}-	rv := max(int64(currentRV)+offset, 1)-	return correctness.WatchRequest{ResourceVersion: strconv.FormatInt(rv, 10)}+	return strconv.FormatInt(max(int64(currentRV)+offset, 1), 10) }  func runWatch(ctx context.Context, store storage.Interface, req correctness.WatchRequest, cfg WatchConfig) correctness.WatchResponse {-	// Without SendInitialEvents=false, ResourceVersion="0" or "" causes storage-	// to emit synthetic ADDED events for existing objects at their current RV.-	sendInitialEvents := false-	w, err := store.Watch(ctx, "/pods/", storage.ListOptions{-		ResourceVersion:   req.ResourceVersion,-		Predicate:         storage.Everything,-		Recursive:         true,-		SendInitialEvents: &sendInitialEvents,-	})+	w, err := store.Watch(ctx, req.Key, req.Options) 	if err != nil { 		if _, ok := errors.AsType[*storage.StorageError](err); ok { 			return correctness.WatchResponse{Err: err}