target: add pagination support

Adds pagination support to the target domain

internal/target/options.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/hashicorp/boundary/globals"
 	intglobals "github.com/hashicorp/boundary/internal/globals"
+	"github.com/hashicorp/boundary/internal/pagination"
 	"github.com/hashicorp/boundary/internal/perms"
 )
 
@@ -52,6 +53,7 @@ type options struct {
 	WithStorageBucketId        string
 	WithEnableSessionRecording bool
 	WithNetResolver            intglobals.NetIpResolver
+	WithStartPageAfterItem     pagination.Item
 }
 
 func getDefaultOptions() options {
@@ -265,3 +267,11 @@ func WithNetResolver(resolver intglobals.NetIpResolver) Option {
 		o.WithNetResolver = resolver
 	}
 }
+
+// WithStartPageAfterItem is used to paginate over the results.
+// The next page will start after the provided item.
+func WithStartPageAfterItem(item pagination.Item) Option {
+	return func(o *options) {
+		o.WithStartPageAfterItem = item
+	}
+}

internal/target/options_test.go

@@ -5,14 +5,31 @@ package target
 
 import (
 	"testing"
+	"time"
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/credential"
+	"github.com/hashicorp/boundary/internal/db/timestamp"
+	"github.com/hashicorp/boundary/internal/pagination"
 	"github.com/hashicorp/boundary/internal/perms"
 	"github.com/hashicorp/boundary/internal/target/store"
 	"github.com/stretchr/testify/assert"
 )
 
+type fakeItem struct {
+	pagination.Item
+	publicId   string
+	updateTime time.Time
+}
+
+func (p *fakeItem) GetPublicId() string {
+	return p.publicId
+}
+
+func (p *fakeItem) GetUpdateTime() *timestamp.Timestamp {
+	return timestamp.New(p.updateTime)
+}
+
 // Test_GetOpts provides unit tests for GetOpts and all the options
 func Test_GetOpts(t *testing.T) {
 	t.Parallel()
@@ -233,4 +250,11 @@ func Test_GetOpts(t *testing.T) {
 		testOpts.WithEnableSessionRecording = true
 		assert.Equal(opts, testOpts)
 	})
+	t.Run("WithStartPageAfterItem", func(t *testing.T) {
+		assert := assert.New(t)
+		updateTime := time.Now()
+		opts := GetOpts(WithStartPageAfterItem(&fakeItem{nil, "s_1", updateTime}))
+		assert.Equal(opts.WithStartPageAfterItem.GetPublicId(), "s_1")
+		assert.Equal(opts.WithStartPageAfterItem.GetUpdateTime(), timestamp.New(updateTime))
+	})
 }

internal/target/query.go

@@ -57,5 +57,9 @@ order by action, source_id;
 select public_id, project_id from target
 %s
 ;
+`
+
+	estimateCountTargets = `
+select sum(reltuples::bigint) as estimate from pg_class where oid in ('target_tcp'::regclass, 'target_ssh'::regclass)
 `
 )

internal/target/repository.go

@@ -8,6 +8,7 @@ import (
 	"database/sql"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/hashicorp/boundary/internal/boundary"
 	"github.com/hashicorp/boundary/internal/db"
@@ -227,12 +228,14 @@ 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 in a project based on the data in the WithPermissions option
 // provided to the Repository constructor. If no permissions are available, this function
 // is a no-op.
-// Supports WithLimit which overrides the limit set in the Repository object.
-func (r *Repository) ListTargets(ctx context.Context, opt ...Option) ([]Target, error) {
-	const op = "target.(Repository).ListTargets"
+// 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"
 
 	if len(r.permissions) == 0 {
 		return []Target{}, nil
@@ -242,19 +245,39 @@ func (r *Repository) ListTargets(ctx context.Context, opt ...Option) ([]Target,
 	if len(where) == 0 {
 		return []Target{}, nil
 	}
-
+	whereClause := 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"
+	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.
+		args = append(args,
+			sql.Named("after_item_update_time", opts.WithStartPageAfterItem.GetUpdateTime()),
+			sql.Named("after_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,
+	err := r.reader.SearchWhere(
+		ctx,
 		&foundTargets,
-		strings.Join(where, " or "),
+		whereClause,
 		args,
 		db.WithLimit(limit),
+		db.WithOrder(withOrder),
 	)
 	if err != nil {
 		return nil, errors.Wrap(ctx, err, op)
@@ -328,6 +351,48 @@ func (r *Repository) listPermissionWhereClauses() ([]string, []any) {
 	return where, args
 }
 
+// listDeletedIds lists the public IDs of any targets deleted since the timestamp provided,
+// and the timestamp of the transaction within which the targets were listed.
+func (r *Repository) listDeletedIds(ctx context.Context, since time.Time) ([]string, time.Time, error) {
+	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"))
+		}
+		var err error
+		transactionTimestamp, err = r.Now(ctx)
+		if err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("failed to get transaction timestamp"))
+		}
+		return nil
+	}); err != nil {
+		return nil, time.Time{}, err
+	}
+	var targetIds []string
+	for _, t := range deleteTargets {
+		targetIds = append(targetIds, t.PublicId)
+	}
+	return targetIds, transactionTimestamp, nil
+}
+
+// estimatedCount returns an estimate of the total number of items across all targets.
+func (r *Repository) estimatedCount(ctx context.Context) (int, error) {
+	const op = "target.(Repository).estimatedCount"
+	rows, err := r.reader.Query(ctx, estimateCountTargets, nil)
+	if err != nil {
+		return 0, errors.Wrap(ctx, err, op, errors.WithMsg("failed to query total targets"))
+	}
+	var count int
+	for rows.Next() {
+		if err := r.reader.ScanRows(ctx, rows, &count); err != nil {
+			return 0, errors.Wrap(ctx, err, op, errors.WithMsg("failed to query total targets"))
+		}
+	}
+	return count, nil
+}
+
 // DeleteTarget will delete a target from the repository.
 func (r *Repository) DeleteTarget(ctx context.Context, publicId string, _ ...Option) (int, error) {
 	const op = "target.(Repository).DeleteTarget"

internal/target/repository_ext_test.go

@@ -5,7 +5,9 @@ package target_test
 
 import (
 	"context"
+	"fmt"
 	"testing"
+	"time"
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/credential"
@@ -15,13 +17,20 @@ import (
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/iam"
 	"github.com/hashicorp/boundary/internal/kms"
+	"github.com/hashicorp/boundary/internal/perms"
 	"github.com/hashicorp/boundary/internal/target"
 	"github.com/hashicorp/boundary/internal/target/store"
 	"github.com/hashicorp/boundary/internal/target/targettest"
+	"github.com/hashicorp/boundary/internal/types/action"
+	"github.com/hashicorp/boundary/internal/types/resource"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func init() {
+	target.Register(targettest.Subtype, hooks{}, globals.TcpTargetPrefix)
+}
+
 type hooks struct{}
 
 func (h hooks) NewTarget(ctx context.Context, projectId string, opt ...target.Option) (target.Target, error) {
@@ -44,9 +53,302 @@ func (h hooks) VetCredentialSources(ctx context.Context, cls []*target.Credentia
 	return targettest.VetCredentialSources(ctx, cls, creds)
 }
 
+func TestListDeletedIds(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	testKms := kms.TestKms(t, conn, wrapper)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	_, proj1 := iam.TestScopes(t, iamRepo)
+
+	rw := db.New(conn)
+	repo, err := target.NewRepository(ctx, rw, rw, testKms)
+	require.NoError(t, err)
+
+	// Expect no entries at the start
+	deletedIds, ttime := target.TestListDeletedIds(t, repo, ctx, time.Now().AddDate(-1, 0, 0))
+	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)))
+
+	// Delete a session
+	tg := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "deleteme")
+	_, err = repo.DeleteTarget(ctx, tg.GetPublicId())
+	require.NoError(t, err)
+
+	// Expect a single entry
+	deletedIds, ttime = target.TestListDeletedIds(t, repo, ctx, time.Now().AddDate(-1, 0, 0))
+	require.Equal(t, []string{tg.GetPublicId()}, deletedIds)
+	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)
+	assert.True(t, time.Now().Before(ttime.Add(10*time.Second)))
+	assert.True(t, time.Now().After(ttime.Add(-10*time.Second)))
+}
+
+func TestEstimatedCount(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	sqlDb, err := conn.SqlDB(ctx)
+	require.NoError(t, err)
+	wrapper := db.TestWrapper(t)
+	testKms := kms.TestKms(t, conn, wrapper)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	_, proj1 := iam.TestScopes(t, iamRepo)
+
+	rw := db.New(conn)
+	repo, err := target.NewRepository(ctx, rw, rw, testKms)
+	require.NoError(t, err)
+
+	// Check total entries at start, expect 0
+	numItems := target.TestEstimatedCount(t, repo, ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 0, numItems)
+
+	// Create a target, expect 1 entries
+	tg := targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), "target1")
+	// Run analyze to update estimate
+	_, err = sqlDb.ExecContext(ctx, "analyze")
+	require.NoError(t, err)
+	numItems = target.TestEstimatedCount(t, repo, ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 1, numItems)
+
+	// Delete the target, expect 0 again
+	_, err = repo.DeleteTarget(ctx, tg.GetPublicId())
+	require.NoError(t, err)
+	// Run analyze to update estimate
+	_, err = sqlDb.ExecContext(ctx, "analyze")
+	require.NoError(t, err)
+	numItems = target.TestEstimatedCount(t, repo, ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 0, numItems)
+}
+
+func TestRepository_ListTargets(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	testKms := kms.TestKms(t, conn, wrapper)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	_, proj1 := iam.TestScopes(t, iamRepo)
+	_, proj2 := iam.TestScopes(t, iamRepo)
+
+	var total int
+	for i := 0; i < 5; i++ {
+		targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), fmt.Sprintf("proj1-%d", i))
+		total++
+		targettest.TestNewTestTarget(ctx, t, conn, proj2.GetPublicId(), fmt.Sprintf("proj2-%d", i))
+		total++
+	}
+
+	rw := db.New(conn)
+	repo, err := target.NewRepository(ctx, rw, rw, testKms,
+		target.WithPermissions([]perms.Permission{
+			{
+				ScopeId:  proj1.PublicId,
+				Resource: resource.Target,
+				Action:   action.List,
+				All:      true,
+			},
+			{
+				ScopeId:  proj2.PublicId,
+				Resource: resource.Target,
+				Action:   action.List,
+				All:      true,
+			},
+		}),
+	)
+	require.NoError(t, err)
+
+	t.Run("no-options", func(t *testing.T) {
+		got := target.TestListTargets(t, repo, ctx)
+		require.NoError(t, err)
+		assert.Equal(t, total, len(got))
+	})
+
+	t.Run("withStartPageAfter", func(t *testing.T) {
+		assert, require := assert.New(t), require.New(t)
+
+		page1 := target.TestListTargets(
+			t,
+			repo,
+			context.Background(),
+			target.WithLimit(2),
+		)
+		require.NoError(err)
+		require.Len(page1, 2)
+		page2 := target.TestListTargets(
+			t,
+			repo,
+			context.Background(),
+			target.WithLimit(2),
+			target.WithStartPageAfterItem(page1[1]),
+		)
+		require.NoError(err)
+		require.Len(page2, 2)
+		for _, item := range page1 {
+			assert.NotEqual(item.GetPublicId(), page2[0].GetPublicId())
+			assert.NotEqual(item.GetPublicId(), page2[1].GetPublicId())
+		}
+		page3 := target.TestListTargets(
+			t,
+			repo,
+			context.Background(),
+			target.WithLimit(2),
+			target.WithStartPageAfterItem(page2[1]),
+		)
+		require.NoError(err)
+		require.Len(page3, 2)
+		for _, item := range page2 {
+			assert.NotEqual(item.GetPublicId(), page3[0].GetPublicId())
+			assert.NotEqual(item.GetPublicId(), page3[1].GetPublicId())
+		}
+		page4 := target.TestListTargets(
+			t,
+			repo,
+			context.Background(),
+			target.WithLimit(2),
+			target.WithStartPageAfterItem(page3[1]),
+		)
+		require.NoError(err)
+		require.Len(page4, 2)
+		for _, item := range page3 {
+			assert.NotEqual(item.GetPublicId(), page4[0].GetPublicId())
+			assert.NotEqual(item.GetPublicId(), page4[1].GetPublicId())
+		}
+		page5 := target.TestListTargets(
+			t,
+			repo,
+			context.Background(),
+			target.WithLimit(2),
+			target.WithStartPageAfterItem(page4[1]),
+		)
+		require.NoError(err)
+		require.Len(page5, 2)
+		for _, item := range page4 {
+			assert.NotEqual(item.GetPublicId(), page5[0].GetPublicId())
+			assert.NotEqual(item.GetPublicId(), page5[1].GetPublicId())
+		}
+		page6 := target.TestListTargets(
+			t,
+			repo,
+			context.Background(),
+			target.WithLimit(2),
+			target.WithStartPageAfterItem(page5[1]),
+		)
+		require.NoError(err)
+		require.Empty(page6)
+
+		// Update the first target and check that it gets listed subsequently
+		page1[0].SetName("new-name")
+		_, _, err = repo.UpdateTarget(ctx, page1[0], page1[0].GetVersion(), []string{"name"})
+		require.NoError(err)
+		page7 := target.TestListTargets(
+			t,
+			repo,
+			context.Background(),
+			target.WithLimit(2),
+			target.WithStartPageAfterItem(page5[1]),
+		)
+		require.NoError(err)
+		require.Len(page7, 1)
+		require.Equal(page7[0].GetPublicId(), page1[0].GetPublicId())
+	})
+}
+
+func TestRepository_ListTargets_Multiple_Scopes(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	testKms := kms.TestKms(t, conn, wrapper)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+
+	_, proj1 := iam.TestScopes(t, iamRepo)
+	_, proj2 := iam.TestScopes(t, iamRepo)
+
+	const numPerScope = 10
+	var total int
+	for i := 0; i < numPerScope; i++ {
+		targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), fmt.Sprintf("proj1-%d", i))
+		total++
+		targettest.TestNewTestTarget(ctx, t, conn, proj2.GetPublicId(), fmt.Sprintf("proj2-%d", i))
+		total++
+	}
+
+	rw := db.New(conn)
+	repo, err := target.NewRepository(ctx, rw, rw, testKms,
+		target.WithPermissions([]perms.Permission{
+			{
+				ScopeId:  proj1.PublicId,
+				Resource: resource.Target,
+				Action:   action.List,
+				All:      true,
+			},
+			{
+				ScopeId:  proj2.PublicId,
+				Resource: resource.Target,
+				Action:   action.List,
+				All:      true,
+			},
+		}),
+	)
+	require.NoError(t, err)
+
+	got := target.TestListTargets(t, repo, ctx)
+	require.NoError(t, err)
+	assert.Equal(t, total, len(got))
+}
+
+func TestRepository_ListRoles_Above_Default_Count(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	testKms := kms.TestKms(t, conn, wrapper)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+
+	_, proj := iam.TestScopes(t, iamRepo)
+
+	numToCreate := db.DefaultLimit + 5
+	var total int
+	for i := 0; i < numToCreate; i++ {
+		targettest.TestNewTestTarget(ctx, t, conn, proj.GetPublicId(), fmt.Sprintf("proj1-%d", i), target.WithAddress("1.2.3.4"))
+		total++
+	}
+	require.Equal(t, numToCreate, total)
+
+	rw := db.New(conn)
+	repo, err := target.NewRepository(ctx, rw, rw, testKms,
+		target.WithPermissions([]perms.Permission{
+			{
+				ScopeId:  proj.PublicId,
+				Resource: resource.Target,
+				Action:   action.List,
+				All:      true,
+			},
+		}))
+	require.NoError(t, err)
+
+	got := target.TestListTargets(t, repo, ctx, target.WithLimit(numToCreate))
+	require.NoError(t, err)
+	assert.Equal(t, total, len(got))
+
+	for _, tar := range got {
+		assert.Equal(t, "1.2.3.4", tar.GetAddress())
+	}
+}
+
 func TestRepository_SetTargetCredentialSources(t *testing.T) {
 	ctx := context.Background()
-	target.Register(targettest.Subtype, hooks{}, globals.TcpTargetPrefix)
 
 	t.Parallel()
 	conn, _ := db.TestSetup(t, "postgres")

internal/target/service_list.go

@@ -0,0 +1,52 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package target
+
+import (
+	"context"
+
+	"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.
+func List(
+	ctx context.Context,
+	grantsHash []byte,
+	pageSize int,
+	filterItemFn pagination.ListFilterFunc[Target],
+	repo *Repository,
+) (*pagination.ListResponse[Target], error) {
+	const op = "target.List"
+
+	if len(grantsHash) == 0 {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
+	}
+	if pageSize < 1 {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
+	}
+	if filterItemFn == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
+	}
+	if repo == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo")
+	}
+
+	listItemsFn := func(ctx context.Context, lastPageItem Target, limit int) ([]Target, error) {
+		opts := []Option{
+			WithLimit(limit),
+		}
+		if lastPageItem != nil {
+			opts = append(opts,
+				WithStartPageAfterItem(lastPageItem),
+			)
+		}
+		return repo.listTargets(ctx, opts...)
+	}
+
+	return pagination.List(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount)
+}

internal/target/service_list_ext_test.go

@@ -0,0 +1,315 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package target_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"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/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"
+	"github.com/hashicorp/boundary/internal/types/action"
+	"github.com/hashicorp/boundary/internal/types/resource"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+// Note: this file imports the targettest package. The only
+// reason this doesn't cause an import cycle is because this is
+// an "external" test package.
+func TestService_List(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	sqlDb, err := conn.SqlDB(ctx)
+	require.NoError(t, err)
+	wrapper := db.TestWrapper(t)
+	testKms := kms.TestKms(t, conn, wrapper)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	_, proj1 := iam.TestScopes(t, iamRepo)
+
+	total := 5
+	var targets []target.Target
+	for i := 0; i < total; i++ {
+		targets = append(targets, targettest.TestNewTestTarget(ctx, t, conn, proj1.GetPublicId(), fmt.Sprintf("proj1-%d", i)))
+	}
+
+	// Run analyze to update target estimate
+	_, err = sqlDb.ExecContext(ctx, "analyze")
+	require.NoError(t, err)
+
+	rw := db.New(conn)
+	repo, err := target.NewRepository(ctx, rw, rw, testKms,
+		target.WithPermissions([]perms.Permission{
+			{
+				ScopeId:  proj1.PublicId,
+				Resource: resource.Target,
+				Action:   action.List,
+				All:      true,
+			},
+		}),
+	)
+	require.NoError(t, err)
+
+	t.Run("List 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
+			}
+			_, err := target.List(ctx, nil, 1, filterFunc, 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
+			}
+			_, err := target.List(ctx, []byte("some hash"), 0, filterFunc, 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
+			}
+			_, err := target.List(ctx, []byte("some hash"), -1, filterFunc, repo)
+			require.ErrorContains(t, err, "page size must be at least 1")
+		})
+		t.Run("nil filter func", func(t *testing.T) {
+			t.Parallel()
+			_, err := target.List(ctx, []byte("some hash"), 1, nil, repo)
+			require.ErrorContains(t, err, "missing filter item callback")
+		})
+		t.Run("nil repo", func(t *testing.T) {
+			t.Parallel()
+			filterFunc := func(_ context.Context, t target.Target) (bool, error) {
+				return true, nil
+			}
+			_, err := target.List(ctx, []byte("some hash"), 1, filterFunc, 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 := refreshtoken.New(ctx, time.Now(), time.Now(), resource.Target, []byte("some hash"), "some-id", time.Now())
+			require.NoError(t, err)
+			_, err = target.ListRefresh(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 := refreshtoken.New(ctx, time.Now(), time.Now(), resource.Target, []byte("some hash"), "some-id", time.Now())
+			require.NoError(t, err)
+			_, err = target.ListRefresh(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 := refreshtoken.New(ctx, time.Now(), time.Now(), resource.Target, []byte("some hash"), "some-id", time.Now())
+			require.NoError(t, err)
+			_, err = target.ListRefresh(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())
+			require.NoError(t, err)
+			_, err = target.ListRefresh(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.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo)
+			require.ErrorContains(t, err, "missing token")
+		})
+		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())
+			require.NoError(t, err)
+			_, err = target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, nil)
+			require.ErrorContains(t, err, "missing repo")
+		})
+	})
+
+	t.Run("simple pagination", func(t *testing.T) {
+		filterFunc := func(_ context.Context, t target.Target) (bool, error) {
+			return true, 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.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{})))
+
+		resp2, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.RefreshToken, repo)
+		require.NoError(t, err)
+		require.Equal(t, resp2.RefreshToken.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{})))
+
+		resp3, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp2.RefreshToken, repo)
+		require.NoError(t, err)
+		require.Equal(t, resp3.RefreshToken.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{})))
+
+		resp4, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp3.RefreshToken, repo)
+		require.NoError(t, err)
+		require.Equal(t, resp4.RefreshToken.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{})))
+
+		resp5, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp4.RefreshToken, repo)
+		require.NoError(t, err)
+		require.Equal(t, resp5.RefreshToken.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{})))
+
+		resp6, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp5.RefreshToken, repo)
+		require.NoError(t, err)
+		require.Equal(t, resp6.RefreshToken.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)
+	})
+
+	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
+		}
+		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.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{})))
+
+		resp2, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.RefreshToken, repo)
+		require.NoError(t, err)
+		require.Equal(t, resp2.RefreshToken.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)
+	})
+
+	t.Run("simple pagination with deletion", func(t *testing.T) {
+		filterFunc := func(_ context.Context, t target.Target) (bool, error) {
+			return true, nil
+		}
+		deletedTargetId := targets[0].GetPublicId()
+		_, err := repo.DeleteTarget(ctx, deletedTargetId)
+		require.NoError(t, err)
+		targets = targets[1:]
+
+		// Run analyze to update target estimate
+		_, err = sqlDb.ExecContext(ctx, "analyze")
+		require.NoError(t, err)
+
+		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.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{})))
+
+		resp2, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.RefreshToken, repo)
+		require.NoError(t, err)
+		require.Equal(t, resp2.RefreshToken.GrantsHash, []byte("some hash"))
+		require.False(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{})))
+
+		deletedTargetId = targets[0].GetPublicId()
+		_, err = repo.DeleteTarget(ctx, deletedTargetId)
+		require.NoError(t, err)
+		targets = targets[1:]
+
+		// Run analyze to update target estimate
+		_, err = sqlDb.ExecContext(ctx, "analyze")
+		require.NoError(t, err)
+
+		resp3, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp2.RefreshToken, repo)
+		require.NoError(t, err)
+		require.Equal(t, resp3.RefreshToken.GrantsHash, []byte("some hash"))
+		require.False(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{})))
+
+		resp4, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp3.RefreshToken, 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{})))
+
+		resp5, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, resp4.RefreshToken, repo)
+		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)
+	})
+}

internal/target/service_list_refresh.go

@@ -0,0 +1,62 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package target
+
+import (
+	"context"
+
+	"github.com/hashicorp/boundary/internal/errors"
+	"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.
+func ListRefresh(
+	ctx context.Context,
+	grantsHash []byte,
+	pageSize int,
+	filterItemFn pagination.ListFilterFunc[Target],
+	tok *refreshtoken.Token,
+	repo *Repository,
+) (*pagination.ListResponse[Target], error) {
+	const op = "target.ListRefresh"
+
+	if len(grantsHash) == 0 {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
+	}
+	if pageSize < 1 {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
+	}
+	if filterItemFn == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
+	}
+	if tok == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token")
+	}
+	if repo == nil {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo")
+	}
+
+	listItemsFn := func(ctx context.Context, tok *refreshtoken.Token, lastPageItem Target, limit int) ([]Target, error) {
+		opts := []Option{
+			WithLimit(limit),
+		}
+		if lastPageItem != nil {
+			opts = append(opts,
+				WithStartPageAfterItem(lastPageItem),
+			)
+		} else {
+			opts = append(opts,
+				WithStartPageAfterItem(tok.LastItem()),
+			)
+		}
+		return repo.listTargets(ctx, opts...)
+	}
+
+	return pagination.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedCount, repo.listDeletedIds, tok)
+}

internal/target/target.go

@@ -189,3 +189,13 @@ func (t *targetView) targetSubtype(ctx context.Context, address string) (Target,
 	tt.SetStorageBucketId(t.StorageBucketId)
 	return tt, nil
 }
+
+type deletedTarget struct {
+	PublicId   string `gorm:"primary_key"`
+	DeleteTime *timestamp.Timestamp
+}
+
+// TableName returns the tablename to override the default gorm table name
+func (s *deletedTarget) TableName() string {
+	return "target_all_subtypes_deleted_view"
+}

internal/target/targettest/target.go

@@ -329,6 +329,7 @@ func New(ctx context.Context, projectId string, opt ...target.Option) (target.Ta
 			EgressWorkerFilter:     opts.WithEgressWorkerFilter,
 			IngressWorkerFilter:    opts.WithIngressWorkerFilter,
 		},
+		Address: opts.WithAddress,
 	}
 	return t, nil
 }
@@ -367,5 +368,10 @@ func TestNewTestTarget(ctx context.Context, t *testing.T, conn *db.DB, projectId
 		err := rw.CreateItems(context.Background(), newCredLibs)
 		require.NoError(err)
 	}
+	if len(opts.WithAddress) != 0 {
+		addr := target.TestNewTargetAddress(id, opts.WithAddress)
+		err := rw.Create(ctx, addr)
+		require.NoError(err)
+	}
 	return tar
 }

internal/target/tcp/repository_test.go

@@ -5,7 +5,6 @@ package tcp_test
 
 import (
 	"context"
-	"fmt"
 	"testing"
 	"time"
 
@@ -15,11 +14,8 @@ import (
 	"github.com/hashicorp/boundary/internal/iam"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/oplog"
-	"github.com/hashicorp/boundary/internal/perms"
 	"github.com/hashicorp/boundary/internal/target"
 	"github.com/hashicorp/boundary/internal/target/tcp"
-	"github.com/hashicorp/boundary/internal/types/action"
-	"github.com/hashicorp/boundary/internal/types/resource"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -141,52 +137,6 @@ func TestRepository_LookupTarget(t *testing.T) {
 	}
 }
 
-func TestRepository_ListRoles_Multiple_Scopes(t *testing.T) {
-	t.Parallel()
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	wrapper := db.TestWrapper(t)
-	testKms := kms.TestKms(t, conn, wrapper)
-	iamRepo := iam.TestRepo(t, conn, wrapper)
-
-	_, proj1 := iam.TestScopes(t, iamRepo)
-	_, proj2 := iam.TestScopes(t, iamRepo)
-
-	db.TestDeleteWhere(t, conn, tcp.NewTestTarget(ctx, ""), "1=1")
-
-	const numPerScope = 10
-	var total int
-	for i := 0; i < numPerScope; i++ {
-		tcp.TestTarget(ctx, t, conn, proj1.GetPublicId(), fmt.Sprintf("proj1-%d", i))
-		total++
-		tcp.TestTarget(ctx, t, conn, proj2.GetPublicId(), fmt.Sprintf("proj2-%d", i))
-		total++
-	}
-
-	rw := db.New(conn)
-	repo, err := target.NewRepository(ctx, rw, rw, testKms,
-		target.WithPermissions([]perms.Permission{
-			{
-				ScopeId:  proj1.PublicId,
-				Resource: resource.Target,
-				Action:   action.List,
-				All:      true,
-			},
-			{
-				ScopeId:  proj2.PublicId,
-				Resource: resource.Target,
-				Action:   action.List,
-				All:      true,
-			},
-		}),
-	)
-	require.NoError(t, err)
-
-	got, err := repo.ListTargets(ctx)
-	require.NoError(t, err)
-	assert.Equal(t, total, len(got))
-}
-
 func TestRepository_DeleteTarget(t *testing.T) {
 	t.Parallel()
 	conn, _ := db.TestSetup(t, "postgres")
@@ -271,42 +221,3 @@ func TestRepository_DeleteTarget(t *testing.T) {
 		})
 	}
 }
-
-func TestRepository_ListRoles_Above_Default_Count(t *testing.T) {
-	t.Parallel()
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	wrapper := db.TestWrapper(t)
-	testKms := kms.TestKms(t, conn, wrapper)
-	iamRepo := iam.TestRepo(t, conn, wrapper)
-
-	_, proj := iam.TestScopes(t, iamRepo)
-
-	numToCreate := db.DefaultLimit + 5
-	var total int
-	for i := 0; i < numToCreate; i++ {
-		tcp.TestTarget(ctx, t, conn, proj.GetPublicId(), fmt.Sprintf("proj1-%d", i), target.WithAddress("1.2.3.4"))
-		total++
-	}
-	require.Equal(t, numToCreate, total)
-
-	rw := db.New(conn)
-	repo, err := target.NewRepository(ctx, rw, rw, testKms,
-		target.WithPermissions([]perms.Permission{
-			{
-				ScopeId:  proj.PublicId,
-				Resource: resource.Target,
-				Action:   action.List,
-				All:      true,
-			},
-		}))
-	require.NoError(t, err)
-
-	got, err := repo.ListTargets(ctx, target.WithLimit(-1))
-	require.NoError(t, err)
-	assert.Equal(t, total, len(got))
-
-	for _, tar := range got {
-		assert.Equal(t, "1.2.3.4", tar.GetAddress())
-	}
-}

internal/target/testing.go

@@ -6,6 +6,7 @@ package target
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/hashicorp/boundary/internal/credential"
 	"github.com/hashicorp/boundary/internal/db"
@@ -73,3 +74,24 @@ func TestTargetAddress(t testing.TB, conn *db.DB, targetId, address string) *Add
 	require.NoError(err)
 	return staticAddress
 }
+
+// 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...)
+	require.NoError(t, err)
+	return targets
+}
+
+// TestListDeletedIds exposes the repo listDeletedIds method for testing purposes.
+func TestListDeletedIds(t testing.TB, repo *Repository, ctx context.Context, since time.Time) ([]string, time.Time) {
+	ids, ttime, err := repo.listDeletedIds(ctx, since)
+	require.NoError(t, err)
+	return ids, ttime
+}
+
+// TestEstimatedCount exposes the repo estimatedCount method for testing purposes.
+func TestEstimatedCount(t testing.TB, repo *Repository, ctx context.Context) int {
+	n, err := repo.estimatedCount(ctx)
+	require.NoError(t, err)
+	return n
+}