internal/target: refactor for new pagination design

The new design uses four separate service functions; List,
ListPage, ListRefresh and ListRefreshPage. The first two are
used during the initial pagination phase and the latter two
are used for any subsequent refresh phases.
This commit is contained in:
Johan Brandhorst-Satzkorn committed 2024-01-05 07:43:20 -08:00
1 parent f0f7e635a0
commit d4d856e712
10 files changed
+1110 -363

No files matched your search

+256
View File
@@ -61,5 +61,261 @@ select public_id, project_id from target
estimateCountTargets = `
select sum(reltuples::bigint) as estimate from pg_class where oid in ('target_tcp'::regclass, 'target_ssh'::regclass)
`
listTargetsTemplate = `
with targets as (
select public_id
from target
-- search condition for applying permissions is constructed
where %s
order by create_time desc, public_id asc
limit %d
),
tcp_targets as (
select *
from target_tcp
where public_id in (select public_id from targets)
),
ssh_targets as (
select *
from target_ssh
where public_id in (select public_id from targets)
),
all_targets as (
select public_id,
project_id,
name,
description,
default_port,
session_max_seconds,
session_connection_limit,
version,
create_time,
update_time,
worker_filter,
egress_worker_filter,
ingress_worker_filter,
default_client_port,
null as storage_bucket_id,
false as enable_session_recording,
'tcp' as type
from tcp_targets
union
select public_id,
project_id,
name,
description,
default_port,
session_max_seconds,
session_connection_limit,
version,
create_time,
update_time,
worker_filter,
egress_worker_filter,
ingress_worker_filter,
default_client_port,
storage_bucket_id,
enable_session_recording,
'ssh' as type
from ssh_targets
)
select *
from all_targets
order by create_time desc, public_id asc;
`
listTargetsPageTemplate = `
with targets as (
select public_id
from target
where (create_time, public_id) < (@last_item_create_time, @last_item_id)
-- search condition for applying permissions is constructed
and %s
order by create_time desc, public_id asc
limit %d
),
tcp_targets as (
select *
from target_tcp
where public_id in (select public_id from targets)
),
ssh_targets as (
select *
from target_ssh
where public_id in (select public_id from targets)
),
all_targets as (
select public_id,
project_id,
name,
description,
default_port,
session_max_seconds,
session_connection_limit,
version,
create_time,
update_time,
worker_filter,
egress_worker_filter,
ingress_worker_filter,
default_client_port,
null as storage_bucket_id,
false as enable_session_recording,
'tcp' as type
from tcp_targets
union
select public_id,
project_id,
name,
description,
default_port,
session_max_seconds,
session_connection_limit,
version,
create_time,
update_time,
worker_filter,
egress_worker_filter,
ingress_worker_filter,
default_client_port,
storage_bucket_id,
enable_session_recording,
'ssh' as type
from ssh_targets
)
select *
from all_targets
order by create_time desc, public_id asc;
`
refreshTargetsTemplate = `
with targets as (
select public_id
from target
where update_time > @updated_after_time
-- search condition for applying permissions is constructed
and %s
order by update_time desc, public_id asc
limit %d
),
tcp_targets as (
select *
from target_tcp
where public_id in (select public_id from targets)
),
ssh_targets as (
select *
from target_ssh
where public_id in (select public_id from targets)
),
all_targets as (
select public_id,
project_id,
name,
description,
default_port,
session_max_seconds,
session_connection_limit,
version,
create_time,
update_time,
worker_filter,
egress_worker_filter,
ingress_worker_filter,
default_client_port,
null as storage_bucket_id,
false as enable_session_recording,
'tcp' as type
from tcp_targets
union
select public_id,
project_id,
name,
description,
default_port,
session_max_seconds,
session_connection_limit,
version,
create_time,
update_time,
worker_filter,
egress_worker_filter,
ingress_worker_filter,
default_client_port,
storage_bucket_id,
enable_session_recording,
'ssh' as type
from ssh_targets
)
select *
from all_targets
order by update_time desc, public_id asc;
`
refreshTargetsPageTemplate = `
with targets as (
select public_id
from target
where update_time > @updated_after_time
and (update_time, public_id) < (@last_item_update_time, @last_item_id)
-- search condition for applying permissions is constructed
and %s
order by update_time desc, public_id asc
limit %d
),
tcp_targets as (
select *
from target_tcp
where public_id in (select public_id from targets)
),
ssh_targets as (
select *
from target_ssh
where public_id in (select public_id from targets)
),
all_targets as (
select public_id,
project_id,
name,
description,
default_port,
session_max_seconds,
session_connection_limit,
version,
create_time,
update_time,
worker_filter,
egress_worker_filter,
ingress_worker_filter,
default_client_port,
null as storage_bucket_id,
false as enable_session_recording,
'tcp' as type
from tcp_targets
union
select public_id,
project_id,
name,
description,
default_port,
session_max_seconds,
session_connection_limit,
version,
create_time,
update_time,
worker_filter,
egress_worker_filter,
ingress_worker_filter,
default_client_port,
storage_bucket_id,
enable_session_recording,
'ssh' as type
from ssh_targets
)
select *
from all_targets
order by update_time desc, public_id asc;
`
)
+105 -68
View File
@@ -12,6 +12,7 @@ import (
"github.com/hashicorp/boundary/internal/boundary"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/db/timestamp"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/oplog"
@@ -228,101 +229,136 @@ func (r *Repository) FetchAuthzProtectedEntitiesByScope(ctx context.Context, pro
return targetsMap, nil
}
// listTargets lists targets in a project based on the data in the WithPermissions option
// listTargets lists targets based on the data in the WithPermissions option
// provided to the Repository constructor. If no permissions are available, this function
// is a no-op.
// Supported options:
// - WithLimit which overrides the limit set in the Repository object
// - WithStartPageAfterItem which sets where to start listing from
func (r *Repository) listTargets(ctx context.Context, opt ...Option) ([]Target, error) {
const op = "target.(Repository).listTargets"
func (r *Repository) listTargets(ctx context.Context, opt ...Option) ([]Target, time.Time, error) {
if len(r.permissions) == 0 {
return []Target{}, nil
return []Target{}, time.Time{}, nil
}
where, args := r.listPermissionWhereClauses()
if len(where) == 0 {
return []Target{}, nil
}
whereClause := strings.Join(where, " or ")
permissionWhereClause := "(" + strings.Join(where, " or ") + ")"
opts := GetOpts(opt...)
limit := r.defaultLimit
if opts.WithLimit != 0 {
limit = opts.WithLimit
}
// Ordering and pagination are tightly coupled.
// We order by update_time ascending so that new
// and updated items appear at the end of the pagination.
// We need to further order by public_id to distinguish items
// with identical update times.
withOrder := "update_time asc, public_id asc"
query := fmt.Sprintf(listTargetsTemplate, permissionWhereClause, limit)
if opts.WithStartPageAfterItem != nil {
// Now that the order is defined, we can use a simple where
// clause to only include items updated since the specified
// start of the page. We use greater than or equal for the update
// time as there may be items with identical update_times. We
// then use public_id as a tiebreaker.
query = fmt.Sprintf(listTargetsPageTemplate, permissionWhereClause, limit)
args = append(args,
sql.Named("after_item_update_time", opts.WithStartPageAfterItem.GetUpdateTime()),
sql.Named("after_item_id", opts.WithStartPageAfterItem.GetPublicId()),
sql.Named("last_item_create_time", opts.WithStartPageAfterItem.GetCreateTime()),
sql.Named("last_item_id", opts.WithStartPageAfterItem.GetPublicId()),
)
whereClause = "(" + whereClause + ") and (update_time > @after_item_update_time or (update_time = @after_item_update_time and public_id > @after_item_id))"
}
var foundTargets []*targetView
err := r.reader.SearchWhere(
ctx,
&foundTargets,
whereClause,
args,
db.WithLimit(limit),
db.WithOrder(withOrder),
return r.queryTargets(ctx, query, args, limit)
}
// listTargetsRefresh lists targets based on the data in the WithPermissions option
// provided to the Repository constructor. If no permissions are available, this function
// is a no-op.
// Supported options:
// - WithLimit which overrides the limit set in the Repository object
// - WithStartPageAfterItem which sets where to start listing from
func (r *Repository) listTargetsRefresh(ctx context.Context, updatedAfter time.Time, opt ...Option) ([]Target, time.Time, error) {
const op = "target.(Repository).listTargetsRefresh"
if updatedAfter.IsZero() {
return nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "missing updated after time")
}
if len(r.permissions) == 0 {
return []Target{}, time.Time{}, nil
}
where, args := r.listPermissionWhereClauses()
permissionWhereClause := "(" + strings.Join(where, " or ") + ")"
opts := GetOpts(opt...)
limit := r.defaultLimit
if opts.WithLimit != 0 {
limit = opts.WithLimit
}
query := fmt.Sprintf(refreshTargetsTemplate, permissionWhereClause, limit)
args = append(args,
sql.Named("updated_after_time", timestamp.New(updatedAfter)),
)
if err != nil {
return nil, errors.Wrap(ctx, err, op)
if opts.WithStartPageAfterItem != nil {
query = fmt.Sprintf(refreshTargetsPageTemplate, permissionWhereClause, limit)
args = append(args,
sql.Named("last_item_update_time", opts.WithStartPageAfterItem.GetUpdateTime()),
sql.Named("last_item_id", opts.WithStartPageAfterItem.GetPublicId()),
)
}
var targetIds []string
for _, t := range foundTargets {
targetIds = append(targetIds, t.GetPublicId())
}
return r.queryTargets(ctx, query, args, limit)
}
addresses := map[string]string{}
var foundAddresses []*Address
err = r.reader.SearchWhere(ctx,
&foundAddresses,
"target_id in (?)",
[]any{targetIds},
db.WithLimit(limit),
)
if err != nil {
return nil, errors.Wrap(ctx, err, op)
}
for _, addr := range foundAddresses {
addresses[addr.TargetId()] = addr.Address()
}
func (r *Repository) queryTargets(ctx context.Context, query string, args []any, limit int) ([]Target, time.Time, error) {
const op = "target.(Repository).queryTargets"
targets := make([]Target, 0, len(foundTargets))
for _, t := range foundTargets {
var address string
if v, ok := addresses[t.GetPublicId()]; ok {
address = v
}
subtype, err := t.targetSubtype(ctx, address)
if errors.Is(err, errTargetSubtypeNotFound) {
// In cases where we have mixed target types and the controller
// doesn't support all of them, we want to ignore if we can't find
// the target subtype and continue listing the others we do support.
continue
}
var targets []Target
var transactionTimestamp time.Time
if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error {
rows, err := r.Query(ctx, query, args)
if err != nil {
return nil, errors.Wrap(ctx, err, op)
return err
}
targets = append(targets, subtype)
defer rows.Close()
var foundTargets []*targetView
for rows.Next() {
if err := r.ScanRows(ctx, rows, &foundTargets); err != nil {
return err
}
}
var targetIds []string
for _, t := range foundTargets {
targetIds = append(targetIds, t.GetPublicId())
}
var foundAddresses []*Address
if err := r.SearchWhere(ctx,
&foundAddresses,
"target_id in (?)",
[]any{targetIds},
db.WithLimit(limit),
); err != nil {
return err
}
addresses := make(map[string]string)
for _, addr := range foundAddresses {
addresses[addr.TargetId()] = addr.Address()
}
for _, t := range foundTargets {
var address string
if v, ok := addresses[t.GetPublicId()]; ok {
address = v
}
subtype, err := t.targetSubtype(ctx, address)
if errors.Is(err, errTargetSubtypeNotFound) {
// In cases where we have mixed target types and the controller
// doesn't support all of them, we want to ignore if we can't find
// the target subtype and continue listing the others we do support.
continue
}
if err != nil {
return err
}
targets = append(targets, subtype)
}
transactionTimestamp, err = r.Now(ctx)
return err
}); err != nil {
return nil, time.Time{}, errors.Wrap(ctx, err, op)
}
return targets, nil
return targets, transactionTimestamp, nil
}
func (r *Repository) listPermissionWhereClauses() ([]string, []any) {
@@ -357,6 +393,7 @@ func (r *Repository) listDeletedIds(ctx context.Context, since time.Time) ([]str
const op = "target.(Repository).listDeletedIds"
var deleteTargets []*deletedTarget
var transactionTimestamp time.Time
if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, _ db.Writer) error {
if err := r.SearchWhere(ctx, &deleteTargets, "delete_time >= ?", []any{since}); err != nil {
return errors.Wrap(ctx, err, op, errors.WithMsg("failed to query deleted targets"))
+57 -14
View File
@@ -81,12 +81,14 @@ func TestListDeletedIds(t *testing.T) {
// Expect a single entry
deletedIds, ttime = target.TestListDeletedIds(t, repo, ctx, time.Now().AddDate(-1, 0, 0))
require.Equal(t, []string{tg.GetPublicId()}, deletedIds)
// Transaction timestamp should be within ~10 seconds of now
assert.True(t, time.Now().Before(ttime.Add(10*time.Second)))
assert.True(t, time.Now().After(ttime.Add(-10*time.Second)))
// Try again with the time set to now, expect no entries
deletedIds, ttime = target.TestListDeletedIds(t, repo, ctx, time.Now())
require.Empty(t, deletedIds)
// Transaction timestamp should be within ~10 seconds of now
assert.True(t, time.Now().Before(ttime.Add(10*time.Second)))
assert.True(t, time.Now().After(ttime.Add(-10*time.Second)))
}
@@ -169,15 +171,18 @@ func TestRepository_ListTargets(t *testing.T) {
require.NoError(t, err)
t.Run("no-options", func(t *testing.T) {
got := target.TestListTargets(t, repo, ctx)
got, ttime := target.TestListTargets(t, repo, ctx)
require.NoError(t, err)
assert.Equal(t, total, len(got))
// Transaction timestamp should be within ~10 seconds of now
assert.True(t, time.Now().Before(ttime.Add(10*time.Second)))
assert.True(t, time.Now().After(ttime.Add(-10*time.Second)))
})
t.Run("withStartPageAfter", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
page1 := target.TestListTargets(
page1, ttime := target.TestListTargets(
t,
repo,
context.Background(),
@@ -185,7 +190,10 @@ func TestRepository_ListTargets(t *testing.T) {
)
require.NoError(err)
require.Len(page1, 2)
page2 := target.TestListTargets(
// Transaction timestamp should be within ~10 seconds of now
assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
page2, ttime := target.TestListTargets(
t,
repo,
context.Background(),
@@ -194,11 +202,13 @@ func TestRepository_ListTargets(t *testing.T) {
)
require.NoError(err)
require.Len(page2, 2)
assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
for _, item := range page1 {
assert.NotEqual(item.GetPublicId(), page2[0].GetPublicId())
assert.NotEqual(item.GetPublicId(), page2[1].GetPublicId())
}
page3 := target.TestListTargets(
page3, ttime := target.TestListTargets(
t,
repo,
context.Background(),
@@ -207,11 +217,13 @@ func TestRepository_ListTargets(t *testing.T) {
)
require.NoError(err)
require.Len(page3, 2)
assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
for _, item := range page2 {
assert.NotEqual(item.GetPublicId(), page3[0].GetPublicId())
assert.NotEqual(item.GetPublicId(), page3[1].GetPublicId())
}
page4 := target.TestListTargets(
page4, ttime := target.TestListTargets(
t,
repo,
context.Background(),
@@ -220,11 +232,13 @@ func TestRepository_ListTargets(t *testing.T) {
)
require.NoError(err)
require.Len(page4, 2)
assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
for _, item := range page3 {
assert.NotEqual(item.GetPublicId(), page4[0].GetPublicId())
assert.NotEqual(item.GetPublicId(), page4[1].GetPublicId())
}
page5 := target.TestListTargets(
page5, ttime := target.TestListTargets(
t,
repo,
context.Background(),
@@ -233,11 +247,13 @@ func TestRepository_ListTargets(t *testing.T) {
)
require.NoError(err)
require.Len(page5, 2)
assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
for _, item := range page4 {
assert.NotEqual(item.GetPublicId(), page5[0].GetPublicId())
assert.NotEqual(item.GetPublicId(), page5[1].GetPublicId())
}
page6 := target.TestListTargets(
page6, ttime := target.TestListTargets(
t,
repo,
context.Background(),
@@ -246,21 +262,44 @@ func TestRepository_ListTargets(t *testing.T) {
)
require.NoError(err)
require.Empty(page6)
assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
// Update the first target and check that it gets listed subsequently
// Update the first two targets
page1[0].SetName("new-name")
_, _, err = repo.UpdateTarget(ctx, page1[0], page1[0].GetVersion(), []string{"name"})
require.NoError(err)
page7 := target.TestListTargets(
page1[1].SetName("newer-name")
_, _, err = repo.UpdateTarget(ctx, page1[1], page1[1].GetVersion(), []string{"name"})
require.NoError(err)
// since it will return newest to oldest, we get page1[1] first
page7, ttime := target.TestListTargetsRefresh(
t,
repo,
context.Background(),
target.WithLimit(2),
target.WithStartPageAfterItem(page5[1]),
time.Now().Add(-1*time.Second),
target.WithLimit(1),
)
require.NoError(err)
require.Len(page7, 1)
require.Equal(page7[0].GetPublicId(), page1[0].GetPublicId())
require.Equal(page7[0].GetPublicId(), page1[1].GetPublicId())
assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
page8, ttime := target.TestListTargetsRefresh(
t,
repo,
context.Background(),
time.Now().Add(-1*time.Second),
target.WithLimit(1),
target.WithStartPageAfterItem(page7[0]),
)
require.NoError(err)
require.Len(page8, 1)
require.Equal(page8[0].GetPublicId(), page1[0].GetPublicId())
assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
})
}
@@ -303,9 +342,11 @@ func TestRepository_ListTargets_Multiple_Scopes(t *testing.T) {
)
require.NoError(t, err)
got := target.TestListTargets(t, repo, ctx)
got, ttime := target.TestListTargets(t, repo, ctx)
require.NoError(t, err)
assert.Equal(t, total, len(got))
assert.True(t, time.Now().Before(ttime.Add(10*time.Second)))
assert.True(t, time.Now().After(ttime.Add(-10*time.Second)))
}
func TestRepository_ListRoles_Above_Default_Count(t *testing.T) {
@@ -338,9 +379,11 @@ func TestRepository_ListRoles_Above_Default_Count(t *testing.T) {
}))
require.NoError(t, err)
got := target.TestListTargets(t, repo, ctx, target.WithLimit(numToCreate))
got, ttime := target.TestListTargets(t, repo, ctx, target.WithLimit(numToCreate))
require.NoError(t, err)
assert.Equal(t, total, len(got))
assert.True(t, time.Now().Before(ttime.Add(10*time.Second)))
assert.True(t, time.Now().After(ttime.Add(-10*time.Second)))
for _, tar := range got {
assert.Equal(t, "1.2.3.4", tar.GetAddress())
+14 -170
View File
@@ -5,13 +5,12 @@ package target
import (
"context"
"database/sql"
"testing"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/perms"
"github.com/hashicorp/boundary/internal/types/action"
"github.com/hashicorp/boundary/internal/types/resource"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -143,172 +142,17 @@ func TestNewRepository(t *testing.T) {
}
}
func TestRepositoryListPermissionWhereClauses(t *testing.T) {
tests := []struct {
name string
perms []perms.Permission
expWhere []string
expArgs []any
}{
{
name: "nilPerms",
perms: nil,
expWhere: []string{},
expArgs: []any{},
},
{
name: "emptyPerms",
perms: []perms.Permission{},
expWhere: []string{},
expArgs: []any{},
},
{
name: "noListActionPerms",
perms: []perms.Permission{
{
ScopeId: "scope_a",
Action: action.Create,
},
{
ScopeId: "scope_b",
Action: action.Read,
},
{
ScopeId: "scope_c",
Action: action.Delete,
},
},
expWhere: []string{},
expArgs: []any{},
},
{
name: "onePermissionAllResources",
perms: []perms.Permission{
{
ScopeId: "scope_a",
Action: action.List,
},
},
expWhere: []string{"(project_id = @project_id_1)"},
expArgs: []any{sql.Named("project_id_1", "scope_a")},
},
{
name: "onePermissionAllResourcesNonListIgnored",
perms: []perms.Permission{
{
ScopeId: "scope_a",
Action: action.List,
},
{
ScopeId: "scope_b",
Action: action.Create,
},
},
expWhere: []string{"(project_id = @project_id_1)"},
expArgs: []any{sql.Named("project_id_1", "scope_a")},
},
{
name: "onePermissionResourceIds",
perms: []perms.Permission{
{
ScopeId: "scope_a",
Action: action.List,
ResourceIds: []string{"resourceid1", "resourceid2"},
},
},
expWhere: []string{"(project_id = @project_id_1 and public_id = any(@public_id_1))"},
expArgs: []any{
sql.Named("project_id_1", "scope_a"),
sql.Named("public_id_1", "{resourceid1,resourceid2}"),
},
},
{
name: "multiplePermissionsAllResources",
perms: []perms.Permission{
{ScopeId: "scope_a", Action: action.List},
{ScopeId: "scope_b", Action: action.List},
{ScopeId: "scope_c", Action: action.List},
{ScopeId: "scope_d", Action: action.List},
},
expWhere: []string{
"(project_id = @project_id_1)",
"(project_id = @project_id_2)",
"(project_id = @project_id_3)",
"(project_id = @project_id_4)",
},
expArgs: []any{
sql.Named("project_id_1", "scope_a"),
sql.Named("project_id_2", "scope_b"),
sql.Named("project_id_3", "scope_c"),
sql.Named("project_id_4", "scope_d"),
},
},
{
name: "multiplePermissionsResourceIds",
perms: []perms.Permission{
{
ScopeId: "scope_a",
Action: action.List,
ResourceIds: []string{"resourceid1", "resourceid2"},
},
{
ScopeId: "scope_b",
Action: action.List,
ResourceIds: []string{"resourceid3", "resourceid4"},
},
},
expWhere: []string{
"(project_id = @project_id_1 and public_id = any(@public_id_1))",
"(project_id = @project_id_2 and public_id = any(@public_id_2))",
},
expArgs: []any{
sql.Named("project_id_1", "scope_a"),
sql.Named("project_id_2", "scope_b"),
sql.Named("public_id_1", "{resourceid1,resourceid2}"),
sql.Named("public_id_2", "{resourceid3,resourceid4}"),
},
},
{
name: "multiplePermissionsMix",
perms: []perms.Permission{
{
ScopeId: "scope_a",
Action: action.List,
ResourceIds: []string{"resourceid1", "resourceid2"},
},
{
ScopeId: "scope_b",
Action: action.List,
ResourceIds: []string{"resourceid3", "resourceid4"},
},
{ScopeId: "scope_c", Action: action.List},
{ScopeId: "scope_d", Action: action.List},
},
expWhere: []string{
"(project_id = @project_id_1 and public_id = any(@public_id_1))",
"(project_id = @project_id_2 and public_id = any(@public_id_2))",
"(project_id = @project_id_3)",
"(project_id = @project_id_4)",
},
expArgs: []any{
sql.Named("project_id_1", "scope_a"),
sql.Named("public_id_1", "{resourceid1,resourceid2}"),
sql.Named("project_id_2", "scope_b"),
sql.Named("public_id_2", "{resourceid3,resourceid4}"),
sql.Named("project_id_3", "scope_c"),
sql.Named("project_id_4", "scope_d"),
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
repo := Repository{}
repo.permissions = tt.perms
where, args := repo.listPermissionWhereClauses()
require.ElementsMatch(t, tt.expWhere, where)
require.ElementsMatch(t, tt.expArgs, args)
})
}
func Test_listTargetsRefresh(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
testKms := kms.TestKms(t, conn, wrapper)
repo, err := NewRepository(context.Background(), rw, rw, testKms)
require.NoError(t, err)
t.Run("missing updated after", func(t *testing.T) {
t.Parallel()
_, _, err := repo.listTargetsRefresh(context.Background(), time.Time{})
require.ErrorContains(t, err, "missing updated after")
})
}
+13 -15
View File
@@ -5,15 +5,17 @@ package target
import (
"context"
"time"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/pagination"
)
// List lists targets according to the page size,
// filtering out entries that do not
// pass the filter item function. It returns a new refresh token
// based on the grants hash and the returned targets.
// List lists up to page size targets, filtering out entries that
// do not pass the filter item function. It will automatically request
// more targets from the database, at page size chunks, to fill the page.
// It returns a new list token used to continue pagination or refresh items.
// Targets are ordered by create time descending (most recently created first).
func List(
ctx context.Context,
grantsHash []byte,
@@ -23,27 +25,23 @@ func List(
) (*pagination.ListResponse[Target], error) {
const op = "target.List"
if len(grantsHash) == 0 {
switch {
case len(grantsHash) == 0:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
}
if pageSize < 1 {
case pageSize < 1:
return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
}
if filterItemFn == nil {
case filterItemFn == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
}
if repo == nil {
case repo == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo")
}
listItemsFn := func(ctx context.Context, lastPageItem Target, limit int) ([]Target, error) {
listItemsFn := func(ctx context.Context, lastPageItem Target, limit int) ([]Target, time.Time, error) {
opts := []Option{
WithLimit(limit),
}
if lastPageItem != nil {
opts = append(opts,
WithStartPageAfterItem(lastPageItem),
)
opts = append(opts, WithStartPageAfterItem(lastPageItem))
}
return repo.listTargets(ctx, opts...)
}
+480 -67
View File
@@ -6,17 +6,19 @@ package target_test
import (
"context"
"fmt"
"slices"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/hashicorp/boundary/globals"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/db/timestamp"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/listtoken"
"github.com/hashicorp/boundary/internal/perms"
"github.com/hashicorp/boundary/internal/refreshtoken"
"github.com/hashicorp/boundary/internal/target"
"github.com/hashicorp/boundary/internal/target/targettest"
"github.com/hashicorp/boundary/internal/target/targettest/store"
@@ -30,7 +32,12 @@ import (
// reason this doesn't cause an import cycle is because this is
// an "external" test package.
func TestService_List(t *testing.T) {
t.Parallel()
// Set database read timeout to avoid duplicates in response
oldReadTimeout := globals.RefreshReadLookbackDuration
globals.RefreshReadLookbackDuration = 0
t.Cleanup(func() {
globals.RefreshReadLookbackDuration = oldReadTimeout
})
ctx := context.Background()
conn, _ := db.TestSetup(t, "postgres")
sqlDb, err := conn.SqlDB(ctx)
@@ -39,6 +46,7 @@ func TestService_List(t *testing.T) {
testKms := kms.TestKms(t, conn, wrapper)
iamRepo := iam.TestRepo(t, conn, wrapper)
_, proj1 := iam.TestScopes(t, iamRepo)
fiveDaysAgo := time.Now()
total := 5
var targets []target.Target
@@ -46,6 +54,9 @@ func TestService_List(t *testing.T) {
targets = append(targets, targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), fmt.Sprintf("proj1-%d", i)))
}
// since we sort ascending, we need to reverse targets
slices.Reverse(targets)
// Run analyze to update target estimate
_, err = sqlDb.ExecContext(ctx, "analyze")
require.NoError(t, err)
@@ -63,6 +74,8 @@ func TestService_List(t *testing.T) {
)
require.NoError(t, err)
cmpOpts := cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})
t.Run("List validation", func(t *testing.T) {
t.Parallel()
t.Run("missing grants hash", func(t *testing.T) {
@@ -103,17 +116,16 @@ func TestService_List(t *testing.T) {
require.ErrorContains(t, err, "missing repo")
})
})
t.Run("ListRefresh validation", func(t *testing.T) {
t.Run("ListPage validation", func(t *testing.T) {
t.Parallel()
t.Run("missing grants hash", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := refreshtoken.New(ctx, time.Now(), time.Now(), resource.Target, []byte("some hash"), "some-id", time.Now())
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefresh(ctx, nil, 1, filterFunc, tok, repo)
_, err = target.ListPage(ctx, nil, 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "missing grants hash")
})
t.Run("zero page size", func(t *testing.T) {
@@ -121,9 +133,9 @@ func TestService_List(t *testing.T) {
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := refreshtoken.New(ctx, time.Now(), time.Now(), resource.Target, []byte("some hash"), "some-id", time.Now())
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefresh(ctx, []byte("some hash"), 0, filterFunc, tok, repo)
_, err = target.ListPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("negative page size", func(t *testing.T) {
@@ -131,16 +143,16 @@ func TestService_List(t *testing.T) {
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := refreshtoken.New(ctx, time.Now(), time.Now(), resource.Target, []byte("some hash"), "some-id", time.Now())
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefresh(ctx, []byte("some hash"), -1, filterFunc, tok, repo)
_, err = target.ListPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("nil filter func", func(t *testing.T) {
t.Parallel()
tok, err := refreshtoken.New(ctx, time.Now(), time.Now(), resource.Target, []byte("some hash"), "some-id", time.Now())
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefresh(ctx, []byte("some hash"), 1, nil, tok, repo)
_, err = target.ListPage(ctx, []byte("some hash"), 1, nil, tok, repo)
require.ErrorContains(t, err, "missing filter item callback")
})
t.Run("nil token", func(t *testing.T) {
@@ -148,17 +160,299 @@ func TestService_List(t *testing.T) {
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
_, err = target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
_, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
require.ErrorContains(t, err, "missing token")
})
t.Run("wrong token type", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "token did not have a pagination token component")
})
t.Run("nil repo", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := refreshtoken.New(ctx, time.Now(), time.Now(), resource.Target, []byte("some hash"), "some-id", time.Now())
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, nil)
_, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil)
require.ErrorContains(t, err, "missing repo")
})
})
t.Run("ListRefresh validation", func(t *testing.T) {
t.Parallel()
t.Run("missing grants hash", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, nil, 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "missing grants hash")
})
t.Run("zero page size", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("negative page size", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("nil filter func", func(t *testing.T) {
t.Parallel()
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), 1, nil, tok, repo)
require.ErrorContains(t, err, "missing filter item callback")
})
t.Run("nil token", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
_, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
require.ErrorContains(t, err, "missing token")
})
t.Run("wrong token type", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "token did not have a start-refresh token component")
})
t.Run("nil repo", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil)
require.ErrorContains(t, err, "missing repo")
})
})
t.Run("ListRefresh validation", func(t *testing.T) {
t.Parallel()
t.Run("missing grants hash", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, nil, 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "missing grants hash")
})
t.Run("zero page size", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("negative page size", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("nil filter func", func(t *testing.T) {
t.Parallel()
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), 1, nil, tok, repo)
require.ErrorContains(t, err, "missing filter item callback")
})
t.Run("nil token", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
_, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
require.ErrorContains(t, err, "missing token")
})
t.Run("wrong token type", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "token did not have a start-refresh token component")
})
t.Run("nil repo", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil)
require.ErrorContains(t, err, "missing repo")
})
})
t.Run("ListRefreshPage validation", func(t *testing.T) {
t.Parallel()
t.Run("missing grants hash", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, nil, 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "missing grants hash")
})
t.Run("zero page size", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("negative page size", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("nil filter func", func(t *testing.T) {
t.Parallel()
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 1, nil, tok, repo)
require.ErrorContains(t, err, "missing filter item callback")
})
t.Run("nil token", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
require.ErrorContains(t, err, "missing token")
})
t.Run("wrong token type", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "token did not have a refresh token component")
})
t.Run("nil repo", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil)
require.ErrorContains(t, err, "missing repo")
})
})
t.Run("ListRefreshPage validation", func(t *testing.T) {
t.Parallel()
t.Run("missing grants hash", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, nil, 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "missing grants hash")
})
t.Run("zero page size", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("negative page size", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo)
require.ErrorContains(t, err, "page size must be at least 1")
})
t.Run("nil filter func", func(t *testing.T) {
t.Parallel()
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 1, nil, tok, repo)
require.ErrorContains(t, err, "missing filter item callback")
})
t.Run("nil token", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
require.ErrorContains(t, err, "missing token")
})
t.Run("wrong token type", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo)
require.ErrorContains(t, err, "token did not have a refresh token component")
})
t.Run("nil repo", func(t *testing.T) {
t.Parallel()
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo)
require.NoError(t, err)
_, err = target.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil)
require.ErrorContains(t, err, "missing repo")
})
})
@@ -169,82 +463,185 @@ func TestService_List(t *testing.T) {
}
resp, err := target.List(ctx, []byte("some hash"), 1, filterFunc, repo)
require.NoError(t, err)
require.NotNil(t, resp.RefreshToken)
require.Equal(t, resp.RefreshToken.GrantsHash, []byte("some hash"))
require.NotNil(t, resp.ListToken)
require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash"))
require.False(t, resp.CompleteListing)
require.Equal(t, resp.EstimatedItemCount, 5)
require.Empty(t, resp.DeletedIds)
require.Len(t, resp.Items, 1)
require.Empty(t, cmp.Diff(resp.Items[0], targets[0], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Empty(t, cmp.Diff(resp.Items[0], targets[0], cmpOpts))
resp2, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.RefreshToken, repo)
resp2, err := target.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp2.RefreshToken.GrantsHash, []byte("some hash"))
require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash"))
require.False(t, resp2.CompleteListing)
require.Equal(t, resp2.EstimatedItemCount, 5)
require.Empty(t, resp2.DeletedIds)
require.Len(t, resp2.Items, 1)
require.Empty(t, cmp.Diff(resp2.Items[0], targets[1], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Empty(t, cmp.Diff(resp2.Items[0], targets[1], cmpOpts))
resp3, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp2.RefreshToken, repo)
resp3, err := target.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp2.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp3.RefreshToken.GrantsHash, []byte("some hash"))
require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash"))
require.False(t, resp3.CompleteListing)
require.Equal(t, resp3.EstimatedItemCount, 5)
require.Empty(t, resp3.DeletedIds)
require.Len(t, resp3.Items, 1)
require.Empty(t, cmp.Diff(resp3.Items[0], targets[2], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Empty(t, cmp.Diff(resp3.Items[0], targets[2], cmpOpts))
resp4, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp3.RefreshToken, repo)
resp4, err := target.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp3.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp4.RefreshToken.GrantsHash, []byte("some hash"))
require.Equal(t, resp4.ListToken.GrantsHash, []byte("some hash"))
require.False(t, resp4.CompleteListing)
require.Equal(t, resp4.EstimatedItemCount, 5)
require.Empty(t, resp4.DeletedIds)
require.Len(t, resp4.Items, 1)
require.Empty(t, cmp.Diff(resp4.Items[0], targets[3], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Empty(t, cmp.Diff(resp4.Items[0], targets[3], cmpOpts))
resp5, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp4.RefreshToken, repo)
resp5, err := target.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp4.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp5.RefreshToken.GrantsHash, []byte("some hash"))
require.Equal(t, resp5.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp5.CompleteListing)
require.Equal(t, resp5.EstimatedItemCount, 5)
require.Empty(t, resp5.DeletedIds)
require.Len(t, resp5.Items, 1)
require.Empty(t, cmp.Diff(resp5.Items[0], targets[4], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Empty(t, cmp.Diff(resp5.Items[0], targets[4], cmpOpts))
resp6, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp5.RefreshToken, repo)
// Finished initial pagination phase, request refresh
// Expect no results.
resp6, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp5.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp6.RefreshToken.GrantsHash, []byte("some hash"))
require.Equal(t, resp6.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp6.CompleteListing)
require.Equal(t, resp6.EstimatedItemCount, 5)
require.Empty(t, resp6.DeletedIds)
require.Empty(t, resp6.Items)
// Create some new targets
newT1 := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "proj1-5")
newT2 := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "proj1-6")
t.Cleanup(func() {
_, err = repo.DeleteTarget(ctx, newT1.GetPublicId())
require.NoError(t, err)
_, err = repo.DeleteTarget(ctx, newT2.GetPublicId())
require.NoError(t, err)
// Run analyze to update target estimate
_, err = sqlDb.ExecContext(ctx, "analyze")
require.NoError(t, err)
})
// Run analyze to update target estimate
_, err = sqlDb.ExecContext(ctx, "analyze")
require.NoError(t, err)
// Refresh again, should get newT2
resp7, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp6.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp7.ListToken.GrantsHash, []byte("some hash"))
require.False(t, resp7.CompleteListing)
require.Equal(t, resp7.EstimatedItemCount, 7)
require.Empty(t, resp7.DeletedIds)
require.Len(t, resp7.Items, 1)
require.Empty(t, cmp.Diff(resp7.Items[0], newT2, cmpOpts))
// Refresh again, should get newT1
resp8, err := target.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, resp6.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp8.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp8.CompleteListing)
require.Equal(t, resp8.EstimatedItemCount, 7)
require.Empty(t, resp8.DeletedIds)
require.Len(t, resp8.Items, 1)
require.Empty(t, cmp.Diff(resp8.Items[0], newT1, cmpOpts))
// Refresh again, should get no results
resp9, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp6.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp9.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp9.CompleteListing)
require.Equal(t, resp9.EstimatedItemCount, 7)
require.Empty(t, resp9.DeletedIds)
require.Empty(t, resp9.Items)
})
t.Run("simple pagination with aggressive filtering", func(t *testing.T) {
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return t.GetPublicId() == targets[len(targets)-1].GetPublicId(), nil
return t.GetPublicId() == targets[1].GetPublicId() ||
t.GetPublicId() == targets[len(targets)-1].GetPublicId(), nil
}
resp, err := target.List(ctx, []byte("some hash"), 1, filterFunc, repo)
require.NoError(t, err)
require.NotNil(t, resp.RefreshToken)
require.Equal(t, resp.RefreshToken.GrantsHash, []byte("some hash"))
require.True(t, resp.CompleteListing)
require.Equal(t, resp.EstimatedItemCount, 1)
require.NotNil(t, resp.ListToken)
require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash"))
require.False(t, resp.CompleteListing)
require.Equal(t, resp.EstimatedItemCount, 5)
require.Empty(t, resp.DeletedIds)
require.Len(t, resp.Items, 1)
require.Empty(t, cmp.Diff(resp.Items[0], targets[4], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Empty(t, cmp.Diff(resp.Items[0], targets[1], cmpOpts))
resp2, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.RefreshToken, repo)
resp2, err := target.ListPage(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp2.RefreshToken.GrantsHash, []byte("some hash"))
require.NotNil(t, resp2.ListToken)
require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp2.CompleteListing)
// Note: this might be surprising, but there isn't any way for the refresh
// call to know that the last call got a different number.
require.Equal(t, resp2.EstimatedItemCount, 5)
require.Empty(t, resp2.DeletedIds)
require.Empty(t, resp2.Items)
require.Len(t, resp2.Items, 1)
require.Empty(t, cmp.Diff(resp2.Items[0], targets[len(targets)-1], cmpOpts))
// request a refresh, nothing should be returned
resp3, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp3.CompleteListing)
require.Equal(t, resp3.EstimatedItemCount, 5)
require.Empty(t, resp3.DeletedIds)
require.Empty(t, resp3.Items)
// Create some new targets
newT1 := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "proj1-7")
newT2 := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "proj1-8")
newT3 := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "proj1-9")
newT4 := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "proj1-10")
// Run analyze to update target estimate
_, err = sqlDb.ExecContext(ctx, "analyze")
require.NoError(t, err)
t.Cleanup(func() {
_, err = repo.DeleteTarget(ctx, newT1.GetPublicId())
require.NoError(t, err)
_, err = repo.DeleteTarget(ctx, newT2.GetPublicId())
require.NoError(t, err)
_, err = repo.DeleteTarget(ctx, newT3.GetPublicId())
require.NoError(t, err)
_, err = repo.DeleteTarget(ctx, newT4.GetPublicId())
require.NoError(t, err)
// Run analyze to update target estimate
_, err = sqlDb.ExecContext(ctx, "analyze")
require.NoError(t, err)
})
filterFunc = func(_ context.Context, t target.Target) (bool, error) {
return t.GetPublicId() == newT3.GetPublicId() ||
t.GetPublicId() == newT1.GetPublicId(), nil
}
// Refresh again, should get newT3
resp4, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp3.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp4.ListToken.GrantsHash, []byte("some hash"))
require.False(t, resp4.CompleteListing)
require.Equal(t, resp4.EstimatedItemCount, 9)
require.Empty(t, resp4.DeletedIds)
require.Len(t, resp4.Items, 1)
require.Empty(t, cmp.Diff(resp4.Items[0], newT3, cmpOpts))
// Refresh again, should get newT1
resp5, err := target.ListRefreshPage(ctx, []byte("some hash"), 1, filterFunc, resp4.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp5.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp5.CompleteListing)
require.Equal(t, resp5.EstimatedItemCount, 9)
require.Empty(t, resp5.DeletedIds)
require.Len(t, resp5.Items, 1)
require.Empty(t, cmp.Diff(resp5.Items[0], newT1, cmpOpts))
})
t.Run("simple pagination with deletion", func(t *testing.T) {
@@ -262,22 +659,23 @@ func TestService_List(t *testing.T) {
resp, err := target.List(ctx, []byte("some hash"), 1, filterFunc, repo)
require.NoError(t, err)
require.NotNil(t, resp.RefreshToken)
require.Equal(t, resp.RefreshToken.GrantsHash, []byte("some hash"))
require.NotNil(t, resp.ListToken)
require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash"))
require.False(t, resp.CompleteListing)
require.Equal(t, resp.EstimatedItemCount, 4)
require.Empty(t, resp.DeletedIds)
require.Len(t, resp.Items, 1)
require.Empty(t, cmp.Diff(resp.Items[0], targets[0], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Empty(t, cmp.Diff(resp.Items[0], targets[0], cmpOpts))
resp2, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.RefreshToken, repo)
// request remaining results
resp2, err := target.ListPage(ctx, []byte("some hash"), 3, filterFunc, resp.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp2.RefreshToken.GrantsHash, []byte("some hash"))
require.False(t, resp2.CompleteListing)
require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp2.CompleteListing)
require.Equal(t, resp2.EstimatedItemCount, 4)
require.Empty(t, resp2.DeletedIds)
require.Len(t, resp2.Items, 1)
require.Empty(t, cmp.Diff(resp2.Items[0], targets[1], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Len(t, resp2.Items, 3)
require.Empty(t, cmp.Diff(resp2.Items, targets[1:], cmpOpts))
deletedTargetId = targets[0].GetPublicId()
_, err = repo.DeleteTarget(ctx, deletedTargetId)
@@ -288,28 +686,43 @@ func TestService_List(t *testing.T) {
_, err = sqlDb.ExecContext(ctx, "analyze")
require.NoError(t, err)
resp3, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp2.RefreshToken, repo)
// request a refresh, nothing should be returned except the deleted id
resp3, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp2.ListToken, repo)
require.NoError(t, err)
require.Equal(t, resp3.RefreshToken.GrantsHash, []byte("some hash"))
require.False(t, resp3.CompleteListing)
require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp3.CompleteListing)
require.Equal(t, resp3.EstimatedItemCount, 3)
require.Contains(t, resp3.DeletedIds, deletedTargetId)
require.Len(t, resp3.Items, 1)
require.Empty(t, cmp.Diff(resp3.Items[0], targets[1], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.Empty(t, resp3.Items)
})
resp4, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp3.RefreshToken, repo)
t.Run("simple pagination with updated items", func(t *testing.T) {
filterFunc := func(_ context.Context, t target.Target) (bool, error) {
return true, nil
}
resp, err := target.List(ctx, []byte("some hash"), 3, filterFunc, repo)
require.NoError(t, err)
require.Equal(t, resp4.RefreshToken.GrantsHash, []byte("some hash"))
require.True(t, resp4.CompleteListing)
require.Equal(t, resp4.EstimatedItemCount, 3)
require.Len(t, resp4.Items, 1)
require.Empty(t, cmp.Diff(resp4.Items[0], targets[2], cmpopts.IgnoreUnexported(targettest.Target{}, store.Target{}, timestamp.Timestamp{}, timestamppb.Timestamp{})))
require.NotNil(t, resp.ListToken)
require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp.CompleteListing)
require.Equal(t, resp.EstimatedItemCount, 3)
require.Empty(t, resp.DeletedIds)
require.Len(t, resp.Items, 3)
require.Empty(t, cmp.Diff(resp.Items, targets, cmpOpts))
resp5, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp4.RefreshToken, repo)
// create a new target and run analyze to update target estimate
newTarget := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "proj1-new")
_, err = sqlDb.ExecContext(ctx, "analyze")
require.NoError(t, err)
require.Equal(t, resp5.RefreshToken.GrantsHash, []byte("some hash"))
require.True(t, resp5.CompleteListing)
require.Equal(t, resp5.EstimatedItemCount, 3)
require.Empty(t, resp5.Items)
resp2, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo)
require.NoError(t, err)
require.NotNil(t, resp2.ListToken)
require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash"))
require.True(t, resp2.CompleteListing)
require.Equal(t, resp2.EstimatedItemCount, 4)
require.Empty(t, resp2.DeletedIds)
require.Len(t, resp2.Items, 1)
require.Empty(t, cmp.Diff(resp2.Items[0], newTarget, cmpOpts))
})
}
+64
View File
@@ -0,0 +1,64 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1
package target
import (
"context"
"time"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/listtoken"
"github.com/hashicorp/boundary/internal/pagination"
)
// ListPage lists up to page size targets, filtering out entries that
// do not pass the filter item function. It will automatically request
// more targets from the database, at page size chunks, to fill the page.
// It will start its paging based on the information in the token.
// It returns a new list token used to continue pagination or refresh items.
// Targets are ordered by create time descending (most recently created first).
func ListPage(
ctx context.Context,
grantsHash []byte,
pageSize int,
filterItemFn pagination.ListFilterFunc[Target],
tok *listtoken.Token,
repo *Repository,
) (*pagination.ListResponse[Target], error) {
const op = "target.ListPage"
switch {
case len(grantsHash) == 0:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
case pageSize < 1:
return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
case filterItemFn == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
case tok == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token")
case repo == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo")
}
if _, ok := tok.Subtype.(*listtoken.PaginationToken); !ok {
return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a pagination token component")
}
listItemsFn := func(ctx context.Context, lastPageItem Target, limit int) ([]Target, time.Time, error) {
opts := []Option{
WithLimit(limit),
}
if lastPageItem != nil {
opts = append(opts, WithStartPageAfterItem(lastPageItem))
} else {
lastItem, err := tok.LastItem(ctx)
if err != nil {
return nil, time.Time{}, err
}
opts = append(opts, WithStartPageAfterItem(lastItem))
}
return repo.listTargets(ctx, opts...)
}
return pagination.ListPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount, tok)
}
+34 -26
View File
@@ -5,58 +5,66 @@ package target
import (
"context"
"time"
"github.com/hashicorp/boundary/globals"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/listtoken"
"github.com/hashicorp/boundary/internal/pagination"
"github.com/hashicorp/boundary/internal/refreshtoken"
)
// ListRefresh lists targets according to the page size
// and refresh token, filtering out entries that do not
// pass the filter item fn. It returns a new refresh token
// based on the old one, the grants hash, and the returned
// targets.
// ListRefresh lists up to page size targets, filtering out entries that
// do not pass the filter item function. It will automatically request
// more targets from the database, at page size chunks, to fill the page.
// It will start its paging based on the information in the token.
// It returns a new list token used to continue pagination or refresh items.
// Targets are ordered by update time descending (most recently updated first).
// Targets may contain items that were already returned during the initial
// pagination phase. It also returns a list of any targets deleted since the
// start of the initial pagination phase or last response.
func ListRefresh(
ctx context.Context,
grantsHash []byte,
pageSize int,
filterItemFn pagination.ListFilterFunc[Target],
tok *refreshtoken.Token,
tok *listtoken.Token,
repo *Repository,
) (*pagination.ListResponse[Target], error) {
const op = "target.ListRefresh"
if len(grantsHash) == 0 {
switch {
case len(grantsHash) == 0:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
}
if pageSize < 1 {
case pageSize < 1:
return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
}
if filterItemFn == nil {
case filterItemFn == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
}
if tok == nil {
case tok == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token")
}
if repo == nil {
case repo == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo")
}
rt, ok := tok.Subtype.(*listtoken.StartRefreshToken)
if !ok {
return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a start-refresh token component")
}
listItemsFn := func(ctx context.Context, tok *refreshtoken.Token, lastPageItem Target, limit int) ([]Target, error) {
listItemsFn := func(ctx context.Context, lastPageItem Target, limit int) ([]Target, time.Time, error) {
opts := []Option{
WithLimit(limit),
}
if lastPageItem != nil {
opts = append(opts,
WithStartPageAfterItem(lastPageItem),
)
} else {
opts = append(opts,
WithStartPageAfterItem(tok.LastItem()),
)
opts = append(opts, WithStartPageAfterItem(lastPageItem))
}
return repo.listTargets(ctx, opts...)
// Add the database read timeout to account for any creations missed due to concurrent
// transactions in the initial pagination phase.
return repo.listTargetsRefresh(ctx, rt.PreviousPhaseUpperBound.Add(-globals.RefreshReadLookbackDuration), opts...)
}
listDeletedIdsFn := func(ctx context.Context, since time.Time) ([]string, time.Time, error) {
// Add the database read timeout to account for any deletions missed due to concurrent
// transactions in previous requests.
return repo.listDeletedIds(ctx, since.Add(-globals.RefreshReadLookbackDuration))
}
return pagination.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount, repo.listDeletedIds, tok)
return pagination.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount, listDeletedIdsFn, tok)
}
@@ -0,0 +1,77 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1
package target
import (
"context"
"time"
"github.com/hashicorp/boundary/globals"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/listtoken"
"github.com/hashicorp/boundary/internal/pagination"
)
// ListRefreshPage lists up to page size targets, filtering out entries that
// do not pass the filter item function. It will automatically request
// more targets from the database, at page size chunks, to fill the page.
// It will start its paging based on the information in the token.
// It returns a new list token used to continue pagination or refresh items.
// Targets are ordered by update time descending (most recently updated first).
// Targets may contain items that were already returned during the initial
// pagination phase. It also returns a list of any targets deleted since the
// last response.
func ListRefreshPage(
ctx context.Context,
grantsHash []byte,
pageSize int,
filterItemFn pagination.ListFilterFunc[Target],
tok *listtoken.Token,
repo *Repository,
) (*pagination.ListResponse[Target], error) {
const op = "target.ListRefreshPage"
switch {
case len(grantsHash) == 0:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
case pageSize < 1:
return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
case filterItemFn == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
case tok == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token")
case repo == nil:
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo")
}
rt, ok := tok.Subtype.(*listtoken.RefreshToken)
if !ok {
return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a refresh token component")
}
listItemsFn := func(ctx context.Context, lastPageItem Target, limit int) ([]Target, time.Time, error) {
opts := []Option{
WithLimit(limit),
}
if lastPageItem != nil {
opts = append(opts, WithStartPageAfterItem(lastPageItem))
} else {
lastItem, err := tok.LastItem(ctx)
if err != nil {
return nil, time.Time{}, err
}
opts = append(opts, WithStartPageAfterItem(lastItem))
}
// Add the database read timeout to account for any creations missed due to concurrent
// transactions in the original list pagination phase.
return repo.listTargetsRefresh(ctx, rt.PhaseLowerBound.Add(-globals.RefreshReadLookbackDuration), opts...)
}
listDeletedIdsFn := func(ctx context.Context, since time.Time) ([]string, time.Time, error) {
// Add the database read timeout to account for any deletes missed due to concurrent
// transactions in the original list pagination phase.
return repo.listDeletedIds(ctx, since.Add(-globals.RefreshReadLookbackDuration))
}
return pagination.ListRefreshPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount, listDeletedIdsFn, tok)
}
+10 -3
View File
@@ -76,10 +76,17 @@ func TestTargetAddress(t testing.TB, conn *db.DB, targetId, address string) *Add
}
// TestListTargets exposes the repo listTargets method for testing purposes.
func TestListTargets(t testing.TB, repo *Repository, ctx context.Context, opts ...Option) []Target {
targets, err := repo.listTargets(ctx, opts...)
func TestListTargets(t testing.TB, repo *Repository, ctx context.Context, opts ...Option) ([]Target, time.Time) {
targets, ttime, err := repo.listTargets(ctx, opts...)
require.NoError(t, err)
return targets
return targets, ttime
}
// TestListTargetsRefresh exposes the repo listTargetsRefresh method for testing purposes.
func TestListTargetsRefresh(t testing.TB, repo *Repository, ctx context.Context, updateAfter time.Time, opts ...Option) ([]Target, time.Time) {
targets, ttime, err := repo.listTargetsRefresh(ctx, updateAfter, opts...)
require.NoError(t, err)
return targets, ttime
}
// TestListDeletedIds exposes the repo listDeletedIds method for testing purposes.