diff --git a/pkg/storage/unified/resource/continue.go b/pkg/storage/unified/resource/continue.go index c8825ccde88..32d960e785a 100644 --- a/pkg/storage/unified/resource/continue.go +++ b/pkg/storage/unified/resource/continue.go @@ -6,10 +6,18 @@ import ( "fmt" ) +// ContinueToken represents a pagination token for list operations. type ContinueToken struct { - StartOffset int64 `json:"o"` + // Namespace is the namespace to continue from. Only set for cross-namespace list queries. + Namespace string `json:"ns,omitempty"` + // Name is the name to continue from. Required for list resources, empty for list history. + Name string `json:"n,omitempty"` + // ResourceVersion is the resource version for pagination. + // For list resources: the RV the list was performed at. + // For list history: the last seen RV for pagination. ResourceVersion int64 `json:"v"` - SortAscending bool `json:"s"` + // SortAscending indicates the sort order (used by list history). + SortAscending bool `json:"s,omitempty"` } func (c ContinueToken) String() string { diff --git a/pkg/storage/unified/resource/continue_test.go b/pkg/storage/unified/resource/continue_test.go index 3f67740efaf..1a2dee2e071 100644 --- a/pkg/storage/unified/resource/continue_test.go +++ b/pkg/storage/unified/resource/continue_test.go @@ -4,13 +4,54 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestContinueToken(t *testing.T) { - token := &ContinueToken{ - ResourceVersion: 100, - StartOffset: 50, - SortAscending: false, - } - assert.Equal(t, "eyJvIjo1MCwidiI6MTAwLCJzIjpmYWxzZX0=", token.String()) + t.Run("round-trip with namespace (cross-namespace query)", func(t *testing.T) { + original := ContinueToken{ + Namespace: "my-namespace", + Name: "my-resource", + ResourceVersion: 200, + } + decoded, err := GetContinueToken(original.String()) + require.NoError(t, err) + assert.Equal(t, original.Namespace, decoded.Namespace) + assert.Equal(t, original.Name, decoded.Name) + assert.Equal(t, original.ResourceVersion, decoded.ResourceVersion) + }) + + t.Run("round-trip without namespace (single-namespace query)", func(t *testing.T) { + original := ContinueToken{ + Name: "test-resource", + ResourceVersion: 100, + } + decoded, err := GetContinueToken(original.String()) + require.NoError(t, err) + assert.Equal(t, "", decoded.Namespace) + assert.Equal(t, original.Name, decoded.Name) + assert.Equal(t, original.ResourceVersion, decoded.ResourceVersion) + }) + + t.Run("history token (no name, uses ResourceVersion for pagination)", func(t *testing.T) { + original := ContinueToken{ + ResourceVersion: 500, + SortAscending: true, + } + decoded, err := GetContinueToken(original.String()) + require.NoError(t, err) + assert.Equal(t, "", decoded.Name) + assert.Equal(t, int64(500), decoded.ResourceVersion) + assert.True(t, decoded.SortAscending) + }) + + t.Run("rejects invalid base64", func(t *testing.T) { + _, err := GetContinueToken("not-valid-base64!") + assert.Error(t, err) + }) + + t.Run("rejects invalid json", func(t *testing.T) { + _, err := GetContinueToken("bm90LWpzb24=") // "not-json" in base64 + assert.Error(t, err) + }) } diff --git a/pkg/storage/unified/resource/datastore.go b/pkg/storage/unified/resource/datastore.go index 83f1d4b4cbf..82f1e1e59d5 100644 --- a/pkg/storage/unified/resource/datastore.go +++ b/pkg/storage/unified/resource/datastore.go @@ -303,7 +303,7 @@ func (d *dataStore) GetResourceKeyAtRevision(ctx context.Context, key GetRequest listKey := ListRequestKey(key) - iter := d.ListResourceKeysAtRevision(ctx, listKey, rv) + iter := d.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: rv}) for dataKey, err := range iter { if err != nil { return DataKey{}, err @@ -313,32 +313,77 @@ func (d *dataStore) GetResourceKeyAtRevision(ctx context.Context, key GetRequest return DataKey{}, ErrNotFound } +type ListRequestOptions struct { + // Key defines the range to query (Group/Resource/Namespace/Name prefix). + Key ListRequestKey + // ContinueNamespace is the namespace to continue from. + // Only used when Key.Namespace is empty (cross-namespace query). + ContinueNamespace string + // ContinueName is the name to continue from. + ContinueName string + ResourceVersion int64 +} + +// Validate checks that the ListRequestOptions are valid. +func (o ListRequestOptions) Validate() error { + if err := o.Key.Validate(); err != nil { + return fmt.Errorf("invalid list request key: %w", err) + } + // ContinueNamespace is only valid for cross-namespace queries + if o.ContinueNamespace != "" && o.Key.Namespace != "" { + return fmt.Errorf("continue namespace %q not allowed when request namespace is set to %q", o.ContinueNamespace, o.Key.Namespace) + } + return nil +} + // ListLatestResourceKeys returns an iterator over the data keys for the latest versions of resources. // Only returns keys for resources that are not deleted. func (d *dataStore) ListLatestResourceKeys(ctx context.Context, key ListRequestKey) iter.Seq2[DataKey, error] { - return d.ListResourceKeysAtRevision(ctx, key, 0) + return d.ListResourceKeysAtRevision(ctx, ListRequestOptions{ + Key: key, + }) } // ListResourceKeysAtRevision returns an iterator over data keys for resources at a specific revision. // If rv is 0, it returns the latest versions. Only returns keys for resources that are not deleted at the given revision. -func (d *dataStore) ListResourceKeysAtRevision(ctx context.Context, key ListRequestKey, rv int64) iter.Seq2[DataKey, error] { - if err := key.Validate(); err != nil { +func (d *dataStore) ListResourceKeysAtRevision(ctx context.Context, options ListRequestOptions) iter.Seq2[DataKey, error] { + if err := options.Validate(); err != nil { return func(yield func(DataKey, error) bool) { - yield(DataKey{}, fmt.Errorf("invalid list request key: %w", err)) + yield(DataKey{}, err) } } + rv := options.ResourceVersion + prefix := options.Key.Prefix() + + startKey := prefix + if options.ContinueName != "" { + // Build the start key from the continue position + continueKey := ListRequestKey{ + Group: options.Key.Group, + Resource: options.Key.Resource, + Namespace: options.Key.Namespace, + Name: options.ContinueName, + } + // For cross-namespace queries, use the continue namespace + if options.Key.Namespace == "" && options.ContinueNamespace != "" { + continueKey.Namespace = options.ContinueNamespace + } + startKey = continueKey.Prefix() + } + + listOptions := ListOptions{ + StartKey: startKey, + EndKey: PrefixRangeEnd(prefix), + Sort: SortOrderAsc, + } + if rv == 0 { rv = math.MaxInt64 } - prefix := key.Prefix() // List all keys in the prefix. - iter := d.kv.Keys(ctx, dataSection, ListOptions{ - StartKey: prefix, - EndKey: PrefixRangeEnd(prefix), - Sort: SortOrderAsc, - }) + iter := d.kv.Keys(ctx, dataSection, listOptions) return func(yield func(DataKey, error) bool) { var candidateKey *DataKey // The current candidate key we are iterating over diff --git a/pkg/storage/unified/resource/datastore_test.go b/pkg/storage/unified/resource/datastore_test.go index 9ee16f651d2..fbfb828f133 100644 --- a/pkg/storage/unified/resource/datastore_test.go +++ b/pkg/storage/unified/resource/datastore_test.go @@ -2022,7 +2022,7 @@ func TestDataStore_ListResourceKeysAtRevision(t *testing.T) { t.Run("list at revision rv1 - should return only resource1 initial version", func(t *testing.T) { resultKeys := make([]DataKey, 0, 2) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, listKey, rv1) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: rv1}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } @@ -2035,7 +2035,7 @@ func TestDataStore_ListResourceKeysAtRevision(t *testing.T) { t.Run("list at revision rv2 - should return resource1, resource2 and resource4", func(t *testing.T) { resultKeys := make([]DataKey, 0, 3) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, listKey, rv2) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: rv2}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } @@ -2053,7 +2053,7 @@ func TestDataStore_ListResourceKeysAtRevision(t *testing.T) { t.Run("list at revision rv3 - should return resource1, resource2 and resource4", func(t *testing.T) { resultKeys := make([]DataKey, 0, 3) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, listKey, rv3) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: rv3}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } @@ -2074,7 +2074,7 @@ func TestDataStore_ListResourceKeysAtRevision(t *testing.T) { t.Run("list at revision rv4 - should return all resources", func(t *testing.T) { resultKeys := make([]DataKey, 0, 4) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, listKey, rv4) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: rv4}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } @@ -2093,7 +2093,7 @@ func TestDataStore_ListResourceKeysAtRevision(t *testing.T) { t.Run("list at revision rv5 - should exclude deleted resource4", func(t *testing.T) { resultKeys := make([]DataKey, 0, 3) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, listKey, rv5) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: rv5}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } @@ -2119,7 +2119,7 @@ func TestDataStore_ListResourceKeysAtRevision(t *testing.T) { } resultKeys := make([]DataKey, 0, 2) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, specificListKey, rv3) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: specificListKey, ResourceVersion: rv3}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } @@ -2132,7 +2132,7 @@ func TestDataStore_ListResourceKeysAtRevision(t *testing.T) { t.Run("list at revision 0 should use MaxInt64", func(t *testing.T) { resultKeys := make([]DataKey, 0, 4) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, listKey, 0) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: 0}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } @@ -2185,7 +2185,7 @@ func TestDataStore_ListResourceKeysAtRevision_ValidationErrors(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - for _, err := range ds.ListResourceKeysAtRevision(ctx, tt.key, 0) { + for _, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: tt.key, ResourceVersion: 0}) { require.Error(t, err) return } @@ -2204,7 +2204,7 @@ func TestDataStore_ListResourceKeysAtRevision_EmptyResults(t *testing.T) { } resultKeys := make([]DataKey, 0, 1) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, listKey, 0) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: 0}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } @@ -2238,7 +2238,7 @@ func TestDataStore_ListResourceKeysAtRevision_ResourcesNewerThanRevision(t *test // List at a revision before the resource was created resultKeys := make([]DataKey, 0, 1) - for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, listKey, rv-1000) { + for dataKey, err := range ds.ListResourceKeysAtRevision(ctx, ListRequestOptions{Key: listKey, ResourceVersion: rv - 1000}) { require.NoError(t, err) resultKeys = append(resultKeys, dataKey) } diff --git a/pkg/storage/unified/resource/storage_backend.go b/pkg/storage/unified/resource/storage_backend.go index 838c0814750..0f65867f351 100644 --- a/pkg/storage/unified/resource/storage_backend.go +++ b/pkg/storage/unified/resource/storage_backend.go @@ -472,15 +472,30 @@ func (k *kvStorageBackend) ListIterator(ctx context.Context, req *resourcepb.Lis namespace := convertEmptyToClusterNamespace(req.Options.Key.Namespace, k.withExperimentalClusterScope) // Parse continue token if provided - offset := int64(0) - resourceVersion := req.ResourceVersion + listOptions := ListRequestOptions{ + Key: ListRequestKey{ + Group: req.Options.Key.Group, + Resource: req.Options.Key.Resource, + Namespace: namespace, + Name: req.Options.Key.Name, + }, + ResourceVersion: req.ResourceVersion, + } + if req.NextPageToken != "" { token, err := GetContinueToken(req.NextPageToken) if err != nil { return 0, fmt.Errorf("invalid continue token: %w", err) } - offset = token.StartOffset - resourceVersion = token.ResourceVersion + if token.Name == "" { + return 0, fmt.Errorf("invalid continue token: name is required for list resources") + } + // Only use token namespace for cross-namespace queries (when request namespace is empty) + if req.Options.Key.Namespace == "" { + listOptions.ContinueNamespace = token.Namespace + } + listOptions.ContinueName = token.Name + listOptions.ResourceVersion = token.ResourceVersion } // We set the listRV to the last event resource version. @@ -492,27 +507,17 @@ func (k *kvStorageBackend) ListIterator(ctx context.Context, req *resourcepb.Lis return 0, fmt.Errorf("failed to fetch last event: %w", err) } - if resourceVersion > 0 { - listRV = resourceVersion + if listOptions.ResourceVersion > 0 { + listRV = listOptions.ResourceVersion } // Fetch the latest objects keys := make([]DataKey, 0, min(defaultListBufferSize, req.Limit+1)) - idx := 0 - for dataKey, err := range k.dataStore.ListResourceKeysAtRevision(ctx, ListRequestKey{ - Group: req.Options.Key.Group, - Resource: req.Options.Key.Resource, - Namespace: namespace, - Name: req.Options.Key.Name, - }, resourceVersion) { + for dataKey, err := range k.dataStore.ListResourceKeysAtRevision(ctx, listOptions) { if err != nil { return 0, err } - // Skip the first offset items. This is not efficient, but it's a simple way to implement it for now. - if idx < int(offset) { - idx++ - continue - } + keys = append(keys, dataKey) // Only fetch the first limit items + 1 to get the next token. if req.Limit > 0 && len(keys) >= int(req.Limit+1) { @@ -524,9 +529,9 @@ func (k *kvStorageBackend) ListIterator(ctx context.Context, req *resourcepb.Lis defer stop() iter := kvListIterator{ - listRV: listRV, - offset: offset, - next: next, + listRV: listRV, + isCrossNamespace: req.Options.Key.Namespace == "", + next: next, } err := cb(&iter) if err != nil { @@ -538,38 +543,44 @@ func (k *kvStorageBackend) ListIterator(ctx context.Context, req *resourcepb.Lis // kvListIterator implements ListIterator for KV storage type kvListIterator struct { - listRV int64 - offset int64 + listRV int64 + isCrossNamespace bool // pull-style iterator next func() (DataObj, error, bool) // current item state - currentDataObj *DataObj + started bool + currentDataObj DataObj value []byte err error + nextDataObj DataObj + nextErr error + hasMore bool } func (i *kvListIterator) Next() bool { - // Pull next item from the iterator - dataObj, err, ok := i.next() - if !ok { - return false + if !i.started { + i.started = true + + i.nextDataObj, i.nextErr, i.hasMore = i.next() } - if err != nil { - i.err = err + + if !i.hasMore { return false } - i.currentDataObj = &dataObj - - i.value, err = readAndClose(dataObj.Value) - if err != nil { - i.err = err + i.currentDataObj, i.err = i.nextDataObj, i.nextErr + if i.err != nil { return false } - i.offset++ + i.value, i.err = readAndClose(i.currentDataObj.Value) + if i.err != nil { + return false + } + + i.nextDataObj, i.nextErr, i.hasMore = i.next() return true } @@ -579,38 +590,31 @@ func (i *kvListIterator) Error() error { } func (i *kvListIterator) ContinueToken() string { - return ContinueToken{ - StartOffset: i.offset, + token := ContinueToken{ + Name: i.nextDataObj.Key.Name, ResourceVersion: i.listRV, - }.String() + } + // Only store namespace in token for cross-namespace queries + if i.isCrossNamespace { + token.Namespace = i.nextDataObj.Key.Namespace + } + return token.String() } func (i *kvListIterator) ResourceVersion() int64 { - if i.currentDataObj != nil { - return i.currentDataObj.Key.ResourceVersion - } - return 0 + return i.currentDataObj.Key.ResourceVersion } func (i *kvListIterator) Namespace() string { - if i.currentDataObj != nil { - return convertClusterNamespaceToEmpty(i.currentDataObj.Key.Namespace) - } - return "" + return convertClusterNamespaceToEmpty(i.currentDataObj.Key.Namespace) } func (i *kvListIterator) Name() string { - if i.currentDataObj != nil { - return i.currentDataObj.Key.Name - } - return "" + return i.currentDataObj.Key.Name } func (i *kvListIterator) Folder() string { - if i.currentDataObj != nil { - return i.currentDataObj.Key.Folder - } - return "" + return i.currentDataObj.Key.Folder } func (i *kvListIterator) Value() []byte { @@ -1129,7 +1133,6 @@ func (i *kvHistoryIterator) ContinueToken() string { } rv := i.currentDataObj.Key.ResourceVersion token := ContinueToken{ - StartOffset: rv, ResourceVersion: rv, SortAscending: i.sortAscending, } diff --git a/pkg/storage/unified/sql/backend.go b/pkg/storage/unified/sql/backend.go index 0abc2fd7329..2416609761d 100644 --- a/pkg/storage/unified/sql/backend.go +++ b/pkg/storage/unified/sql/backend.go @@ -730,7 +730,7 @@ func (b *backend) listAtRevision(ctx context.Context, req *resourcepb.ListReques // Get the RV iter := &listIter{listRV: req.ResourceVersion, sortAsc: false} if req.NextPageToken != "" { - continueToken, err := resource.GetContinueToken(req.NextPageToken) + continueToken, err := GetContinueToken(req.NextPageToken) if err != nil { return 0, fmt.Errorf("get continue token (%q): %w", req.NextPageToken, err) } @@ -832,7 +832,7 @@ func (b *backend) getHistory(ctx context.Context, req *resourcepb.ListRequest, c useCurrentRV: true, // use the current RV for the continue token instead of the listRV } if req.NextPageToken != "" { - continueToken, err := resource.GetContinueToken(req.NextPageToken) + continueToken, err := GetContinueToken(req.NextPageToken) if err != nil { return 0, fmt.Errorf("get continue token (%q): %w", req.NextPageToken, err) } diff --git a/pkg/storage/unified/sql/backend_test.go b/pkg/storage/unified/sql/backend_test.go index 65e64a76ac2..f7ecd755f32 100644 --- a/pkg/storage/unified/sql/backend_test.go +++ b/pkg/storage/unified/sql/backend_test.go @@ -720,7 +720,7 @@ func TestBackend_getHistoryPagination(t *testing.T) { // Define all pages we want to test pages := []struct { versions []int64 - token *resource.ContinueToken + token *ContinueToken }{ { versions: []int64{rv51, rv52, rv53, rv54}, @@ -728,7 +728,7 @@ func TestBackend_getHistoryPagination(t *testing.T) { }, { versions: []int64{rv55, rv56, rv57, rv58}, - token: &resource.ContinueToken{ + token: &ContinueToken{ ResourceVersion: rv54, StartOffset: 4, SortAscending: true, @@ -736,7 +736,7 @@ func TestBackend_getHistoryPagination(t *testing.T) { }, { versions: []int64{rv59, rv60}, - token: &resource.ContinueToken{ + token: &ContinueToken{ ResourceVersion: rv58, StartOffset: 8, SortAscending: true, diff --git a/pkg/storage/unified/sql/continue.go b/pkg/storage/unified/sql/continue.go new file mode 100644 index 00000000000..8a9fc387540 --- /dev/null +++ b/pkg/storage/unified/sql/continue.go @@ -0,0 +1,33 @@ +package sql + +import ( + "encoding/base64" + "encoding/json" + "fmt" +) + +type ContinueToken struct { + StartOffset int64 `json:"o"` + ResourceVersion int64 `json:"v"` + SortAscending bool `json:"s"` +} + +func (c ContinueToken) String() string { + b, _ := json.Marshal(c) + return base64.StdEncoding.EncodeToString(b) +} + +func GetContinueToken(token string) (*ContinueToken, error) { + continueVal, err := base64.StdEncoding.DecodeString(token) + if err != nil { + return nil, fmt.Errorf("error decoding continue token") + } + + t := &ContinueToken{} + err = json.Unmarshal(continueVal, t) + if err != nil { + return nil, err + } + + return t, nil +} diff --git a/pkg/storage/unified/sql/list_iterator.go b/pkg/storage/unified/sql/list_iterator.go index 8689dc20f45..95aa90598f1 100644 --- a/pkg/storage/unified/sql/list_iterator.go +++ b/pkg/storage/unified/sql/list_iterator.go @@ -31,9 +31,9 @@ type listIter struct { // ContinueToken implements resource.ListIterator. func (l *listIter) ContinueToken() string { if l.useCurrentRV { - return resource.ContinueToken{ResourceVersion: l.rv, StartOffset: l.offset, SortAscending: l.sortAsc}.String() + return ContinueToken{ResourceVersion: l.rv, StartOffset: l.offset, SortAscending: l.sortAsc}.String() } - return resource.ContinueToken{ResourceVersion: l.listRV, StartOffset: l.offset, SortAscending: l.sortAsc}.String() + return ContinueToken{ResourceVersion: l.listRV, StartOffset: l.offset, SortAscending: l.sortAsc}.String() } func (l *listIter) Error() error { diff --git a/pkg/storage/unified/sql/list_iterator_test.go b/pkg/storage/unified/sql/list_iterator_test.go index 7edafd675a8..3b567ca0f3e 100644 --- a/pkg/storage/unified/sql/list_iterator_test.go +++ b/pkg/storage/unified/sql/list_iterator_test.go @@ -214,14 +214,14 @@ func TestIntegrationListIter(t *testing.T) { token := iter.ContinueToken() - var actual resource.ContinueToken + var actual ContinueToken b, err := base64.StdEncoding.DecodeString(token) require.NoError(t, err) err = json.Unmarshal(b, &actual) require.NoError(t, err) - expected := resource.ContinueToken{ + expected := ContinueToken{ ResourceVersion: 300, StartOffset: 1, SortAscending: true, @@ -250,14 +250,14 @@ func TestIntegrationListIter(t *testing.T) { token := iter.ContinueToken() - var actual resource.ContinueToken + var actual ContinueToken b, err := base64.StdEncoding.DecodeString(token) require.NoError(t, err) err = json.Unmarshal(b, &actual) require.NoError(t, err) - expected := resource.ContinueToken{ + expected := ContinueToken{ ResourceVersion: 100, StartOffset: 1, SortAscending: true, diff --git a/pkg/storage/unified/testing/storage_backend.go b/pkg/storage/unified/testing/storage_backend.go index b3cc08b57c0..ce36d8836bc 100644 --- a/pkg/storage/unified/testing/storage_backend.go +++ b/pkg/storage/unified/testing/storage_backend.go @@ -440,16 +440,14 @@ func runTestIntegrationBackendList(t *testing.T, backend resource.StorageBackend require.NoError(t, err) require.Nil(t, res.Error) require.Len(t, res.Items, 3) - continueToken, err := resource.GetContinueToken(res.NextPageToken) - require.NoError(t, err) require.Contains(t, string(res.Items[0].Value), "item1 ADDED") require.Contains(t, string(res.Items[1].Value), "item2 MODIFIED") require.Contains(t, string(res.Items[2].Value), "item4 ADDED") - require.GreaterOrEqual(t, continueToken.ResourceVersion, rv8) + require.NotEmpty(t, res.NextPageToken) res, err = server.List(ctx, &resourcepb.ListRequest{ Limit: 3, - NextPageToken: continueToken.String(), + NextPageToken: res.NextPageToken, Options: &resourcepb.ListOptions{ Key: &resourcepb.ResourceKey{ Namespace: ns, @@ -460,6 +458,8 @@ func runTestIntegrationBackendList(t *testing.T, backend resource.StorageBackend }) require.NoError(t, err) require.Nil(t, res.Error) + require.Contains(t, string(res.Items[0].Value), "item5 ADDED") + require.Contains(t, string(res.Items[1].Value), "item6 ADDED") require.Len(t, res.Items, 2) require.Empty(t, res.NextPageToken) }) @@ -504,15 +504,12 @@ func runTestIntegrationBackendList(t *testing.T, backend resource.StorageBackend require.Contains(t, string(res.Items[0].Value), "item1 ADDED") require.Contains(t, string(res.Items[1].Value), "item2 MODIFIED") require.Contains(t, string(res.Items[2].Value), "item4 ADDED") - - continueToken, err := resource.GetContinueToken(res.NextPageToken) - require.NoError(t, err) - require.Equal(t, rv7, continueToken.ResourceVersion) + require.NotEmpty(t, res.NextPageToken) res, err = server.List(ctx, &resourcepb.ListRequest{ Limit: 3, ResourceVersion: rv7, - NextPageToken: continueToken.String(), + NextPageToken: res.NextPageToken, Options: &resourcepb.ListOptions{ Key: &resourcepb.ResourceKey{ Namespace: ns, @@ -756,10 +753,7 @@ func runTestIntegrationBackendListHistory(t *testing.T, backend resource.Storage require.GreaterOrEqual(t, res.ResourceVersion, tc.minExpectedHeadRV) // Check continue token - continueToken, err := resource.GetContinueToken(res.NextPageToken) - require.NoError(t, err) - require.Equal(t, tc.expectedContinueRV, continueToken.ResourceVersion) - require.Equal(t, tc.expectedSortAsc, continueToken.SortAscending) + require.NotEmpty(t, res.NextPageToken) }) } @@ -775,10 +769,7 @@ func runTestIntegrationBackendListHistory(t *testing.T, backend resource.Storage } firstPageRes, err := server.List(ctx, firstRequest) require.NoError(t, err) - - // Get continue token for second page - continueToken, err := resource.GetContinueToken(firstPageRes.NextPageToken) - require.NoError(t, err) + require.NotEmpty(t, firstPageRes.NextPageToken) // Get second page secondPageRes, err := server.List(ctx, &resourcepb.ListRequest{ @@ -786,12 +777,13 @@ func runTestIntegrationBackendListHistory(t *testing.T, backend resource.Storage Source: resourcepb.ListRequest_HISTORY, ResourceVersion: rv1, VersionMatchV2: resourcepb.ResourceVersionMatchV2_NotOlderThan, - NextPageToken: continueToken.String(), + NextPageToken: firstPageRes.NextPageToken, Options: &resourcepb.ListOptions{Key: baseKey}, }) require.NoError(t, err) require.Nil(t, secondPageRes.Error) require.Len(t, secondPageRes.Items, 3) + require.Empty(t, secondPageRes.NextPageToken) // Second page should continue in ascending order expectedRVs := []int64{rvHistory3, rvHistory4, rvHistory5} @@ -829,7 +821,6 @@ func runTestIntegrationBackendListHistory(t *testing.T, backend resource.Storage t.Run("fetch second page of history at revision", func(t *testing.T) { continueToken := &resource.ContinueToken{ ResourceVersion: rvHistory3, - StartOffset: 2, SortAscending: false, } res, err := server.List(ctx, &resourcepb.ListRequest{