diff --git a/internal/target/query.go b/internal/target/query.go index cdac831df..4b77c73ac 100644 --- a/internal/target/query.go +++ b/internal/target/query.go @@ -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; ` ) diff --git a/internal/target/repository.go b/internal/target/repository.go index d083b3b75..e0015a278 100644 --- a/internal/target/repository.go +++ b/internal/target/repository.go @@ -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")) diff --git a/internal/target/repository_ext_test.go b/internal/target/repository_ext_test.go index 7c6702cfe..53a5004c1 100644 --- a/internal/target/repository_ext_test.go +++ b/internal/target/repository_ext_test.go @@ -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()) diff --git a/internal/target/repository_test.go b/internal/target/repository_test.go index 77984a0a6..4663fae12 100644 --- a/internal/target/repository_test.go +++ b/internal/target/repository_test.go @@ -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") + }) } diff --git a/internal/target/service_list.go b/internal/target/service_list.go index aaefff276..4ad17779c 100644 --- a/internal/target/service_list.go +++ b/internal/target/service_list.go @@ -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...) } diff --git a/internal/target/service_list_ext_test.go b/internal/target/service_list_ext_test.go index 25457652c..5416cf5ea 100644 --- a/internal/target/service_list_ext_test.go +++ b/internal/target/service_list_ext_test.go @@ -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)) }) } diff --git a/internal/target/service_list_page.go b/internal/target/service_list_page.go new file mode 100644 index 000000000..4e5d900b4 --- /dev/null +++ b/internal/target/service_list_page.go @@ -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) +} diff --git a/internal/target/service_list_refresh.go b/internal/target/service_list_refresh.go index 3f3ddd149..60afac82a 100644 --- a/internal/target/service_list_refresh.go +++ b/internal/target/service_list_refresh.go @@ -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) } diff --git a/internal/target/service_list_refresh_page.go b/internal/target/service_list_refresh_page.go new file mode 100644 index 000000000..71a92b345 --- /dev/null +++ b/internal/target/service_list_refresh_page.go @@ -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) +} diff --git a/internal/target/testing.go b/internal/target/testing.go index e3e5dca19..ae299afd6 100644 --- a/internal/target/testing.go +++ b/internal/target/testing.go @@ -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.