diff --git a/internal/host/plugin/host.go b/internal/host/plugin/host.go index c42e32cb9..23bd96713 100644 --- a/internal/host/plugin/host.go +++ b/internal/host/plugin/host.go @@ -196,3 +196,13 @@ func (agg *hostAgg) TableName() string { func (agg *hostAgg) GetPublicId() string { return agg.PublicId } + +type deletedHost struct { + PublicId string `gorm:"primary_key"` + DeleteTime *timestamp.Timestamp +} + +// TableName returns the tablename to override the default gorm table name +func (s *deletedHost) TableName() string { + return "host_plugin_host_deleted" +} diff --git a/internal/host/plugin/host_set_member_test.go b/internal/host/plugin/host_set_member_test.go index c6ead9f83..b01f54ab5 100644 --- a/internal/host/plugin/host_set_member_test.go +++ b/internal/host/plugin/host_set_member_test.go @@ -162,7 +162,7 @@ func TestHostSetMember_InsertDelete(t *testing.T) { require.Len(t, hosts, 2) // Base case the count by catalog ID - hosts, _, err = repo.ListHostsByCatalogId(ctx, blueCat.PublicId) + hosts, _, _, err = repo.listHosts(ctx, blueCat.PublicId) require.NoError(t, err) assert.Len(t, hosts, 4) @@ -179,7 +179,7 @@ func TestHostSetMember_InsertDelete(t *testing.T) { require.NoError(t, err) assert.Equal(t, 1, count) assert.NoError(t, db.TestVerifyOplog(t, rw, blueHost1.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_DELETE), db.WithCreateNotBefore(10*time.Second))) - hosts, _, err = repo.ListHostsByCatalogId(ctx, blueCat.PublicId) + hosts, _, _, err = repo.listHosts(ctx, blueCat.PublicId) require.NoError(t, err) require.Len(t, hosts, 3) @@ -194,7 +194,7 @@ func TestHostSetMember_InsertDelete(t *testing.T) { require.NoError(t, err) assert.Equal(t, 1, count) assert.NoError(t, db.TestVerifyOplog(t, rw, blueHost2.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_DELETE), db.WithCreateNotBefore(10*time.Second))) - hosts, _, err = repo.ListHostsByCatalogId(ctx, blueCat.PublicId) + hosts, _, _, err = repo.listHosts(ctx, blueCat.PublicId) require.NoError(t, err) require.Len(t, hosts, 2) @@ -211,7 +211,7 @@ func TestHostSetMember_InsertDelete(t *testing.T) { assert.Equal(t, 2, count) assert.NoError(t, db.TestVerifyOplog(t, rw, blueHost3.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_DELETE), db.WithCreateNotBefore(10*time.Second))) assert.NoError(t, db.TestVerifyOplog(t, rw, blueHost4.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_DELETE), db.WithCreateNotBefore(10*time.Second))) - hosts, _, err = repo.ListHostsByCatalogId(ctx, blueCat.PublicId) + hosts, _, _, err = repo.listHosts(ctx, blueCat.PublicId) require.NoError(t, err) require.Len(t, hosts, 0) } diff --git a/internal/host/plugin/job_set_sync_test.go b/internal/host/plugin/job_set_sync_test.go index e2394915e..990367fe7 100644 --- a/internal/host/plugin/job_set_sync_test.go +++ b/internal/host/plugin/job_set_sync_test.go @@ -220,7 +220,7 @@ func TestSetSyncJob_Run(t *testing.T) { assert.Equal(1, r.numSets) assert.Equal(1, r.numProcessed) // Check the version number of the host(s) - hosts, _, err := hostRepo.ListHostsByCatalogId(ctx, hsa.CatalogId) + hosts, _, _, err := hostRepo.listHosts(ctx, hsa.CatalogId) require.NoError(err) assert.Len(hosts, 1) for _, host := range hosts { @@ -254,7 +254,7 @@ func TestSetSyncJob_Run(t *testing.T) { assert.Equal(1, r.numSets) assert.Equal(1, r.numProcessed) // Check the version number of the host(s) again - hosts, _, err = hostRepo.ListHostsByCatalogId(ctx, hsa.CatalogId) + hosts, _, _, err = hostRepo.listHosts(ctx, hsa.CatalogId) require.NoError(err) assert.Len(hosts, 1) for _, host := range hosts { diff --git a/internal/host/plugin/query.go b/internal/host/plugin/query.go index 6a32a2809..9132338b1 100644 --- a/internal/host/plugin/query.go +++ b/internal/host/plugin/query.go @@ -53,5 +53,293 @@ set last_sync_time = current_timestamp, need_sync = false where public_id = ? +` + + estimateCountHosts = ` +select sum(reltuples::bigint) as estimate from pg_class where oid in ('host_plugin_host'::regclass) +` + + listHostsTemplate = ` +with hosts as ( + select public_id, + catalog_id, + external_id, + external_name, + name, + description, + create_time, + update_time, + version + from host_plugin_host + where catalog_id = @catalog_id + order by create_time desc, public_id asc + limit %d +), +host_catalog as ( + select project_id, + plugin_id, + public_id + from host_plugin_catalog + where public_id = @catalog_id +), +host_ip_addresses as ( + select string_agg(distinct host(address), '|') as ip_addresses, + host_id + from host_ip_address + where host_id in (select public_id from hosts) + group by host_id +), +host_dns_names as ( + select string_agg(distinct name, '|') as dns_names, + host_id + from host_dns_name + where host_id in (select public_id from hosts) + group by host_id +), +host_set_ids as ( + select string_agg(distinct set_id, '|') as set_ids, + host_id + from host_plugin_set_member + where host_id in (select public_id from hosts) + group by host_id +), +final as ( + select h.public_id, + h.catalog_id, + h.external_id, + h.external_name, + hc.project_id, + hc.plugin_id, + h.name, + h.description, + h.create_time, + h.update_time, + h.version, + hia.ip_addresses, + hdn.dns_names, + hsi.set_ids + from hosts h + join host_catalog hc on hc.public_id = h.catalog_id + left outer join host_ip_addresses hia on hia.host_id = h.public_id + left outer join host_dns_names hdn on hdn.host_id = h.public_id + left outer join host_set_ids hsi on hsi.host_id = h.public_id +) + select * + from final +order by create_time desc, public_id asc; +` + + listHostsPageTemplate = ` +with hosts as ( + select public_id, + catalog_id, + external_id, + external_name, + name, + description, + create_time, + update_time, + version + from host_plugin_host + where catalog_id = @catalog_id + and (create_time, public_id) < (@last_item_create_time, @last_item_id) + order by create_time desc, public_id asc + limit %d +), +host_catalog as ( + select project_id, + plugin_id, + public_id + from host_plugin_catalog + where public_id = @catalog_id +), +host_ip_addresses as ( + select string_agg(distinct host(address), '|') as ip_addresses, + host_id + from host_ip_address + where host_id in (select public_id from hosts) + group by host_id +), +host_dns_names as ( + select string_agg(distinct name, '|') as dns_names, + host_id + from host_dns_name + where host_id in (select public_id from hosts) + group by host_id +), +host_set_ids as ( + select string_agg(distinct set_id, '|') as set_ids, + host_id + from host_plugin_set_member + where host_id in (select public_id from hosts) + group by host_id +), +final as ( + select h.public_id, + h.catalog_id, + h.external_id, + h.external_name, + hc.project_id, + hc.plugin_id, + h.name, + h.description, + h.create_time, + h.update_time, + h.version, + hia.ip_addresses, + hdn.dns_names, + hsi.set_ids + from hosts h + join host_catalog hc on hc.public_id = h.catalog_id + left outer join host_ip_addresses hia on hia.host_id = h.public_id + left outer join host_dns_names hdn on hdn.host_id = h.public_id + left outer join host_set_ids hsi on hsi.host_id = h.public_id +) + select * + from final +order by create_time desc, public_id asc; +` + + listHostsRefreshTemplate = ` +with hosts as ( + select public_id, + catalog_id, + external_id, + external_name, + name, + description, + create_time, + update_time, + version + from host_plugin_host + where catalog_id = @catalog_id + and update_time > @updated_after_time + order by update_time desc, public_id asc + limit %d +), +host_catalog as ( + select project_id, + plugin_id, + public_id + from host_plugin_catalog + where public_id = @catalog_id +), +host_ip_addresses as ( + select string_agg(distinct host(address), '|') as ip_addresses, + host_id + from host_ip_address + where host_id in (select public_id from hosts) + group by host_id +), +host_dns_names as ( + select string_agg(distinct name, '|') as dns_names, + host_id + from host_dns_name + where host_id in (select public_id from hosts) + group by host_id +), +host_set_ids as ( + select string_agg(distinct set_id, '|') as set_ids, + host_id + from host_plugin_set_member + where host_id in (select public_id from hosts) + group by host_id +), +final as ( + select h.public_id, + h.catalog_id, + h.external_id, + h.external_name, + hc.project_id, + hc.plugin_id, + h.name, + h.description, + h.create_time, + h.update_time, + h.version, + hia.ip_addresses, + hdn.dns_names, + hsi.set_ids + from hosts h + join host_catalog hc on hc.public_id = h.catalog_id + left outer join host_ip_addresses hia on hia.host_id = h.public_id + left outer join host_dns_names hdn on hdn.host_id = h.public_id + left outer join host_set_ids hsi on hsi.host_id = h.public_id +) + select * + from final +order by update_time desc, public_id asc; +` + + listHostsRefreshPageTemplate = ` +with hosts as ( + select public_id, + catalog_id, + external_id, + external_name, + name, + description, + create_time, + update_time, + version + from host_plugin_host + where catalog_id = @catalog_id + and update_time > @updated_after_time + and (update_time, public_id) < (@last_item_update_time, @last_item_id) + order by update_time desc, public_id asc + limit %d +), +host_catalog as ( + select project_id, + plugin_id, + public_id + from host_plugin_catalog + where public_id = @catalog_id +), +host_ip_addresses as ( + select string_agg(distinct host(address), '|') as ip_addresses, + host_id + from host_ip_address + where host_id in (select public_id from hosts) + group by host_id +), +host_dns_names as ( + select string_agg(distinct name, '|') as dns_names, + host_id + from host_dns_name + where host_id in (select public_id from hosts) + group by host_id +), +host_set_ids as ( + select string_agg(distinct set_id, '|') as set_ids, + host_id + from host_plugin_set_member + where host_id in (select public_id from hosts) + group by host_id +), +final as ( + select h.public_id, + h.catalog_id, + h.external_id, + h.external_name, + hc.project_id, + hc.plugin_id, + h.name, + h.description, + h.create_time, + h.update_time, + h.version, + hia.ip_addresses, + hdn.dns_names, + hsi.set_ids + from hosts h + join host_catalog hc on hc.public_id = h.catalog_id + left outer join host_ip_addresses hia on hia.host_id = h.public_id + left outer join host_dns_names hdn on hdn.host_id = h.public_id + left outer join host_set_ids hsi on hsi.host_id = h.public_id +) + select * + from final +order by update_time desc, public_id asc; ` ) diff --git a/internal/host/plugin/repository_host.go b/internal/host/plugin/repository_host.go index 2679b5f65..1893031fc 100644 --- a/internal/host/plugin/repository_host.go +++ b/internal/host/plugin/repository_host.go @@ -5,7 +5,9 @@ package plugin import ( "context" + "database/sql" "fmt" + "time" "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/errors" @@ -35,12 +37,14 @@ func (r *Repository) LookupHost(ctx context.Context, publicId string, opt ...Opt return ha.toHost(), plg, nil } -// ListHostsByCatalogId returns a slice of Hosts for the catalogId. -// WithLimit is the only option supported. -func (r *Repository) ListHostsByCatalogId(ctx context.Context, catalogId string, opt ...Option) ([]*Host, *plugin.Plugin, error) { - const op = "plugin.(Repository).ListHostsByCatalogId" +// listHosts returns a slice of Hosts for the catalogId and the associated plugin. +// Supported options: +// - WithLimit which overrides the limit set in the Repository object +// - WithStartPageAfterItem which sets where to start listing from +func (r *Repository) listHosts(ctx context.Context, catalogId string, opt ...Option) ([]*Host, *plugin.Plugin, time.Time, error) { + const op = "plugin.(Repository).listHosts" if catalogId == "" { - return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "no catalog id") + return nil, nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "no catalog id") } opts := getOpts(opt...) limit := r.defaultLimit @@ -48,28 +52,88 @@ func (r *Repository) ListHostsByCatalogId(ctx context.Context, catalogId string, // non-zero signals an override of the default limit for the repo. limit = opts.withLimit } - var hostAggs []*hostAgg - err := r.reader.SearchWhere(ctx, &hostAggs, "catalog_id = ?", []any{catalogId}, db.WithLimit(limit)) + query := fmt.Sprintf(listHostsTemplate, limit) + args := []any{sql.Named("catalog_id", catalogId)} + if opts.withStartPageAfterItem != nil { + query = fmt.Sprintf(listHostsPageTemplate, limit) + args = append(args, + sql.Named("last_item_create_time", opts.withStartPageAfterItem.GetCreateTime()), + sql.Named("last_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) + } + return r.queryHosts(ctx, query, args) +} + +// listHostsRefresh returns a slice of Hosts for the catalogId and the associated plugin. +// Supported options: +// - WithLimit which overrides the limit set in the Repository object +// - WithStartPageAfterItem which sets where to start listing from +func (r *Repository) listHostsRefresh(ctx context.Context, catalogId string, updatedAfter time.Time, opt ...Option) ([]*Host, *plugin.Plugin, time.Time, error) { + const op = "plugin.(Repository).listHostsRefresh" switch { - case err != nil: - return nil, nil, errors.Wrap(ctx, err, op) - case len(hostAggs) == 0: - return nil, nil, nil + case catalogId == "": + return nil, nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "no catalog id") + case updatedAfter.IsZero(): + return nil, nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "missing updated after time") + } + opts := getOpts(opt...) + limit := r.defaultLimit + if opts.withLimit != 0 { + // non-zero signals an override of the default limit for the repo. + limit = opts.withLimit + } + query := fmt.Sprintf(listHostsRefreshTemplate, limit) + args := []any{ + sql.Named("catalog_id", catalogId), + sql.Named("updated_after_time", updatedAfter), + } + if opts.withStartPageAfterItem != nil { + query = fmt.Sprintf(listHostsRefreshPageTemplate, limit) + args = append(args, + sql.Named("last_item_update_time", opts.withStartPageAfterItem.GetUpdateTime()), + sql.Named("last_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) } - pluginId := hostAggs[0].PluginId - plg, err := r.getPlugin(ctx, pluginId) - if err != nil { - return nil, nil, errors.Wrap(ctx, err, op) - } + return r.queryHosts(ctx, query, args) +} - hosts := make([]*Host, 0, len(hostAggs)) - for _, ha := range hostAggs { - hosts = append(hosts, ha.toHost()) - } +func (r *Repository) queryHosts(ctx context.Context, query string, args []any) ([]*Host, *plugin.Plugin, time.Time, error) { + const op = "plugin.(Repository).queryHosts" - return hosts, plg, nil + var hosts []*Host + var plg *plugin.Plugin + 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 err + } + defer rows.Close() + var foundHosts []*hostAgg + for rows.Next() { + if err := r.ScanRows(ctx, rows, &foundHosts); err != nil { + return err + } + } + if len(foundHosts) != 0 { + plg = plugin.NewPlugin() + plg.PublicId = foundHosts[0].PluginId + if err := r.LookupByPublicId(ctx, plg); err != nil { + return err + } + hosts = make([]*Host, 0, len(foundHosts)) + for _, ha := range foundHosts { + hosts = append(hosts, ha.toHost()) + } + } + transactionTimestamp, err = r.Now(ctx) + return err + }); err != nil { + return nil, nil, time.Time{}, errors.Wrap(ctx, err, op) + } + return hosts, plg, transactionTimestamp, nil } // ListHostsBySetId returns a slice of Hosts for the given set IDs. @@ -119,3 +183,45 @@ public_id in return hosts, nil } + +// listDeletedHostIds lists the public IDs of any hosts deleted since the timestamp provided, +// and the timestamp of the transaction within which the hosts were listed. +func (r *Repository) listDeletedHostIds(ctx context.Context, since time.Time) ([]string, time.Time, error) { + const op = "static.(Repository).listDeletedHostIds" + var deleteHosts []*deletedHost + 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, &deleteHosts, "delete_time >= ?", []any{since}); err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("failed to query deleted hosts")) + } + 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 hostIds []string + for _, t := range deleteHosts { + hostIds = append(hostIds, t.PublicId) + } + return hostIds, transactionTimestamp, nil +} + +// estimatedHostCount returns an estimate of the total number of plugin hosts. +func (r *Repository) estimatedHostCount(ctx context.Context) (int, error) { + const op = "plugin.(Repository).estimatedHostCount" + rows, err := r.reader.Query(ctx, estimateCountHosts, nil) + if err != nil { + return 0, errors.Wrap(ctx, err, op, errors.WithMsg("failed to query plugin hosts")) + } + 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 plugin hosts")) + } + } + return count, nil +} diff --git a/internal/host/plugin/repository_host_test.go b/internal/host/plugin/repository_host_test.go index afbec224a..1f9f72a68 100644 --- a/internal/host/plugin/repository_host_test.go +++ b/internal/host/plugin/repository_host_test.go @@ -230,7 +230,7 @@ func TestJob_UpsertHosts(t *testing.T) { // Check again, but via performing an explicit list var gotPlg *plugin.Plugin - got, gotPlg, err = repo.ListHostsByCatalogId(ctx, in.catalog.GetPublicId()) + got, gotPlg, _, err = repo.listHosts(ctx, in.catalog.GetPublicId()) require.NoError(err) assert.Len(got, len(in.phs)) assert.Empty( @@ -297,3 +297,156 @@ func TestJob_UpsertHosts(t *testing.T) { }) } } + +func TestRepository_ListHostsByCatalogId(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + _, proj1 := iam.TestScopes(t, iamRepo) + sched := scheduler.TestScheduler(t, conn, wrapper) + plg := plugin.TestPlugin(t, conn, "test") + plgm := map[string]plgpb.HostPluginServiceClient{ + plg.GetPublicId(): &loopback.WrappingPluginHostClient{Server: &loopback.TestPluginServer{}}, + } + catalog := TestCatalog(t, conn, proj1.GetPublicId(), plg.GetPublicId()) + + total := 5 + for i := 0; i < total; i++ { + TestHost(t, conn, catalog.GetPublicId(), fmt.Sprintf("ext-id-%d", i)) + } + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, kms, sched, plgm) + require.NoError(t, err) + + t.Run("no-options", func(t *testing.T) { + got, retPlg, ttime, err := repo.listHosts(ctx, catalog.GetPublicId()) + require.NoError(t, err) + assert.Equal(t, total, len(got)) + assert.Equal(t, retPlg, plg) + // 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, retPlg, ttime, err := repo.listHosts( + context.Background(), + catalog.GetPublicId(), + WithLimit(2), + ) + require.NoError(err) + require.Len(page1, 2) + assert.Equal(retPlg, plg) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + page2, retPlg, ttime, err := repo.listHosts( + context.Background(), + catalog.GetPublicId(), + WithLimit(2), + WithStartPageAfterItem(page1[1]), + ) + require.NoError(err) + require.Len(page2, 2) + assert.Equal(retPlg, plg) + 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, retPlg, ttime, err := repo.listHosts( + context.Background(), + catalog.GetPublicId(), + WithLimit(2), + WithStartPageAfterItem(page2[1]), + ) + require.NoError(err) + require.Len(page3, 1) + assert.Equal(retPlg, plg) + 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()) + } + page4, retPlg, ttime, err := repo.listHosts( + context.Background(), + catalog.GetPublicId(), + WithLimit(2), + WithStartPageAfterItem(page3[0]), + ) + require.NoError(err) + require.Empty(page4) + require.Empty(retPlg) + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + }) +} + +func Test_listDeletedHostIds(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + wrapper := db.TestWrapper(t) + testKms := kms.TestKms(t, conn, wrapper) + sched := scheduler.TestScheduler(t, conn, wrapper) + plg := plugin.TestPlugin(t, conn, "test") + plgm := map[string]plgpb.HostPluginServiceClient{ + plg.GetPublicId(): &loopback.WrappingPluginHostClient{Server: &loopback.TestPluginServer{}}, + } + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, testKms, sched, plgm) + require.NoError(t, err) + + // Expect no entries + deletedIds, ttime, err := repo.listDeletedHostIds(ctx, time.Now().AddDate(-1, 0, 0)) + require.NoError(t, err) + 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))) + + // It's not possible to delete a plugin host, so this is kinda hard to test +} + +func Test_estimatedHostCount(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) + sched := scheduler.TestScheduler(t, conn, wrapper) + plg := plugin.TestPlugin(t, conn, "test") + plgm := map[string]plgpb.HostPluginServiceClient{ + plg.GetPublicId(): &loopback.WrappingPluginHostClient{Server: &loopback.TestPluginServer{}}, + } + catalog := TestCatalog(t, conn, proj1.GetPublicId(), plg.GetPublicId()) + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, testKms, sched, plgm) + require.NoError(t, err) + + // Check total entries at start, expect 0 + numItems, err := repo.estimatedHostCount(ctx) + require.NoError(t, err) + assert.Equal(t, 0, numItems) + + // Create a host, expect 1 entries + TestHost(t, conn, catalog.GetPublicId(), "ext-id") + // Run analyze to update estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + numItems, err = repo.estimatedHostCount(ctx) + require.NoError(t, err) + assert.Equal(t, 1, numItems) +} diff --git a/internal/host/plugin/service_list_hosts.go b/internal/host/plugin/service_list_hosts.go new file mode 100644 index 000000000..3ec1592a8 --- /dev/null +++ b/internal/host/plugin/service_list_hosts.go @@ -0,0 +1,64 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/pagination" + ppagination "github.com/hashicorp/boundary/internal/pagination/plugin" + "github.com/hashicorp/boundary/internal/plugin" +) + +// ListHosts lists up to page size hosts, filtering out entries that +// do not pass the filter item function. It will automatically request +// more hosts from the database, at page size chunks, to fill the page. +// It returns a new list token used to continue pagination or refresh items. +// Hosts are ordered by create time descending (most recently created first). +func ListHosts( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn ppagination.ListPluginFilterFunc[host.Host], + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Host], *plugin.Plugin, error) { + const op = "plugin.ListHosts" + + switch { + case len(grantsHash) == 0: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case repo == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Host, limit int) ([]host.Host, *plugin.Plugin, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } + pHosts, plg, listTime, err := repo.listHosts(ctx, hostCatalogId, opts...) + if err != nil { + return nil, nil, time.Time{}, err + } + var hosts []host.Host + for _, host := range pHosts { + hosts = append(hosts, host) + } + return hosts, plg, listTime, nil + } + + return ppagination.ListPlugin(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedHostCount) +} diff --git a/internal/host/plugin/service_list_hosts_ext_test.go b/internal/host/plugin/service_list_hosts_ext_test.go new file mode 100644 index 000000000..e7789fd52 --- /dev/null +++ b/internal/host/plugin/service_list_hosts_ext_test.go @@ -0,0 +1,649 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin_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/host" + hostplugin "github.com/hashicorp/boundary/internal/host/plugin" + "github.com/hashicorp/boundary/internal/host/plugin/store" + "github.com/hashicorp/boundary/internal/iam" + "github.com/hashicorp/boundary/internal/kms" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/plugin" + "github.com/hashicorp/boundary/internal/plugin/loopback" + "github.com/hashicorp/boundary/internal/scheduler" + "github.com/hashicorp/boundary/internal/types/resource" + plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestService_ListHosts(t *testing.T) { + // 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) + require.NoError(t, err) + wrapper := db.TestWrapper(t) + testKms := kms.TestKms(t, conn, wrapper) + sche := scheduler.TestScheduler(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + _, proj1 := iam.TestScopes(t, iamRepo) + plg := plugin.TestPlugin(t, conn, "test") + fiveDaysAgo := time.Now().AddDate(0, 0, -5) + catalog := hostplugin.TestCatalog(t, conn, proj1.GetPublicId(), plg.GetPublicId()) + var hosts []host.Host + for i := 0; i < 5; i++ { + host := hostplugin.TestHost(t, conn, catalog.GetPublicId(), fmt.Sprintf("ext-host-%d", i)) + host.PluginId = plg.GetPublicId() + hosts = append(hosts, host) + } + // since we sort by create time descending, we need to reverse the slice + slices.Reverse(hosts) + + // Run analyze to update host estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + rw := db.New(conn) + plgm := map[string]plgpb.HostPluginServiceClient{ + plg.GetPublicId(): loopback.NewWrappingPluginHostClient(&plgpb.UnimplementedHostPluginServiceServer{}), + } + repo, err := hostplugin.NewRepository(ctx, rw, rw, testKms, sche, plgm) + require.NoError(t, err) + + cmpOpts := []cmp.Option{ + cmpopts.IgnoreUnexported( + hostplugin.Host{}, + store.Host{}, + timestamp.Timestamp{}, + timestamppb.Timestamp{}, + ), + } + + t.Run("ListHosts validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.ListHosts(ctx, nil, 1, filterFunc, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.ListHosts(ctx, []byte("some hash"), 0, filterFunc, repo, catalog.GetPublicId()) + 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, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.ListHosts(ctx, []byte("some hash"), -1, filterFunc, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + _, _, err := hostplugin.ListHosts(ctx, []byte("some hash"), 1, nil, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.ListHosts(ctx, []byte("some hash"), 1, filterFunc, nil, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing repo") + }) + t.Run("missing host catalog ID", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.ListHosts(ctx, []byte("some hash"), 1, filterFunc, repo, "") + require.ErrorContains(t, err, "missing host catalog ID") + }) + }) + t.Run("ListHostsPage validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsPage(ctx, nil, 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, catalog.GetPublicId()) + 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, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, catalog.GetPublicId()) + 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.NewPagination(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, nil, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err = hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + 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, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing repo") + }) + t.Run("missing host catalog ID", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, "") + require.ErrorContains(t, err, "missing host catalog ID") + }) + t.Run("wrong token resource type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "token did not have a host resource type") + }) + }) + t.Run("ListHostsRefresh validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefresh(ctx, nil, 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 0, filterFunc, tok, repo, catalog.GetPublicId()) + 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, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefresh(ctx, []byte("some hash"), -1, filterFunc, tok, repo, catalog.GetPublicId()) + 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.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, nil, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err = hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing token") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, nil, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing repo") + }) + t.Run("missing host catalog ID", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, repo, "") + require.ErrorContains(t, err, "missing host catalog ID") + }) + t.Run("wrong token resource type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "token did not have a host resource type") + }) + }) + t.Run("ListHostsRefreshPage validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefreshPage(ctx, nil, 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, catalog.GetPublicId()) + 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, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, catalog.GetPublicId()) + 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.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 1, nil, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err = hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + 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, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing repo") + }) + t.Run("missing credential store id", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Host, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, "") + require.ErrorContains(t, err, "missing host catalog ID") + }) + t.Run("wrong token resource type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, h host.Host, plg *plugin.Plugin) (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 = hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "token did not have a host resource type") + }) + }) + + t.Run("simple pagination", func(t *testing.T) { + filterFunc := func(context.Context, host.Host, *plugin.Plugin) (bool, error) { + return true, nil + } + resp, retPlg, err := hostplugin.ListHosts(ctx, []byte("some hash"), 1, filterFunc, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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], hosts[0], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp2, retPlg, err := hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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], hosts[1], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp3, retPlg, err := hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, resp2.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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], hosts[2], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp4, retPlg, err := hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, resp3.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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], hosts[3], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp5, retPlg, err := hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, resp4.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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], hosts[4], cmpOpts...)) + require.Equal(t, retPlg, plg) + + // Finished initial pagination phase, request refresh + // Expect no results. + resp6, retPlg, err := hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp5.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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) + require.Empty(t, retPlg) + + // Create some new hosts + host1 := hostplugin.TestHost(t, conn, catalog.GetPublicId(), "some-host-id") + host1.PluginId = plg.GetPublicId() + host2 := hostplugin.TestHost(t, conn, catalog.GetPublicId(), "some-host-id-2") + host2.PluginId = plg.GetPublicId() + t.Cleanup(func() { + // Note: we can't delete plugin hosts via the repo, so we use this hack. + _, err := sqlDb.ExecContext(ctx, "delete from host_plugin_host where public_id = $1", host1.GetPublicId()) + require.NoError(t, err) + _, err = sqlDb.ExecContext(ctx, "delete from host_plugin_host where public_id = $1", host2.GetPublicId()) + require.NoError(t, err) + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + }) + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + // Refresh again, should get host2 + resp7, retPlg, err := hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp6.ListToken, repo, catalog.GetPublicId()) + 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], host2, cmpOpts...)) + require.Equal(t, retPlg, plg) + + // Refresh again, should get host1 + resp8, retPlg, err := hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, resp7.ListToken, repo, catalog.GetPublicId()) + 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], host1, cmpOpts...)) + require.Equal(t, retPlg, plg) + + // Refresh again, should get no results + resp9, retPlg, err := hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp8.ListToken, repo, catalog.GetPublicId()) + 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) + require.Empty(t, retPlg) + }) + + t.Run("simple pagination with aggressive filtering", func(t *testing.T) { + filterFunc := func(ctx context.Context, h host.Host, plg *plugin.Plugin) (bool, error) { + return h.GetPublicId() == hosts[1].GetPublicId() || + h.GetPublicId() == hosts[len(hosts)-1].GetPublicId(), nil + } + resp, retPlg, err := hostplugin.ListHosts(ctx, []byte("some hash"), 1, filterFunc, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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], hosts[1], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp2, retPlg, err := hostplugin.ListHostsPage(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, catalog.GetPublicId()) + 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, 5) + require.Empty(t, resp2.DeletedIds) + require.Len(t, resp2.Items, 1) + require.Empty(t, cmp.Diff(resp2.Items[0], hosts[len(hosts)-1], cmpOpts...)) + require.Equal(t, retPlg, plg) + + // request a refresh, nothing should be returned + resp3, retPlg, err := hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, catalog.GetPublicId()) + 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) + require.Empty(t, retPlg) + + // Create some new hosts + host1 := hostplugin.TestHost(t, conn, catalog.GetPublicId(), "some-host-id") + host1.PluginId = plg.GetPublicId() + host2 := hostplugin.TestHost(t, conn, catalog.GetPublicId(), "some-host-id-2") + host2.PluginId = plg.GetPublicId() + host3 := hostplugin.TestHost(t, conn, catalog.GetPublicId(), "some-host-id-3") + host3.PluginId = plg.GetPublicId() + host4 := hostplugin.TestHost(t, conn, catalog.GetPublicId(), "some-host-id-4") + host4.PluginId = plg.GetPublicId() + t.Cleanup(func() { + // Note: we can't delete plugin hosts via the repo, so we use this hack. + _, err := sqlDb.ExecContext(ctx, "delete from host_plugin_host where public_id = $1", host1.GetPublicId()) + require.NoError(t, err) + _, err = sqlDb.ExecContext(ctx, "delete from host_plugin_host where public_id = $1", host2.GetPublicId()) + require.NoError(t, err) + _, err = sqlDb.ExecContext(ctx, "delete from host_plugin_host where public_id = $1", host3.GetPublicId()) + require.NoError(t, err) + _, err = sqlDb.ExecContext(ctx, "delete from host_plugin_host where public_id = $1", host4.GetPublicId()) + require.NoError(t, err) + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + }) + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + filterFunc = func(_ context.Context, h host.Host, retPlg *plugin.Plugin) (bool, error) { + return h.GetPublicId() == host3.GetPublicId() || + h.GetPublicId() == host1.GetPublicId(), nil + } + // Refresh again, should get host3 + resp4, retPlg, err := hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp3.ListToken, repo, catalog.GetPublicId()) + 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], host3, cmpOpts...)) + require.Equal(t, retPlg, plg) + + // Refresh again, should get host1 + resp5, retPlg, err := hostplugin.ListHostsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, resp4.ListToken, repo, catalog.GetPublicId()) + 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], host1, cmpOpts...)) + require.Equal(t, retPlg, plg) + }) + + t.Run("simple pagination with deletion", func(t *testing.T) { + filterFunc := func(context.Context, host.Host, *plugin.Plugin) (bool, error) { + return true, nil + } + deletedHostId := hosts[0].GetPublicId() + _, err := sqlDb.ExecContext(ctx, "delete from host_plugin_host where public_id = $1", deletedHostId) + require.NoError(t, err) + hosts = hosts[1:] + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + resp, retPlg, err := hostplugin.ListHosts(ctx, []byte("some hash"), 1, filterFunc, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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], hosts[0], cmpOpts...)) + require.Equal(t, retPlg, plg) + + // request remaining results + resp2, retPlg, err := hostplugin.ListHostsPage(ctx, []byte("some hash"), 3, filterFunc, resp.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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, 3) + require.Empty(t, cmp.Diff(resp2.Items, hosts[1:], cmpOpts...)) + require.Equal(t, retPlg, plg) + + deletedHostId = hosts[0].GetPublicId() + _, err = sqlDb.ExecContext(ctx, "delete from host_plugin_host where public_id = $1", deletedHostId) + require.NoError(t, err) + hosts = hosts[1:] + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + // request a refresh, nothing should be returned except the deleted id + resp3, retPlg, err := hostplugin.ListHostsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp2.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + 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, deletedHostId) + require.Empty(t, resp3.Items) + require.Empty(t, retPlg) + }) +} diff --git a/internal/host/plugin/service_list_hosts_page.go b/internal/host/plugin/service_list_hosts_page.go new file mode 100644 index 000000000..64e17db21 --- /dev/null +++ b/internal/host/plugin/service_list_hosts_page.go @@ -0,0 +1,81 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" + ppagination "github.com/hashicorp/boundary/internal/pagination/plugin" + "github.com/hashicorp/boundary/internal/plugin" + "github.com/hashicorp/boundary/internal/types/resource" +) + +// ListHostsPage lists up to page size hosts, filtering out entries that +// do not pass the filter item function. It will automatically request +// more hosts 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. +// Hosts are ordered by create time descending (most recently created first). +func ListHostsPage( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn ppagination.ListPluginFilterFunc[host.Host], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Host], *plugin.Plugin, error) { + const op = "plugin.ListHostsPage" + + switch { + case len(grantsHash) == 0: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.Host: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a host resource type") + } + if _, ok := tok.Subtype.(*listtoken.PaginationToken); !ok { + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a pagination token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Host, limit int) ([]host.Host, *plugin.Plugin, 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, nil, time.Time{}, err + } + opts = append(opts, WithStartPageAfterItem(lastItem)) + } + pHosts, plg, listTime, err := repo.listHosts(ctx, hostCatalogId, opts...) + if err != nil { + return nil, nil, time.Time{}, err + } + var hosts []host.Host + for _, host := range pHosts { + hosts = append(hosts, host) + } + return hosts, plg, listTime, nil + } + + return ppagination.ListPluginPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedHostCount, tok) +} diff --git a/internal/host/plugin/service_list_hosts_refresh.go b/internal/host/plugin/service_list_hosts_refresh.go new file mode 100644 index 000000000..a73b094e8 --- /dev/null +++ b/internal/host/plugin/service_list_hosts_refresh.go @@ -0,0 +1,83 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" + ppagination "github.com/hashicorp/boundary/internal/pagination/plugin" + "github.com/hashicorp/boundary/internal/plugin" + "github.com/hashicorp/boundary/internal/types/resource" +) + +// ListHostsRefresh lists hosts according to the page size +// and list token, filtering out entries that do not +// pass the filter item fn. It returns a new list token +// based on the old one, the grants hash, and the returned +// hosts. It also returns the plugin associated with the host catalog. +func ListHostsRefresh( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn ppagination.ListPluginFilterFunc[host.Host], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Host], *plugin.Plugin, error) { + const op = "plugin.ListHostsRefresh" + + switch { + case len(grantsHash) == 0: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.Host: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a host resource type") + } + rt, ok := tok.Subtype.(*listtoken.StartRefreshToken) + if !ok { + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a start-refresh token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Host, limit int) ([]host.Host, *plugin.Plugin, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } + // Add the database read timeout to account for any creations missed due to concurrent + // transactions in the initial pagination phase. + pHosts, plg, listTime, err := repo.listHostsRefresh(ctx, hostCatalogId, rt.PreviousPhaseUpperBound.Add(-globals.RefreshReadLookbackDuration), opts...) + if err != nil { + return nil, nil, time.Time{}, err + } + var hosts []host.Host + for _, host := range pHosts { + hosts = append(hosts, host) + } + return hosts, plg, listTime, nil + } + 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.listDeletedHostIds(ctx, since.Add(-globals.RefreshReadLookbackDuration)) + } + + return ppagination.ListPluginRefresh(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedHostCount, listDeletedIdsFn, tok) +} diff --git a/internal/host/plugin/service_list_hosts_refresh_page.go b/internal/host/plugin/service_list_hosts_refresh_page.go new file mode 100644 index 000000000..d4539008f --- /dev/null +++ b/internal/host/plugin/service_list_hosts_refresh_page.go @@ -0,0 +1,93 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" + ppagination "github.com/hashicorp/boundary/internal/pagination/plugin" + "github.com/hashicorp/boundary/internal/plugin" + "github.com/hashicorp/boundary/internal/types/resource" +) + +// ListHostsRefreshPage lists up to page size hosts, filtering out entries that +// do not pass the filter item function. It will automatically request +// more hosts 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. +// Hosts are ordered by update time descending (most recently updated first). +// Hosts may contain items that were already returned during the initial +// pagination phase. It also returns a list of any hosts deleted since the +// last response. +func ListHostsRefreshPage( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn ppagination.ListPluginFilterFunc[host.Host], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Host], *plugin.Plugin, error) { + const op = "credential.ListHostsRefreshPage" + + switch { + case len(grantsHash) == 0: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.Host: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a host resource type") + } + rt, ok := tok.Subtype.(*listtoken.RefreshToken) + if !ok { + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a refresh token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Host, limit int) ([]host.Host, *plugin.Plugin, 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, 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. + pHosts, plg, listTime, err := repo.listHostsRefresh(ctx, hostCatalogId, rt.PhaseLowerBound.Add(-globals.RefreshReadLookbackDuration), opts...) + if err != nil { + return nil, nil, time.Time{}, err + } + var hosts []host.Host + for _, host := range pHosts { + hosts = append(hosts, host) + } + return hosts, plg, listTime, nil + } + 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.listDeletedHostIds(ctx, since.Add(-globals.RefreshReadLookbackDuration)) + } + + return ppagination.ListPluginRefreshPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedHostCount, listDeletedIdsFn, tok) +}