From f630a5860574917c17db340938f6d64e128d746a Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 12:54:50 +0100 Subject: [PATCH 1/9] fix(api): add opt-in keyset pagination to GET /api/recommendations The list endpoint returned the whole filtered recommendations cache in a single body, so a tenant with a few thousand cached rows pushed the response past the 6 MB Lambda payload cap and the page failed with 502. Gzip (#775) raised that ceiling but did not bound it. Callers can now send limit (1-1000) and the returned next_cursor to read the list in pages ordered by recommendation ID. The cursor is an opaque base64url encoding of the last ID seen, so a row deleted between requests does not shift later pages. summary and regions still describe the full filtered set on every page, and compute capacity is only decoded for the returned rows. Requests without limit or cursor keep the existing unpaginated response, so the current frontend is unaffected until it adopts the parameters. openapi.yaml documents the parameters and the actual response shape. Refs #777 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_recommendations.go | 31 ++- internal/api/handler_recommendations_page.go | 136 +++++++++++ .../api/handler_recommendations_page_test.go | 216 ++++++++++++++++++ internal/api/openapi.yaml | 49 +++- internal/api/types.go | 3 + 5 files changed, 422 insertions(+), 13 deletions(-) create mode 100644 internal/api/handler_recommendations_page.go create mode 100644 internal/api/handler_recommendations_page_test.go diff --git a/internal/api/handler_recommendations.go b/internal/api/handler_recommendations.go index 0c2d33e75..ecb97a1fc 100644 --- a/internal/api/handler_recommendations.go +++ b/internal/api/handler_recommendations.go @@ -94,6 +94,10 @@ func (h *Handler) getRecommendations(ctx context.Context, req *events.LambdaFunc if err != nil { return nil, err } + page, err := parseRecommendationsPage(params) + if err != nil { + return nil, err + } // Resolve the account scope once: it decides both whether this read may // kick a background collection (issue #627) and the post-filter below. @@ -122,7 +126,7 @@ func (h *Handler) getRecommendations(ctx context.Context, req *events.LambdaFunc return nil, err } - return buildRecommendationsResponse(recommendations), nil + return buildRecommendationsPageResponse(recommendations, page), nil } // filterRecommendationsByAllowedAccounts filters recommendations to only include @@ -195,17 +199,28 @@ func (h *Handler) filterRecommendationsByEnabledProviders(ctx context.Context, r } // buildRecommendationsResponse calculates summary statistics and collects -// unique regions from a set of recommendations. +// unique regions from a set of recommendations, returning all of them as rows +// (the legacy unpaginated shape). func buildRecommendationsResponse(recommendations []config.RecommendationRecord) *RecommendationsResponse { + return summarizeRecommendations(recommendations, recommendations) +} + +// summarizeRecommendations computes the summary and region list over all (the +// full filtered set) and returns rows as the response body. rows is either all +// itself or one page of it (issue #723), so the totals never shrink to the +// page. +func summarizeRecommendations(all, rows []config.RecommendationRecord) *RecommendationsResponse { regionSet := make(map[string]struct{}) var totalSavings, totalUpfront float64 - for i := range recommendations { - rec := &recommendations[i] + for i := range all { + rec := &all[i] totalSavings += rec.Savings totalUpfront += rec.UpfrontCost if rec.Region != "" { regionSet[rec.Region] = struct{}{} } + } + for i := range rows { // Surface the compute size (VCPU / MemoryGB) at the top level so // the frontend Capacity column can render it without parsing the // opaque Details blob. The canonical values live in the typed @@ -214,8 +229,8 @@ func buildRecommendationsResponse(recommendations []config.RecommendationRecord) // pointers when this is a compute rec with a known size. Non-compute // recs, legacy rows with empty Details, and decode failures all // leave the pointers nil, so JSON omits them and the frontend - // renders "—". - stampComputeCapacity(rec) + // renders "—". Only returned rows are decoded. + stampComputeCapacity(&rows[i]) } regions := make([]string, 0, len(regionSet)) @@ -229,9 +244,9 @@ func buildRecommendationsResponse(recommendations []config.RecommendationRecord) } return &RecommendationsResponse{ - Recommendations: recommendations, + Recommendations: rows, Summary: RecommendationsSummary{ - TotalCount: len(recommendations), + TotalCount: len(all), TotalMonthlySavings: totalSavings, TotalUpfrontCost: totalUpfront, AvgPaybackMonths: avgPayback, diff --git a/internal/api/handler_recommendations_page.go b/internal/api/handler_recommendations_page.go new file mode 100644 index 000000000..7557bedf6 --- /dev/null +++ b/internal/api/handler_recommendations_page.go @@ -0,0 +1,136 @@ +package api + +import ( + "encoding/base64" + "sort" + "strconv" + "strings" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// Opt-in keyset pagination for GET /api/recommendations (issue #723). +// +// The list endpoint used to return the whole filtered cache in one body, +// 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 request carrying neither parameter keeps the existing unpaginated shape +// unchanged, so the current frontend (which sorts, groups and totals the +// full set client-side) keeps working until it adopts pagination. +// +// In both modes `summary` and `regions` describe the full filtered set, not +// just the returned page, so headline totals stay correct while paging. +const ( + // DefaultRecommendationsPageSize is the page size used when a caller + // sends a cursor without a limit. + DefaultRecommendationsPageSize = 500 + + // MaxRecommendationsPageSize caps one page. List rows are roughly + // 500-600 bytes of JSON, so a full page stays far below the 6 MB cap. + MaxRecommendationsPageSize = 1000 + + // maxRecommendationsCursorLen bounds the encoded cursor so a hostile + // query string cannot force a large decode. + maxRecommendationsCursorLen = 1024 +) + +// recommendationsPage is the parsed pagination request. enabled is false when +// the caller sent neither limit nor cursor (legacy unpaginated mode). +type recommendationsPage struct { + enabled bool + limit int + afterID string +} + +// parseRecommendationsPage validates the optional limit and cursor query +// parameters. limit must be an integer in 1..MaxRecommendationsPageSize; +// out-of-range values are rejected rather than clamped so a client bug does +// not silently return a different page size than it asked for. +func parseRecommendationsPage(params map[string]string) (recommendationsPage, error) { + rawLimit := strings.TrimSpace(params["limit"]) + rawCursor := strings.TrimSpace(params["cursor"]) + if rawLimit == "" && rawCursor == "" { + return recommendationsPage{}, nil + } + + page := recommendationsPage{enabled: true, limit: DefaultRecommendationsPageSize} + if rawLimit != "" { + n, err := strconv.Atoi(rawLimit) + if err != nil || n < 1 || n > MaxRecommendationsPageSize { + return recommendationsPage{}, NewClientError(400, + "invalid limit: must be an integer between 1 and "+strconv.Itoa(MaxRecommendationsPageSize)) + } + page.limit = n + } + if rawCursor != "" { + afterID, err := decodeRecommendationsCursor(rawCursor) + if err != nil { + return recommendationsPage{}, err + } + page.afterID = afterID + } + return page, nil +} + +// encodeRecommendationsCursor turns the last row ID of a page into an opaque, +// URL-safe cursor. Recommendation IDs can carry characters such as "|", so +// the raw ID is not used directly in the query string. +func encodeRecommendationsCursor(id string) string { + return base64.RawURLEncoding.EncodeToString([]byte(id)) +} + +// decodeRecommendationsCursor reverses encodeRecommendationsCursor and +// rejects malformed or oversized cursors with a 400. +func decodeRecommendationsCursor(cursor string) (string, error) { + if len(cursor) > maxRecommendationsCursorLen { + return "", NewClientError(400, "invalid cursor: too long") + } + raw, err := base64.RawURLEncoding.DecodeString(cursor) + if err != nil || len(raw) == 0 { + return "", NewClientError(400, "invalid cursor") + } + return string(raw), nil +} + +// paginateRecommendations orders the filtered set by ID and returns the rows +// strictly after page.afterID, at most page.limit of them. recs is sorted in +// place (it is this request's own slice, freshly read from the store) and +// returned as sorted; rows is a sub-slice of it. nextCursor is empty on the +// last page. +// +// The ID is the cache table's primary key, so ID order is a total order and +// no row can be skipped or repeated at a page boundary. +// +// Keyset (rather than offset) paging means a row added or removed between +// two requests does not shift later pages: the next page always resumes +// after the last ID the client saw, even if that row has since been deleted. +func paginateRecommendations(recs []config.RecommendationRecord, page recommendationsPage) (sorted, rows []config.RecommendationRecord, nextCursor string) { + sorted = recs + sort.Slice(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID }) + + start := 0 + if page.afterID != "" { + start = sort.Search(len(sorted), func(i int) bool { return sorted[i].ID > page.afterID }) + } + end := min(start+page.limit, len(sorted)) + rows = sorted[start:end] + if end < len(sorted) { + nextCursor = encodeRecommendationsCursor(sorted[end-1].ID) + } + return sorted, rows, nextCursor +} + +// buildRecommendationsPageResponse builds the list response for either mode. +// Unpaginated requests get exactly the legacy buildRecommendationsResponse +// output; paginated requests get one page of rows plus next_cursor, with the +// summary and regions still computed over the full filtered set. +func buildRecommendationsPageResponse(recs []config.RecommendationRecord, page recommendationsPage) *RecommendationsResponse { + if !page.enabled { + return buildRecommendationsResponse(recs) + } + sorted, rows, next := paginateRecommendations(recs, page) + resp := summarizeRecommendations(sorted, rows) + resp.NextCursor = next + return resp +} diff --git a/internal/api/handler_recommendations_page_test.go b/internal/api/handler_recommendations_page_test.go new file mode 100644 index 000000000..c502db6f0 --- /dev/null +++ b/internal/api/handler_recommendations_page_test.go @@ -0,0 +1,216 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// lambdaResponseCapBytes is the AWS Lambda synchronous response payload cap +// that GET /api/recommendations used to exceed (issue #723). +const lambdaResponseCapBytes = 6 * 1024 * 1024 + +// pageTestRecs returns n recommendations in deliberately non-ID order, each +// padded so a single row is about 1 KB of JSON. +func pageTestRecs(n int) []config.RecommendationRecord { + pad := strings.Repeat("x", 900) + recs := make([]config.RecommendationRecord, n) + for i := range n { + // Reverse order so the handler has to sort to page stably. + idx := n - 1 - i + recs[i] = config.RecommendationRecord{ + ID: fmt.Sprintf("rec-%06d", idx), + Provider: "aws", + Service: "ec2", + Region: "us-east-" + strconv.Itoa(idx%3+1), + ResourceType: "m5.large-" + pad, + Savings: 10, + UpfrontCost: 20, + } + } + return recs +} + +func pageTestHandler(t *testing.T, recs []config.RecommendationRecord) *Handler { + t.Helper() + ms := new(MockScheduler) + ms.On("ListRecommendations", mock.Anything, mock.Anything).Return(recs, nil) + return &Handler{scheduler: ms, apiKey: "test-key"} +} + +func pageTestReq() *events.LambdaFunctionURLRequest { + return &events.LambdaFunctionURLRequest{Headers: map[string]string{"x-api-key": "test-key"}} +} + +// TestGetRecommendations_PaginationBoundsResponse is the #723 regression: a +// tenant whose full list is well over 6 MB must be able to fetch it in pages +// that each fit under the Lambda cap, and walking the cursor must return +// every row exactly once. +func TestGetRecommendations_PaginationBoundsResponse(t *testing.T) { + const total = 8000 + recs := pageTestRecs(total) + h := pageTestHandler(t, recs) + + full, err := json.Marshal(buildRecommendationsResponse(append([]config.RecommendationRecord(nil), recs...))) + require.NoError(t, err) + require.Greater(t, len(full), lambdaResponseCapBytes, "fixture must exceed the cap to be meaningful") + + seen := make(map[string]int, total) + var order []string + params := map[string]string{"limit": strconv.Itoa(MaxRecommendationsPageSize)} + for pages := 0; ; pages++ { + require.Less(t, pages, total, "cursor walk did not terminate") + resp, err := h.getRecommendations(context.Background(), pageTestReq(), params) + require.NoError(t, err) + + body, err := json.Marshal(resp) + require.NoError(t, err) + assert.Less(t, len(body), lambdaResponseCapBytes, "page %d exceeds the Lambda response cap", pages) + assert.LessOrEqual(t, len(resp.Recommendations), MaxRecommendationsPageSize) + + // Summary describes the full filtered set on every page. + assert.Equal(t, total, resp.Summary.TotalCount) + assert.InDelta(t, float64(total*10), resp.Summary.TotalMonthlySavings, 1e-6) + assert.Len(t, resp.Regions, 3) + + for i := range resp.Recommendations { + id := resp.Recommendations[i].ID + seen[id]++ + order = append(order, id) + } + if resp.NextCursor == "" { + break + } + params = map[string]string{"limit": strconv.Itoa(MaxRecommendationsPageSize), "cursor": resp.NextCursor} + } + + assert.Len(t, seen, total, "every row must be returned") + for id, n := range seen { + assert.Equal(t, 1, n, "row %s returned %d times", id, n) + } + assert.IsIncreasing(t, order, "rows must be returned in ID order across pages") +} + +// TestGetRecommendations_UnpaginatedShapeUnchanged pins backward +// compatibility: without limit/cursor the endpoint still returns every row, +// in the order the scheduler gave them, with no next_cursor. +func TestGetRecommendations_UnpaginatedShapeUnchanged(t *testing.T) { + recs := pageTestRecs(5) + h := pageTestHandler(t, recs) + + resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{}) + require.NoError(t, err) + require.Len(t, resp.Recommendations, 5) + for i := range recs { + assert.Equal(t, recs[i].ID, resp.Recommendations[i].ID) + } + assert.Empty(t, resp.NextCursor) + assert.Equal(t, 5, resp.Summary.TotalCount) + + body, err := json.Marshal(resp) + require.NoError(t, err) + assert.NotContains(t, string(body), "next_cursor") +} + +func TestGetRecommendations_PaginationPages(t *testing.T) { + recs := pageTestRecs(5) // rec-000000 .. rec-000004, stored in reverse + h := pageTestHandler(t, recs) + ctx := context.Background() + + first, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"limit": "2"}) + require.NoError(t, err) + require.Len(t, first.Recommendations, 2) + assert.Equal(t, "rec-000000", first.Recommendations[0].ID) + assert.Equal(t, "rec-000001", first.Recommendations[1].ID) + assert.Equal(t, encodeRecommendationsCursor("rec-000001"), first.NextCursor) + assert.Equal(t, 5, first.Summary.TotalCount) + + last, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{ + "limit": "2", "cursor": encodeRecommendationsCursor("rec-000003"), + }) + require.NoError(t, err) + require.Len(t, last.Recommendations, 1) + assert.Equal(t, "rec-000004", last.Recommendations[0].ID) + assert.Empty(t, last.NextCursor, "last page must not carry a cursor") + + // Exact fit: the page ending on the final row has no next cursor. + exact, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"limit": "5"}) + require.NoError(t, err) + assert.Len(t, exact.Recommendations, 5) + assert.Empty(t, exact.NextCursor) + + // Keyset: a cursor naming a row that has since disappeared resumes at + // the next ID instead of failing or restarting. + gone, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{ + "limit": "2", "cursor": encodeRecommendationsCursor("rec-000001x"), + }) + require.NoError(t, err) + require.Len(t, gone.Recommendations, 2) + assert.Equal(t, "rec-000002", gone.Recommendations[0].ID) + + // Cursor past the end yields an empty page, not an error. + past, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{ + "cursor": encodeRecommendationsCursor("rec-999999"), + }) + require.NoError(t, err) + assert.Empty(t, past.Recommendations) + assert.Empty(t, past.NextCursor) + assert.Equal(t, 5, past.Summary.TotalCount) +} + +func TestGetRecommendations_CursorWithoutLimitUsesDefaultPageSize(t *testing.T) { + recs := pageTestRecs(DefaultRecommendationsPageSize + 3) + h := pageTestHandler(t, recs) + + resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{ + "cursor": encodeRecommendationsCursor("rec-"), + }) + require.NoError(t, err) + assert.Len(t, resp.Recommendations, DefaultRecommendationsPageSize) + assert.NotEmpty(t, resp.NextCursor) +} + +func TestGetRecommendations_PaginationRejectsBadParams(t *testing.T) { + cases := map[string]map[string]string{ + "zero limit": {"limit": "0"}, + "negative limit": {"limit": "-1"}, + "limit above max": {"limit": strconv.Itoa(MaxRecommendationsPageSize + 1)}, + "non-numeric limit": {"limit": "abc"}, + "fractional limit": {"limit": "1.5"}, + "non-base64 cursor": {"cursor": "!!!"}, + // Valid base64 (length is a multiple of 4), so only the length cap + // can reject it. + "oversized cursor": {"cursor": strings.Repeat("A", maxRecommendationsCursorLen+4)}, + "bad cursor + limit": {"limit": "10", "cursor": "%%%"}, + } + for name, params := range cases { + t.Run(name, func(t *testing.T) { + // No scheduler: validation must reject before any store read. + h := &Handler{apiKey: "test-key"} + _, err := h.getRecommendations(context.Background(), pageTestReq(), params) + require.Error(t, err) + ce, ok := IsClientError(err) + require.True(t, ok, "expected ClientError, got %T: %v", err, err) + assert.Equal(t, 400, ce.code) + }) + } +} + +func TestGetRecommendations_PaginationMaxLimitAccepted(t *testing.T) { + h := pageTestHandler(t, pageTestRecs(3)) + resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{ + "limit": strconv.Itoa(MaxRecommendationsPageSize), + }) + require.NoError(t, err) + assert.Len(t, resp.Recommendations, 3) + assert.Empty(t, resp.NextCursor) +} diff --git a/internal/api/openapi.yaml b/internal/api/openapi.yaml index 15a10aeb4..ba54fee31 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -224,6 +224,26 @@ paths: schema: type: string - $ref: '#/components/parameters/AccountID' + - name: limit + in: query + description: >- + Opt-in page size (1-1000). When neither limit nor cursor is sent, + every matching recommendation is returned in one response (legacy + behavior). When either is sent, rows are returned in ascending id + order, one page at a time; summary and regions still describe the + full filtered set. + schema: + type: integer + minimum: 1 + maximum: 1000 + - name: cursor + in: query + description: >- + Opaque next_cursor value from the previous page. Implies + pagination; the page size defaults to 500 when limit is omitted. + schema: + type: string + maxLength: 1024 responses: '200': description: Recommendations list @@ -3073,11 +3093,30 @@ components: type: array items: $ref: '#/components/schemas/RecommendationRecord' - total_savings: - type: number - format: double - count: - type: integer + regions: + type: array + items: + type: string + summary: + type: object + description: Totals over the full filtered set, not just the returned page. + properties: + total_count: + type: integer + total_monthly_savings: + type: number + format: double + total_upfront_cost: + type: number + format: double + avg_payback_months: + type: number + format: double + next_cursor: + type: string + description: >- + Present only on a paginated request that has more rows; pass it as + the cursor query parameter to fetch the next page. RecommendationRecord: type: object diff --git a/internal/api/types.go b/internal/api/types.go index 9570dca8d..7e1243128 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -442,6 +442,9 @@ type RecommendationsResponse struct { Recommendations []config.RecommendationRecord `json:"recommendations"` Regions []string `json:"regions"` Summary RecommendationsSummary `json:"summary"` + // NextCursor is set only on a paginated request (limit or cursor sent) + // that has more rows; pass it back as ?cursor= for the next page. + NextCursor string `json:"next_cursor,omitempty"` } // UsagePoint is a single sample in the per-recommendation usage time From 09d9b29cdca1efdda72ea80a78c5dc43daa9d2e6 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 16:03:19 +0100 Subject: [PATCH 2/9] fix(api): cap recommendation pages by serialized bytes and reject empty paging params A page was bounded only by row count, but the opaque Details blob makes a row's size data-dependent, so 1000 large rows could still pass the 6 MiB Lambda response cap. Pages now also stop once the rows, measured as they appear inside the Lambda response envelope (the body re-encoded as a JSON string, so quotes and backslashes count twice), plus the summary, regions, keys and next_cursor would pass a 4 MiB budget. The budget is checked against uncompressed bytes because the #775 gzip only runs when the client accepts it. A page always holds at least one row, so a row that alone is over the budget is returned on its own page with a logged warning instead of stalling the cursor walk. Legacy unpaginated mode is now selected only when neither limit nor cursor is present. A present but empty ?limit= or ?cursor= returns 400 instead of falling back to the unbounded response. Refs #723 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_recommendations.go | 2 +- internal/api/handler_recommendations_page.go | 145 ++++++++++++---- .../api/handler_recommendations_page_test.go | 163 ++++++++++++++++++ internal/api/openapi.yaml | 14 +- 4 files changed, 287 insertions(+), 37 deletions(-) diff --git a/internal/api/handler_recommendations.go b/internal/api/handler_recommendations.go index ecb97a1fc..1f13a0b40 100644 --- a/internal/api/handler_recommendations.go +++ b/internal/api/handler_recommendations.go @@ -126,7 +126,7 @@ func (h *Handler) getRecommendations(ctx context.Context, req *events.LambdaFunc return nil, err } - return buildRecommendationsPageResponse(recommendations, page), nil + return buildRecommendationsPageResponse(recommendations, page) } // filterRecommendationsByAllowedAccounts filters recommendations to only include diff --git a/internal/api/handler_recommendations_page.go b/internal/api/handler_recommendations_page.go index 7557bedf6..9c14bae26 100644 --- a/internal/api/handler_recommendations_page.go +++ b/internal/api/handler_recommendations_page.go @@ -2,10 +2,13 @@ package api import ( "encoding/base64" + "encoding/json" + "fmt" "sort" "strconv" "strings" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" ) @@ -14,7 +17,10 @@ import ( // The list endpoint used to return the whole filtered cache in one body, // 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. +// pages, the `cursor` returned as `next_cursor`) gets one bounded page. 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 // unchanged, so the current frontend (which sorts, groups and totals the // full set client-side) keeps working until it adopts pagination. @@ -26,10 +32,21 @@ const ( // sends a cursor without a limit. DefaultRecommendationsPageSize = 500 - // MaxRecommendationsPageSize caps one page. List rows are roughly - // 500-600 bytes of JSON, so a full page stays far below the 6 MB cap. + // MaxRecommendationsPageSize caps the row count of one page. It is an + // upper bound only: recommendationsPageByteBudget can cut a page shorter. MaxRecommendationsPageSize = 1000 + // recommendationsPageByteBudget caps the bytes one page adds to the + // Lambda response payload. The Lambda limit (6 MiB, 6291456 bytes) + // applies to the whole invoke response: the JSON envelope with status + // code and headers, plus the body re-encoded as a JSON string, which + // escapes every quote and backslash again. Rows are therefore measured + // as they appear inside that string (see envelopeLen), and the budget is + // checked against uncompressed bytes because the #775 gzip only runs + // when the client sends Accept-Encoding: gzip. 4 MiB leaves about 2 MiB + // for the envelope, headers and next_cursor, which are a few KiB. + recommendationsPageByteBudget = 4 << 20 + // maxRecommendationsCursorLen bounds the encoded cursor so a hostile // query string cannot force a large decode. maxRecommendationsCursorLen = 1024 @@ -46,16 +63,23 @@ type recommendationsPage struct { // parseRecommendationsPage validates the optional limit and cursor query // parameters. limit must be an integer in 1..MaxRecommendationsPageSize; // out-of-range values are rejected rather than clamped so a client bug does -// not silently return a different page size than it asked for. +// not silently return a different page size than it asked for. Legacy +// unpaginated mode is selected only when neither parameter is present: a +// present but empty value (?limit= or ?cursor=) is a 400, so a client bug +// cannot silently fall back to the unbounded response. func parseRecommendationsPage(params map[string]string) (recommendationsPage, error) { - rawLimit := strings.TrimSpace(params["limit"]) - rawCursor := strings.TrimSpace(params["cursor"]) - if rawLimit == "" && rawCursor == "" { + rawLimit, hasLimit := params["limit"] + rawCursor, hasCursor := params["cursor"] + if !hasLimit && !hasCursor { return recommendationsPage{}, nil } + // From here a present but empty value fails the checks below: "" is not + // an integer, and an empty cursor does not decode to an ID. + rawLimit = strings.TrimSpace(rawLimit) + rawCursor = strings.TrimSpace(rawCursor) page := recommendationsPage{enabled: true, limit: DefaultRecommendationsPageSize} - if rawLimit != "" { + if hasLimit { n, err := strconv.Atoi(rawLimit) if err != nil || n < 1 || n > MaxRecommendationsPageSize { return recommendationsPage{}, NewClientError(400, @@ -63,7 +87,7 @@ func parseRecommendationsPage(params map[string]string) (recommendationsPage, er } page.limit = n } - if rawCursor != "" { + if hasCursor { afterID, err := decodeRecommendationsCursor(rawCursor) if err != nil { return recommendationsPage{}, err @@ -93,11 +117,9 @@ func decodeRecommendationsCursor(cursor string) (string, error) { return string(raw), nil } -// paginateRecommendations orders the filtered set by ID and returns the rows -// strictly after page.afterID, at most page.limit of them. recs is sorted in -// place (it is this request's own slice, freshly read from the store) and -// returned as sorted; rows is a sub-slice of it. nextCursor is empty on the -// last page. +// sortRecommendationsAfter sorts recs by ID in place (it is this request's +// own slice, freshly read from the store) and returns the index of the first +// row strictly after afterID. // // The ID is the cache table's primary key, so ID order is a total order and // no row can be skipped or repeated at a page boundary. @@ -105,32 +127,93 @@ func decodeRecommendationsCursor(cursor string) (string, error) { // Keyset (rather than offset) paging means a row added or removed between // two requests does not shift later pages: the next page always resumes // after the last ID the client saw, even if that row has since been deleted. -func paginateRecommendations(recs []config.RecommendationRecord, page recommendationsPage) (sorted, rows []config.RecommendationRecord, nextCursor string) { - sorted = recs - sort.Slice(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID }) +func sortRecommendationsAfter(recs []config.RecommendationRecord, afterID string) int { + sort.Slice(recs, func(i, j int) bool { return recs[i].ID < recs[j].ID }) + if afterID == "" { + return 0 + } + return sort.Search(len(recs), func(i int) bool { return recs[i].ID > afterID }) +} - start := 0 - if page.afterID != "" { - start = sort.Search(len(sorted), func(i int) bool { return sorted[i].ID > page.afterID }) +// envelopeLen returns how many bytes v adds to the Lambda invoke response: +// its JSON encoding, re-encoded as the content of the response's body +// string (without the surrounding quotes). +func envelopeLen(v any) (int, error) { + raw, err := json.Marshal(v) + if err != nil { + return 0, err } - end := min(start+page.limit, len(sorted)) - rows = sorted[start:end] - if end < len(sorted) { - nextCursor = encodeRecommendationsCursor(sorted[end-1].ID) + quoted, err := json.Marshal(string(raw)) + if err != nil { + return 0, err } - return sorted, rows, nextCursor + return len(quoted) - 2, nil +} + +// selectRecommendationsPage stamps and measures rows from sorted[start:] and +// returns how many fit in one page: at most limit rows, and no more bytes +// than budget once the fixed overhead (summary, regions, keys) and the next +// cursor are counted. It always takes at least one row when any remain, so a +// row that alone exceeds the budget cannot stall the cursor walk; that row +// is returned on its own page with a logged warning (it only fails at Lambda +// if it is over the 6 MiB hard cap by itself, which no page size could fix). +func selectRecommendationsPage(sorted []config.RecommendationRecord, start, limit, overhead, budget int) (int, error) { + used := overhead + n := 0 + for i := start; i < len(sorted) && n < limit; i++ { + stampComputeCapacity(&sorted[i]) + size, err := envelopeLen(&sorted[i]) + if err != nil { + return 0, fmt.Errorf("failed to size recommendation %q: %w", sorted[i].ID, err) + } + // +1 for the separating comma; the cursor that would point at this + // row is reserved so next_cursor always fits. + cost := size + 1 + cursorLen := base64.RawURLEncoding.EncodedLen(len(sorted[i].ID)) + if n > 0 && used+cost+cursorLen > budget { + break + } + if n == 0 && used+cost+cursorLen > budget { + logging.Warnf("recommendation %q alone is %d bytes, over the %d-byte page budget; returning it on its own page", sorted[i].ID, size, budget) + } + used += cost + n++ + } + return n, nil } // buildRecommendationsPageResponse builds the list response for either mode. // Unpaginated requests get exactly the legacy buildRecommendationsResponse // output; paginated requests get one page of rows plus next_cursor, with the // summary and regions still computed over the full filtered set. -func buildRecommendationsPageResponse(recs []config.RecommendationRecord, page recommendationsPage) *RecommendationsResponse { +func buildRecommendationsPageResponse(recs []config.RecommendationRecord, page recommendationsPage) (*RecommendationsResponse, error) { if !page.enabled { - return buildRecommendationsResponse(recs) + return buildRecommendationsResponse(recs), nil + } + return buildBudgetedPage(recs, page, recommendationsPageByteBudget) +} + +// buildBudgetedPage is the paginated branch of +// buildRecommendationsPageResponse with an explicit byte budget. +func buildBudgetedPage(recs []config.RecommendationRecord, page recommendationsPage, budget int) (*RecommendationsResponse, error) { + start := sortRecommendationsAfter(recs, page.afterID) + resp := summarizeRecommendations(recs, recs[start:start]) + // Overhead: everything except the rows and the cursor value, measured + // with an empty next_cursor key present. + resp.NextCursor = " " + overhead, err := envelopeLen(resp) + if err != nil { + return nil, fmt.Errorf("failed to size recommendations page: %w", err) + } + n, err := selectRecommendationsPage(recs, start, page.limit, overhead, budget) + if err != nil { + return nil, err + } + end := start + n + resp.Recommendations = recs[start:end] + resp.NextCursor = "" + if end < len(recs) { + resp.NextCursor = encodeRecommendationsCursor(recs[end-1].ID) } - sorted, rows, next := paginateRecommendations(recs, page) - resp := summarizeRecommendations(sorted, rows) - resp.NextCursor = next - return resp + return resp, nil } diff --git a/internal/api/handler_recommendations_page_test.go b/internal/api/handler_recommendations_page_test.go index c502db6f0..d550da8a4 100644 --- a/internal/api/handler_recommendations_page_test.go +++ b/internal/api/handler_recommendations_page_test.go @@ -191,6 +191,13 @@ func TestGetRecommendations_PaginationRejectsBadParams(t *testing.T) { // can reject it. "oversized cursor": {"cursor": strings.Repeat("A", maxRecommendationsCursorLen+4)}, "bad cursor + limit": {"limit": "10", "cursor": "%%%"}, + // Present but empty must not fall back to the unbounded legacy shape. + "empty limit": {"limit": ""}, + "blank limit": {"limit": " "}, + "empty cursor": {"cursor": ""}, + "blank cursor": {"cursor": " \t"}, + "valid limit + empty cursor": {"limit": "10", "cursor": ""}, + "empty limit + valid cursor": {"limit": "", "cursor": encodeRecommendationsCursor("rec-000001")}, } for name, params := range cases { t.Run(name, func(t *testing.T) { @@ -214,3 +221,159 @@ func TestGetRecommendations_PaginationMaxLimitAccepted(t *testing.T) { assert.Len(t, resp.Recommendations, 3) assert.Empty(t, resp.NextCursor) } + +// bigDetailsRecs returns n recommendations whose Details blob is detailBytes +// of quote-heavy JSON. Each quote doubles again when the body is re-encoded +// into the Lambda response envelope, so measuring raw JSON would undercount. +func bigDetailsRecs(n, detailBytes int) []config.RecommendationRecord { + recs := pageTestRecs(n) + note := strings.Repeat(`"<&\\`, detailBytes/4) + details, err := json.Marshal(map[string]string{"note": note}) + if err != nil { + panic(err) + } + for i := range recs { + recs[i].ResourceType = "m5.large" + recs[i].Details = details + } + return recs +} + +// lambdaPayloadLen is the size of the Lambda invoke response that would carry +// resp, matching how handler.go and the Lambda runtime encode it. +func lambdaPayloadLen(t *testing.T, resp *RecommendationsResponse) (bodyInEnvelope, total int) { + t.Helper() + body, err := json.Marshal(resp) + require.NoError(t, err) + quoted, err := json.Marshal(string(body)) + require.NoError(t, err) + envelope, err := json.Marshal(events.LambdaFunctionURLResponse{ + StatusCode: 200, + Headers: map[string]string{"Content-Type": "application/json"}, + Body: string(body), + }) + require.NoError(t, err) + return len(quoted) - 2, len(envelope) +} + +// TestGetRecommendations_PageCutByByteBudget: with large Details, limit rows +// would be far over the Lambda cap, so pages must be cut by serialized bytes +// while still visiting every row exactly once, in order, with full totals. +func TestGetRecommendations_PageCutByByteBudget(t *testing.T) { + const total = 1500 + recs := bigDetailsRecs(total, 1500) // ~10 KB per row in the envelope + h := pageTestHandler(t, recs) + + _, fullSize := lambdaPayloadLen(t, buildRecommendationsResponse(append([]config.RecommendationRecord(nil), recs...))) + require.Greater(t, fullSize, 2*lambdaResponseCapBytes, "fixture must be far over the cap") + + seen := make(map[string]int, total) + var order []string + cutByBytes := false + params := map[string]string{"limit": strconv.Itoa(MaxRecommendationsPageSize)} + for pages := 0; ; pages++ { + require.Less(t, pages, total, "cursor walk did not terminate") + resp, err := h.getRecommendations(context.Background(), pageTestReq(), params) + require.NoError(t, err) + + inEnvelope, payload := lambdaPayloadLen(t, resp) + assert.LessOrEqual(t, inEnvelope, recommendationsPageByteBudget, "page %d over the byte budget", pages) + assert.Less(t, payload, lambdaResponseCapBytes, "page %d over the Lambda cap", pages) + require.NotEmpty(t, resp.Recommendations, "a page must never be empty mid-walk") + if resp.NextCursor != "" && len(resp.Recommendations) < MaxRecommendationsPageSize { + cutByBytes = true + } + + assert.Equal(t, total, resp.Summary.TotalCount) + assert.InDelta(t, float64(total*10), resp.Summary.TotalMonthlySavings, 1e-6) + assert.InDelta(t, float64(total*20), resp.Summary.TotalUpfrontCost, 1e-6) + + for i := range resp.Recommendations { + seen[resp.Recommendations[i].ID]++ + order = append(order, resp.Recommendations[i].ID) + } + if resp.NextCursor == "" { + break + } + params = map[string]string{"limit": strconv.Itoa(MaxRecommendationsPageSize), "cursor": resp.NextCursor} + } + + assert.True(t, cutByBytes, "expected at least one page cut by the byte budget before limit") + assert.Len(t, seen, total) + for id, n := range seen { + assert.Equal(t, 1, n, "row %s returned %d times", id, n) + } + assert.IsIncreasing(t, order) +} + +// TestBuildBudgetedPage_OversizedRowAlone: a row that alone exceeds the +// budget is returned on its own page so the walk advances, and the rows +// around it still pack normally. +func TestBuildBudgetedPage_OversizedRowAlone(t *testing.T) { + recs := pageTestRecs(6) // ~1 KB rows: rec-000000 .. rec-000005 + huge := bigDetailsRecs(1, 8000)[0] + recs[2].Details = huge.Details // rec-000003 (pageTestRecs is reversed) + const budget = 4000 // fits a few small rows, never the big one + + var pages [][]string + page := recommendationsPage{enabled: true, limit: MaxRecommendationsPageSize} + for range len(recs) + 1 { + resp, err := buildBudgetedPage(append([]config.RecommendationRecord(nil), recs...), page, budget) + require.NoError(t, err) + require.NotEmpty(t, resp.Recommendations) + ids := make([]string, 0, len(resp.Recommendations)) + for i := range resp.Recommendations { + ids = append(ids, resp.Recommendations[i].ID) + } + pages = append(pages, ids) + if len(ids) > 1 { + inEnvelope, _ := lambdaPayloadLen(t, resp) + assert.LessOrEqual(t, inEnvelope, budget, "multi-row page over budget: %v", ids) + } + if resp.NextCursor == "" { + break + } + cursor, err := decodeRecommendationsCursor(resp.NextCursor) + require.NoError(t, err) + page.afterID = cursor + } + + assert.Contains(t, pages, []string{"rec-000003"}, "oversized row must be alone on its page") + all := make([]string, 0, len(recs)) + for _, p := range pages { + all = append(all, p...) + } + assert.Equal(t, []string{"rec-000000", "rec-000001", "rec-000002", "rec-000003", "rec-000004", "rec-000005"}, all) + assert.Greater(t, len(pages[0]), 1, "small rows before the big one must still share a page") +} + +// TestBuildBudgetedPage_TightBudgetSweep sweeps the budget one byte at a time +// so the accounting is checked at its exact edges: every multi-row page must +// fit (counting the summary, regions, keys and next_cursor), and the page +// must be maximal, i.e. one more row would not have fit. +func TestBuildBudgetedPage_TightBudgetSweep(t *testing.T) { + recs := pageTestRecs(6) + page := recommendationsPage{enabled: true, limit: MaxRecommendationsPageSize} + for budget := 1500; budget <= 7000; budget++ { + resp, err := buildBudgetedPage(append([]config.RecommendationRecord(nil), recs...), page, budget) + require.NoError(t, err) + n := len(resp.Recommendations) + require.GreaterOrEqual(t, n, 1) + if n > 1 { + inEnvelope, _ := lambdaPayloadLen(t, resp) + require.LessOrEqual(t, inEnvelope, budget, "budget %d: %d rows do not fit", budget, n) + } + if n < len(recs) { + // The same page plus the next row, with that row's cursor. + bigger, err := buildBudgetedPage(append([]config.RecommendationRecord(nil), recs...), + recommendationsPage{enabled: true, limit: n + 1}, 1<<30) + require.NoError(t, err) + bigger.NextCursor = encodeRecommendationsCursor(bigger.Recommendations[n].ID) + inEnvelope, _ := lambdaPayloadLen(t, bigger) + // The paginator overcounts by at most 2 bytes (one comma and the + // one-byte cursor placeholder in the overhead), so anything it + // rejected must have been within 2 bytes of overflowing. + require.Greater(t, inEnvelope+2, budget, "budget %d: page of %d rows is not maximal", budget, n) + } + } +} diff --git a/internal/api/openapi.yaml b/internal/api/openapi.yaml index ba54fee31..00fbdfb46 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -227,11 +227,15 @@ paths: - name: limit in: query description: >- - Opt-in page size (1-1000). When neither limit nor cursor is sent, - every matching recommendation is returned in one response (legacy - behavior). When either is sent, rows are returned in ascending id - order, one page at a time; summary and regions still describe the - full filtered set. + Opt-in maximum page size in rows (1-1000). When neither limit nor + cursor is present, every matching recommendation is returned in + one response (legacy behavior); a present but empty value is a + 400. When either is sent, rows are returned in ascending id order, + one page at a time, and a page can hold fewer than limit rows + (follow next_cursor, not the row count) because each page is also + capped at 4 MiB of serialized JSON. A single row larger than that + is returned alone. summary and regions still describe the full + filtered set. schema: type: integer minimum: 1 From 11ddfea57e19cb2bfecc844daf593b25a90a0012 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Wed, 21 Oct 2026 09:38:18 +0100 Subject: [PATCH 3/9] fix(api): reject duplicate recommendation IDs when paging and document 400s - Correct the ordering comment: the paging key is the application recommendation ID built from the natural-key columns, not the cache table's UUID primary key. A paginated request now fails if two rows share an ID, since the strict "after" comparison would skip one of them at a page boundary. - Describe the feature as opt-in in-memory cursor slicing, not a SQL keyset (store-side paging is #981). - TestGetRecommendations_UnpaginatedShapeUnchanged copies the expected IDs before the call so an in-place reorder in legacy mode fails it. - Add TestGetRecommendations_PaginationRejectsDuplicateIDs. - openapi: document the 400 response for bad limit/cursor/filter values, and say next_cursor's absence is the only end-of-list signal. Refs #723 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_recommendations_page.go | 38 +++++++++++++------ .../api/handler_recommendations_page_test.go | 32 +++++++++++++++- internal/api/openapi.yaml | 16 +++++++- 3 files changed, 72 insertions(+), 14 deletions(-) diff --git a/internal/api/handler_recommendations_page.go b/internal/api/handler_recommendations_page.go index 9c14bae26..dd40dbb04 100644 --- a/internal/api/handler_recommendations_page.go +++ b/internal/api/handler_recommendations_page.go @@ -12,12 +12,15 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/config" ) -// Opt-in keyset pagination for GET /api/recommendations (issue #723). +// Opt-in cursor pagination for GET /api/recommendations (refs issue #723). // // The list endpoint used to return the whole filtered cache in one body, // 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 +// pages, the `cursor` returned as `next_cursor`) gets one bounded page. +// This bounds the response only: every request still reads, filters and +// sorts the whole filtered set in memory and then slices the page from it +// (store-side paging is tracked in #981). 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. @@ -121,18 +124,28 @@ func decodeRecommendationsCursor(cursor string) (string, error) { // own slice, freshly read from the store) and returns the index of the first // row strictly after afterID. // -// The ID is the cache table's primary key, so ID order is a total order and -// no row can be skipped or repeated at a page boundary. +// The ID is the application recommendation ID that the scheduler builds from +// the natural-key columns (recommendationID in internal/scheduler), not the +// cache table's UUID primary key. Paging needs ID order to be a total order: +// two rows sharing an ID would let the strict "after" comparison skip the +// second one at a page boundary. A duplicate is therefore an error rather +// than a silently short walk. // -// Keyset (rather than offset) paging means a row added or removed between -// two requests does not shift later pages: the next page always resumes -// after the last ID the client saw, even if that row has since been deleted. -func sortRecommendationsAfter(recs []config.RecommendationRecord, afterID string) int { +// Resuming after the last ID (rather than at an offset) means a row added or +// removed between two requests does not shift later pages: the next page +// always resumes after the last ID the client saw, even if that row has +// since been deleted. +func sortRecommendationsAfter(recs []config.RecommendationRecord, afterID string) (int, error) { sort.Slice(recs, func(i, j int) bool { return recs[i].ID < recs[j].ID }) + for i := 1; i < len(recs); i++ { + if recs[i].ID == recs[i-1].ID { + return 0, fmt.Errorf("duplicate recommendation ID %q: cannot page the recommendations list", recs[i].ID) + } + } if afterID == "" { - return 0 + return 0, nil } - return sort.Search(len(recs), func(i int) bool { return recs[i].ID > afterID }) + return sort.Search(len(recs), func(i int) bool { return recs[i].ID > afterID }), nil } // envelopeLen returns how many bytes v adds to the Lambda invoke response: @@ -196,7 +209,10 @@ func buildRecommendationsPageResponse(recs []config.RecommendationRecord, page r // buildBudgetedPage is the paginated branch of // buildRecommendationsPageResponse with an explicit byte budget. func buildBudgetedPage(recs []config.RecommendationRecord, page recommendationsPage, budget int) (*RecommendationsResponse, error) { - start := sortRecommendationsAfter(recs, page.afterID) + start, err := sortRecommendationsAfter(recs, page.afterID) + if err != nil { + return nil, err + } resp := summarizeRecommendations(recs, recs[start:start]) // Overhead: everything except the rows and the cursor value, measured // with an empty next_cursor key present. diff --git a/internal/api/handler_recommendations_page_test.go b/internal/api/handler_recommendations_page_test.go index d550da8a4..101986cfc 100644 --- a/internal/api/handler_recommendations_page_test.go +++ b/internal/api/handler_recommendations_page_test.go @@ -105,14 +105,22 @@ func TestGetRecommendations_PaginationBoundsResponse(t *testing.T) { // in the order the scheduler gave them, with no next_cursor. func TestGetRecommendations_UnpaginatedShapeUnchanged(t *testing.T) { recs := pageTestRecs(5) + // Copy the expected order before the call: the mock hands the handler + // this same backing array, so an in-place sort would also reorder recs. + wantIDs := make([]string, len(recs)) + for i := range recs { + wantIDs[i] = recs[i].ID + } h := pageTestHandler(t, recs) resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{}) require.NoError(t, err) require.Len(t, resp.Recommendations, 5) - for i := range recs { - assert.Equal(t, recs[i].ID, resp.Recommendations[i].ID) + gotIDs := make([]string, len(resp.Recommendations)) + for i := range resp.Recommendations { + gotIDs[i] = resp.Recommendations[i].ID } + assert.Equal(t, wantIDs, gotIDs, "legacy mode must keep the scheduler's order") assert.Empty(t, resp.NextCursor) assert.Equal(t, 5, resp.Summary.TotalCount) @@ -121,6 +129,26 @@ func TestGetRecommendations_UnpaginatedShapeUnchanged(t *testing.T) { assert.NotContains(t, string(body), "next_cursor") } +// TestGetRecommendations_PaginationRejectsDuplicateIDs pins the total-order +// guard: two rows sharing an ID would let the strict "after" comparison skip +// one of them at a page boundary (a,b,b,c with limit=2 used to return only +// three rows), so a paginated request fails instead of walking short. Legacy +// mode does not sort or page and still returns every row. +func TestGetRecommendations_PaginationRejectsDuplicateIDs(t *testing.T) { + recs := pageTestRecs(4) + recs[1].ID = recs[2].ID + h := pageTestHandler(t, recs) + ctx := context.Background() + + _, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"limit": "2"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "duplicate recommendation ID") + + legacy, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{}) + require.NoError(t, err) + assert.Len(t, legacy.Recommendations, 4) +} + func TestGetRecommendations_PaginationPages(t *testing.T) { recs := pageTestRecs(5) // rec-000000 .. rec-000004, stored in reverse h := pageTestHandler(t, recs) diff --git a/internal/api/openapi.yaml b/internal/api/openapi.yaml index abba72eb9..f3ce19c76 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -325,6 +325,17 @@ paths: application/json: schema: $ref: '#/components/schemas/RecommendationsResponse' + '400': + description: > + Invalid query parameter: a limit that is empty, not an integer or + outside 1-1000; a cursor that is empty, longer than 1024 + characters or not a cursor this endpoint issued; or an invalid + provider, service, region, account_ids, min_savings_usd or + min_savings_pct filter. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' '401': $ref: '#/components/responses/Unauthorized' '403': @@ -3479,7 +3490,10 @@ components: type: string description: >- Present only on a paginated request that has more rows; pass it as - the cursor query parameter to fetch the next page. + the cursor query parameter to fetch the next page. Its absence is + the only end-of-list signal: a page with fewer rows than limit is + not necessarily the last one, because each page is also capped by + serialized size. RecommendationRecord: type: object From 9ba77c121c452c5368522bc01b4b4ed7fb4d5c70 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 22 Oct 2026 01:39:46 +0100 Subject: [PATCH 4/9] perf(db): index recommendations by byte-wise payload id for keyset scans Migration 000142 adds recommendations_scan_key_idx on (COALESCE(payload->>'id', '') COLLATE "C", id). The paginated GET /api/recommendations scan (issue #981) walks the cache in chunks in that order and resumes each chunk after the last (id, primary key) it read; without the index every chunk is a full scan plus a sort. COLLATE "C" is byte order, the same order Go's string comparison uses, so SQL paging keeps the order the API already returns. Refs #981 Co-Authored-By: Claude Opus 5.5 --- ...0142_recommendations_scan_key_idx.down.sql | 1 + ...000142_recommendations_scan_key_idx.up.sql | 21 ++++++++ ...00142_recommendations_scan_key_idx_test.go | 52 +++++++++++++++++++ 3 files changed, 74 insertions(+) create mode 100644 internal/database/postgres/migrations/000142_recommendations_scan_key_idx.down.sql create mode 100644 internal/database/postgres/migrations/000142_recommendations_scan_key_idx.up.sql create mode 100644 internal/database/postgres/migrations/000142_recommendations_scan_key_idx_test.go diff --git a/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.down.sql b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.down.sql new file mode 100644 index 000000000..77faacddf --- /dev/null +++ b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS recommendations_scan_key_idx; diff --git a/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.up.sql b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.up.sql new file mode 100644 index 000000000..551d0b428 --- /dev/null +++ b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.up.sql @@ -0,0 +1,21 @@ +-- Index the keyset order GET /api/recommendations pages in (issue #981). +-- +-- A paginated request walks the recommendations cache in chunks ordered by the +-- application recommendation ID (payload->>'id'), with the table's UUID +-- primary key as the tie-breaker, and resumes each chunk after the last row it +-- read. Without an index on that order every chunk is a full scan plus a +-- top-N sort, so a walk of N rows in chunks of C costs O(N^2 / C). +-- +-- COLLATE "C" compares bytes, which is the order Go's string comparison uses, +-- so the SQL order matches the order the API has always paged in. COALESCE +-- maps a missing or null id to '' (what the Go decoder produces), so no row +-- falls outside the order. The expression must stay identical to +-- recommendationScanKeyExpr in internal/config/store_postgres_recommendations.go +-- or the planner cannot use the index. +-- +-- The table is a regenerable cache rewritten every collection cycle, so a +-- plain CREATE INDEX (which blocks writes while it builds) is acceptable; the +-- migration runner wraps every migration in a transaction, which rules out +-- CREATE INDEX CONCURRENTLY anyway. +CREATE INDEX IF NOT EXISTS recommendations_scan_key_idx + ON recommendations ((COALESCE(payload->>'id', '') COLLATE "C"), id); diff --git a/internal/database/postgres/migrations/000142_recommendations_scan_key_idx_test.go b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx_test.go new file mode 100644 index 000000000..1e7bf95fb --- /dev/null +++ b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx_test.go @@ -0,0 +1,52 @@ +//go:build integration + +package migrations_test + +import ( + "context" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/stretchr/testify/require" +) + +// TestMigration_RecommendationsScanKeyIndex checks that 000142 creates the +// keyset index the paginated recommendations scan relies on (issue #981), in +// byte-wise ("C") order with the primary key as tie-breaker, and that the down +// migration removes it. +func TestMigration_RecommendationsScanKeyIndex(t *testing.T) { + ctx := context.Background() + container := testhelpers.RequirePostgresContainer(ctx, t) + t.Cleanup(func() { require.NoError(t, container.Cleanup(context.Background())) }) + pool, path := container.DB.Pool(), getMigrationsPath() + + const version = 142 + prev := previousMigrationVersion(t, version) + + indexDef := func() string { + t.Helper() + var def string + require.NoError(t, pool.QueryRow(ctx, ` + SELECT COALESCE((SELECT indexdef FROM pg_indexes + WHERE tablename = 'recommendations' + AND indexname = 'recommendations_scan_key_idx'), '')`).Scan(&def)) + return def + } + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, prev)) + require.Empty(t, indexDef(), "before 000142 the index must not exist") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version)) + def := indexDef() + require.Contains(t, def, `COALESCE((payload ->> 'id'::text), ''::text) COLLATE "C", id)`, + "the index must be on the byte-wise id, then the primary key") + require.NotContains(t, def, "UNIQUE", "rows may share an application id; the primary key breaks ties") + require.NotContains(t, def, "WHERE", "a partial index cannot serve an unfiltered scan") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, prev)) + require.Empty(t, indexDef(), "down must drop the index") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version)) + require.NotEmpty(t, indexDef(), "up must re-apply cleanly after down") +} From 5d6b5491a1b1cf2ded00c4581e9a3d251f18e281 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 22 Oct 2026 01:39:46 +0100 Subject: [PATCH 5/9] perf(config): read stored recommendations in keyset chunks ListStoredRecommendationsChunk returns at most limit rows matching a RecommendationFilter, ordered by the byte-wise recommendation ID and then the table's UUID primary key, resuming strictly after a given position. The primary key keeps rows that share an ID in a total order, so a chunk boundary inside such a run neither skips nor repeats a row. The MinSavingsPct floor stays in-process (shared with ListStoredRecommendations via decodeStoredRecommendation); rows it drops still advance the position. Integration tests pin the order to Go's sort.Strings on IDs that a linguistic collation orders differently, cover duplicate IDs across every chunk size, the pct floor emptying a whole chunk, the empty table, an EXPLAIN check that recommendations_scan_key_idx serves the query without a Sort node, and walks racing concurrent upserts. Refs #981 Co-Authored-By: Claude Opus 5.5 --- internal/analytics/collector_test.go | 3 + .../export_recommendations_scan_test.go | 6 + internal/config/interfaces.go | 4 + .../config/store_postgres_recommendations.go | 33 ++- .../store_postgres_recommendations_scan.go | 92 ++++++ ...tore_postgres_recommendations_scan_test.go | 263 ++++++++++++++++++ internal/config/types.go | 20 ++ internal/mocks/stores.go | 13 + internal/server/test_helpers_test.go | 3 + 9 files changed, 426 insertions(+), 11 deletions(-) create mode 100644 internal/config/export_recommendations_scan_test.go create mode 100644 internal/config/store_postgres_recommendations_scan.go create mode 100644 internal/config/store_postgres_recommendations_scan_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 2400b1ad9..39876c4fc 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -489,6 +489,9 @@ func (m *mockConfigStore) UpsertRecommendations(_ context.Context, _ time.Time, func (m *mockConfigStore) ListStoredRecommendations(_ context.Context, _ config.RecommendationFilter) ([]config.RecommendationRecord, error) { return nil, nil } +func (m *mockConfigStore) ListStoredRecommendationsChunk(_ context.Context, _ config.RecommendationFilter, _ *config.RecommendationScanKey, _ int) (config.RecommendationChunk, error) { + return config.RecommendationChunk{Done: true}, nil +} func (m *mockConfigStore) GetRecommendationsFreshness(_ context.Context) (*config.RecommendationsFreshness, error) { return &config.RecommendationsFreshness{}, nil } diff --git a/internal/config/export_recommendations_scan_test.go b/internal/config/export_recommendations_scan_test.go new file mode 100644 index 000000000..cf1fb3e76 --- /dev/null +++ b/internal/config/export_recommendations_scan_test.go @@ -0,0 +1,6 @@ +package config + +// BuildRecommendationChunkQuery exposes the keyset chunk query to the external +// integration tests, which EXPLAIN it to prove recommendations_scan_key_idx +// (migration 000142) can serve it. +var BuildRecommendationChunkQuery = buildRecommendationChunkQuery diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index f2e627890..75dabdc3f 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -548,6 +548,10 @@ type StoreInterface interface { ReplaceRecommendations(ctx context.Context, collectedAt time.Time, recs []RecommendationRecord) error UpsertRecommendations(ctx context.Context, collectedAt time.Time, recs []RecommendationRecord, successfulCollects []SuccessfulCollect) error ListStoredRecommendations(ctx context.Context, filter RecommendationFilter) ([]RecommendationRecord, error) + // ListStoredRecommendationsChunk is ListStoredRecommendations for a + // 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) 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/store_postgres_recommendations.go b/internal/config/store_postgres_recommendations.go index d95877c47..87513e902 100644 --- a/internal/config/store_postgres_recommendations.go +++ b/internal/config/store_postgres_recommendations.go @@ -410,19 +410,13 @@ func (s *PostgresStore) ListStoredRecommendations(ctx context.Context, filter Re if err := rows.Scan(&payload); err != nil { return nil, fmt.Errorf("failed to scan payload: %w", err) } - var rec RecommendationRecord - if err := json.Unmarshal(payload, &rec); err != nil { - return nil, fmt.Errorf("failed to unmarshal payload: %w", err) + rec, keep, err := decodeStoredRecommendation(payload, filter) + if err != nil { + return nil, err } - // Apply the in-process percentage floor. Recs where the percentage - // cannot be computed (missing on-demand baseline) pass through -- we - // only drop a rec when pct is computable AND below the threshold. - if filter.MinSavingsPct > 0 { - if pct, ok := recEffectiveSavingsPct(&rec); ok && pct < filter.MinSavingsPct { - continue - } + if keep { + out = append(out, rec) } - out = append(out, rec) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("rows iteration: %w", err) @@ -430,6 +424,23 @@ func (s *PostgresStore) ListStoredRecommendations(ctx context.Context, filter Re return out, nil } +// decodeStoredRecommendation unmarshals one stored payload and applies the +// in-process MinSavingsPct floor. keep is false when the floor drops the rec. +// Recs where the percentage cannot be computed (missing on-demand baseline) +// pass through: a rec is only dropped when pct is computable AND below the +// threshold. +func decodeStoredRecommendation(payload []byte, filter RecommendationFilter) (rec RecommendationRecord, keep bool, err error) { + if err := json.Unmarshal(payload, &rec); err != nil { + return RecommendationRecord{}, false, fmt.Errorf("failed to unmarshal payload: %w", err) + } + if filter.MinSavingsPct > 0 { + if pct, ok := recEffectiveSavingsPct(&rec); ok && pct < filter.MinSavingsPct { + return rec, false, nil + } + } + return rec, true, nil +} + // GetRecommendationsFreshness returns the singleton freshness row. The // table is seeded with id=1 by the migration so a row always exists; // LastCollectedAt, LastCollectionError, LastCollectionStartedAt and diff --git a/internal/config/store_postgres_recommendations_scan.go b/internal/config/store_postgres_recommendations_scan.go new file mode 100644 index 000000000..a6bdf67f5 --- /dev/null +++ b/internal/config/store_postgres_recommendations_scan.go @@ -0,0 +1,92 @@ +package config + +import ( + "context" + "errors" + "fmt" +) + +// recommendationScanKeyExpr is the application recommendation ID in the +// byte-wise order the keyset scan walks (issue #981). COLLATE "C" compares +// bytes, the same order as Go's string comparison, so paging through SQL +// returns rows in the order the API used to produce by sorting in Go. +// COALESCE maps a missing or JSON-null id to "", which is what decoding the +// payload yields. It must stay identical to the expression indexed by +// migration 000142 (recommendations_scan_key_idx), or the planner falls back +// to a full scan and sort for every chunk. +const recommendationScanKeyExpr = `(COALESCE(payload->>'id', '') COLLATE "C")` + +// ListStoredRecommendationsChunk reads at most limit rows matching filter, +// ordered by recommendation ID (byte-wise) and then by the table's UUID +// primary key, starting strictly after the position after (nil means from +// the beginning). The primary key gives rows that share an ID a total order, +// so resuming from Last never skips or repeats a row, even at a run of +// duplicate IDs. +// +// The SQL-pushed conditions are the same as ListStoredRecommendations; the +// MinSavingsPct floor is still applied in-process, so Recs can be shorter +// than the rows read. Last is the position of the last row read (before that +// floor), and Done reports that fewer than limit rows were read, meaning +// nothing matches after Last. Each call is a single statement, so a chunk is +// one consistent snapshot; consecutive chunks are not. +func (s *PostgresStore) ListStoredRecommendationsChunk(ctx context.Context, filter RecommendationFilter, after *RecommendationScanKey, limit int) (RecommendationChunk, error) { + if limit < 1 { + return RecommendationChunk{}, errors.New("recommendations chunk limit must be positive") + } + if err := s.checkAzurePricingMigration(ctx, filter.RequireAzurePricingMigration); err != nil { + return RecommendationChunk{}, err + } + query, args := buildRecommendationChunkQuery(filter, after, limit) + rows, err := s.db.Query(ctx, query, args...) + if err != nil { + return RecommendationChunk{}, fmt.Errorf("failed to query recommendations chunk: %w", err) + } + defer rows.Close() + + var chunk RecommendationChunk + read := 0 + for rows.Next() { + var key RecommendationScanKey + var payload []byte + if err := rows.Scan(&key.RowID, &key.ID, &payload); err != nil { + return RecommendationChunk{}, fmt.Errorf("failed to scan recommendations chunk row: %w", err) + } + read++ + chunk.Last = &key + rec, keep, err := decodeStoredRecommendation(payload, filter) + if err != nil { + return RecommendationChunk{}, err + } + if keep { + chunk.Recs = append(chunk.Recs, rec) + } + } + if err := rows.Err(); err != nil { + return RecommendationChunk{}, fmt.Errorf("recommendations chunk rows iteration: %w", err) + } + chunk.Done = read < limit + return chunk, nil +} + +// buildRecommendationChunkQuery appends the keyset bound, ORDER BY and LIMIT +// to the ListStoredRecommendations filter. The row comparison against +// (ID, primary key) is what recommendations_scan_key_idx can serve directly. +func buildRecommendationChunkQuery(filter RecommendationFilter, after *RecommendationScanKey, limit int) (query string, args []any) { + where, args := buildRecommendationFilter(filter) + if after != nil { + cond := fmt.Sprintf("(%s, id) > ($%d, $%d::uuid)", recommendationScanKeyExpr, len(args)+1, len(args)+2) + if where == "" { + where = " WHERE " + cond + } else { + where += " AND " + cond + } + args = append(args, after.ID, after.RowID) + } + args = append(args, limit) + // The output columns are aliased and the primary key in ORDER BY is + // table-qualified: a bare "id" there would resolve to the id::text output + // column, a different sort key the index cannot serve. + query = fmt.Sprintf(`SELECT recommendations.id::text AS row_id, %s AS rec_id, payload FROM recommendations%s ORDER BY %s, recommendations.id LIMIT $%d`, + recommendationScanKeyExpr, where, recommendationScanKeyExpr, len(args)) + return query, args +} diff --git a/internal/config/store_postgres_recommendations_scan_test.go b/internal/config/store_postgres_recommendations_scan_test.go new file mode 100644 index 000000000..89d16539d --- /dev/null +++ b/internal/config/store_postgres_recommendations_scan_test.go @@ -0,0 +1,263 @@ +//go:build integration +// +build integration + +package config_test + +import ( + "context" + "fmt" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// scanRec is an AWS rec with the given application ID. resourceType must be +// unique per row: it is the part of the natural key these tests vary. +func scanRec(id, resourceType string) config.RecommendationRecord { + return awsRec(id, "ec2", "us-east-1", resourceType, 10) +} + +// walkChunks reads every chunk for filter, chunkSize rows at a time, and +// returns the IDs of the recs in scan order plus the number of reads. +func walkChunks(ctx context.Context, t *testing.T, store *config.PostgresStore, filter config.RecommendationFilter, chunkSize int) (ids []string, reads int) { + t.Helper() + var after *config.RecommendationScanKey + for { + reads++ + require.Less(t, reads, 10000, "chunk walk did not terminate") + chunk, err := store.ListStoredRecommendationsChunk(ctx, filter, after, chunkSize) + require.NoError(t, err) + for i := range chunk.Recs { + ids = append(ids, chunk.Recs[i].ID) + } + if chunk.Done { + return ids, reads + } + require.NotNil(t, chunk.Last, "a chunk that is not done must say where to resume") + after = chunk.Last + } +} + +// TestListStoredRecommendationsChunk_OrderMatchesGoStringOrder pins the scan +// order to Go's byte-wise string order (issue #981), which is the order the +// paginated API sorted rows into before the keyset moved into SQL. The IDs mix +// case, "|", "-", "/", spaces, non-ASCII and the empty ID, where a linguistic +// collation (en_US and friends) orders differently from bytes. +func TestListStoredRecommendationsChunk_OrderMatchesGoStringOrder(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + ids := []string{ + "aws|b", "AWS|a", "aws-a", "aws a", "aws/a", "aws|a|Linux/UNIX|default", + "aws|a", "Zeta", "alpha", "élan", "elan", "_x", "", "aws|a|Windows|default", + } + recs := make([]config.RecommendationRecord, len(ids)) + for i, id := range ids { + recs[i] = scanRec(id, fmt.Sprintf("t%02d", i)) + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + + want := append([]string(nil), ids...) + sort.Strings(want) + + for _, chunkSize := range []int{1, 2, 3, len(ids), len(ids) + 1} { + t.Run(fmt.Sprintf("chunk=%d", chunkSize), func(t *testing.T) { + got, reads := walkChunks(ctx, t, store, config.RecommendationFilter{}, chunkSize) + assert.Equal(t, want, got, "scan order must equal Go's sort.Strings order") + // The walk ends on the first short chunk: exactly enough reads to + // cover every row plus, when the count divides evenly, one empty + // read to learn there is nothing after the last row. + assert.Equal(t, len(ids)/chunkSize+1, reads) + }) + } +} + +// TestListStoredRecommendationsChunk_DuplicateIDsNeverSkipped pins the +// tie-break: rows that share an application ID are ordered by primary key, and +// a chunk boundary that falls inside such a run neither skips nor repeats a +// row, so a caller further up sees every duplicate and can reject it. +func TestListStoredRecommendationsChunk_DuplicateIDsNeverSkipped(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + recs := []config.RecommendationRecord{ + scanRec("a", "t1"), scanRec("b", "t2"), scanRec("b", "t3"), scanRec("b", "t4"), scanRec("c", "t5"), + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + + for chunkSize := 1; chunkSize <= len(recs)+1; chunkSize++ { + got, _ := walkChunks(ctx, t, store, config.RecommendationFilter{}, chunkSize) + assert.Equal(t, []string{"a", "b", "b", "b", "c"}, got, "chunk size %d", chunkSize) + } +} + +// TestListStoredRecommendationsChunk_FiltersAndPctFloor checks the SQL filters +// still apply, and that rows dropped by the in-process MinSavingsPct floor +// still advance the position, so a chunk whose rows are all below the floor is +// empty but not the end of the walk. +func TestListStoredRecommendationsChunk_FiltersAndPctFloor(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + pct := func(id, resourceType string, p float64) config.RecommendationRecord { + rec := scanRec(id, resourceType) + rec.SavingsPercentage = &p + return rec + } + other := scanRec("r00", "t0") + other.Region = "eu-west-1" + recs := []config.RecommendationRecord{ + other, + pct("r01", "t1", 5), pct("r02", "t2", 5), // a whole chunk below the floor + pct("r03", "t3", 50), pct("r04", "t4", 5), pct("r05", "t5", 60), + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + + filter := config.RecommendationFilter{Region: "us-east-1", MinSavingsPct: 30} + first, err := store.ListStoredRecommendationsChunk(ctx, filter, nil, 2) + require.NoError(t, err) + assert.Empty(t, first.Recs, "both rows of the first chunk are below the floor") + assert.False(t, first.Done, "a full chunk is not the end even when the floor emptied it") + require.NotNil(t, first.Last) + assert.Equal(t, "r02", first.Last.ID) + assert.NotEmpty(t, first.Last.RowID) + + got, _ := walkChunks(ctx, t, store, filter, 2) + assert.Equal(t, []string{"r03", "r05"}, got) + + _, err = store.ListStoredRecommendationsChunk(ctx, filter, nil, 0) + require.Error(t, err, "a non-positive limit is a caller bug, not an empty result") +} + +// TestListStoredRecommendationsChunk_EmptyTable returns a done, empty chunk +// with no position. +func TestListStoredRecommendationsChunk_EmptyTable(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + chunk, err := store.ListStoredRecommendationsChunk(ctx, config.RecommendationFilter{}, nil, 10) + require.NoError(t, err) + assert.Empty(t, chunk.Recs) + assert.Nil(t, chunk.Last) + assert.True(t, chunk.Done) +} + +// TestListStoredRecommendationsChunk_UsesScanKeyIndex proves the chunk query +// matches the expression migration 000142 indexes: with sequential and bitmap +// scans disabled the plan reads recommendations_scan_key_idx in order and has no +// Sort node. A drift between recommendationScanKeyExpr and the index (for +// example dropping COLLATE "C") brings the full scan plus sort back for every +// chunk. +func TestListStoredRecommendationsChunk_UsesScanKeyIndex(t *testing.T) { + ctx := context.Background() + db := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath()).DB.Pool() + + for name, after := range map[string]*config.RecommendationScanKey{ + "first chunk": nil, + "resume": {ID: "aws|x", RowID: "00000000-0000-0000-0000-000000000001"}, + } { + t.Run(name, func(t *testing.T) { + query, args := config.BuildRecommendationChunkQuery(config.RecommendationFilter{}, after, 500) + plan := explainWithoutSeqScan(ctx, t, db, query, args) + assert.Contains(t, plan, "recommendations_scan_key_idx", plan) + assert.NotContains(t, plan, "Sort", plan) + }) + } +} + +func explainWithoutSeqScan(ctx context.Context, t *testing.T, db interface { + Begin(context.Context) (pgx.Tx, error) +}, query string, args []any) string { + t.Helper() + tx, err := db.Begin(ctx) + require.NoError(t, err) + defer func() { _ = tx.Rollback(ctx) }() + // An empty test table makes any plan cheap, so steer the planner away + // from the scans that do not return rows in index order; the question + // is whether the ordered index scan can serve the query at all. + _, err = tx.Exec(ctx, `SET LOCAL enable_seqscan = off; SET LOCAL enable_bitmapscan = off`) + require.NoError(t, err) + rows, err := tx.Query(ctx, "EXPLAIN "+query, args...) + require.NoError(t, err) + defer rows.Close() + var lines []string + for rows.Next() { + var line string + require.NoError(t, rows.Scan(&line)) + lines = append(lines, line) + } + require.NoError(t, rows.Err()) + return strings.Join(lines, "\n") +} + +// TestListStoredRecommendationsChunk_ConcurrentUpsertsDuringWalk runs chunk +// walks while collections keep upserting the same rows (new payload values, +// same natural keys) and inserting and evicting other rows. Every row that +// exists for the whole walk must come back exactly once and in order: an +// upsert keeps the primary key, so the (ID, primary key) position stays +// valid between chunks. +func TestListStoredRecommendationsChunk_ConcurrentUpsertsDuringWalk(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + const stable = 300 + base := make([]config.RecommendationRecord, stable) + wantStable := make([]string, stable) + for i := range base { + base[i] = scanRec(fmt.Sprintf("s%04d", i), fmt.Sprintf("s%04d", i)) + wantStable[i] = base[i].ID + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), base, nil)) + + stop := make(chan struct{}) + var wg sync.WaitGroup + writerErr := make(chan error, 1) + wg.Add(1) + go func() { + defer wg.Done() + for round := 0; ; round++ { + select { + case <-stop: + return + default: + } + recs := append([]config.RecommendationRecord(nil), base...) + for i := range recs { + recs[i].Savings = float64(round) + } + // A churning row that sorts between stable ones. + recs = append(recs, scanRec(fmt.Sprintf("s%04d-churn-%d", round%stable, round), fmt.Sprintf("c%06d", round))) + if err := store.UpsertRecommendations(ctx, time.Now().UTC(), recs, []config.SuccessfulCollect{{Provider: "aws"}}); err != nil { + writerErr <- err + return + } + } + }() + + for walk := 0; walk < 20; walk++ { + got, _ := walkChunks(ctx, t, store, config.RecommendationFilter{}, 7) + assert.True(t, sort.StringsAreSorted(got), "walk %d out of order", walk) + var gotStable []string + for _, id := range got { + if !strings.Contains(id, "-churn-") { + gotStable = append(gotStable, id) + } + } + require.Equal(t, wantStable, gotStable, "walk %d must return every stable row exactly once", walk) + } + close(stop) + wg.Wait() + select { + case err := <-writerErr: + require.NoError(t, err) + default: + } +} diff --git a/internal/config/types.go b/internal/config/types.go index 18d5c1916..de4330d7f 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -764,6 +764,26 @@ type RecommendationFilter struct { ID string // "" = all ids; non-empty = exact match on the id column } +// RecommendationScanKey is a stored recommendation's position in the keyset +// scan order of ListStoredRecommendationsChunk (issue #981): the application +// recommendation ID compared byte-wise, then the cache table's UUID primary +// key (RowID), which keeps rows that share an ID in a total order. +type RecommendationScanKey struct { + ID string + RowID string +} + +// RecommendationChunk is one keyset read of the recommendations cache. +// Recs holds the rows that passed the filter, in scan order. Last is the +// position of the last row read (nil when no row was read) and is where the +// next chunk resumes. Done is true when fewer rows than the limit were read, +// so nothing matches after Last. +type RecommendationChunk struct { + Recs []RecommendationRecord + Last *RecommendationScanKey + Done bool +} + // PurchasePlanFilter parameterises ListPurchasePlans. Zero-value means "no // filter" (all plans are returned). Non-empty AccountIDs restricts the result // to plans that reference at least one of the given account IDs via the diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 38b64862f..b30640689 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1612,6 +1612,19 @@ func (m *MockConfigStore) ListStoredRecommendations(ctx context.Context, filter return v, args.Error(1) } +func (m *MockConfigStore) ListStoredRecommendationsChunk(ctx context.Context, filter config.RecommendationFilter, after *config.RecommendationScanKey, limit int) (config.RecommendationChunk, error) { + m.record("ListStoredRecommendationsChunk", ctx, filter, after, limit) + if !isExpected(&m.Mock, "ListStoredRecommendationsChunk") { + return config.RecommendationChunk{Done: true}, nil + } + args := m.Called(ctx, filter, after, limit) + v, ok := args.Get(0).(config.RecommendationChunk) + if !ok { + panic(fmt.Sprintf("mock: expected config.RecommendationChunk, got %T", 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/server/test_helpers_test.go b/internal/server/test_helpers_test.go index f58bee1df..97dbb1ebd 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -340,6 +340,9 @@ func (m *mockConfigStoreForHealth) UpsertRecommendations(_ context.Context, _ ti func (m *mockConfigStoreForHealth) ListStoredRecommendations(_ context.Context, _ config.RecommendationFilter) ([]config.RecommendationRecord, error) { return nil, nil } +func (m *mockConfigStoreForHealth) ListStoredRecommendationsChunk(_ context.Context, _ config.RecommendationFilter, _ *config.RecommendationScanKey, _ int) (config.RecommendationChunk, error) { + return config.RecommendationChunk{Done: true}, nil +} func (m *mockConfigStoreForHealth) GetRecommendationsFreshness(_ context.Context) (*config.RecommendationsFreshness, error) { return &config.RecommendationsFreshness{}, nil } From b128ce3d55165955629625a74ce869cd591f8eed Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 22 Oct 2026 01:41:30 +0100 Subject: [PATCH 6/9] perf(scheduler): scan cached recommendations chunk by chunk ScanRecommendations walks the recommendations cache through ListStoredRecommendationsChunk in byte-wise ID order and applies the same suppressions and account overrides as ListRecommendations to each chunk, handing the survivors to a callback. It holds one chunk at a time instead of the whole filtered set. Suppressions are read once per scan and each (account, provider, service) override is resolved once per scan, not once per chunk. Read failures over-show exactly like ListRecommendations. The freshness read, cold-start kick and stale-while-revalidate kick are pulled out of ListRecommendations into startCacheRead/finishCacheRead so both reads share them unchanged. Tests pin the scan to ListRecommendations' output (sorted by ID) for chunk sizes 1 to 31 with suppressions and overrides active. Refs #981 Co-Authored-By: Claude Opus 5.5 --- internal/scheduler/scheduler.go | 50 ++++-- internal/scheduler/scheduler_scan.go | 144 +++++++++++++++ internal/scheduler/scheduler_scan_test.go | 206 ++++++++++++++++++++++ 3 files changed, 383 insertions(+), 17 deletions(-) create mode 100644 internal/scheduler/scheduler_scan.go create mode 100644 internal/scheduler/scheduler_scan_test.go diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 1d6953042..c1299b002 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -1197,14 +1197,9 @@ func recommendationID(rec config.RecommendationRecord) (string, error) { func (s *Scheduler) ListRecommendations(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) { logging.Info("Reading recommendations from cache...") - freshness, err := s.config.GetRecommendationsFreshness(ctx) + freshness, err := s.startCacheRead(ctx) if err != nil { - return nil, fmt.Errorf("failed to read recommendations freshness: %w", err) - } - - mayCollect := BackgroundCollectionAllowed(ctx) - if mayCollect && freshness.LastCollectedAt == nil { - s.kickColdStartCollection(ctx) + return nil, err } recs, err := s.config.ListStoredRecommendations(ctx, filter) @@ -1229,19 +1224,40 @@ func (s *Scheduler) ListRecommendations(ctx context.Context, filter config.Recom logging.Errorf("failed to apply account overrides; returning un-filtered recs: %v", err) } - // Background refresh is only attempted on non-Lambda runtimes; skip - // the DB config read entirely on Lambda so GetGlobalConfig is not called - // on the hot read path (Lambda has no persistent goroutines). - if mayCollect && !s.isLambda { - ttl, disabled := s.resolveEffectiveCacheTTL(ctx) - if disabled { - return recs, nil - } - s.maybeKickBackgroundRefresh(freshness, ttl) - } + s.finishCacheRead(ctx, freshness) return recs, nil } +// startCacheRead is steps 1 and 2 of ListRecommendations (and of +// ScanRecommendations): read freshness and, for a caller allowed to trigger +// collection, kick the cold-start collection. +func (s *Scheduler) startCacheRead(ctx context.Context) (*config.RecommendationsFreshness, error) { + freshness, err := s.config.GetRecommendationsFreshness(ctx) + if err != nil { + return nil, fmt.Errorf("failed to read recommendations freshness: %w", err) + } + if BackgroundCollectionAllowed(ctx) && freshness.LastCollectedAt == nil { + s.kickColdStartCollection(ctx) + } + return freshness, nil +} + +// finishCacheRead is step 4 of ListRecommendations (and of +// ScanRecommendations): the stale-while-revalidate kick. Background refresh is +// only attempted on non-Lambda runtimes; the DB config read is skipped +// entirely on Lambda so GetGlobalConfig is not called on the hot read path +// (Lambda has no persistent goroutines). +func (s *Scheduler) finishCacheRead(ctx context.Context, freshness *config.RecommendationsFreshness) { + if !BackgroundCollectionAllowed(ctx) || s.isLambda { + return + } + ttl, disabled := s.resolveEffectiveCacheTTL(ctx) + if disabled { + return + } + s.maybeKickBackgroundRefresh(freshness, ttl) +} + // ListRecommendationsStrict is ListRecommendations for the money path: the // same stored rows, suppressions and account overrides, but a suppression or // override read failure is returned instead of logged. Over-showing is fine diff --git a/internal/scheduler/scheduler_scan.go b/internal/scheduler/scheduler_scan.go new file mode 100644 index 000000000..f2632d8d3 --- /dev/null +++ b/internal/scheduler/scheduler_scan.go @@ -0,0 +1,144 @@ +package scheduler + +import ( + "context" + "errors" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// ScanRecommendations is ListRecommendations for a caller that pages through +// the cache without holding all of it (issue #981). It walks the cached +// recommendations in chunks of at most chunkSize rows, in byte-wise ID order +// (the order of Go's string comparison), applies the same suppressions and +// account overrides as ListRecommendations to each chunk, and calls fn with +// each chunk's surviving rows in that order. fn may keep what it needs; the +// scan itself holds one chunk at a time. An error from fn stops the scan and +// is returned. +// +// It has the same cold-start and stale-while-revalidate side effects as +// ListRecommendations, and the same over-show trade-off: a suppression or +// override read failure is logged and the affected rows pass through. +// +// Each chunk is one consistent read; the walk as a whole is not one snapshot. +// A row upserted in place keeps its position, so it is visited once. A row +// inserted behind the scan position during the walk is missed by this walk, +// and a row evicted ahead of it is not visited. +func (s *Scheduler) ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error { + if chunkSize < 1 { + return errors.New("recommendations scan chunk size must be positive") + } + logging.Info("Scanning recommendations from cache...") + + freshness, err := s.startCacheRead(ctx) + if err != nil { + return err + } + + suppressions := s.loadSuppressionIndex(ctx) + overrides := newOverrideScanFilter(s.config) + var after *config.RecommendationScanKey + for { + chunk, err := s.config.ListStoredRecommendationsChunk(ctx, filter, after, chunkSize) + if err != nil { + return err + } + recs := chunk.Recs + if len(suppressions) > 0 { + recs = applySuppressionIndex(recs, suppressions) + } + recs = overrides.apply(ctx, recs) + if len(recs) > 0 { + if err := fn(recs); err != nil { + return err + } + } + if chunk.Done || chunk.Last == nil { + break + } + after = chunk.Last + } + + s.finishCacheRead(ctx, freshness) + return nil +} + +// loadSuppressionIndex reads the active suppressions once for a whole scan. +// A read failure is logged and yields no index, so the scan over-shows like +// ListRecommendations does. +func (s *Scheduler) loadSuppressionIndex(ctx context.Context) map[suppressionKey]*suppressionAgg { + sups, err := s.config.ListActiveSuppressions(ctx) + if err != nil { + logging.Errorf("failed to apply suppressions; returning un-suppressed recs: %v", err) + return nil + } + if len(sups) == 0 { + return nil + } + return aggregateSuppressions(sups) +} + +// overrideScanFilter applies account overrides chunk by chunk, resolving each +// (account, provider, service) triple once per scan instead of once per chunk. +type overrideScanFilter struct { + store config.AccountConfigReader + resolved map[string]*config.ServiceConfig + seen map[string]struct{} +} + +func newOverrideScanFilter(store config.AccountConfigReader) *overrideScanFilter { + return &overrideScanFilter{ + store: store, + resolved: make(map[string]*config.ServiceConfig), + seen: make(map[string]struct{}), + } +} + +// apply resolves the triples recs introduces and filters recs with every +// configuration resolved so far. On a resolution error the chunk passes +// through unfiltered (logged), and its new triples stay unresolved so a later +// chunk retries them. +func (o *overrideScanFilter) apply(ctx context.Context, recs []config.RecommendationRecord) []config.RecommendationRecord { + if len(recs) == 0 { + return recs + } + var unseen []config.RecommendationRecord + var unseenKeys []string + for i := range recs { + key := recAccountConfigKey(&recs[i]) + if _, ok := o.seen[key]; ok { + continue + } + o.seen[key] = struct{}{} + unseen = append(unseen, recs[i]) + unseenKeys = append(unseenKeys, key) + } + if len(unseen) > 0 { + resolved, err := config.ResolveAccountConfigsForRecs(ctx, o.store, unseen) + if err != nil { + for _, key := range unseenKeys { + delete(o.seen, key) + } + logging.Errorf("failed to apply account overrides; returning un-filtered recs: %v", err) + return recs + } + for key, cfg := range resolved { + o.resolved[key] = cfg + } + } + if len(o.resolved) == 0 { + return recs + } + return filterRecsByResolvedConfigs(recs, o.resolved) +} + +// recAccountConfigKey is the AccountConfigKey filterRecsByResolvedConfigs +// looks a rec up by. +func recAccountConfigKey(rec *config.RecommendationRecord) string { + accountID := "" + if rec.CloudAccountID != nil { + accountID = *rec.CloudAccountID + } + return config.AccountConfigKey(accountID, rec.Provider, rec.Service) +} diff --git a/internal/scheduler/scheduler_scan_test.go b/internal/scheduler/scheduler_scan_test.go new file mode 100644 index 000000000..688f3e4a8 --- /dev/null +++ b/internal/scheduler/scheduler_scan_test.go @@ -0,0 +1,206 @@ +package scheduler + +import ( + "context" + "errors" + "fmt" + "sort" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// scanStore serves ListStoredRecommendationsChunk from recs the way the +// Postgres store does: ordered by ID (byte-wise) then by row position, +// resuming strictly after the given key. It also counts reads so tests can +// check that override lookups are not repeated per chunk. +type scanStore struct { + mockOverrideStore + sups []config.PurchaseSuppression + supsErr error + chunkErr error + chunkReads int + overrideLookup map[string]int +} + +func (m *scanStore) ListActiveSuppressions(_ context.Context) ([]config.PurchaseSuppression, error) { + return m.sups, m.supsErr +} + +func (m *scanStore) GetAccountServiceOverride(ctx context.Context, accountID, provider, service string) (*config.AccountServiceOverride, error) { + if m.overrideLookup == nil { + m.overrideLookup = map[string]int{} + } + m.overrideLookup[accountID+"|"+provider+"|"+service]++ + return m.mockOverrideStore.GetAccountServiceOverride(ctx, accountID, provider, service) +} + +func (m *scanStore) ListStoredRecommendationsChunk(_ context.Context, _ config.RecommendationFilter, after *config.RecommendationScanKey, limit int) (config.RecommendationChunk, error) { + m.chunkReads++ + if m.chunkErr != nil { + return config.RecommendationChunk{}, m.chunkErr + } + type row struct { + key config.RecommendationScanKey + rec config.RecommendationRecord + } + rows := make([]row, len(m.recs)) + for i := range m.recs { + rows[i] = row{config.RecommendationScanKey{ID: m.recs[i].ID, RowID: fmt.Sprintf("%08d", i)}, m.recs[i]} + } + less := func(a, b config.RecommendationScanKey) bool { + if a.ID != b.ID { + return a.ID < b.ID + } + return a.RowID < b.RowID + } + sort.Slice(rows, func(i, j int) bool { return less(rows[i].key, rows[j].key) }) + var chunk config.RecommendationChunk + for i := range rows { + if after != nil && !less(*after, rows[i].key) { + continue + } + if len(chunk.Recs) == limit { + return chunk, nil + } + chunk.Recs = append(chunk.Recs, rows[i].rec) + k := rows[i].key + chunk.Last = &k + } + // Like the store: only a short read proves there is nothing after Last. + chunk.Done = len(chunk.Recs) < limit + return chunk, nil +} + +// scanFixture is a cache where suppressions and account overrides each drop +// some rows and change others, spread over several chunks. +func scanFixture() *scanStore { + var recs []config.RecommendationRecord + for i := 9; i >= 0; i-- { // reverse order: the scan must not rely on input order + for _, acct := range []string{"acct-A", "acct-B", "acct-C"} { + rec := rdsRec(acct, "us-east-1", "mysql") + rec.ID = fmt.Sprintf("%s|%02d", acct, i) + rec.ResourceType = fmt.Sprintf("db.t3.r%02d", i) + rec.Count = 2 + recs = append(recs, rec) + } + } + future := time.Now().Add(time.Hour) + return &scanStore{ + mockOverrideStore: mockOverrideStore{ + recs: recs, + globals: map[string]*config.ServiceConfig{ + "aws|rds": {Provider: "aws", Service: "rds", Enabled: true}, + }, + overrides: map[string]*config.AccountServiceOverride{ + "acct-B|aws|rds": {Enabled: boolPtr(false)}, + }, + }, + sups: []config.PurchaseSuppression{ + // Drops acct-C|03 (2 - 2 = 0) and annotates acct-A|05 (2 - 1 = 1). + {ExecutionID: "e1", AccountID: "acct-C", Provider: "aws", Service: "rds", Region: "us-east-1", + ResourceType: "db.t3.r03", Engine: "mysql", SuppressedCount: 2, ExpiresAt: future}, + {ExecutionID: "e2", AccountID: "acct-A", Provider: "aws", Service: "rds", Region: "us-east-1", + ResourceType: "db.t3.r05", Engine: "mysql", SuppressedCount: 1, ExpiresAt: future}, + }, + } +} + +func collectScan(t *testing.T, s *Scheduler, chunkSize int) ([]config.RecommendationRecord, int) { + t.Helper() + var got []config.RecommendationRecord + calls := 0 + err := s.ScanRecommendations(context.Background(), config.RecommendationFilter{}, chunkSize, func(chunk []config.RecommendationRecord) error { + calls++ + require.NotEmpty(t, chunk, "fn must not be called with an empty chunk") + got = append(got, chunk...) + return nil + }) + require.NoError(t, err) + return got, calls +} + +// TestScanRecommendations_MatchesListRecommendations pins the scan to the +// legacy read: for every chunk size it yields exactly the rows +// ListRecommendations returns, with the same suppression counts and override +// drops, in ID order. +func TestScanRecommendations_MatchesListRecommendations(t *testing.T) { + store := scanFixture() + s := &Scheduler{config: store} + + want, err := s.ListRecommendations(context.Background(), config.RecommendationFilter{}) + require.NoError(t, err) + sort.Slice(want, func(i, j int) bool { return want[i].ID < want[j].ID }) + require.Len(t, want, 19, "fixture: 30 rows, acct-B's 10 overridden away, one suppressed away") + + for _, chunkSize := range []int{1, 2, 3, 7, 30, 31} { + t.Run(fmt.Sprintf("chunk=%d", chunkSize), func(t *testing.T) { + store.overrideLookup = nil + got, _ := collectScan(t, s, chunkSize) + assert.Equal(t, want, got) + for key, n := range store.overrideLookup { + assert.Equal(t, 1, n, "override for %s looked up %d times in one scan", key, n) + } + }) + } +} + +// TestScanRecommendations_SkipsEmptyChunks checks that a chunk whose rows are +// all filtered away is not passed to fn, and that the walk still continues +// past it. +func TestScanRecommendations_SkipsEmptyChunks(t *testing.T) { + store := scanFixture() + s := &Scheduler{config: store} + got, calls := collectScan(t, s, 1) + assert.Len(t, got, 19) + assert.Equal(t, 19, calls, "one call per surviving row with chunk size 1") + assert.Equal(t, 31, store.chunkReads, "30 one-row chunks and the empty final read") +} + +// TestScanRecommendations_StopsOnCallbackError propagates fn's error and +// reads no further chunks. +func TestScanRecommendations_StopsOnCallbackError(t *testing.T) { + store := scanFixture() + s := &Scheduler{config: store} + boom := errors.New("boom") + err := s.ScanRecommendations(context.Background(), config.RecommendationFilter{}, 5, func([]config.RecommendationRecord) error { + return boom + }) + require.ErrorIs(t, err, boom) + assert.Equal(t, 1, store.chunkReads) +} + +// TestScanRecommendations_StoreErrorFails returns a chunk read error instead +// of a short walk. +func TestScanRecommendations_StoreErrorFails(t *testing.T) { + store := scanFixture() + store.chunkErr = errors.New("db down") + s := &Scheduler{config: store} + err := s.ScanRecommendations(context.Background(), config.RecommendationFilter{}, 5, func([]config.RecommendationRecord) error { + t.Fatal("fn must not run when the read fails") + return nil + }) + require.ErrorContains(t, err, "db down") +} + +// TestScanRecommendations_ReadFailuresOverShow mirrors ListRecommendations' +// trade-off: a suppression or override read failure is logged and the rows +// pass through rather than failing the page. +func TestScanRecommendations_ReadFailuresOverShow(t *testing.T) { + store := scanFixture() + store.supsErr = errors.New("sups down") + store.getOverrideErr = errors.New("overrides down") + s := &Scheduler{config: store} + got, _ := collectScan(t, s, 4) + assert.Len(t, got, 30, "nothing suppressed or overridden away") +} + +// TestScanRecommendations_RejectsBadChunkSize guards the loop bound. +func TestScanRecommendations_RejectsBadChunkSize(t *testing.T) { + s := &Scheduler{config: scanFixture()} + err := s.ScanRecommendations(context.Background(), config.RecommendationFilter{}, 0, func([]config.RecommendationRecord) error { return nil }) + require.Error(t, err) +} From 402d43d089afebc777b962b6444fe930e4f719dd Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 22 Oct 2026 01:49:57 +0100 Subject: [PATCH 7/9] perf(api): page GET /api/recommendations from the chunked scan A paginated request (limit or cursor sent) no longer loads the whole filtered set through ListRecommendations and sorts it in memory. It runs Scheduler.ScanRecommendations, applies the enabled-providers and account scope filters to each chunk (their inputs read once per request), and feeds the survivors to recommendationsPageCollector. The collector totals every row into the summary and keeps only the rows after the cursor up to limit+1 (the page plus one row proving a next page exists), then cuts the page with the existing byte budget. Peak memory is one scan chunk plus one page instead of the whole tenant. Output is unchanged: rows arrive in byte-wise ID order (Go string order), duplicate IDs still fail the request, and the summary still covers the full filtered set. The unpaginated path is untouched. The in-memory sortRecommendationsAfter/buildBudgetedPage are removed; the existing #723 pagination tests now run against the scan path, with buildBudgetedPage kept as a test helper over the collector. Refs #981 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_recommendations.go | 79 +++++++--- internal/api/handler_recommendations_page.go | 73 +-------- .../api/handler_recommendations_page_test.go | 17 +++ internal/api/handler_recommendations_scan.go | 142 ++++++++++++++++++ internal/api/mocks_test.go | 21 +++ internal/api/types.go | 4 + internal/server/interfaces.go | 3 + internal/testutil/mocks.go | 9 ++ 8 files changed, 257 insertions(+), 91 deletions(-) create mode 100644 internal/api/handler_recommendations_scan.go diff --git a/internal/api/handler_recommendations.go b/internal/api/handler_recommendations.go index 1f13a0b40..2a0fe1570 100644 --- a/internal/api/handler_recommendations.go +++ b/internal/api/handler_recommendations.go @@ -106,6 +106,10 @@ func (h *Handler) getRecommendations(ctx context.Context, req *events.LambdaFunc return nil, fmt.Errorf("failed to get allowed accounts: %w", err) } + if page.enabled { + return h.getRecommendationsPage(ctx, scope, filter, page) + } + recommendations, err := h.scheduler.ListRecommendations(recommendationsReadContext(ctx, scope), filter) if err != nil { return nil, fmt.Errorf("failed to get recommendations: %w", err) @@ -126,7 +130,7 @@ func (h *Handler) getRecommendations(ctx context.Context, req *events.LambdaFunc return nil, err } - return buildRecommendationsPageResponse(recommendations, page) + return buildRecommendationsResponse(recommendations), nil } // filterRecommendationsByAllowedAccounts filters recommendations to only include @@ -146,10 +150,18 @@ func (h *Handler) filterRecommendationsByScope(ctx context.Context, allowedAccou if allowedAccounts.AllowsAll() { return recs, nil } + nameByID, err := h.cloudAccountNamesByID(ctx) + if err != nil { + return nil, err + } + return filterRecommendationsInScope(allowedAccounts, nameByID, recs), nil +} - // Resolve account names so the allowed list can match on either ID or - // display name. Recommendations only carry CloudAccountID, so the name - // lookup needs a one-time fetch of the account list. +// cloudAccountNamesByID resolves account names so an allowed-accounts list +// can match on either ID or display name. Recommendations only carry +// CloudAccountID, so the name lookup needs a one-time fetch of the account +// list. +func (h *Handler) cloudAccountNamesByID(ctx context.Context) (map[string]string, error) { accounts, err := h.config.ListCloudAccounts(ctx, config.CloudAccountFilter{}) if err != nil { return nil, fmt.Errorf("failed to list accounts for filter: %w", err) @@ -159,7 +171,12 @@ func (h *Handler) filterRecommendationsByScope(ctx context.Context, allowedAccou a := accounts[_rvc] nameByID[a.ID] = a.Name } + return nameByID, nil +} +// filterRecommendationsInScope keeps the recs whose account allowedAccounts +// allows, in place and in order. Recs without an account are dropped. +func filterRecommendationsInScope(allowedAccounts auth.AccountScope, nameByID map[string]string, recs []config.RecommendationRecord) []config.RecommendationRecord { filtered := recs[:0] for _rvc := range recs { rec := recs[_rvc] @@ -171,7 +188,7 @@ func (h *Handler) filterRecommendationsByScope(ctx context.Context, allowedAccou filtered = append(filtered, rec) } } - return filtered, nil + return filtered } // filterRecommendationsByEnabledProviders removes recommendations for providers @@ -210,15 +227,9 @@ func buildRecommendationsResponse(recommendations []config.RecommendationRecord) // itself or one page of it (issue #723), so the totals never shrink to the // page. func summarizeRecommendations(all, rows []config.RecommendationRecord) *RecommendationsResponse { - regionSet := make(map[string]struct{}) - var totalSavings, totalUpfront float64 + var summary recommendationsSummary for i := range all { - rec := &all[i] - totalSavings += rec.Savings - totalUpfront += rec.UpfrontCost - if rec.Region != "" { - regionSet[rec.Region] = struct{}{} - } + summary.add(&all[i]) } for i := range rows { // Surface the compute size (VCPU / MemoryGB) at the top level so @@ -232,23 +243,51 @@ func summarizeRecommendations(all, rows []config.RecommendationRecord) *Recommen // renders "—". Only returned rows are decoded. stampComputeCapacity(&rows[i]) } + return summary.response(rows) +} + +// recommendationsSummary accumulates the full-set totals and region list of +// a list response one rec at a time, so a paginated scan can total every +// filtered row without holding them all (issue #981). +type recommendationsSummary struct { + count int + totalSavings float64 + totalUpfront float64 + regionSet map[string]struct{} +} + +func (s *recommendationsSummary) add(rec *config.RecommendationRecord) { + s.count++ + s.totalSavings += rec.Savings + s.totalUpfront += rec.UpfrontCost + if rec.Region == "" { + return + } + if s.regionSet == nil { + s.regionSet = make(map[string]struct{}) + } + s.regionSet[rec.Region] = struct{}{} +} - regions := make([]string, 0, len(regionSet)) - for r := range regionSet { +// response returns rows as the response body with the accumulated summary +// and regions. +func (s *recommendationsSummary) response(rows []config.RecommendationRecord) *RecommendationsResponse { + regions := make([]string, 0, len(s.regionSet)) + for r := range s.regionSet { regions = append(regions, r) } var avgPayback float64 - if totalSavings > 0 { - avgPayback = totalUpfront / totalSavings + if s.totalSavings > 0 { + avgPayback = s.totalUpfront / s.totalSavings } return &RecommendationsResponse{ Recommendations: rows, Summary: RecommendationsSummary{ - TotalCount: len(all), - TotalMonthlySavings: totalSavings, - TotalUpfrontCost: totalUpfront, + TotalCount: s.count, + TotalMonthlySavings: s.totalSavings, + TotalUpfrontCost: s.totalUpfront, AvgPaybackMonths: avgPayback, }, Regions: regions, diff --git a/internal/api/handler_recommendations_page.go b/internal/api/handler_recommendations_page.go index dd40dbb04..68a23a9ba 100644 --- a/internal/api/handler_recommendations_page.go +++ b/internal/api/handler_recommendations_page.go @@ -4,7 +4,6 @@ import ( "encoding/base64" "encoding/json" "fmt" - "sort" "strconv" "strings" @@ -18,9 +17,8 @@ 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. -// This bounds the response only: every request still reads, filters and -// sorts the whole filtered set in memory and then slices the page from it -// (store-side paging is tracked in #981). A +// 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 // (recommendationsPageByteBudget), because the opaque Details blob makes the // size of a row data-dependent. @@ -120,34 +118,6 @@ func decodeRecommendationsCursor(cursor string) (string, error) { return string(raw), nil } -// sortRecommendationsAfter sorts recs by ID in place (it is this request's -// own slice, freshly read from the store) and returns the index of the first -// row strictly after afterID. -// -// The ID is the application recommendation ID that the scheduler builds from -// the natural-key columns (recommendationID in internal/scheduler), not the -// cache table's UUID primary key. Paging needs ID order to be a total order: -// two rows sharing an ID would let the strict "after" comparison skip the -// second one at a page boundary. A duplicate is therefore an error rather -// than a silently short walk. -// -// Resuming after the last ID (rather than at an offset) means a row added or -// removed between two requests does not shift later pages: the next page -// always resumes after the last ID the client saw, even if that row has -// since been deleted. -func sortRecommendationsAfter(recs []config.RecommendationRecord, afterID string) (int, error) { - sort.Slice(recs, func(i, j int) bool { return recs[i].ID < recs[j].ID }) - for i := 1; i < len(recs); i++ { - if recs[i].ID == recs[i-1].ID { - return 0, fmt.Errorf("duplicate recommendation ID %q: cannot page the recommendations list", recs[i].ID) - } - } - if afterID == "" { - return 0, nil - } - return sort.Search(len(recs), func(i int) bool { return recs[i].ID > afterID }), nil -} - // envelopeLen returns how many bytes v adds to the Lambda invoke response: // its JSON encoding, re-encoded as the content of the response's body // string (without the surrounding quotes). @@ -194,42 +164,3 @@ func selectRecommendationsPage(sorted []config.RecommendationRecord, start, limi } return n, nil } - -// buildRecommendationsPageResponse builds the list response for either mode. -// Unpaginated requests get exactly the legacy buildRecommendationsResponse -// output; paginated requests get one page of rows plus next_cursor, with the -// summary and regions still computed over the full filtered set. -func buildRecommendationsPageResponse(recs []config.RecommendationRecord, page recommendationsPage) (*RecommendationsResponse, error) { - if !page.enabled { - return buildRecommendationsResponse(recs), nil - } - return buildBudgetedPage(recs, page, recommendationsPageByteBudget) -} - -// buildBudgetedPage is the paginated branch of -// buildRecommendationsPageResponse with an explicit byte budget. -func buildBudgetedPage(recs []config.RecommendationRecord, page recommendationsPage, budget int) (*RecommendationsResponse, error) { - start, err := sortRecommendationsAfter(recs, page.afterID) - if err != nil { - return nil, err - } - resp := summarizeRecommendations(recs, recs[start:start]) - // Overhead: everything except the rows and the cursor value, measured - // with an empty next_cursor key present. - resp.NextCursor = " " - overhead, err := envelopeLen(resp) - if err != nil { - return nil, fmt.Errorf("failed to size recommendations page: %w", err) - } - n, err := selectRecommendationsPage(recs, start, page.limit, overhead, budget) - if err != nil { - return nil, err - } - end := start + n - resp.Recommendations = recs[start:end] - resp.NextCursor = "" - if end < len(recs) { - resp.NextCursor = encodeRecommendationsCursor(recs[end-1].ID) - } - return resp, nil -} diff --git a/internal/api/handler_recommendations_page_test.go b/internal/api/handler_recommendations_page_test.go index 101986cfc..e5f3ebd01 100644 --- a/internal/api/handler_recommendations_page_test.go +++ b/internal/api/handler_recommendations_page_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "sort" "strconv" "strings" "testing" @@ -43,10 +44,26 @@ func pageTestRecs(n int) []config.RecommendationRecord { func pageTestHandler(t *testing.T, recs []config.RecommendationRecord) *Handler { t.Helper() ms := new(MockScheduler) + // Unpaginated requests read the full list; paginated ones scan it in + // ID order (issue #981). ms.On("ListRecommendations", mock.Anything, mock.Anything).Return(recs, nil) + ms.On("ScanRecommendations", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(recs, nil) return &Handler{scheduler: ms, apiKey: "test-key"} } +// buildBudgetedPage pages recs the way a paginated request does: the rows +// reach the collector in ID order, as the store scan delivers them, and the +// page is cut with the given byte budget. +func buildBudgetedPage(recs []config.RecommendationRecord, page recommendationsPage, budget int) (*RecommendationsResponse, error) { + sorted := append([]config.RecommendationRecord(nil), recs...) + sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID }) + collector := newRecommendationsPageCollector(page) + if err := collector.add(sorted); err != nil { + return nil, err + } + return collector.response(budget) +} + func pageTestReq() *events.LambdaFunctionURLRequest { return &events.LambdaFunctionURLRequest{Headers: map[string]string{"x-api-key": "test-key"}} } diff --git a/internal/api/handler_recommendations_scan.go b/internal/api/handler_recommendations_scan.go new file mode 100644 index 000000000..4ef8aae48 --- /dev/null +++ b/internal/api/handler_recommendations_scan.go @@ -0,0 +1,142 @@ +package api + +import ( + "context" + "fmt" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// 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. +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. +// +// 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. +func (h *Handler) getRecommendationsPage(ctx context.Context, scope auth.AccountScope, filter config.RecommendationFilter, page recommendationsPage) (*RecommendationsResponse, error) { + keep, err := h.recommendationsRowFilter(ctx, scope) + if err != nil { + return nil, err + } + collector := newRecommendationsPageCollector(page) + err = h.scheduler.ScanRecommendations(recommendationsReadContext(ctx, scope), filter, recommendationsScanChunkSize, + func(chunk []config.RecommendationRecord) error { + return collector.add(keep(chunk)) + }) + if err != nil { + return nil, fmt.Errorf("failed to get recommendations: %w", err) + } + return collector.response(recommendationsPageByteBudget) +} + +// recommendationsRowFilter returns the handler-side row filters 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) { + 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) + } + enabledProviders = globalCfg.EnabledProviders + } + var nameByID map[string]string + if !scope.AllowsAll() { + var err error + if nameByID, err = h.cloudAccountNamesByID(ctx); err != nil { + return nil, err + } + } + return func(recs []config.RecommendationRecord) []config.RecommendationRecord { + recs = config.FilterRecommendationsByEnabledProviders(recs, enabledProviders) + if scope.AllowsAll() { + return recs + } + return filterRecommendationsInScope(scope, nameByID, recs) + }, nil +} + +// 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. +type recommendationsPageCollector struct { + page recommendationsPage + summary recommendationsSummary + seenAny bool + lastID string + rows []config.RecommendationRecord +} + +func newRecommendationsPageCollector(page recommendationsPage) *recommendationsPageCollector { + return &recommendationsPageCollector{page: page, rows: []config.RecommendationRecord{}} +} + +// add consumes the next rows of the scan. Paging needs ID order to be a total +// order: two rows sharing an ID would let the strict "after" comparison skip +// the second one at a page boundary, so a duplicate fails the request rather +// than silently walking short. The ID is the application recommendation ID +// built from the natural-key columns (recommendationID in +// internal/scheduler), not the cache table's UUID primary key. A row out of +// order means the scan broke its contract and would make the cursor skip +// rows, so it fails too. +func (c *recommendationsPageCollector) add(recs []config.RecommendationRecord) error { + for i := range recs { + rec := &recs[i] + if c.seenAny && rec.ID <= c.lastID { + if rec.ID == c.lastID { + return fmt.Errorf("duplicate recommendation ID %q: cannot page the recommendations list", rec.ID) + } + return fmt.Errorf("recommendations scan out of ID order: %q after %q", rec.ID, c.lastID) + } + c.seenAny, c.lastID = true, rec.ID + c.summary.add(rec) + // Resuming after the last ID the client saw (rather than at an + // offset) means a row added or removed between two requests does not + // shift later pages, even if the cursor row itself is gone. + if len(c.rows) <= c.page.limit && (c.page.afterID == "" || rec.ID > c.page.afterID) { + c.rows = append(c.rows, *rec) + } + } + 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. +func (c *recommendationsPageCollector) response(budget int) (*RecommendationsResponse, error) { + resp := c.summary.response(c.rows[:0]) + // Overhead: everything except the rows and the cursor value, measured + // with an empty next_cursor key present. + resp.NextCursor = " " + overhead, err := envelopeLen(resp) + if err != nil { + return nil, fmt.Errorf("failed to size recommendations page: %w", err) + } + n, err := selectRecommendationsPage(c.rows, 0, c.page.limit, overhead, budget) + if err != nil { + return nil, err + } + resp.Recommendations = c.rows[:n] + resp.NextCursor = "" + if n < len(c.rows) { + resp.NextCursor = encodeRecommendationsCursor(c.rows[n-1].ID) + } + return resp, nil +} diff --git a/internal/api/mocks_test.go b/internal/api/mocks_test.go index 2851eb04c..2f9c05d57 100644 --- a/internal/api/mocks_test.go +++ b/internal/api/mocks_test.go @@ -3,6 +3,7 @@ package api import ( "context" "fmt" + "sort" "sync" "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" @@ -138,6 +139,26 @@ func (m *MockScheduler) ListRecommendations(ctx context.Context, filter config.R return args.Get(0).([]config.RecommendationRecord), args.Error(1) } +// ScanRecommendations replays the []config.RecommendationRecord registered +// for it the way the store-backed scan does: sorted by ID (a stable sort, so +// duplicate IDs keep their relative order), chunkSize rows per call to fn, +// stopping at fn's first error. The second return value is the scan's own +// error. +func (m *MockScheduler) ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error { + args := m.Called(ctx, filter, chunkSize, fn) + if recs, ok := args.Get(0).([]config.RecommendationRecord); ok { + sorted := append([]config.RecommendationRecord(nil), recs...) + sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID }) + for start := 0; start < len(sorted); start += chunkSize { + end := min(start+chunkSize, len(sorted)) + if err := fn(sorted[start:end]); err != nil { + return err + } + } + } + return args.Error(1) +} + 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 3c9a860cd..ff69ba36a 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -170,6 +170,10 @@ type PurchaseManagerInterface interface { type SchedulerInterface interface { CollectRecommendations(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) ListRecommendations(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) + // ScanRecommendations walks the same rows as ListRecommendations in + // 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 // 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/server/interfaces.go b/internal/server/interfaces.go index ebdbb78f6..2a072b0f4 100644 --- a/internal/server/interfaces.go +++ b/internal/server/interfaces.go @@ -13,6 +13,9 @@ import ( type SchedulerInterface interface { CollectRecommendations(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) ListRecommendations(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) + // 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 // 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 bf6559c14..646a8155b 100644 --- a/internal/testutil/mocks.go +++ b/internal/testutil/mocks.go @@ -14,6 +14,7 @@ type MockScheduler struct { CollectRecommendationsFunc func(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) 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 } func (m *MockScheduler) CollectRecommendations(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) { @@ -30,6 +31,14 @@ func (m *MockScheduler) ListRecommendations(ctx context.Context, filter config.R return []config.RecommendationRecord{}, nil } +// ScanRecommendations defaults to an empty cache: fn is never called. +func (m *MockScheduler) ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error { + if m.ScanRecommendationsFunc != nil { + return m.ScanRecommendationsFunc(ctx, filter, chunkSize, fn) + } + return nil +} + func (m *MockScheduler) GetRecommendationByID(ctx context.Context, id string) (*config.RecommendationRecord, []string, error) { if m.GetRecommendationByIDFunc != nil { return m.GetRecommendationByIDFunc(ctx, id) From cf7baa1191037adeb21b3a56f3ef8c874caa1473 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 22 Oct 2026 01:49:57 +0100 Subject: [PATCH 8/9] test(api): pin paginated recommendations order, ties and last page - PaginatedRequestDoesNotLoadFullList: the #981 regression; a paged request must use ScanRecommendations, never ListRecommendations. - PagedWalkMatchesLegacyOrder: for an unrestricted and a scoped reader, every chunk size and page size (including the exact-fit last page), the walk returns the legacy rows sorted by Go string order in consecutive pages, next_cursor on all but the last page, and the legacy summary on every page. IDs mix case, "|", "-", "/", spaces and accents. - PagedDuplicateIDs: ties fail every paged request at any chunk/page boundary; a twin removed by the scope filter is not a tie. - Background-collection scope (#627), the one-page memory bound, the out-of-order guard and the empty cache. - An integration test walks a real Postgres cache spanning several scan chunks with the min_savings_pct floor and matches the legacy output. Refs #981 Co-Authored-By: Claude Opus 5.5 --- ...r_recommendations_scan_integration_test.go | 90 +++++++ .../api/handler_recommendations_scan_test.go | 254 ++++++++++++++++++ 2 files changed, 344 insertions(+) create mode 100644 internal/api/handler_recommendations_scan_integration_test.go create mode 100644 internal/api/handler_recommendations_scan_test.go diff --git a/internal/api/handler_recommendations_scan_integration_test.go b/internal/api/handler_recommendations_scan_integration_test.go new file mode 100644 index 000000000..a66c2a2a7 --- /dev/null +++ b/internal/api/handler_recommendations_scan_integration_test.go @@ -0,0 +1,90 @@ +//go:build integration + +package api + +import ( + "context" + "fmt" + "sort" + "strconv" + "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/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" +) + +// TestGetRecommendations_PagedScanMatchesLegacyOnPostgres runs the paginated +// GET /api/recommendations end to end on a real migrated database (issue +// #981): handler, scheduler scan and the keyset chunk query. The cache holds +// more rows than one scan chunk, with IDs whose byte order differs from a +// linguistic order, and the request carries the in-process +// min_savings_pct floor. Walking the cursor must return exactly the legacy +// unpaginated rows in Go string order, with the legacy summary on every page. +func TestGetRecommendations_PagedScanMatchesLegacyOnPostgres(t *testing.T) { + ctx := t.Context() + pg, err := testhelpers.SetupPostgresContainer(ctx, t) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, pg.Cleanup(context.Background())) }) + require.NoError(t, migrations.RunMigrations(ctx, pg.DB.Pool(), "../database/postgres/migrations", "", "")) + + store := config.NewPostgresStore(pg.DB) + prefixes := []string{"aws", "AWS", "aws-x", "aws x", "Aws", "élan", "elan", "_z", "aws/x"} + const total = 2*recommendationsScanChunkSize + 37 + recs := make([]config.RecommendationRecord, total) + for i := range recs { + pct := float64(10 + i%50) // rows with pct < 20 fall under the floor below + recs[i] = config.RecommendationRecord{ + ID: fmt.Sprintf("%s|%04d", prefixes[i%len(prefixes)], total-i), + Provider: "aws", + Service: "ec2", + Region: "us-east-" + strconv.Itoa(i%3+1), + ResourceType: fmt.Sprintf("m5.r%04d", i), + Count: 1, + Term: 1, + Payment: "no-upfront", + Savings: float64(i % 7), + UpfrontCost: float64(i % 5), + SavingsPercentage: &pct, + } + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + + 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) + want := make([]string, len(legacy.Recommendations)) + for i := range legacy.Recommendations { + want[i] = legacy.Recommendations[i].ID + } + sort.Strings(want) + require.Greater(t, len(want), recommendationsScanChunkSize, "the walk must span several scan chunks") + require.Less(t, len(want), total, "the pct floor must drop some rows") + + const limit = 97 + var got []string + params := map[string]string{"min_savings_pct": "20", "limit": strconv.Itoa(limit)} + for page := 0; ; page++ { + require.LessOrEqual(t, page, len(want)/limit+1, "cursor walk did not terminate") + resp, err := h.getRecommendations(ctx, pageTestReq(), params) + require.NoError(t, err) + assert.Equal(t, legacy.Summary, resp.Summary, "page %d summary", page) + assert.ElementsMatch(t, legacy.Regions, resp.Regions) + for i := range resp.Recommendations { + got = append(got, resp.Recommendations[i].ID) + } + if resp.NextCursor == "" { + break + } + require.Len(t, resp.Recommendations, limit, "only the last page may be short") + params = map[string]string{"min_savings_pct": "20", "limit": strconv.Itoa(limit), "cursor": resp.NextCursor} + } + assert.Equal(t, want, got) +} diff --git a/internal/api/handler_recommendations_scan_test.go b/internal/api/handler_recommendations_scan_test.go new file mode 100644 index 000000000..7a43669e1 --- /dev/null +++ b/internal/api/handler_recommendations_scan_test.go @@ -0,0 +1,254 @@ +package api + +import ( + "context" + "sort" + "strconv" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// chunkScanScheduler serves ScanRecommendations from recs like the store +// scan does (ID order, chunk rows per call) but with a chunk size the test +// picks, so page and chunk boundaries can be made to cross. Every other +// scheduler method goes to the embedded mock, which has no expectations: a +// paginated request that falls back to ListRecommendations panics. +type chunkScanScheduler struct { + *MockScheduler + recs []config.RecommendationRecord + chunk int + ctxs []context.Context +} + +func (s *chunkScanScheduler) ScanRecommendations(ctx context.Context, _ config.RecommendationFilter, _ int, fn func([]config.RecommendationRecord) error) error { + s.ctxs = append(s.ctxs, ctx) + sorted := append([]config.RecommendationRecord(nil), s.recs...) + sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID }) + for start := 0; start < len(sorted); start += s.chunk { + end := min(start+s.chunk, len(sorted)) + // A fresh slice per chunk, as the store returns. + if err := fn(append([]config.RecommendationRecord(nil), sorted[start:end]...)); err != nil { + return err + } + } + return nil +} + +// scanFilterRecs mixes IDs that a linguistic collation orders differently +// from bytes with rows the handler-side filters drop: an Azure row (only AWS +// is enabled in recsReadStore), a row in an account outside the scoped +// user's scope and a row with no account. +func scanFilterRecs() []config.RecommendationRecord { + in, out := scopedInAccount, "22222222-2222-4222-8222-222222222222" + mk := func(id, provider string, account *string, region string, savings float64) config.RecommendationRecord { + return config.RecommendationRecord{ + ID: id, Provider: provider, Service: "ec2", Region: region, ResourceType: "m5.large", + CloudAccountID: account, Savings: savings, UpfrontCost: 2 * savings, + } + } + return []config.RecommendationRecord{ + mk("aws|b", "aws", &in, "us-east-1", 1), + mk("AWS|a", "aws", &in, "us-east-2", 2), + mk("aws-a", "aws", &in, "eu-west-1", 3), + mk("aws a", "aws", &in, "us-east-1", 4), + mk("azure|x", "azure", &in, "westeurope", 5), + mk("aws/a", "aws", &out, "ap-south-1", 6), + mk("aws|a", "aws", &in, "us-east-1", 7), + mk("Zeta", "aws", &in, "us-west-2", 8), + mk("ambient", "aws", nil, "us-east-1", 9), + mk("élan", "aws", &in, "us-east-1", 10), + mk("elan", "aws", &in, "sa-east-1", 11), + } +} + +func scanTestHandler(recs []config.RecommendationRecord, chunk int, scoped bool) (*Handler, *chunkScanScheduler) { + sched := &chunkScanScheduler{MockScheduler: new(MockScheduler), recs: recs, chunk: chunk} + accounts := []string(nil) + if scoped { + accounts = []string{scopedInAccount} + } + return &Handler{config: recsReadStore(), scheduler: sched, auth: scopedAuth(accounts)}, sched +} + +// legacyResponse is the unpaginated response for recs, read through +// ListRecommendations on a separate handler with its own copy (the legacy +// scope filter compacts its input in place). +func legacyResponse(t *testing.T, recs []config.RecommendationRecord, scoped bool) *RecommendationsResponse { + t.Helper() + ms := new(MockScheduler) + ms.On("ListRecommendations", mock.Anything, mock.Anything). + Return(append([]config.RecommendationRecord(nil), recs...), nil) + accounts := []string(nil) + if scoped { + accounts = []string{scopedInAccount} + } + h := &Handler{config: recsReadStore(), scheduler: ms, auth: scopedAuth(accounts)} + resp, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{}) + require.NoError(t, err) + return resp +} + +// TestGetRecommendations_PaginatedRequestDoesNotLoadFullList is the #981 +// regression: a paginated request must page from the chunked scan and never +// read the whole filtered set through ListRecommendations. +func TestGetRecommendations_PaginatedRequestDoesNotLoadFullList(t *testing.T) { + recs := pageTestRecs(5) + ms := new(MockScheduler) + ms.On("ScanRecommendations", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(recs, nil) + h := &Handler{scheduler: ms, apiKey: "test-key"} + + resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{"limit": "2"}) + require.NoError(t, err) + require.Len(t, resp.Recommendations, 2) + assert.Equal(t, 5, resp.Summary.TotalCount) + ms.AssertNotCalled(t, "ListRecommendations", mock.Anything, mock.Anything) + ms.AssertCalled(t, "ScanRecommendations", mock.Anything, mock.Anything, recommendationsScanChunkSize, mock.Anything) +} + +// TestGetRecommendations_PagedWalkMatchesLegacyOrder pins the paginated +// output to what the in-memory paging returned: the legacy (unpaginated) rows +// after every filter, sorted by Go string order, cut into consecutive pages +// of limit rows, with next_cursor on every page but the last and the +// full-set summary on every page. It runs for an unrestricted and a scoped +// reader, across chunk sizes that put chunk boundaries before, on and after +// page boundaries, including the exact-fit last page. +func TestGetRecommendations_PagedWalkMatchesLegacyOrder(t *testing.T) { + recs := scanFilterRecs() + for _, scoped := range []bool{false, true} { + legacy := legacyResponse(t, recs, scoped) + want := make([]string, len(legacy.Recommendations)) + for i := range legacy.Recommendations { + want[i] = legacy.Recommendations[i].ID + } + sort.Strings(want) + require.NotEmpty(t, want) + if scoped { + require.Len(t, want, 8, "fixture: Azure, out-of-scope and account-less rows dropped") + } + n := len(want) + for _, chunk := range []int{1, 2, 3, n, n + 1} { + for _, limit := range []int{1, 2, 3, n - 1, n, n + 1} { + t.Run("scoped="+strconv.FormatBool(scoped)+"/chunk="+strconv.Itoa(chunk)+"/limit="+strconv.Itoa(limit), func(t *testing.T) { + h, _ := scanTestHandler(recs, chunk, scoped) + var got []string + params := map[string]string{"limit": strconv.Itoa(limit)} + for page := 0; ; page++ { + require.LessOrEqual(t, page, n, "cursor walk did not terminate") + resp, err := h.getRecommendations(context.Background(), scopedRequest(""), params) + require.NoError(t, err) + + start := page * limit + end := min(start+limit, n) + ids := make([]string, len(resp.Recommendations)) + for i := range resp.Recommendations { + ids[i] = resp.Recommendations[i].ID + } + require.Equal(t, want[start:end], ids, "page %d", page) + assert.Equal(t, legacy.Summary, resp.Summary, "summary must cover the full filtered set") + assert.ElementsMatch(t, legacy.Regions, resp.Regions) + got = append(got, ids...) + if end == n { + assert.Empty(t, resp.NextCursor, "the last page carries no cursor") + break + } + require.Equal(t, encodeRecommendationsCursor(want[end-1]), resp.NextCursor) + params = map[string]string{"limit": strconv.Itoa(limit), "cursor": resp.NextCursor} + } + assert.Equal(t, want, got) + }) + } + } + } +} + +// TestGetRecommendations_PagedDuplicateIDs pins the tie handling to the +// in-memory paging: two surviving rows sharing an ID fail every paginated +// request, wherever the chunk and page boundaries fall and even when the +// duplicate lies beyond the requested page, while a duplicate whose twin a +// filter removes is not a tie at all. +func TestGetRecommendations_PagedDuplicateIDs(t *testing.T) { + recs := scanFilterRecs() + dup := recs[0] // "aws|b", in scope + dup.Region = "us-west-1" + recs = append(recs, dup) + + for _, chunk := range []int{1, 2, 3, len(recs)} { + for _, limit := range []string{"1", "2", "100"} { + h, _ := scanTestHandler(recs, chunk, false) + _, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{"limit": limit}) + require.Error(t, err, "chunk=%d limit=%s", chunk, limit) + assert.Contains(t, err.Error(), `duplicate recommendation ID "aws|b"`) + } + } + + // The twin is in an account outside the scoped reader's scope, so the + // reader sees one "aws|b" and pages normally. + outside := "22222222-2222-4222-8222-222222222222" + recs[len(recs)-1].CloudAccountID = &outside + h, _ := scanTestHandler(recs, 1, true) + resp, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{"limit": "100"}) + require.NoError(t, err) + assert.Len(t, resp.Recommendations, 8) +} + +// TestGetRecommendations_PagedReadBackgroundCollectionScope keeps issue #627 +// on the paginated path: only an unrestricted reader's scan may kick a +// background collection. +func TestGetRecommendations_PagedReadBackgroundCollectionScope(t *testing.T) { + for _, scoped := range []bool{false, true} { + h, sched := scanTestHandler(scanFilterRecs(), 4, scoped) + _, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{"limit": "3"}) + require.NoError(t, err) + require.Len(t, sched.ctxs, 1, "one scan per request") + assert.Equal(t, !scoped, scheduler.BackgroundCollectionAllowed(sched.ctxs[0]), "scoped=%v", scoped) + } +} + +// TestRecommendationsPageCollector_HoldsOnePage checks the memory bound: after +// a long scan the collector holds the page plus one look-ahead row, while +// the summary counts every row. +func TestRecommendationsPageCollector_HoldsOnePage(t *testing.T) { + recs := pageTestRecs(2000) + sort.Slice(recs, func(i, j int) bool { return recs[i].ID < recs[j].ID }) + c := newRecommendationsPageCollector(recommendationsPage{enabled: true, limit: 10, afterID: recs[99].ID}) + for start := 0; start < len(recs); start += 37 { + require.NoError(t, c.add(recs[start:min(start+37, len(recs))])) + } + assert.Len(t, c.rows, 11, "page plus one look-ahead row") + assert.Equal(t, recs[100].ID, c.rows[0].ID, "the page starts strictly after the cursor") + + resp, err := c.response(recommendationsPageByteBudget) + require.NoError(t, err) + assert.Len(t, resp.Recommendations, 10) + assert.Equal(t, 2000, resp.Summary.TotalCount) + assert.Equal(t, encodeRecommendationsCursor(recs[109].ID), resp.NextCursor) +} + +// TestRecommendationsPageCollector_RejectsOutOfOrderScan fails loudly when the +// scan breaks its ordering contract, which would make the cursor skip rows. +func TestRecommendationsPageCollector_RejectsOutOfOrderScan(t *testing.T) { + c := newRecommendationsPageCollector(recommendationsPage{enabled: true, limit: 10}) + require.NoError(t, c.add([]config.RecommendationRecord{{ID: "b"}})) + err := c.add([]config.RecommendationRecord{{ID: "a"}}) + require.ErrorContains(t, err, "out of ID order") +} + +// TestGetRecommendations_PagedEmptyCache returns an empty page as a JSON +// array, with zero totals and no cursor. +func TestGetRecommendations_PagedEmptyCache(t *testing.T) { + h, _ := scanTestHandler(nil, 3, false) + resp, err := h.getRecommendations(context.Background(), &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer scoped-token"}, + }, map[string]string{"limit": "5"}) + require.NoError(t, err) + require.NotNil(t, resp.Recommendations) + assert.Empty(t, resp.Recommendations) + assert.Empty(t, resp.NextCursor) + assert.Equal(t, 0, resp.Summary.TotalCount) +} From 9273549be5a8f5249e0a33df99ed8cc4433d2930 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 23 Oct 2026 05:30:16 +0000 Subject: [PATCH 9/9] fix(scheduler): scan with main's variant-aware suppression index Merging main brought #1110 (issue #816), which changed aggregateSuppressions to return the raw rows grouped by 6-tuple (map[suppressionKey][]config.PurchaseSuppression) and moved the per-rec aggregate into applySuppressionIndex, because which rows apply now depends on the rec's variant (AZ, platform, tenancy). ScanRecommendations still held the old map[suppressionKey]*suppressionAgg, so the merged tree did not build. loadSuppressionIndex now returns main's grouped index and the scan passes it to the same applySuppressionIndex as ListRecommendations. Matching is per rec, so applying it chunk by chunk suppresses exactly what the non-scan path does. A new test pins the scan to ListRecommendations at every chunk size on an EC2 fixture with a variant purchase, a fully covered variant, a legacy wildcard row and a rec with unreadable Details, and checks the expected per-variant counts. Stripping variants from the scan's index fails it. Co-Authored-By: Claude Opus 5.5 --- internal/scheduler/scheduler_scan.go | 7 +- internal/scheduler/scheduler_scan_test.go | 92 +++++++++++++++++++++++ 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/internal/scheduler/scheduler_scan.go b/internal/scheduler/scheduler_scan.go index f2632d8d3..cef7057de 100644 --- a/internal/scheduler/scheduler_scan.go +++ b/internal/scheduler/scheduler_scan.go @@ -64,10 +64,13 @@ func (s *Scheduler) ScanRecommendations(ctx context.Context, filter config.Recom return nil } -// loadSuppressionIndex reads the active suppressions once for a whole scan. +// loadSuppressionIndex reads the active suppressions once for a whole scan and +// groups them by 6-tuple, as applySuppressions does. Which rows of a group +// apply to a rec depends only on that rec's variant (issue #816), so applying +// the index chunk by chunk suppresses exactly what ListRecommendations does. // A read failure is logged and yields no index, so the scan over-shows like // ListRecommendations does. -func (s *Scheduler) loadSuppressionIndex(ctx context.Context) map[suppressionKey]*suppressionAgg { +func (s *Scheduler) loadSuppressionIndex(ctx context.Context) map[suppressionKey][]config.PurchaseSuppression { sups, err := s.config.ListActiveSuppressions(ctx) if err != nil { logging.Errorf("failed to apply suppressions; returning un-suppressed recs: %v", err) diff --git a/internal/scheduler/scheduler_scan_test.go b/internal/scheduler/scheduler_scan_test.go index 688f3e4a8..304ba86b9 100644 --- a/internal/scheduler/scheduler_scan_test.go +++ b/internal/scheduler/scheduler_scan_test.go @@ -148,6 +148,98 @@ func TestScanRecommendations_MatchesListRecommendations(t *testing.T) { } } +// variantScanFixture is an EC2 cache exercising the variant-aware suppression +// rules of issue #816: a purchase of one variant (AZ, platform, tenancy) must +// reduce only that variant, a purchase covering a variant fully hides it, a +// legacy row with no variant is a wildcard over its 6-tuple, and a rec whose +// Details cannot be read matches every row of its 6-tuple. +func variantScanFixture(t *testing.T) *scanStore { + t.Helper() + bought := ec2VariantRec("ec2|bought", "us-east-1a", "Linux/UNIX", "default") + covered := ec2VariantRec("ec2|covered", "us-east-1c", "Linux/UNIX", "default") + unreadable := ec2VariantRec("ec2|unreadable", "us-east-1b", "Windows", "default") + unreadable.Details = []byte(`{"platform":42}`) + unreadable.Count = 10 + legacyA := ec2VariantRec("ec2|legacy-a", "us-east-1a", "Linux/UNIX", "default") + legacyA.ResourceType = "m5.xlarge" + legacyB := ec2VariantRec("ec2|legacy-b", "us-east-1b", "Windows", "dedicated") + legacyB.ResourceType = "m5.xlarge" + recs := []config.RecommendationRecord{ + legacyB, + ec2VariantRec("ec2|sib-windows", "us-east-1a", "Windows", "default"), + unreadable, + ec2VariantRec("ec2|sib-other-az", "us-east-1b", "Linux/UNIX", "default"), + covered, + ec2VariantRec("ec2|sib-dedicated", "us-east-1a", "Linux/UNIX", "dedicated"), + bought, + ec2VariantRec("ec2|sib-regional", "", "Linux/UNIX", "default"), + legacyA, + } + + boughtPurchase := bought + boughtPurchase.Count = 3 + coveredPurchase := covered // the full count of 5 + sups := config.BuildSuppressions([]config.RecommendationRecord{boughtPurchase, coveredPurchase}, "e-variant", + &config.GlobalConfig{GracePeriodDays: map[string]int{"aws": 7}}, time.Now()) + require.Len(t, sups, 2) + for i := range sups { + require.NotNil(t, sups[i].Variant, "BuildSuppressions must write variant rows") + } + sups = append(sups, config.PurchaseSuppression{ExecutionID: "e-legacy", AccountID: "acct-1", Provider: "aws", + Service: "ec2", Region: "us-east-1", ResourceType: "m5.xlarge", SuppressedCount: 1, + ExpiresAt: time.Now().Add(time.Hour)}) + + return &scanStore{ + mockOverrideStore: mockOverrideStore{ + recs: recs, + globals: map[string]*config.ServiceConfig{ + "aws|ec2": {Provider: "aws", Service: "ec2", Enabled: true}, + }, + }, + sups: sups, + } +} + +// TestScanRecommendations_VariantSuppressionsMatchListRecommendations pins +// the scan to ListRecommendations under issue #816's variant-aware +// suppressions, at every chunk size, and checks the expected per-variant +// result so the equality is not vacuous. +func TestScanRecommendations_VariantSuppressionsMatchListRecommendations(t *testing.T) { + store := variantScanFixture(t) + s := &Scheduler{config: store} + + want, err := s.ListRecommendations(context.Background(), config.RecommendationFilter{}) + require.NoError(t, err) + sort.Slice(want, func(i, j int) bool { return want[i].ID < want[j].ID }) + + for chunkSize := 1; chunkSize <= len(store.recs)+1; chunkSize++ { + t.Run(fmt.Sprintf("chunk=%d", chunkSize), func(t *testing.T) { + got, _ := collectScan(t, s, chunkSize) + assert.Equal(t, want, got) + + byID := map[string]config.RecommendationRecord{} + for _, r := range got { + byID[r.ID] = r + } + require.Len(t, byID, 8, "only the fully covered variant is hidden") + assert.NotContains(t, byID, "ec2|covered") + assert.Equal(t, 2, byID["ec2|bought"].Count, "5 - 3 on the purchased variant") + assert.Equal(t, 3, byID["ec2|bought"].SuppressedCount) + for _, id := range []string{"ec2|sib-windows", "ec2|sib-other-az", "ec2|sib-dedicated", "ec2|sib-regional"} { + assert.Equal(t, 5, byID[id].Count, "%s: a sibling variant keeps its full count", id) + assert.Zero(t, byID[id].SuppressedCount, id) + assert.Nil(t, byID[id].PrimarySuppressionExecutionID, id) + } + assert.Equal(t, 2, byID["ec2|unreadable"].Count, "10 - 3 - 5: unreadable Details match every row") + for _, id := range []string{"ec2|legacy-a", "ec2|legacy-b"} { + assert.Equal(t, 4, byID[id].Count, "%s: the legacy row is a wildcard over its 6-tuple", id) + require.NotNil(t, byID[id].PrimarySuppressionExecutionID, id) + assert.Equal(t, "e-legacy", *byID[id].PrimarySuppressionExecutionID, id) + } + }) + } +} + // TestScanRecommendations_SkipsEmptyChunks checks that a chunk whose rows are // all filtered away is not passed to fn, and that the walk still continues // past it.