diff --git a/internal/api/handler_recommendations_page.go b/internal/api/handler_recommendations_page.go index 68a23a9b..83df39e1 100644 --- a/internal/api/handler_recommendations_page.go +++ b/internal/api/handler_recommendations_page.go @@ -17,9 +17,9 @@ import ( // which passes the 6 MB Lambda response cap once a tenant accumulates a few // thousand recommendations. A caller that sends `limit` (and, for later // pages, the `cursor` returned as `next_cursor`) gets one bounded page. -// A paginated request scans the filtered cache in ID-ordered chunks and keeps -// only the page (issue #981, see handler_recommendations_scan.go). A -// page is bounded twice: by row count (limit) and by serialized bytes +// A paginated request reads only the page (plus one row) from the store, with +// the summary totaled by an SQL aggregate (issue #981, see +// handler_recommendations_scan.go). A page is bounded twice: by row count (limit) and by serialized bytes // (recommendationsPageByteBudget), because the opaque Details blob makes the // size of a row data-dependent. // A request carrying neither parameter keeps the existing unpaginated shape diff --git a/internal/api/handler_recommendations_page_bound_integration_test.go b/internal/api/handler_recommendations_page_bound_integration_test.go new file mode 100644 index 00000000..a0d78b3c --- /dev/null +++ b/internal/api/handler_recommendations_page_bound_integration_test.go @@ -0,0 +1,139 @@ +//go:build integration + +package api + +import ( + "context" + "fmt" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" +) + +// chunkCountingStore counts the rows the Go-side scan pulls out of Postgres. +type chunkCountingStore struct { + *config.PostgresStore + chunkRows atomic.Int64 +} + +func (s *chunkCountingStore) ListStoredRecommendationsChunk(ctx context.Context, filter config.RecommendationFilter, after *config.RecommendationScanKey, limit int) (config.RecommendationChunk, error) { + chunk, err := s.PostgresStore.ListStoredRecommendationsChunk(ctx, filter, after, limit) + s.chunkRows.Add(int64(len(chunk.Recs))) + return chunk, err +} + +// TestGetRecommendations_PageDoesNotReadEveryRow is the issue #981 +// regression: a paginated GET /api/recommendations must not read the whole +// filtered cache into the process to total the summary. Before the fix every +// page request pulled every row through ListStoredRecommendationsChunk; now +// the summary is an SQL aggregate and only the page reaches Go. The walk +// must still return the legacy rows and totals. +func TestGetRecommendations_PageDoesNotReadEveryRow(t *testing.T) { + ctx := t.Context() + f := newPageSQLFixture(ctx, t) + store := &chunkCountingStore{PostgresStore: f.store} + sched := scheduler.NewScheduler(scheduler.SchedulerConfig{ConfigStore: store, IsLambda: true}) + h := &Handler{config: store, scheduler: sched, apiKey: "test-key"} + + legacy, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"min_savings_pct": "20"}) + require.NoError(t, err) + require.Greater(t, len(legacy.Recommendations), 100) + + const limit = 50 + params := map[string]string{"min_savings_pct": "20", "limit": strconv.Itoa(limit)} + walked := 0 + for page := 0; ; page++ { + require.Less(t, page, len(legacy.Recommendations), "cursor walk did not terminate") + resp, err := h.getRecommendations(ctx, pageTestReq(), params) + require.NoError(t, err) + assert.Equal(t, legacy.Summary.TotalCount, resp.Summary.TotalCount) + // The unpaginated total sums in table order, the paginated one in ID + // order, so the float totals may differ in the last bits. + assert.InDelta(t, legacy.Summary.TotalMonthlySavings, resp.Summary.TotalMonthlySavings, 1e-6*legacy.Summary.TotalMonthlySavings) + assert.ElementsMatch(t, legacy.Regions, resp.Regions) + walked += len(resp.Recommendations) + if resp.NextCursor == "" { + break + } + params["cursor"] = resp.NextCursor + } + assert.Equal(t, len(legacy.Recommendations), walked) + assert.Zero(t, store.chunkRows.Load(), "a paginated request must not scan the cache into Go") +} + +// TestGetRecommendations_PageAndSummaryShareOneSnapshot races paginated reads +// against collections that rewrite every row in place. Each collection +// commits atomically, so every response must describe one collection: a +// summary from one and rows from another, or a summary mixing both, would +// show totals the cache never held. The SQL page read takes its summary and +// its rows from one REPEATABLE READ snapshot; a read made of separately +// committed statements (as the chunked scan was) can mix them. +func TestGetRecommendations_PageAndSummaryShareOneSnapshot(t *testing.T) { + ctx := t.Context() + pg := testhelpers.RequireMigratedDatabase(ctx, t, "../database/postgres/migrations") + store := config.NewPostgresStore(pg.DB) + const n = 1200 + collection := func(savings float64) []config.RecommendationRecord { + recs := make([]config.RecommendationRecord, n) + for i := range recs { + recs[i] = config.RecommendationRecord{ID: fmt.Sprintf("rec|%05d", i), Provider: "aws", Service: "rds", Region: "us-east-1", + ResourceType: fmt.Sprintf("db.r%05d", i), Count: 1, Term: 1, Payment: "no-upfront", Savings: savings} + } + return recs + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now(), collection(1), nil)) + h := &Handler{config: store, scheduler: scheduler.NewScheduler(scheduler.SchedulerConfig{ConfigStore: store, IsLambda: true}), apiKey: "test-key"} + + raceCtx, cancel := context.WithTimeout(ctx, 4*time.Second) + defer cancel() + var wg sync.WaitGroup + var writes atomic.Int64 + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; raceCtx.Err() == nil; i++ { + if err := store.UpsertRecommendations(ctx, time.Now(), collection(float64(1+i%2)), nil); err != nil { + t.Errorf("upsert: %v", err) + return + } + writes.Add(1) + } + }() + var reads atomic.Int64 + for r := 0; r < 4; r++ { + wg.Add(1) + go func(r int) { + defer wg.Done() + cursor := encodeRecommendationsCursor(fmt.Sprintf("rec|%05d", (r*271)%n)) + for raceCtx.Err() == nil { + resp, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"limit": "100", "cursor": cursor}) + if !assert.NoError(t, err) { + return + } + total := resp.Summary.TotalMonthlySavings + if !assert.Contains(t, []float64{n, 2 * n}, total, "summary mixes two collections") { + return + } + for i := range resp.Recommendations { + if !assert.Equal(t, total/n, resp.Recommendations[i].Savings, "page row from another collection than its summary") { + return + } + } + reads.Add(1) + } + }(r) + } + wg.Wait() + t.Logf("%d page reads raced %d collections", reads.Load(), writes.Load()) + require.Positive(t, writes.Load()) + require.Positive(t, reads.Load()) +} diff --git a/internal/api/handler_recommendations_page_sql_fixture_integration_test.go b/internal/api/handler_recommendations_page_sql_fixture_integration_test.go new file mode 100644 index 00000000..09ff90af --- /dev/null +++ b/internal/api/handler_recommendations_page_sql_fixture_integration_test.go @@ -0,0 +1,272 @@ +//go:build integration + +package api + +import ( + "context" + "encoding/json" + "fmt" + "math/rand" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" +) + +// pageSQLFixture is a recommendations cache built to exercise every filter +// the SQL page read re-implements (issue #981): registered and missing +// accounts, three providers, legacy and canonical service slugs, +// Savings Plans, empty regions, engines, every shape of the savings-pct +// inputs, fractional dollar amounts (so a summation order change would show), +// IDs whose byte order differs from linguistic order, suppressions with and +// without variants (including a legacy "shared" tenancy and unreadable +// Details), service configs and account overrides. +type pageSQLFixture struct { + store *config.PostgresStore + sched *scheduler.Scheduler + h *Handler + accounts []config.CloudAccount // alpha, beta, gamma (registered) + unreg string // an account ID no cloud_accounts row has (scope-only: rows carry registered accounts by FK) + recs []config.RecommendationRecord +} + +func newPageSQLFixture(ctx context.Context, t *testing.T) *pageSQLFixture { + t.Helper() + pg := testhelpers.RequireMigratedDatabase(ctx, t, "../database/postgres/migrations") + f := &pageSQLFixture{store: config.NewPostgresStore(pg.DB), unreg: uuid.NewString()} + for i, name := range []string{"alpha", "beta", "gamma"} { + a := config.CloudAccount{ID: uuid.NewString(), Name: name, Provider: "aws", ExternalID: fmt.Sprintf("11111111111%d", i), Enabled: true} + require.NoError(t, f.store.CreateCloudAccount(ctx, &a)) + f.accounts = append(f.accounts, a) + } + f.recs = f.buildRecs() + require.NoError(t, f.store.UpsertRecommendations(ctx, time.Now().UTC(), f.recs, nil)) + f.seedSuppressions(ctx, t, pg) + f.seedOverrides(ctx, t) + f.sched = scheduler.NewScheduler(scheduler.SchedulerConfig{ConfigStore: f.store, IsLambda: true}) + f.h = &Handler{config: f.store, scheduler: f.sched} + return f +} + +func pfloat(v float64) *float64 { return &v } + +// pick returns one of options, chosen by rng. +func pick[T any](rng *rand.Rand, options ...T) T { return options[rng.Intn(len(options))] } + +func (f *pageSQLFixture) buildRecs() []config.RecommendationRecord { + rng := rand.New(rand.NewSource(981)) + accounts := []*string{nil, &f.accounts[0].ID, &f.accounts[1].ID, &f.accounts[2].ID} + type svc struct{ provider, service string } + services := []svc{ + {"aws", "compute"}, {"aws", "ec2"}, {"aws", "rds"}, {"aws", "relational-db"}, + {"aws", "savings-plans-compute"}, {"aws", "savingsplans-compute"}, {"azure", "compute"}, {"gcp", "compute"}, + } + prefixes := []string{"aws", "AWS", "aws-x", "aws x", "élan", "elan", "_z", "Zeta"} + seen := map[string]bool{} + var recs []config.RecommendationRecord + for i := 0; len(recs) < 900; i++ { + s := pick(rng, services...) + rec := config.RecommendationRecord{ + ID: fmt.Sprintf("%s|%05d", pick(rng, prefixes...), rng.Intn(100000)), + Provider: s.provider, + Service: s.service, + Region: pick(rng, "us-east-1", "us-west-2", "eu-west-1", ""), + ResourceType: pick(rng, "m5.large", "c5.xlarge", "db.r6g.large"), + Engine: pick(rng, "", "", "mysql", "postgres"), + Count: 1 + rng.Intn(12), + Term: pick(rng, 0, 1, 1, 3), + Payment: pick(rng, "no-upfront", "all-upfront", "partial-upfront"), + Savings: float64(rng.Intn(500000)) / 7, + UpfrontCost: pick(rng, 0, float64(rng.Intn(9000000))/13), + CloudAccountID: pick(rng, accounts...), + } + f.setPctInputs(rng, &rec) + if s.service == "compute" || s.service == "ec2" { + rec.Details = json.RawMessage(pick(rng, + `{"platform":"Linux/UNIX","tenancy":"default"}`, `{"platform":"Linux/UNIX","tenancy":"shared"}`, + `{"platform":"Windows","tenancy":"dedicated"}`, `"unreadable"`, `{"availability_zone":"us-east-1a"}`, ``)) + rec.AvailabilityZone = pick(rng, "", "", "us-east-1b") + } + key := naturalKeyOf(&rec) + if seen[key] || seen["id:"+rec.ID] { + continue + } + seen[key], seen["id:"+rec.ID] = true, true + recs = append(recs, rec) + } + return append(recs, pctBoundaryRecs()...) +} + +// pctBoundaryRecs sit exactly on the fixture's pct floors: a provider +// percentage equal to the floor (kept, as Go's check is "pct < floor"), and +// 57/100*100, which float64 rounds to 56.99999999999999 (dropped at a floor +// of 57 by Go, kept by exact decimal arithmetic), through both the on-demand +// and the reconstructed baseline. +func pctBoundaryRecs() []config.RecommendationRecord { + mk := func(id, provider, service string, count int) config.RecommendationRecord { + return config.RecommendationRecord{ID: id, Provider: provider, Service: service, Region: "us-east-1", + ResourceType: "boundary-" + id, Engine: "postgres", Count: count, Term: 1, Payment: "no-upfront"} + } + at20 := mk("bound|20", "aws", "rds", 5) + at20.SavingsPercentage, at20.Savings = pfloat(20), 300 + at35 := mk("bound|35", "aws", "rds", 5) + at35.SavingsPercentage, at35.Savings = pfloat(35), 200 + at57 := mk("bound|57", "aws", "rds", 5) + at57.SavingsPercentage = pfloat(57) + od57 := mk("bound|od57", "aws", "rds", 5) + od57.Savings, od57.OnDemandCost = 57, pfloat(100) + mc57 := mk("bound|mc57", "gcp", "compute", 7) + mc57.Savings, mc57.MonthlyCost = 57, pfloat(43) + return []config.RecommendationRecord{at20, at35, at57, od57, mc57} +} + +// naturalKeyOf mirrors the cache's natural key, so the fixture never upserts +// two rows onto one key: the account (nil and missing collapse), the +// identity columns, and availability zone, platform and tenancy read +// leniently from Details as the generated columns do. +func naturalKeyOf(rec *config.RecommendationRecord) string { + account := "" + if rec.CloudAccountID != nil { + account = *rec.CloudAccountID + } + var d struct { + Platform string `json:"platform"` + Tenancy string `json:"tenancy"` + AZ string `json:"availability_zone"` + } + _ = json.Unmarshal(rec.Details, &d) // a non-object reads as no platform or tenancy + az := rec.AvailabilityZone + if az == "" { + az = d.AZ + } + return fmt.Sprintf("%s|%s|%s|%s|%s|%s|%d|%s|%s|%s|%s", account, rec.Provider, rec.Service, rec.Region, + rec.ResourceType, rec.Engine, rec.Term, rec.Payment, az, d.Platform, d.Tenancy) +} + +// setPctInputs covers each branch of the savings-pct computation: a provider +// percentage, an on-demand baseline (positive or zero), a reconstructed +// baseline from monthly cost, and AWS rows without a baseline. +func (f *pageSQLFixture) setPctInputs(rng *rand.Rand, rec *config.RecommendationRecord) { + switch rng.Intn(5) { + case 0: + rec.SavingsPercentage = pfloat(float64(rng.Intn(9000)) / 100) + case 1: + rec.OnDemandCost = pfloat(rec.Savings * float64(1+rng.Intn(9)) / 3) + case 2: + rec.OnDemandCost = pfloat(0) + rec.MonthlyCost = pfloat(float64(rng.Intn(400000)) / 9) + case 3: + rec.MonthlyCost = pfloat(float64(rng.Intn(400000)) / 9) + } +} + +// seedSuppressions adds active suppressions on the 6-tuples of some recs: +// matching and non-matching variants, legacy rows without a variant, the +// legacy "ec2" slug against "compute" recs, counts below and above the rec's +// count, and an expired row that must not apply. +func (f *pageSQLFixture) seedSuppressions(ctx context.Context, t *testing.T, pg *testhelpers.PostgresContainer) { + t.Helper() + rng := rand.New(rand.NewSource(1981)) + for n := 0; n < 60; n++ { + rec := f.recs[rng.Intn(len(f.recs))] + account := "" + if rec.CloudAccountID != nil { + account = *rec.CloudAccountID + } + service := rec.Service + if rec.Provider == "aws" && service == "compute" && rng.Intn(2) == 0 { + service = "ec2" + } + var az, platform, tenancy any + if v, err := rec.SuppressionVariant(); err == nil && rng.Intn(3) > 0 { + az, platform, tenancy = v.AvailabilityZone, v.Platform, v.Tenancy + if rng.Intn(4) == 0 { + tenancy = "shared" + } + if rng.Intn(5) == 0 { + platform = "other" + } + } + expires := "NOW() + INTERVAL '1 day'" + if n%13 == 0 { + expires = "NOW() - INTERVAL '1 day'" + } + _, err := pg.DB.Pool().Exec(ctx, ` + INSERT INTO purchase_suppressions (execution_id, account_id, provider, service, region, resource_type, engine, + suppressed_count, expires_at, availability_zone, platform, tenancy) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, `+expires+`, $9, $10, $11)`, + uuid.NewString(), account, rec.Provider, service, rec.Region, rec.ResourceType, rec.Engine, + 1+rng.Intn(rec.Count+2), az, platform, tenancy) + require.NoError(t, err) + } +} + +// seedOverrides adds global service configurations and per-account +// overrides on every dimension the override filter has. +func (f *pageSQLFixture) seedOverrides(ctx context.Context, t *testing.T) { + t.Helper() + for _, c := range []config.ServiceConfig{ + {Provider: "aws", Service: "rds", Enabled: true, ExcludeEngines: []string{"mysql"}, MinCount: 3}, + {Provider: "aws", Service: "compute", Enabled: true, IncludeRegions: []string{"us-east-1", "us-west-2"}, ExcludeTypes: []string{"c5.xlarge"}}, + {Provider: "aws", Service: "savings-plans-compute", Enabled: true, MinCount: 50}, + {Provider: "azure", Service: "compute", Enabled: false}, + {Provider: "gcp", Service: "compute", Enabled: true, IncludeEngines: []string{"postgres"}, MinCount: 6}, + } { + c := c + require.NoError(t, f.store.SaveServiceConfig(ctx, &c)) + } + yes, no := true, false + for _, o := range []config.AccountServiceOverride{ + {AccountID: f.accounts[0].ID, Provider: "aws", Service: "rds", Enabled: &no}, + {AccountID: f.accounts[1].ID, Provider: "aws", Service: "compute", IncludeRegions: []string{"eu-west-1"}}, + {AccountID: f.accounts[2].ID, Provider: "azure", Service: "compute", Enabled: &yes, ExcludeRegions: []string{"us-west-2"}}, + {AccountID: f.accounts[0].ID, Provider: "aws", Service: "ec2", IncludeTypes: []string{"m5.large"}}, + } { + o := o + require.NoError(t, f.store.SaveAccountServiceOverride(ctx, &o)) + } +} + +// scopes are the account scopes the comparison runs under: unrestricted, +// one registered account, a registered and an unregistered account, an +// account the caller can name only by external ID, and nothing at all. +func (f *pageSQLFixture) scopes() map[string]auth.AccountScope { + registered := make([]auth.RegisteredAccount, len(f.accounts)) + for i, a := range f.accounts { + registered[i] = auth.RegisteredAccount{ID: a.ID, Name: a.Name, ExternalID: a.ExternalID} + } + scoped := func(ids ...string) auth.AccountScope { + return auth.ScopeForAccounts(ids).WithRegisteredAccounts(registered) + } + return map[string]auth.AccountScope{ + "unrestricted": auth.UnrestrictedScope(), + "alpha": scoped(f.accounts[0].ID), + "beta+unreg": scoped(f.accounts[1].ID, f.unreg), + "external-id": scoped(f.accounts[2].ExternalID, f.accounts[2].Name), + "nothing": auth.ScopeForAccounts(nil), + "alpha-unscoped": auth.ScopeForAccounts([]string{f.accounts[0].ID}), + } +} + +// filters are the query-string filters the comparison runs with. +func (f *pageSQLFixture) filters() map[string]map[string]string { + return map[string]map[string]string{ + "none": {}, + "aws": {"provider": "aws"}, + "compute": {"service": "compute"}, + "region": {"region": "us-east-1"}, + "accounts": {"account_ids": f.accounts[0].ID + "," + f.accounts[2].ID}, + "usd": {"min_savings_usd": "20000"}, + "pct": {"min_savings_pct": "20"}, + "pct-aws-usd": {"min_savings_pct": "35", "provider": "aws", "min_savings_usd": "100"}, + "pct-high": {"min_savings_pct": "95"}, + "pct-57": {"min_savings_pct": "57"}, + "azure-account": {"provider": "azure", "account_ids": f.accounts[2].ID}, + } +} diff --git a/internal/api/handler_recommendations_page_sql_integration_test.go b/internal/api/handler_recommendations_page_sql_integration_test.go new file mode 100644 index 00000000..563cbfed --- /dev/null +++ b/internal/api/handler_recommendations_page_sql_integration_test.go @@ -0,0 +1,198 @@ +//go:build integration + +package api + +import ( + "context" + "fmt" + "sort" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" +) + +// pageWalkStats records what a comparison walk covered, so the test can prove +// the fixture exercised the filters rather than comparing empty pages. +type pageWalkStats struct { + rows, suppressed int +} + +// comparePageWalk walks every page of one request through the SQL page read +// and through the Go-side scan (the pre-#981 path, kept as the oracle), page +// by page from the same cursor, and requires identical responses: rows, +// suppression badges, summary totals to the bit, regions and next_cursor. +func comparePageWalk(ctx context.Context, t *testing.T, f *pageSQLFixture, scope auth.AccountScope, params map[string]string, limit int) pageWalkStats { + t.Helper() + filter, err := parseRecommendationFilter(params) + require.NoError(t, err) + rows, err := f.h.recommendationsRowScope(ctx, scope, filter) + require.NoError(t, err) + var stats pageWalkStats + afterID := "" + for page := 0; ; page++ { + require.Less(t, page, len(f.recs)+2, "cursor walk did not terminate") + req := recommendationsPage{enabled: true, limit: limit, afterID: afterID} + got, err := f.h.getRecommendationsPageFromSQL(ctx, scope, rows, req) + require.NoError(t, err, "page %d: the SQL page read must not fall back on this fixture", page) + want, err := f.h.getRecommendationsPageByScan(ctx, scope, filter, rows.keep, req) + require.NoError(t, err) + sort.Strings(got.Regions) + sort.Strings(want.Regions) + require.Equal(t, want, got, "page %d", page) + for i := range got.Recommendations { + rec := &got.Recommendations[i] + if !scope.AllowsAll() { + // An account-scoped caller must never see an out-of-scope row. + require.NotNil(t, rec.CloudAccountID) + require.Contains(t, rows.sql.AllowedAccountIDs, *rec.CloudAccountID) + } + if rec.SuppressedCount > 0 { + stats.suppressed++ + } + } + stats.rows += len(got.Recommendations) + if got.NextCursor == "" { + assert.Equal(t, got.Summary.TotalCount, stats.rows, "the walk returns exactly the counted rows") + return stats + } + afterID, err = decodeRecommendationsCursor(got.NextCursor) + require.NoError(t, err) + } +} + +// TestRecommendationsPageSQL_MatchesScanOracle is the equivalence proof for +// issue #981: on a varied Postgres cache, every combination of enabled +// providers, account scope, query-string filter and page size returns, page +// by page, exactly what the Go-side scan returns. Account-scoped callers are +// included, and the summary they get must equal the scan's, which only +// counts in-scope rows. +func TestRecommendationsPageSQL_MatchesScanOracle(t *testing.T) { + ctx := t.Context() + f := newPageSQLFixture(ctx, t) + var total pageWalkStats + for _, enabled := range [][]string{nil, {"aws", "gcp"}} { + require.NoError(t, f.store.SaveGlobalConfig(ctx, &config.GlobalConfig{EnabledProviders: enabled, DefaultTerm: 1, DefaultPayment: "all-upfront"})) + for scopeName, scope := range f.scopes() { + for filterName, params := range f.filters() { + limits := []int{97, 1000} + if filterName == "none" || filterName == "pct" { + limits = append(limits, 7) + } + for _, limit := range limits { + name := fmt.Sprintf("enabled=%v/scope=%s/filter=%s/limit=%d", enabled, scopeName, filterName, limit) + t.Run(name, func(t *testing.T) { + stats := comparePageWalk(ctx, t, f, scope, params, limit) + total.rows += stats.rows + total.suppressed += stats.suppressed + }) + } + } + } + } + assert.Positive(t, total.rows) + assert.Positive(t, total.suppressed, "the fixture must reach rows whose count a suppression reduced") +} + +// TestRecommendationsPageSQL_FixtureFiltersBite checks that the fixture's +// filters drop rows, so the oracle comparison is not between two unfiltered +// lists. +func TestRecommendationsPageSQL_FixtureFiltersBite(t *testing.T) { + ctx := t.Context() + f := newPageSQLFixture(ctx, t) + scope := auth.UnrestrictedScope() + rows, err := f.h.recommendationsRowScope(ctx, scope, config.RecommendationFilter{}) + require.NoError(t, err) + resp, err := f.h.getRecommendationsPageFromSQL(ctx, scope, rows, recommendationsPage{enabled: true, limit: 1}) + require.NoError(t, err) + assert.Positive(t, resp.Summary.TotalCount) + assert.Less(t, resp.Summary.TotalCount, len(f.recs), "suppressions and overrides must drop rows") +} + +// pageFallbackHandler stores recs in a fresh database, runs edit (raw SQL +// for payload shapes the writer cannot produce) when set, and returns a +// handler reading them. +func pageFallbackHandler(ctx context.Context, t *testing.T, recs []config.RecommendationRecord, edit string) *Handler { + t.Helper() + pg := testhelpers.RequireMigratedDatabase(ctx, t, "../database/postgres/migrations") + store := config.NewPostgresStore(pg.DB) + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + if edit != "" { + _, err := pg.DB.Pool().Exec(ctx, edit) + require.NoError(t, err) + } + return &Handler{config: store, scheduler: scheduler.NewScheduler(scheduler.SchedulerConfig{ConfigStore: store, IsLambda: true})} +} + +func fallbackRec(id, resourceType string, savings float64) config.RecommendationRecord { + return config.RecommendationRecord{ID: id, Provider: "azure", Service: "compute", Region: "westeurope", ResourceType: resourceType, + Count: 1, Term: 1, Payment: "no-upfront", Savings: savings} +} + +// TestRecommendationsPageSQL_FallsBackToScan covers the inputs SQL cannot +// handle like Go. Each must make the SQL read report that it needs the scan, +// and the endpoint must then answer exactly as the scan does. +func TestRecommendationsPageSQL_FallsBackToScan(t *testing.T) { + ctx := t.Context() + cases := []struct { + name string + recs func() []config.RecommendationRecord + params map[string]string + afterID string + edit string + }{ + {"pct outside float-safe bounds", func() []config.RecommendationRecord { + r := fallbackRec("a", "x1", 5) + r.SavingsPercentage = pfloat(1e200) + return []config.RecommendationRecord{r, fallbackRec("b", "x2", 6)} + }, map[string]string{"min_savings_pct": "10"}, "", ""}, + {"savings outside float-safe bounds", func() []config.RecommendationRecord { + return []config.RecommendationRecord{fallbackRec("a", "x1", 1e-200), fallbackRec("b", "x2", 6)} + }, map[string]string{}, "", ""}, + {"cursor is not UTF-8", func() []config.RecommendationRecord { + return []config.RecommendationRecord{fallbackRec("a", "x1", 5), fallbackRec("\xc3", "x2", 6), fallbackRec("z", "x3", 7)} + }, map[string]string{}, "a\xff", ""}, + {"empty but present account", func() []config.RecommendationRecord { + return []config.RecommendationRecord{fallbackRec("a", "x1", 5), fallbackRec("b", "x2", 6)} + }, map[string]string{}, "", `UPDATE recommendations SET payload = jsonb_set(payload, '{cloud_account_id}', '""') WHERE payload->>'id' = 'a'`}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + h := pageFallbackHandler(ctx, t, tc.recs(), tc.edit) + filter, err := parseRecommendationFilter(tc.params) + require.NoError(t, err) + scope := auth.UnrestrictedScope() + rows, err := h.recommendationsRowScope(ctx, scope, filter) + require.NoError(t, err) + req := recommendationsPage{enabled: true, limit: 1, afterID: tc.afterID} + _, err = h.getRecommendationsPageFromSQL(ctx, scope, rows, req) + require.ErrorIs(t, err, scheduler.ErrRecommendationsPageNeedsScan) + got, err := h.getRecommendationsPage(ctx, scope, filter, req) + require.NoError(t, err) + want, err := h.getRecommendationsPageByScan(ctx, scope, filter, rows.keep, req) + require.NoError(t, err) + assert.Equal(t, want, got) + }) + } +} + +// TestRecommendationsPageSQL_DuplicateIDsStillFail keeps the scan's refusal +// to page a cache where two rows share an ID: the SQL read hands the request +// to the scan, which fails it. +func TestRecommendationsPageSQL_DuplicateIDsStillFail(t *testing.T) { + ctx := t.Context() + h := pageFallbackHandler(ctx, t, []config.RecommendationRecord{fallbackRec("dup", "x1", 5), fallbackRec("dup", "x2", 6)}, "") + scope := auth.UnrestrictedScope() + rows, err := h.recommendationsRowScope(ctx, scope, config.RecommendationFilter{}) + require.NoError(t, err) + _, err = h.getRecommendationsPageFromSQL(ctx, scope, rows, recommendationsPage{enabled: true, limit: 5}) + require.ErrorIs(t, err, scheduler.ErrRecommendationsPageNeedsScan) + _, err = h.getRecommendationsPage(ctx, scope, config.RecommendationFilter{}, recommendationsPage{enabled: true, limit: 5}) + require.ErrorContains(t, err, "duplicate recommendation ID") +} diff --git a/internal/api/handler_recommendations_page_sql_test.go b/internal/api/handler_recommendations_page_sql_test.go new file mode 100644 index 00000000..81a08053 --- /dev/null +++ b/internal/api/handler_recommendations_page_sql_test.go @@ -0,0 +1,134 @@ +package api + +import ( + "context" + "errors" + "sort" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" +) + +// TestGetRecommendations_PagedReadUsesSQLPage checks the issue #981 path: a +// paginated request takes its page and summary from PageRecommendations, +// passes the handler-side filters down (enabled providers and the scoped +// reader's exact account set), asks for one row past the page, and never +// scans the cache. +func TestGetRecommendations_PagedReadUsesSQLPage(t *testing.T) { + h, sched := scanTestHandler(scanFilterRecs(), 4, true) + in := scopedInAccount + rows := []config.RecommendationRecord{ + {ID: "a", Provider: "aws", CloudAccountID: &in, Region: "us-east-1"}, + {ID: "b", Provider: "aws", CloudAccountID: &in, Region: "us-east-1"}, + {ID: "c", Provider: "aws", CloudAccountID: &in, Region: "us-east-1"}, + } + sched.On("PageRecommendations", mock.Anything, mock.Anything, "Z", 3).Return(&config.RecommendationPage{ + TotalCount: 42, TotalSavings: 10, TotalUpfront: 25, Regions: []string{"us-east-1", "eu-west-1"}, Rows: rows, + }, nil) + + resp, err := h.getRecommendations(context.Background(), scopedRequest(""), + map[string]string{"limit": "2", "cursor": encodeRecommendationsCursor("Z"), "provider": "aws"}) + require.NoError(t, err) + + assert.Empty(t, sched.ctxs, "the paginated read must not scan") + assert.Equal(t, RecommendationsSummary{TotalCount: 42, TotalMonthlySavings: 10, TotalUpfrontCost: 25, AvgPaybackMonths: 2.5}, resp.Summary) + assert.ElementsMatch(t, []string{"us-east-1", "eu-west-1"}, resp.Regions) + require.Len(t, resp.Recommendations, 2) + assert.Equal(t, encodeRecommendationsCursor("b"), resp.NextCursor) + + call := sched.Calls[len(sched.Calls)-1] + assert.False(t, scheduler.BackgroundCollectionAllowed(call.Arguments.Get(0).(context.Context)), "a scoped reader never kicks a collection") + scope := call.Arguments.Get(1).(config.RecommendationListScope) + assert.Equal(t, "aws", scope.Filter.Provider) + assert.Equal(t, []string{"aws"}, scope.EnabledProviders) + assert.True(t, scope.Restricted) + assert.Equal(t, []string{scopedInAccount}, scope.AllowedAccountIDs) +} + +// TestGetRecommendations_SQLPageFallsBackToScan checks every way the SQL +// page hands the request to the scan: the scheduler says so, a row the +// handler's own filters reject, or rows out of cursor order. The response is +// then exactly the scan's. +func TestGetRecommendations_SQLPageFallsBackToScan(t *testing.T) { + in, out := scopedInAccount, "22222222-2222-4222-8222-222222222222" + cases := map[string]struct { + page *config.RecommendationPage + err error + }{ + "scheduler needs scan": {nil, scheduler.ErrRecommendationsPageNeedsScan}, + "row out of scope": {&config.RecommendationPage{TotalCount: 1, Rows: []config.RecommendationRecord{ + {ID: "a", Provider: "aws", CloudAccountID: &out}}}, nil}, + "row of a disabled provider": {&config.RecommendationPage{TotalCount: 1, Rows: []config.RecommendationRecord{ + {ID: "a", Provider: "azure", CloudAccountID: &in}}}, nil}, + "rows out of order": {&config.RecommendationPage{TotalCount: 2, Rows: []config.RecommendationRecord{ + {ID: "b", Provider: "aws", CloudAccountID: &in}, {ID: "a", Provider: "aws", CloudAccountID: &in}}}, nil}, + "rows not after the cursor": {&config.RecommendationPage{TotalCount: 1, Rows: []config.RecommendationRecord{ + {ID: "", Provider: "aws", CloudAccountID: &in}}}, nil}, + } + params := map[string]string{"limit": "3"} + scanOnly, _ := scanTestHandler(scanFilterRecs(), 4, true) + want, err := scanOnly.getRecommendations(context.Background(), scopedRequest(""), params) + require.NoError(t, err) + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + h, sched := scanTestHandler(scanFilterRecs(), 4, true) + sched.On("PageRecommendations", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(tc.page, tc.err) + if name == "rows not after the cursor" { + params = map[string]string{"limit": "3", "cursor": encodeRecommendationsCursor("A")} + var err error + want, err = scanOnly.getRecommendations(context.Background(), scopedRequest(""), params) + require.NoError(t, err) + } + got, err := h.getRecommendations(context.Background(), scopedRequest(""), params) + require.NoError(t, err) + assert.Len(t, sched.ctxs, 1, "the request is served by the scan") + sort.Strings(got.Regions) + sort.Strings(want.Regions) + assert.Equal(t, want, got) + }) + } +} + +// TestGetRecommendations_SQLPageErrorFails keeps a real read failure a +// failure: only ErrRecommendationsPageNeedsScan falls back. +func TestGetRecommendations_SQLPageErrorFails(t *testing.T) { + h, sched := scanTestHandler(scanFilterRecs(), 4, false) + sched.On("PageRecommendations", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, errors.New("apply suppressions: db down")) + _, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{"limit": "3"}) + require.ErrorContains(t, err, "db down") + assert.Empty(t, sched.ctxs) +} + +// TestScopeAllowedAccountIDs checks that the SQL account set is exactly what +// AccountScope.Allows grants a rec's account ID: every non-empty allow-list +// entry taken literally (registered or not), and no registered account the +// list only names by external ID or display name. +func TestScopeAllowedAccountIDs(t *testing.T) { + registered := []auth.RegisteredAccount{ + {ID: "id-a", Name: "alpha", ExternalID: "ext-a"}, + {ID: "id-b", Name: "beta", ExternalID: "ext-b"}, + } + nameByID := map[string]string{"id-a": "alpha", "id-b": "beta"} + scope := auth.ScopeForAccounts([]string{"id-a", "unregistered", "ext-b", "beta", ""}).WithRegisteredAccounts(registered) + got := scopeAllowedAccountIDs(scope, nameByID) + assert.Equal(t, []string{"beta", "ext-b", "id-a", "unregistered"}, got) + for _, id := range []string{"id-a", "id-b", "unregistered", "ext-b", "beta", ""} { + assert.Equal(t, scope.Allows(id, nameByID[id]), contains(got, id), id) + } + assert.Empty(t, scopeAllowedAccountIDs(auth.ScopeForAccounts(nil), nameByID)) +} + +func contains(list []string, s string) bool { + for _, v := range list { + if v == s { + return true + } + } + return false +} diff --git a/internal/api/handler_recommendations_scan.go b/internal/api/handler_recommendations_scan.go index 4ef8aae4..cdc58d27 100644 --- a/internal/api/handler_recommendations_scan.go +++ b/internal/api/handler_recommendations_scan.go @@ -2,38 +2,75 @@ package api import ( "context" + "errors" "fmt" + "sort" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" ) -// recommendationsScanChunkSize is how many cache rows a paginated -// GET /api/recommendations reads per store query. Peak memory is about one -// chunk plus one page, whatever the tenant's size. +// recommendationsScanChunkSize is how many cache rows the Go-side scan of a +// paginated GET /api/recommendations (the fallback of the SQL page read) +// reads per store query. Peak memory is about one chunk plus one page, +// whatever the tenant's size. const recommendationsScanChunkSize = 500 // getRecommendationsPage serves a paginated GET /api/recommendations (issue -// #981). Instead of loading and sorting the whole filtered set, it scans the -// cache in ID-ordered chunks through the scheduler (which applies -// suppressions and account overrides), drops rows for disabled providers and -// accounts outside the caller's scope, and hands the survivors to a collector -// that keeps only the requested page while totaling every row for the -// summary. +// #981). It asks the scheduler for the page with every list filter evaluated +// in SQL: the store totals the summary and regions in one aggregate query and +// returns only the page plus one row, so neither the rows read into Go nor +// the memory held grow with the tenant. // -// The summary still needs every filtered row, so each request still reads the -// whole filtered set from the store; what no longer grows with the tenant is -// the memory held and the in-process sort. Pushing the summary into SQL would -// need suppressions, account overrides, the provider and scope filters and -// the savings-percentage floor in SQL too; until then a SQL aggregate would -// count rows the response hides. +// When SQL cannot reproduce a Go filter faithfully for this request (see +// scheduler.PageRecommendations) the page is served by the Go-side scan +// instead, which reads every filtered row the way the endpoint did before. func (h *Handler) getRecommendationsPage(ctx context.Context, scope auth.AccountScope, filter config.RecommendationFilter, page recommendationsPage) (*RecommendationsResponse, error) { - keep, err := h.recommendationsRowFilter(ctx, scope) + rows, err := h.recommendationsRowScope(ctx, scope, filter) if err != nil { return nil, err } + resp, err := h.getRecommendationsPageFromSQL(ctx, scope, rows, page) + if !errors.Is(err, scheduler.ErrRecommendationsPageNeedsScan) { + return resp, err + } + logging.Warnf("paginated recommendations: %v; serving the page from the scan", err) + return h.getRecommendationsPageByScan(ctx, scope, filter, rows.keep, page) +} + +// getRecommendationsPageFromSQL reads the page and the full-set summary from +// the scheduler's SQL page read. The rows are checked against the handler's +// own filters and the page order once more; a disagreement returns +// scheduler.ErrRecommendationsPageNeedsScan rather than a wrong page. +func (h *Handler) getRecommendationsPageFromSQL(ctx context.Context, scope auth.AccountScope, rows recommendationsRowScope, page recommendationsPage) (*RecommendationsResponse, error) { + pg, err := h.scheduler.PageRecommendations(recommendationsReadContext(ctx, scope), rows.sql, page.afterID, page.limit+1) + if err != nil { + if errors.Is(err, scheduler.ErrRecommendationsPageNeedsScan) { + return nil, err + } + return nil, fmt.Errorf("failed to get recommendations: %w", err) + } + if kept := rows.keep(append([]config.RecommendationRecord(nil), pg.Rows...)); len(kept) != len(pg.Rows) { + return nil, fmt.Errorf("%w: SQL kept a row outside the enabled providers or account scope", scheduler.ErrRecommendationsPageNeedsScan) + } collector := newRecommendationsPageCollector(page) - err = h.scheduler.ScanRecommendations(recommendationsReadContext(ctx, scope), filter, recommendationsScanChunkSize, + if err := collector.setPage(pg); err != nil { + return nil, fmt.Errorf("%w: %w", scheduler.ErrRecommendationsPageNeedsScan, err) + } + return collector.response(recommendationsPageByteBudget) +} + +// getRecommendationsPageByScan is the Go-side page: it scans the cache in +// ID-ordered chunks through the scheduler (which applies suppressions and +// account overrides), drops rows the handler-side filters reject, and hands +// the survivors to a collector that keeps only the requested page while +// totaling every row for the summary. Each request reads the whole filtered +// set; only the memory held stays bounded. +func (h *Handler) getRecommendationsPageByScan(ctx context.Context, scope auth.AccountScope, filter config.RecommendationFilter, keep func([]config.RecommendationRecord) []config.RecommendationRecord, page recommendationsPage) (*RecommendationsResponse, error) { + collector := newRecommendationsPageCollector(page) + err := h.scheduler.ScanRecommendations(recommendationsReadContext(ctx, scope), filter, recommendationsScanChunkSize, func(chunk []config.RecommendationRecord) error { return collector.add(keep(chunk)) }) @@ -43,39 +80,78 @@ func (h *Handler) getRecommendationsPage(ctx context.Context, scope auth.Account return collector.response(recommendationsPageByteBudget) } -// recommendationsRowFilter returns the handler-side row filters of +// recommendationsRowScope is the handler-side row filter of // getRecommendations (enabled providers, issue #1409, then the caller's -// account scope) as one function, with their inputs read once up front -// rather than once per chunk. -func (h *Handler) recommendationsRowFilter(ctx context.Context, scope auth.AccountScope) (func([]config.RecommendationRecord) []config.RecommendationRecord, error) { +// account scope) in two forms: keep applies it in Go, and sql carries it, +// with the request's filter, to the SQL page read. +type recommendationsRowScope struct { + keep func([]config.RecommendationRecord) []config.RecommendationRecord + sql config.RecommendationListScope +} + +// recommendationsRowScope reads the filters' inputs once up front rather +// than once per chunk. +func (h *Handler) recommendationsRowScope(ctx context.Context, scope auth.AccountScope, filter config.RecommendationFilter) (recommendationsRowScope, error) { var enabledProviders []string if h.config != nil { globalCfg, err := h.config.GetGlobalConfig(ctx) if err != nil { - return nil, fmt.Errorf("failed to read global config for enabled-providers filter: %w", err) + return recommendationsRowScope{}, fmt.Errorf("failed to read global config for enabled-providers filter: %w", err) } enabledProviders = globalCfg.EnabledProviders } + out := recommendationsRowScope{sql: config.RecommendationListScope{Filter: filter, EnabledProviders: enabledProviders}} var nameByID map[string]string if !scope.AllowsAll() { var err error if nameByID, err = h.cloudAccountNamesByID(ctx); err != nil { - return nil, err + return recommendationsRowScope{}, err } + out.sql.Restricted = true + out.sql.AllowedAccountIDs = scopeAllowedAccountIDs(scope, nameByID) } - return func(recs []config.RecommendationRecord) []config.RecommendationRecord { + out.keep = func(recs []config.RecommendationRecord) []config.RecommendationRecord { recs = config.FilterRecommendationsByEnabledProviders(recs, enabledProviders) if scope.AllowsAll() { return recs } return filterRecommendationsInScope(scope, nameByID, recs) - }, nil + } + return out, nil +} + +// scopeAllowedAccountIDs is the exact set of account IDs +// filterRecommendationsInScope keeps for a restricted scope. It keeps a rec +// whose account is id when scope.Allows(id, nameByID[id]), which can only +// hold when id is on the allow-list or carries a registered name, so testing +// those candidates finds every allowed ID. +func scopeAllowedAccountIDs(scope auth.AccountScope, nameByID map[string]string) []string { + ids := []string{} + seen := make(map[string]struct{}, len(scope.Accounts)+len(nameByID)) + consider := func(id string) { + if _, ok := seen[id]; ok { + return + } + seen[id] = struct{}{} + if scope.Allows(id, nameByID[id]) { + ids = append(ids, id) + } + } + for _, id := range scope.Accounts { + consider(id) + } + for id := range nameByID { + consider(id) + } + sort.Strings(ids) + return ids } // recommendationsPageCollector builds one page from a scan's surviving rows, // which arrive in strictly increasing ID order. It totals every row into the // summary but keeps at most limit+1 rows after the cursor: the page, plus one -// row that proves a next page exists. +// row that proves a next page exists. The SQL page read fills it with +// setPage instead. type recommendationsPageCollector struct { page recommendationsPage summary recommendationsSummary @@ -117,6 +193,31 @@ func (c *recommendationsPageCollector) add(recs []config.RecommendationRecord) e return nil } +// setPage loads a page the store already filtered and totaled. The rows must +// be the first limit+1 after the cursor in strictly increasing ID order; any +// other order breaks the cursor walk and is an error. +func (c *recommendationsPageCollector) setPage(pg *config.RecommendationPage) error { + prev := c.page.afterID + for i := range pg.Rows { + if id := pg.Rows[i].ID; (i > 0 || prev != "") && id <= prev { + return fmt.Errorf("recommendations page out of ID order: %q after %q", id, prev) + } + prev = pg.Rows[i].ID + } + if len(pg.Rows) > c.page.limit+1 { + return fmt.Errorf("recommendations page returned %d rows for limit %d", len(pg.Rows), c.page.limit) + } + c.rows = pg.Rows + c.summary = recommendationsSummary{count: pg.TotalCount, totalSavings: pg.TotalSavings, totalUpfront: pg.TotalUpfront} + for _, r := range pg.Regions { + if c.summary.regionSet == nil { + c.summary.regionSet = make(map[string]struct{}, len(pg.Regions)) + } + c.summary.regionSet[r] = struct{}{} + } + return nil +} + // response cuts the collected rows to one page by row count and serialized // bytes (see selectRecommendationsPage) and sets next_cursor when any row is // left after it. diff --git a/internal/api/mocks_test.go b/internal/api/mocks_test.go index b840c4c9..1f124b30 100644 --- a/internal/api/mocks_test.go +++ b/internal/api/mocks_test.go @@ -160,6 +160,20 @@ func (m *MockScheduler) ScanRecommendations(ctx context.Context, filter config.R return args.Error(1) } +// PageRecommendations returns the registered page when the test sets an +// expectation; otherwise it sends the handler to the scan, so tests written +// against ScanRecommendations keep exercising the paginated list. +func (m *MockScheduler) PageRecommendations(ctx context.Context, scope config.RecommendationListScope, afterID string, limit int) (*config.RecommendationPage, error) { + for _, call := range m.ExpectedCalls { + if call.Method == "PageRecommendations" { + args := m.Called(ctx, scope, afterID, limit) + page, _ := args.Get(0).(*config.RecommendationPage) + return page, args.Error(1) + } + } + return nil, scheduler.ErrRecommendationsPageNeedsScan +} + func (m *MockScheduler) GetRecommendationByID(ctx context.Context, id string) (*config.RecommendationRecord, []string, error) { args := m.Called(ctx, id) var rec *config.RecommendationRecord diff --git a/internal/api/types.go b/internal/api/types.go index 43b0fb0a..f1eaea97 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -172,6 +172,11 @@ type SchedulerInterface interface { // byte-wise ID order, chunkSize store rows at a time, calling fn with // each chunk's survivors (issue #981). ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error + // PageRecommendations returns one page of the list and the summary of + // the whole filtered set with every filter evaluated in SQL (issue + // #981). An error wrapping scheduler.ErrRecommendationsPageNeedsScan + // means the request must be served by ScanRecommendations instead. + PageRecommendations(ctx context.Context, scope config.RecommendationListScope, afterID string, limit int) (*config.RecommendationPage, error) // GetRecommendationByID fetches a single rec by its application-level id, // bypassing account-override filtering so deep-linked URLs to override- // hidden recs resolve. hiddenBy is non-nil when the rec would be dropped by diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index f34129c3..acbea186 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -682,6 +682,14 @@ type StoreInterface interface { // keyset walk: at most limit rows after the position after, in // byte-wise ID order then primary-key order (issue #981). ListStoredRecommendationsChunk(ctx context.Context, filter RecommendationFilter, after *RecommendationScanKey, limit int) (RecommendationChunk, error) + // ListStoredRecommendationTriples lists the distinct (account, + // provider, service) keys of the rows scope selects, for resolving + // account overrides before PageStoredRecommendations (issue #981). + ListStoredRecommendationTriples(ctx context.Context, scope RecommendationListScope) ([]RecommendationTriple, error) + // PageStoredRecommendations reads one page of the recommendations list + // and the summary of the whole filtered set with every list filter + // evaluated in SQL, in one snapshot (issue #981). + PageStoredRecommendations(ctx context.Context, q RecommendationPageQuery) (RecommendationPage, error) GetRecommendationsFreshness(ctx context.Context) (*RecommendationsFreshness, error) SetRecommendationsCollectionError(ctx context.Context, errMsg string) error // SetRecommendationsCollectionPartial records (or, with nil, clears) the diff --git a/internal/config/recommendations_page.go b/internal/config/recommendations_page.go new file mode 100644 index 00000000..d213aad7 --- /dev/null +++ b/internal/config/recommendations_page.go @@ -0,0 +1,122 @@ +package config + +// Types of the SQL-side paginated recommendations read (issue #981). A +// paginated GET /api/recommendations used to read every filtered row into Go +// to total the summary; PageStoredRecommendations instead evaluates every +// list filter in SQL, totals the summary in one aggregate query and returns +// only the requested page. + +// RecommendationListScope is the row filter of one recommendations list +// request: the query-string filter plus the handler-side filters, all applied +// to the decoded payload the way the Go filters see it. +type RecommendationListScope struct { + Filter RecommendationFilter + // EnabledProviders keeps only rows whose provider is listed. Empty keeps + // every provider (FilterRecommendationsByEnabledProviders). + EnabledProviders []string + // Restricted keeps only rows whose cloud_account_id is one of + // AllowedAccountIDs; rows without an account are dropped. The caller + // resolves the account scope to the exact set of IDs it allows. + Restricted bool + AllowedAccountIDs []string +} + +// RecommendationTriple is an (account, provider, service) combination of the +// stored rows, the key account overrides are resolved by. AccountID is "" +// when the payload has no cloud_account_id; HasAccount tells that apart from +// a present but empty one, which override resolution treats differently. +type RecommendationTriple struct { + AccountID string + Provider string + Service string + HasAccount bool +} + +// RecommendationOverrideRule is the resolved account-override configuration +// of one AccountConfigKey. Config is nil when neither a global service +// configuration nor an account override exists, so the rows pass unfiltered. +// MinCountExempt is set for Savings Plans services, whose count is not an +// instance count. +type RecommendationOverrideRule struct { + Key string + Config *ServiceConfig + MinCountExempt bool +} + +// StoredRecommendation is a decoded cache row with its table primary key. +type StoredRecommendation struct { + RowID string + Rec RecommendationRecord +} + +// RecommendationPageQuery parameterizes PageStoredRecommendations. +// +// Rules must hold an entry for every AccountConfigKey the scope's rows can +// produce (ListStoredRecommendationTriples); a row whose key has no entry +// makes the page fall back, because its overrides were never resolved. +// +// SuppressionKeys are the suppression 6-tuples (see SuppressionTupleKey) +// with every stored spelling of the service. Rows on one of them are read +// first and handed to DecideSuppressed, which returns the primary keys of +// the rows that survive suppressions and overrides; the other rows are +// filtered by Rules in SQL. +type RecommendationPageQuery struct { + Scope RecommendationListScope + Rules []RecommendationOverrideRule + SuppressionKeys []string + DecideSuppressed func(rows []StoredRecommendation) (keepRowIDs []string, err error) + // AfterID returns only rows whose ID sorts byte-wise after it; "" starts + // at the first row. + AfterID string + // Limit caps the returned rows. + Limit int +} + +// RecommendationPage is one page of the filtered recommendations with the +// summary of the whole filtered set. TotalSavings and TotalUpfront are summed +// in ID order, the order the Go summary added them in. Regions holds each +// non-empty region once. +// +// Fallback is non-empty when SQL cannot reproduce the Go filters for this +// read (for example a value whose float arithmetic SQL would reject, or a +// row whose overrides were not resolved); the caller must then serve the +// request from the Go-side scan, and the other fields are unset. +type RecommendationPage struct { + TotalCount int + TotalSavings float64 + TotalUpfront float64 + Regions []string + Rows []RecommendationRecord + Fallback string +} + +// SuppressionTupleKey joins a suppression 6-tuple into the string +// PageStoredRecommendations matches rows against. Two distinct tuples can +// only produce the same key when a value contains the separator; such a row +// is merely decided in Go instead of SQL, which is still exact. +func SuppressionTupleKey(accountID, provider, service, region, resourceType, engine string) string { + const sep = "\x1f" + return accountID + sep + provider + sep + service + sep + region + sep + resourceType + sep + engine +} + +// SuppressionServiceSpellings returns every service spelling a stored rec +// can carry that CanonicalSuppressionService maps to canonical for provider: +// canonical itself (when it maps to itself) and its legacy and dash-free +// aliases. +func SuppressionServiceSpellings(provider, canonical string) []string { + var out []string + if CanonicalSuppressionService(provider, canonical) == canonical { + out = append(out, canonical) + } + if provider != "aws" { + return out + } + for _, aliases := range []map[string]string{awsLegacyServiceSlugs, awsSavingsPlansSpellings} { + for alias := range aliases { + if alias != canonical && CanonicalSuppressionService(provider, alias) == canonical { + out = append(out, alias) + } + } + } + return out +} diff --git a/internal/config/store_postgres_recommendations_page.go b/internal/config/store_postgres_recommendations_page.go new file mode 100644 index 00000000..72b290eb --- /dev/null +++ b/internal/config/store_postgres_recommendations_page.go @@ -0,0 +1,197 @@ +package config + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "unicode/utf8" + + "github.com/jackc/pgx/v5" +) + +// ListStoredRecommendationTriples returns the distinct (account, provider, +// service) combinations of the rows scope selects, read from the payload, +// so the caller can resolve account overrides for exactly those keys before +// PageStoredRecommendations (issue #981). +func (s *PostgresStore) ListStoredRecommendationTriples(ctx context.Context, scope RecommendationListScope) ([]RecommendationTriple, error) { + if err := s.checkAzurePricingMigration(ctx, scope.Filter.RequireAzurePricingMigration); err != nil { + return nil, err + } + query, args := buildPageTriplesQuery(scope) + rows, err := s.db.Query(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to query recommendation override keys: %w", err) + } + defer rows.Close() + var out []RecommendationTriple + for rows.Next() { + var t RecommendationTriple + if err := rows.Scan(&t.AccountID, &t.HasAccount, &t.Provider, &t.Service); err != nil { + return nil, fmt.Errorf("failed to scan recommendation override key: %w", err) + } + out = append(out, t) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("recommendation override keys iteration: %w", err) + } + return out, nil +} + +// PageStoredRecommendations reads one page of the recommendations list with +// every list filter evaluated in SQL (issue #981): the query-string filter, +// the MinSavingsPct floor, the enabled providers, the account scope, account +// overrides (q.Rules) and suppressions (rows on q.SuppressionKeys are decided +// by q.DecideSuppressed). It returns the summary of the whole filtered set +// from one aggregate query and at most q.Limit rows after q.AfterID, so the +// rows handed to Go are the page, not the tenant. +// +// All reads run in one REPEATABLE READ, READ ONLY transaction: the rows +// DecideSuppressed sees, the summary and the page come from the same +// snapshot, so a collection committing mid-request cannot make the page +// disagree with its summary. +// +// When SQL cannot reproduce the Go filters faithfully the page comes back +// with Fallback set instead (see RecommendationPage). +func (s *PostgresStore) PageStoredRecommendations(ctx context.Context, q RecommendationPageQuery) (RecommendationPage, error) { + if fallback, err := checkPageQuery(&q); fallback != "" || err != nil { + return RecommendationPage{Fallback: fallback}, err + } + if err := s.checkAzurePricingMigration(ctx, q.Scope.Filter.RequireAzurePricingMigration); err != nil { + return RecommendationPage{}, err + } + rules, err := pageRulesJSON(q.Rules) + if err != nil { + return RecommendationPage{}, fmt.Errorf("failed to encode override rules: %w", err) + } + + tx, err := s.db.Begin(ctx) + if err != nil { + return RecommendationPage{}, fmt.Errorf("failed to begin recommendations page transaction: %w", err) + } + defer rollbackTx(ctx, tx, "PageStoredRecommendations") + if _, err = tx.Exec(ctx, "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY"); err != nil { + return RecommendationPage{}, fmt.Errorf("failed to set recommendations page isolation: %w", err) + } + page, err := pageInSnapshot(ctx, tx, &q, rules) + if err != nil { + return RecommendationPage{}, err + } + if err = tx.Commit(ctx); err != nil { + return RecommendationPage{}, fmt.Errorf("failed to commit recommendations page transaction: %w", err) + } + return page, nil +} + +// checkPageQuery rejects a malformed query and reports a fallback for a +// cursor Postgres cannot compare: text cannot hold NUL or invalid UTF-8, but +// a client-supplied cursor can, and the Go scan compares it byte-wise all +// the same. +func checkPageQuery(q *RecommendationPageQuery) (fallback string, err error) { + switch { + case q.Limit < 1: + return "", errors.New("recommendations page limit must be positive") + case len(q.SuppressionKeys) > 0 && q.DecideSuppressed == nil: + return "", errors.New("recommendations page has suppression keys but no decider") + case !utf8.ValidString(q.AfterID) || strings.ContainsRune(q.AfterID, 0): + return "cursor is not valid UTF-8 text", nil + } + return "", nil +} + +// pageInSnapshot is PageStoredRecommendations' reads inside its transaction. +func pageInSnapshot(ctx context.Context, tx pgx.Tx, q *RecommendationPageQuery, rules []byte) (RecommendationPage, error) { + keepRowIDs, err := decideTouchedRows(ctx, tx, q) + if err != nil { + return RecommendationPage{}, err + } + page, err := queryPageAggregate(ctx, tx, q, rules, keepRowIDs) + if err != nil || page.Fallback != "" { + return page, err + } + query, args := buildPageRowsQuery(q, rules, keepRowIDs) + rows, err := queryStoredRecommendations(ctx, tx, query, args) + if err != nil { + return RecommendationPage{}, err + } + page.Rows = make([]RecommendationRecord, 0, len(rows)) + for i := range rows { + // The MinSavingsPct floor is checked again in Go on the page: a row + // SQL kept but Go drops means the two disagree, and the request is + // served by the Go scan rather than with a wrong row. + if belowSavingsFloor(&rows[i].Rec, q.Scope.Filter.MinSavingsPct) { + return RecommendationPage{Fallback: fmt.Sprintf("SQL kept recommendation %q below the savings floor", rows[i].Rec.ID)}, nil + } + page.Rows = append(page.Rows, rows[i].Rec) + } + return page, nil +} + +// decideTouchedRows reads the rows on a suppression 6-tuple and returns the +// primary keys q.DecideSuppressed keeps; nil when there are no suppressions. +func decideTouchedRows(ctx context.Context, tx pgx.Tx, q *RecommendationPageQuery) ([]string, error) { + if len(q.SuppressionKeys) == 0 { + return nil, nil + } + query, args := buildPageTouchedQuery(q) + touched, err := queryStoredRecommendations(ctx, tx, query, args) + if err != nil { + return nil, err + } + return q.DecideSuppressed(touched) +} + +// belowSavingsFloor is decodeStoredRecommendation's MinSavingsPct check. +func belowSavingsFloor(rec *RecommendationRecord, floor float64) bool { + if floor <= 0 { + return false + } + pct, ok := recEffectiveSavingsPct(rec) + return ok && pct < floor +} + +// queryPageAggregate runs the summary query and reports a fallback when any +// in-scope row is unsure or two surviving rows share an ID. +func queryPageAggregate(ctx context.Context, tx pgx.Tx, q *RecommendationPageQuery, rules []byte, keepRowIDs []string) (RecommendationPage, error) { + query, args := buildPageAggregateQuery(q, rules, keepRowIDs) + var unsure, distinct int + var page RecommendationPage + err := tx.QueryRow(ctx, query, args...).Scan(&unsure, &page.TotalCount, &distinct, &page.TotalSavings, &page.TotalUpfront, &page.Regions) + if err != nil { + return RecommendationPage{}, fmt.Errorf("failed to aggregate recommendations: %w", err) + } + switch { + case unsure > 0: + return RecommendationPage{Fallback: fmt.Sprintf("%d rows need the Go filters", unsure)}, nil + case distinct != page.TotalCount: + return RecommendationPage{Fallback: "recommendation IDs are not unique"}, nil + } + return page, nil +} + +// queryStoredRecommendations runs a query selecting (id::text, payload) and +// decodes each payload. +func queryStoredRecommendations(ctx context.Context, tx pgx.Tx, query string, args []any) ([]StoredRecommendation, error) { + rows, err := tx.Query(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to query recommendations page rows: %w", err) + } + defer rows.Close() + var out []StoredRecommendation + for rows.Next() { + var row StoredRecommendation + var payload []byte + if err := rows.Scan(&row.RowID, &payload); err != nil { + return nil, fmt.Errorf("failed to scan recommendations page row: %w", err) + } + if err := json.Unmarshal(payload, &row.Rec); err != nil { + return nil, fmt.Errorf("failed to unmarshal payload: %w", err) + } + out = append(out, row) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("recommendations page rows iteration: %w", err) + } + return out, nil +} diff --git a/internal/config/store_postgres_recommendations_page_sql.go b/internal/config/store_postgres_recommendations_page_sql.go new file mode 100644 index 00000000..b553a147 --- /dev/null +++ b/internal/config/store_postgres_recommendations_page_sql.go @@ -0,0 +1,341 @@ +package config + +import ( + "encoding/json" + "fmt" + "strings" +) + +// SQL builders for PageStoredRecommendations (issue #981). Every expression +// reads the decoded payload the way the Go filters do: a missing or JSON-null +// string field is "". +// +// The per-row work is staged in two subqueries, each fenced with OFFSET 0 so +// the planner cannot inline it: a reads each payload field once into a +// column, b derives the filter decisions from a's columns. Inlining would +// substitute every field expression into every decision and multiply the +// expression size (and per-row cost) a hundredfold. See pageStages for how +// the aggregate and the page read join the stages. +// +// The expressions must never raise an error on a row the Go path would have +// handled, because a single bad row would then fail every page. Numbers are +// therefore type-checked with jsonb_typeof before any cast, and float +// arithmetic only runs on magnitudes where it cannot overflow or underflow +// (Postgres raises on both, Go does not). A row outside those bounds is +// reported as "unsure" and the whole request falls back to the Go scan. + +const ( + pageAccountExpr = `COALESCE(r.payload->>'cloud_account_id', '')` + pageProviderExpr = `COALESCE(r.payload->>'provider', '')` + pageServiceExpr = `COALESCE(r.payload->>'service', '')` + pageRegionExpr = `COALESCE(r.payload->>'region', '')` + pageTypeExpr = `COALESCE(r.payload->>'resource_type', '')` + pageEngineExpr = `COALESCE(r.payload->>'engine', '')` + // pageKeyExpr is recommendationScanKeyExpr on the aliased table, so the + // page read can walk recommendations_scan_key_idx. + pageKeyExpr = `(COALESCE(r.payload->>'id', '') COLLATE "C")` +) + +// pageNumericField is a payload number the filters read; stage a holds it as +// t_ (its jsonb_typeof) and n_ (numeric, NULL unless a number). +type pageNumericField struct{ name, field string } + +// pageSummaryFields are read for every request, pagePctFields only for one +// with a MinSavingsPct floor. +var ( + pageSummaryFields = []pageNumericField{{"s", "savings"}, {"up", "upfront_cost"}, {"cnt", "count"}} + pagePctFields = []pageNumericField{{"sp", "savings_percentage"}, {"term", "term"}, {"od", "on_demand_cost"}, {"mc", "monthly_cost"}} +) + +// pct states of the stage-b pct column. +const ( + pagePctDrop = 1 + pagePctUnsure = 2 +) + +// pageArgs numbers query parameters as they are added. +type pageArgs struct{ args []any } + +func (p *pageArgs) add(v any) string { + p.args = append(p.args, v) + return fmt.Sprintf("$%d", len(p.args)) +} + +// pageStageACols selects each payload value the filters need once. rules +// and touchKeys are parameter references (touchKeys is "" when there are no +// suppression keys); withPct adds the pct floor's inputs. +func pageStageACols(rules, touchKeys string, withPct bool) string { + cols := make([]string, 0, 10+2*len(pageSummaryFields)+2*len(pagePctFields)) + cols = append(cols, + "r.id AS rid", pageKeyExpr+" AS sk", + pageAccountExpr+" AS acct", pageProviderExpr+" AS prov", pageServiceExpr+" AS svc", + pageRegionExpr+" AS reg", pageTypeExpr+" AS rtype", pageEngineExpr+" AS eng", + fmt.Sprintf(`%s::jsonb -> (%s || '|' || %s || '|' || %s) AS rule`, rules, pageAccountExpr, pageProviderExpr, pageServiceExpr), + ) + fields := pageSummaryFields + if withPct { + fields = append(append([]pageNumericField(nil), fields...), pagePctFields...) + } + for _, f := range fields { + cols = append(cols, + fmt.Sprintf(`COALESCE(jsonb_typeof(r.payload->'%s'), 'null') AS t_%s`, f.field, f.name), + fmt.Sprintf(`CASE WHEN jsonb_typeof(r.payload->'%[1]s') = 'number' THEN (r.payload->'%[1]s')::numeric END AS n_%[2]s`, f.field, f.name)) + } + touched := "false" + if touchKeys != "" { + touched = pageTouchedExpr(touchKeys) + } + return strings.Join(append(cols, touched+" AS touched"), ", ") +} + +// pageTouchedExpr is true for rows on one of the suppression 6-tuples. +func pageTouchedExpr(touchKeys string) string { + return fmt.Sprintf(`(%s || E'\x1f' || %s || E'\x1f' || %s || E'\x1f' || %s || E'\x1f' || %s || E'\x1f' || %s) = ANY(%s::text[])`, + pageAccountExpr, pageProviderExpr, pageServiceExpr, pageRegionExpr, pageTypeExpr, pageEngineExpr, touchKeys) +} + +// pageSafe is true when the numeric x is 0 or within magnitudes where the +// float8 cast and the pct arithmetic can neither overflow nor underflow. +func pageSafe(x string) string { + return fmt.Sprintf(`(%[1]s = 0 OR abs(%[1]s) BETWEEN 1e-100 AND 1e100)`, x) +} + +const ( + pageSavings = `COALESCE(a.n_s, 0)` + pageUpfront = `COALESCE(a.n_up, 0)` + // pageCount is the payload count as Go decodes it (absent or null is 0); + // a count that is not a plain integer is unsure. + pageCount = `(CASE WHEN a.n_cnt = trunc(a.n_cnt) AND abs(a.n_cnt) <= 1e15 THEN a.n_cnt::bigint ELSE 0 END)` +) + +// pageBadType is true when a numeric payload field holds a non-number, which +// the Go decoder rejects. +func pageBadType(name string) string { + return fmt.Sprintf(`a.t_%s NOT IN ('number', 'null')`, name) +} + +// pageNumbersUnsure flags rows whose summed or compared numbers SQL cannot +// handle like Go: a non-number in a numeric field, a savings or upfront value +// outside pageSafe, or a count that is not a plain integer. +var pageNumbersUnsure = `(` + strings.Join([]string{ + pageBadType("s"), pageBadType("up"), pageBadType("cnt"), + `NOT ` + pageSafe(pageSavings), `NOT ` + pageSafe(pageUpfront), + `(a.n_cnt IS NOT NULL AND (a.n_cnt <> trunc(a.n_cnt) OR abs(a.n_cnt) > 1e15))`, +}, " OR ") + `)` + +// pagePctState mirrors decodeStoredRecommendation's MinSavingsPct floor +// (recEffectiveSavingsPct and recOnDemandBaseline) in float8, operation for +// operation, so the rounding matches Go's: 1 where Go drops the row, 0 where +// it keeps it and 2 where the inputs are outside the bounds SQL can compute +// safely. floor is a float8 parameter reference. +func pagePctState(floor string) string { + f8 := func(x string) string { return "(" + x + ")::float8" } + onDemand := fmt.Sprintf(`((%s + %s) + %s / ((a.n_term)::bigint * 12)::float8)`, f8("a.n_mc"), f8(pageSavings), f8(pageUpfront)) + bad := strings.Join([]string{ + pageBadType("sp"), pageBadType("term"), pageBadType("od"), pageBadType("mc"), pageBadType("s"), pageBadType("up"), + }, " OR ") + return fmt.Sprintf(`CASE + WHEN %[1]s THEN 2 + WHEN a.n_sp IS NOT NULL THEN (CASE WHEN NOT %[2]s THEN 2 WHEN %[3]s < %[4]s THEN 1 ELSE 0 END) + WHEN COALESCE(a.n_term, 0) = 0 THEN 0 + WHEN a.n_term <> trunc(a.n_term) OR abs(a.n_term) > 1e9 THEN 2 + WHEN a.n_od > 0 THEN (CASE WHEN NOT %[5]s OR NOT %[6]s THEN 2 WHEN (%[7]s / %[8]s) * 100::float8 < %[4]s THEN 1 ELSE 0 END) + WHEN a.n_mc IS NULL THEN 0 + WHEN a.prov = 'aws' THEN 0 + WHEN NOT %[9]s OR NOT %[6]s OR NOT %[10]s THEN 2 + WHEN %[11]s = 0 THEN 0 + WHEN (%[7]s / %[11]s) * 100::float8 < %[4]s THEN 1 + ELSE 0 +END`, + bad, pageSafe("a.n_sp"), f8("a.n_sp"), floor, + pageSafe("a.n_od"), pageSafe(pageSavings), f8(pageSavings), f8("a.n_od"), + pageSafe("a.n_mc"), pageSafe(pageUpfront), onDemand) +} + +// pageInList mirrors inListRule: exclude wins, an empty include list allows +// anything else. +func pageInList(value, include, exclude string) string { + return fmt.Sprintf(`(NOT COALESCE((a.rule->'%[3]s') ? %[1]s, false) AND (jsonb_array_length(a.rule->'%[2]s') = 0 OR COALESCE((a.rule->'%[2]s') ? %[1]s, false)))`, + value, include, exclude) +} + +// pageOverridePass mirrors passesOverrideDimensions under the row's resolved +// configuration: true when the key has no configuration and NULL when it was +// never resolved. +var pageOverridePass = fmt.Sprintf(`CASE WHEN NOT (a.rule->>'c')::boolean THEN true ELSE ( + (a.rule->>'en')::boolean + AND (a.eng = '' OR %s) + AND %s + AND %s + AND ((a.rule->>'sp')::boolean OR (a.rule->>'mc')::bigint <= 0 OR %s >= (a.rule->>'mc')::bigint) +) END`, + pageInList("a.eng", "ie", "ee"), pageInList("a.reg", "ir", "er"), pageInList("a.rtype", "it", "et"), pageCount) + +// pageStageBCols derives the decisions from stage a: pct (the floor's +// state), numunsure, pass (the override filter), and the float8 summands. +// floor is "" when there is no MinSavingsPct floor. +func pageStageBCols(floor string) string { + pct := "0" + if floor != "" { + pct = pagePctState(floor) + } + return fmt.Sprintf(`a.rid, a.sk, a.reg, a.touched, a.rule, %s AS pct, %s AS numunsure, %s AS pass, + CASE WHEN %s THEN (%s)::float8 ELSE 0 END AS sv, + CASE WHEN %s THEN (%s)::float8 ELSE 0 END AS upv`, + pct, pageNumbersUnsure, pageOverridePass, + pageSafe(pageSavings), pageSavings, pageSafe(pageUpfront), pageUpfront) +} + +// pageRuleJSON is one resolved override configuration as the SQL filter reads +// it. Lists are never null so the jsonb ? checks and lengths are defined. +type pageRuleJSON struct { + Configured bool `json:"c"` + Enabled bool `json:"en"` + IncludeEngines []string `json:"ie"` + ExcludeEngines []string `json:"ee"` + IncludeRegions []string `json:"ir"` + ExcludeRegions []string `json:"er"` + IncludeTypes []string `json:"it"` + ExcludeTypes []string `json:"et"` + MinCount int `json:"mc"` + MinCountExempt bool `json:"sp"` +} + +func nonNilStrings(s []string) []string { + if s == nil { + return []string{} + } + return s +} + +// pageRulesJSON encodes rules as one JSON object keyed by AccountConfigKey, +// the same key the Go filter looks configurations up by. +func pageRulesJSON(rules []RecommendationOverrideRule) ([]byte, error) { + obj := make(map[string]pageRuleJSON, len(rules)) + for _, rule := range rules { + r := pageRuleJSON{} + if c := rule.Config; c != nil { + r = pageRuleJSON{ + Configured: true, Enabled: c.Enabled, + IncludeEngines: nonNilStrings(c.IncludeEngines), ExcludeEngines: nonNilStrings(c.ExcludeEngines), + IncludeRegions: nonNilStrings(c.IncludeRegions), ExcludeRegions: nonNilStrings(c.ExcludeRegions), + IncludeTypes: nonNilStrings(c.IncludeTypes), ExcludeTypes: nonNilStrings(c.ExcludeTypes), + MinCount: c.MinCount, MinCountExempt: rule.MinCountExempt, + } + } + obj[rule.Key] = r + } + return json.Marshal(obj) +} + +// pageScopeCond is the scope's row filter: the query-string filter on the +// table columns (shared with ListStoredRecommendations), then the enabled +// providers and the account scope on the payload. It must be the first thing +// added to p, because buildRecommendationFilter numbers its parameters from +// $1. +func pageScopeCond(p *pageArgs, scope RecommendationListScope) string { + where, args := buildRecommendationFilter(scope.Filter) + conds := []string{"TRUE"} + if where != "" { + conds[0] = strings.TrimPrefix(where, " WHERE ") + p.args = append(p.args, args...) + } + if len(scope.EnabledProviders) > 0 { + conds = append(conds, pageProviderExpr+" = ANY("+p.add(scope.EnabledProviders)+"::text[])") + } + if scope.Restricted { + conds = append(conds, "r.payload->>'cloud_account_id' = ANY("+p.add(nonNilStrings(scope.AllowedAccountIDs))+"::text[])") + } + return "(" + strings.Join(conds, " AND ") + ")" +} + +// pageStages are the staged row sources shared by the aggregate and the page +// read, built against one parameter list. The aggregate streams every row +// through nested subqueries (from); the page read joins the stages per row +// as LATERAL subqueries (lateral), so the planner can walk +// recommendations_scan_key_idx in order and stop at the LIMIT instead of +// staging and sorting every row. keep and unsure read stage b. +type pageStages struct { + scope string // the scope's row filter, on r + from string // "(stage b over stage a over r) b" for the aggregate + lateral string // "r CROSS JOIN LATERAL a CROSS JOIN LATERAL b" for the page + keep string // the row survives every filter (NULL when unsure) + unsure string // SQL cannot decide the row the way Go does +} + +func buildPageStages(p *pageArgs, q *RecommendationPageQuery, rules []byte, keepRowIDs []string) pageStages { + scope := pageScopeCond(p, q.Scope) + floor := "" + if q.Scope.Filter.MinSavingsPct > 0 { + floor = p.add(q.Scope.Filter.MinSavingsPct) + "::float8" + } + rulesParam := p.add(string(rules)) + touchKeys, override := "", "b.pass" + if len(q.SuppressionKeys) > 0 { + touchKeys = p.add(q.SuppressionKeys) + override = fmt.Sprintf(`(CASE WHEN b.touched THEN b.rid = ANY(%s::uuid[]) ELSE b.pass END)`, p.add(nonNilStrings(keepRowIDs))) + } + aCols, bCols := pageStageACols(rulesParam, touchKeys, floor != ""), pageStageBCols(floor) + return pageStages{ + scope: scope, + from: fmt.Sprintf(`(SELECT %s FROM (SELECT %s FROM recommendations r WHERE %s OFFSET 0) a OFFSET 0) b`, + bCols, aCols, scope), + lateral: fmt.Sprintf(`recommendations r CROSS JOIN LATERAL (SELECT %s OFFSET 0) a CROSS JOIN LATERAL (SELECT %s OFFSET 0) b`, + aCols, bCols), + keep: fmt.Sprintf(`(b.pct <> %d AND %s)`, pagePctDrop, override), + unsure: fmt.Sprintf(`(b.pct = %d OR b.rule IS NULL OR b.numunsure)`, pagePctUnsure), + } +} + +// buildPageAggregateQuery totals the whole filtered set. The sums are ordered +// aggregates in scan order, so float8 addition happens in the order the Go +// summary used; count(DISTINCT) exposes duplicate IDs, which the Go path +// rejects. +func buildPageAggregateQuery(q *RecommendationPageQuery, rules []byte, keepRowIDs []string) (query string, args []any) { + p := &pageArgs{} + e := buildPageStages(p, q, rules, keepRowIDs) + query = fmt.Sprintf(`SELECT + count(*) FILTER (WHERE u), + count(*) FILTER (WHERE k), + count(DISTINCT sk) FILTER (WHERE k), + COALESCE(sum(sv ORDER BY sk, rid) FILTER (WHERE k), 0), + COALESCE(sum(upv ORDER BY sk, rid) FILTER (WHERE k), 0), + COALESCE(array_agg(DISTINCT reg ORDER BY reg) FILTER (WHERE k AND reg <> ''), '{}') +FROM ( + SELECT b.rid, b.sk, b.sv, b.upv, b.reg, %s AS u, (NOT %s AND COALESCE(%s, false)) AS k + FROM %s + OFFSET 0 +) d`, + e.unsure, e.unsure, e.keep, e.from) + return query, p.args +} + +// buildPageRowsQuery reads the page: at most limit surviving rows after +// afterID in scan order. +func buildPageRowsQuery(q *RecommendationPageQuery, rules []byte, keepRowIDs []string) (query string, args []any) { + p := &pageArgs{} + e := buildPageStages(p, q, rules, keepRowIDs) + where := e.scope + " AND COALESCE(" + e.keep + ", false)" + if q.AfterID != "" { + where += " AND " + pageKeyExpr + " > " + p.add(q.AfterID) + } + query = fmt.Sprintf(`SELECT r.id::text, r.payload FROM %s WHERE %s ORDER BY %s, r.id LIMIT %s`, + e.lateral, where, pageKeyExpr, p.add(q.Limit)) + return query, p.args +} + +// buildPageTouchedQuery reads the in-scope rows on a suppression 6-tuple. +func buildPageTouchedQuery(q *RecommendationPageQuery) (query string, args []any) { + p := &pageArgs{} + scope := pageScopeCond(p, q.Scope) + keys := p.add(q.SuppressionKeys) + return fmt.Sprintf(`SELECT r.id::text, r.payload FROM recommendations r WHERE %s AND %s`, scope, pageTouchedExpr(keys)), p.args +} + +// buildPageTriplesQuery lists the distinct override keys of the scope's rows. +func buildPageTriplesQuery(scope RecommendationListScope) (query string, args []any) { + p := &pageArgs{} + cond := pageScopeCond(p, scope) + return fmt.Sprintf(`SELECT DISTINCT %s, COALESCE(jsonb_typeof(r.payload->'cloud_account_id'), 'null') <> 'null', %s, %s FROM recommendations r WHERE %s`, + pageAccountExpr, pageProviderExpr, pageServiceExpr, cond), p.args +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index ce62f622..76c8a79c 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1859,6 +1859,37 @@ func (m *MockConfigStore) ListStoredRecommendationsChunk(ctx context.Context, fi return v, args.Error(1) } +func (m *MockConfigStore) ListStoredRecommendationTriples(ctx context.Context, scope config.RecommendationListScope) ([]config.RecommendationTriple, error) { + m.record("ListStoredRecommendationTriples", ctx, scope) + if !isExpected(&m.Mock, "ListStoredRecommendationTriples") { + return nil, nil + } + args := m.Called(ctx, scope) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).([]config.RecommendationTriple) + if !ok { + m.reporter.wrongReturnType("ListStoredRecommendationTriples", "[]config.RecommendationTriple", args.Get(0)) + } + return v, args.Error(1) +} + +// PageStoredRecommendations defaults to an empty cache: an empty page with a +// zero summary. +func (m *MockConfigStore) PageStoredRecommendations(ctx context.Context, q config.RecommendationPageQuery) (config.RecommendationPage, error) { + m.record("PageStoredRecommendations", ctx, q) + if !isExpected(&m.Mock, "PageStoredRecommendations") { + return config.RecommendationPage{}, nil + } + args := m.Called(ctx, q) + v, ok := args.Get(0).(config.RecommendationPage) + if !ok { + m.reporter.wrongReturnType("PageStoredRecommendations", "config.RecommendationPage", args.Get(0)) + } + return v, args.Error(1) +} + func (m *MockConfigStore) GetRecommendationsFreshness(ctx context.Context) (*config.RecommendationsFreshness, error) { m.record("GetRecommendationsFreshness", ctx) if !isExpected(&m.Mock, "GetRecommendationsFreshness") { diff --git a/internal/scheduler/scheduler_page.go b/internal/scheduler/scheduler_page.go new file mode 100644 index 00000000..eae378f9 --- /dev/null +++ b/internal/scheduler/scheduler_page.go @@ -0,0 +1,174 @@ +package scheduler + +import ( + "context" + "errors" + "fmt" + "sort" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// ErrRecommendationsPageNeedsScan reports that PageRecommendations cannot +// reproduce the Go-side list filters for this request in SQL. The caller +// serves the request from ScanRecommendations instead, which applies them in +// Go exactly as before. +var ErrRecommendationsPageNeedsScan = errors.New("recommendations page needs the Go-side scan") + +func needsScan(reason string) error { + return fmt.Errorf("%w: %s", ErrRecommendationsPageNeedsScan, reason) +} + +// PageRecommendations returns one page of the recommendations list, at most +// limit rows after afterID in byte-wise ID order, with the summary of the +// whole filtered set (issue #981). Unlike ScanRecommendations, which hands +// every row to the caller, the filters run in SQL: the store totals the +// summary in an aggregate query and returns only the page. +// +// The filters are the ones ScanRecommendations and the handler apply: +// scope's filter, enabled providers and account scope, plus suppressions and +// account overrides. Overrides are resolved here, for the (account, +// provider, service) keys the scope's rows have, with the same resolver as +// the scan. Rows on a suppressed 6-tuple are few (recent purchases), so the +// store hands them to applySuppressionIndex and the override filter here +// rather than re-implementing the variant match in SQL; every other row has +// no suppression and is filtered by the resolved overrides in SQL. The page +// rows are then run through the same Go filters once more, which also stamps +// their suppression badge fields; a row the Go filters would drop means SQL +// and Go disagree, and the request falls back rather than show it. +// +// It returns an error wrapping ErrRecommendationsPageNeedsScan when SQL +// cannot reproduce the Go filters for this request (override resolution +// failed, values SQL cannot compute like Go, duplicate IDs, a cursor that is +// not valid text); the caller then uses ScanRecommendations. The cold-start +// kick has already run by then, and the scan's own kick is a no-op (the +// collection marker or Lambda guard stops it). A suppression read failure +// fails the request, as it does for the scan (issue #111). +func (s *Scheduler) PageRecommendations(ctx context.Context, scope config.RecommendationListScope, afterID string, limit int) (*config.RecommendationPage, error) { + freshness, err := s.startCacheRead(ctx) + if err != nil { + return nil, err + } + suppressions, err := s.loadSuppressionIndex(ctx) + if err != nil { + return nil, err + } + rules, resolved, err := s.resolvePageOverrides(ctx, scope) + if err != nil { + return nil, err + } + q := config.RecommendationPageQuery{ + Scope: scope, Rules: rules, SuppressionKeys: suppressionPageKeys(suppressions), + AfterID: afterID, Limit: limit, + } + if len(q.SuppressionKeys) > 0 { + q.DecideSuppressed = func(rows []config.StoredRecommendation) ([]string, error) { + return decideSuppressedRows(rows, suppressions, resolved), nil + } + } + page, err := s.config.PageStoredRecommendations(ctx, q) + if err != nil { + return nil, fmt.Errorf("failed to page recommendations: %w", err) + } + if page.Fallback != "" { + return nil, needsScan(page.Fallback) + } + rows, ok := filterPageRows(page.Rows, suppressions, resolved) + if !ok { + return nil, needsScan("SQL kept a row the Go filters drop") + } + page.Rows = rows + s.finishCacheRead(ctx, freshness) + return &page, nil +} + +// resolvePageOverrides resolves account overrides for every key of the +// scope's rows and returns them as SQL rules plus the map the Go filter +// uses. A resolution failure, which the scan logs and passes through +// unfiltered chunk by chunk, falls back to the scan so it behaves the same. +// So do two rare shapes the rules cannot express: an empty but present +// cloud_account_id (resolution then depends on which row the scan meets +// first) and two keys joining to the same AccountConfigKey string. +func (s *Scheduler) resolvePageOverrides(ctx context.Context, scope config.RecommendationListScope) ([]config.RecommendationOverrideRule, map[string]*config.ServiceConfig, error) { + triples, err := s.config.ListStoredRecommendationTriples(ctx, scope) + if err != nil { + return nil, nil, err + } + recs := make([]config.RecommendationRecord, 0, len(triples)) + rules := make([]config.RecommendationOverrideRule, 0, len(triples)) + keys := make(map[string]struct{}, len(triples)) + for _, t := range triples { + if t.HasAccount && t.AccountID == "" { + return nil, nil, needsScan("a recommendation has an empty cloud_account_id") + } + key := config.AccountConfigKey(t.AccountID, t.Provider, t.Service) + if _, dup := keys[key]; dup { + return nil, nil, needsScan("two override keys join to " + key) + } + keys[key] = struct{}{} + rec := config.RecommendationRecord{Provider: t.Provider, Service: t.Service} + if t.HasAccount { + id := t.AccountID + rec.CloudAccountID = &id + } + recs = append(recs, rec) + rules = append(rules, config.RecommendationOverrideRule{ + Key: key, MinCountExempt: common.IsSavingsPlan(common.ServiceType(t.Service)), + }) + } + resolved, err := config.ResolveAccountConfigsForRecs(ctx, s.config, recs) + if err != nil { + logging.Warnf("recommendations page: failed to resolve account overrides, using the scan: %v", err) + return nil, nil, needsScan("account override resolution failed") + } + for i := range rules { + rules[i].Config = resolved[rules[i].Key] + } + return rules, resolved, nil +} + +// suppressionPageKeys lists the suppression 6-tuples as the store matches +// rows against them: one key per stored spelling of each canonical service. +func suppressionPageKeys(index map[suppressionKey][]config.PurchaseSuppression) []string { + var keys []string + for k := range index { + for _, service := range config.SuppressionServiceSpellings(k.provider, k.service) { + keys = append(keys, config.SuppressionTupleKey(k.accountID, k.provider, service, k.region, k.resourceType, k.engine)) + } + } + sort.Strings(keys) + return keys +} + +// decideSuppressedRows returns the primary keys of the rows that survive +// suppressions and then account overrides, decided by the same functions +// the scan uses, one row at a time. +func decideSuppressedRows(rows []config.StoredRecommendation, index map[suppressionKey][]config.PurchaseSuppression, resolved map[string]*config.ServiceConfig) []string { + keep := make([]string, 0, len(rows)) + for i := range rows { + out := applySuppressionIndex([]config.RecommendationRecord{rows[i].Rec}, index) + if len(out) == 1 && len(resolved) > 0 { + out = filterRecsByResolvedConfigs(out, resolved) + } + if len(out) == 1 { + keep = append(keep, rows[i].RowID) + } + } + return keep +} + +// filterPageRows runs the page through the scan's suppression and override +// filters, which stamps the suppression badge fields on the rows. ok is false +// when the filters drop a row SQL kept. +func filterPageRows(rows []config.RecommendationRecord, index map[suppressionKey][]config.PurchaseSuppression, resolved map[string]*config.ServiceConfig) ([]config.RecommendationRecord, bool) { + out := rows + if len(index) > 0 { + out = applySuppressionIndex(out, index) + } + if len(resolved) > 0 && len(filterRecsByResolvedConfigs(out, resolved)) != len(out) { + return nil, false + } + return out, len(out) == len(rows) +} diff --git a/internal/scheduler/scheduler_page_test.go b/internal/scheduler/scheduler_page_test.go new file mode 100644 index 00000000..44c27045 --- /dev/null +++ b/internal/scheduler/scheduler_page_test.go @@ -0,0 +1,151 @@ +package scheduler + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// pageStore serves PageRecommendations' store calls: the override keys of +// triples, the suppressions in sups, and a page built by page, which sees the +// query (and can call its DecideSuppressed) like the Postgres store does. +type pageStore struct { + mockOverrideStore + triples []config.RecommendationTriple + sups []config.PurchaseSuppression + supsErr error + page func(q config.RecommendationPageQuery) (config.RecommendationPage, error) + queries []config.RecommendationPageQuery +} + +func (m *pageStore) ListActiveSuppressions(_ context.Context) ([]config.PurchaseSuppression, error) { + return m.sups, m.supsErr +} + +func (m *pageStore) ListStoredRecommendationTriples(_ context.Context, _ config.RecommendationListScope) ([]config.RecommendationTriple, error) { + return m.triples, nil +} + +func (m *pageStore) PageStoredRecommendations(_ context.Context, q config.RecommendationPageQuery) (config.RecommendationPage, error) { + m.queries = append(m.queries, q) + return m.page(q) +} + +const pageAccount = "11111111-1111-4111-8111-111111111111" + +func pageRec(id, service, resourceType string, count int) config.RecommendationRecord { + account := pageAccount + return config.RecommendationRecord{ID: id, Provider: "aws", Service: service, Region: "us-east-1", + ResourceType: resourceType, Count: count, CloudAccountID: &account} +} + +func pageSuppression(service, resourceType string, count int) config.PurchaseSuppression { + return config.PurchaseSuppression{ExecutionID: "exec-" + resourceType, AccountID: pageAccount, Provider: "aws", + Service: service, Region: "us-east-1", ResourceType: resourceType, SuppressedCount: count, + ExpiresAt: time.Now().Add(time.Hour), CreatedAt: time.Now()} +} + +func newPageScheduler(store *pageStore) *Scheduler { + return &Scheduler{config: store, isLambda: true} +} + +// TestPageRecommendations_SuppressionsDecidedInGo checks the suppression +// split: the store is asked to match every stored spelling of a suppressed +// service, rows on those tuples are decided by applySuppressionIndex and the +// override filter, and the page rows come back with their badge fields. +func TestPageRecommendations_SuppressionsDecidedInGo(t *testing.T) { + store := &pageStore{ + triples: []config.RecommendationTriple{{AccountID: pageAccount, Provider: "aws", Service: "ec2", HasAccount: true}}, + sups: []config.PurchaseSuppression{pageSuppression("ec2", "m5.large", 2), pageSuppression("compute", "c5.large", 5)}, + } + store.globals = map[string]*config.ServiceConfig{"aws|ec2": {Provider: "aws", Service: "ec2", Enabled: true, MinCount: 2}} + var decided []string + store.page = func(q config.RecommendationPageQuery) (config.RecommendationPage, error) { + require.NotNil(t, q.DecideSuppressed) + var err error + decided, err = q.DecideSuppressed([]config.StoredRecommendation{ + {RowID: "keep", Rec: pageRec("a", "compute", "m5.large", 6)}, // 6-2=4 >= min_count 2 + {RowID: "floor", Rec: pageRec("b", "ec2", "m5.large", 3)}, // 3-2=1 < min_count 2 + {RowID: "gone", Rec: pageRec("c", "ec2", "c5.large", 5)}, // fully suppressed + }) + require.NoError(t, err) + return config.RecommendationPage{TotalCount: 1, Rows: []config.RecommendationRecord{pageRec("a", "compute", "m5.large", 6)}}, nil + } + page, err := newPageScheduler(store).PageRecommendations(context.Background(), config.RecommendationListScope{}, "", 10) + require.NoError(t, err) + + assert.Equal(t, []string{"keep"}, decided) + require.Len(t, store.queries, 1) + keys := store.queries[0].SuppressionKeys + for _, service := range []string{"compute", "ec2"} { + assert.Contains(t, keys, config.SuppressionTupleKey(pageAccount, "aws", service, "us-east-1", "m5.large", "")) + } + require.Len(t, store.queries[0].Rules, 1) + assert.Equal(t, config.AccountConfigKey(pageAccount, "aws", "ec2"), store.queries[0].Rules[0].Key) + assert.NotNil(t, store.queries[0].Rules[0].Config) + require.Len(t, page.Rows, 1) + assert.Equal(t, 4, page.Rows[0].Count, "the page row carries the suppressed count") + assert.Equal(t, 2, page.Rows[0].SuppressedCount) +} + +// TestPageRecommendations_FallsBackToScan covers each reason the SQL page +// hands the request back to the Go scan. +func TestPageRecommendations_FallsBackToScan(t *testing.T) { + triple := config.RecommendationTriple{AccountID: pageAccount, Provider: "aws", Service: "rds", HasAccount: true} + cases := []struct { + name string + setup func(*pageStore) + }{ + {"override resolution fails", func(s *pageStore) { s.getGlobalErr = errors.New("db down") }}, + {"empty but present account", func(s *pageStore) { + s.triples = []config.RecommendationTriple{{Provider: "aws", Service: "rds", HasAccount: true}} + }}, + {"two keys join to one string", func(s *pageStore) { + s.triples = []config.RecommendationTriple{ + {AccountID: "a", Provider: "b|c", Service: "d", HasAccount: true}, + {AccountID: "a|b", Provider: "c", Service: "d", HasAccount: true}, + } + }}, + {"store reports a fallback", func(s *pageStore) { + s.page = func(config.RecommendationPageQuery) (config.RecommendationPage, error) { + return config.RecommendationPage{Fallback: "unsure rows"}, nil + } + }}, + {"page row the override filter drops", func(s *pageStore) { + s.globals = map[string]*config.ServiceConfig{"aws|rds": {Provider: "aws", Service: "rds", Enabled: false}} + }}, + {"page row fully suppressed", func(s *pageStore) { + s.sups = []config.PurchaseSuppression{pageSuppression("rds", "db.r5.large", 9)} + s.page = func(q config.RecommendationPageQuery) (config.RecommendationPage, error) { + return config.RecommendationPage{TotalCount: 1, Rows: []config.RecommendationRecord{pageRec("a", "rds", "db.r5.large", 1)}}, nil + } + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + store := &pageStore{triples: []config.RecommendationTriple{triple}} + store.page = func(config.RecommendationPageQuery) (config.RecommendationPage, error) { + return config.RecommendationPage{TotalCount: 1, Rows: []config.RecommendationRecord{pageRec("a", "rds", "db.r5.large", 1)}}, nil + } + tc.setup(store) + _, err := newPageScheduler(store).PageRecommendations(context.Background(), config.RecommendationListScope{}, "", 10) + require.ErrorIs(t, err, ErrRecommendationsPageNeedsScan) + }) + } +} + +// TestPageRecommendations_SuppressionReadFailsClosed keeps issue #111: a +// suppression read failure fails the request instead of falling back. +func TestPageRecommendations_SuppressionReadFailsClosed(t *testing.T) { + store := &pageStore{supsErr: errors.New("db down")} + _, err := newPageScheduler(store).PageRecommendations(context.Background(), config.RecommendationListScope{}, "", 10) + require.Error(t, err) + assert.NotErrorIs(t, err, ErrRecommendationsPageNeedsScan) + assert.Empty(t, store.queries) +} diff --git a/internal/server/interfaces.go b/internal/server/interfaces.go index bae10f86..e3d9d9ec 100644 --- a/internal/server/interfaces.go +++ b/internal/server/interfaces.go @@ -16,6 +16,9 @@ type SchedulerInterface interface { // ScanRecommendations walks the same rows as ListRecommendations in // ID-ordered chunks for the paginated list endpoint (issue #981). ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error + // PageRecommendations reads one list page and the full-set summary in + // SQL for the paginated list endpoint (issue #981). + PageRecommendations(ctx context.Context, scope config.RecommendationListScope, afterID string, limit int) (*config.RecommendationPage, error) // GetRecommendationByID fetches a single rec by application-level id, // bypassing account-override filtering. hiddenBy is non-nil when the rec // exists but would be dropped by the override filter. Returns nil, nil, diff --git a/internal/testutil/mocks.go b/internal/testutil/mocks.go index 6f5c16e4..3cb45bd6 100644 --- a/internal/testutil/mocks.go +++ b/internal/testutil/mocks.go @@ -15,6 +15,7 @@ type MockScheduler struct { ListRecommendationsFunc func(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) GetRecommendationByIDFunc func(ctx context.Context, id string) (*config.RecommendationRecord, []string, error) ScanRecommendationsFunc func(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error + PageRecommendationsFunc func(ctx context.Context, scope config.RecommendationListScope, afterID string, limit int) (*config.RecommendationPage, error) } func (m *MockScheduler) CollectRecommendations(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) { @@ -39,6 +40,15 @@ func (m *MockScheduler) ScanRecommendations(ctx context.Context, filter config.R return nil } +// PageRecommendations defaults to sending the caller to the scan, so a test +// that only sets ScanRecommendationsFunc keeps driving the paginated list. +func (m *MockScheduler) PageRecommendations(ctx context.Context, scope config.RecommendationListScope, afterID string, limit int) (*config.RecommendationPage, error) { + if m.PageRecommendationsFunc != nil { + return m.PageRecommendationsFunc(ctx, scope, afterID, limit) + } + return nil, scheduler.ErrRecommendationsPageNeedsScan +} + func (m *MockScheduler) GetRecommendationByID(ctx context.Context, id string) (*config.RecommendationRecord, []string, error) { if m.GetRecommendationByIDFunc != nil { return m.GetRecommendationByIDFunc(ctx, id)