From fb26345e2de289946c438860ff23a2bdabfc28c2 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 23 Oct 2026 06:15:58 +0000 Subject: [PATCH] perf(ladder): read the whole timeline scope from one snapshot The ladder timeline endpoints returned 100-row pages, and every page re-aggregated the whole account scope, so a client load of n rows cost about n/100 statements each scanning n rows. The client also compares total_count and the totals across pages, so an amendment, tranche or run committed between two page requests failed the whole timeline load ("The ladder timeline changed while loading"). Both GET /api/ladder/tranches and /api/ladder/runs now return every scoped row, with the counts and totals, from one statement, so a load is one snapshot and one scan. No next_cursor is issued, so the web client's loop finishes after its first request; a cursor from the earlier API is still honored as a lower bound for clients mid-load across a deploy. Closes #668 Co-Authored-By: Claude Opus 5.5 --- internal/config/ladder_timeline.go | 12 +- .../config/store_postgres_ladder_timeline.go | 46 ++- ...stgres_ladder_timeline_integration_test.go | 25 +- ...dder_timeline_snapshot_integration_test.go | 309 ++++++++++++++++++ .../store_postgres_ladder_timeline_test.go | 52 +-- 5 files changed, 379 insertions(+), 65 deletions(-) create mode 100644 internal/config/store_postgres_ladder_timeline_snapshot_integration_test.go diff --git a/internal/config/ladder_timeline.go b/internal/config/ladder_timeline.go index 55c04a2c5..8ec990ac3 100644 --- a/internal/config/ladder_timeline.go +++ b/internal/config/ladder_timeline.go @@ -7,8 +7,10 @@ import ( "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" ) -const LadderTimelinePageSize = 100 - +// LadderTimelineCursor is an optional (created_at, id) lower bound on the rows +// a timeline read returns. Reads never issue one: each returns the whole scope +// from one snapshot (#668), and the bound is only honored for a client that +// still follows a next_cursor from the earlier paginated API. type LadderTimelineCursor struct { CreatedAt time.Time `json:"created_at"` ID string `json:"id"` @@ -35,7 +37,6 @@ type LadderTimelineEvent struct { type LadderTimelinePage struct { Events []LadderTimelineEvent `json:"events"` - NextCursor *LadderTimelineCursor `json:"next_cursor,omitempty"` TotalUSDHr string `json:"total_usd_hr"` TotalODUSDHr string `json:"total_od_usd_hr"` TotalCount int64 `json:"total_count"` @@ -63,9 +64,8 @@ type LadderTimelineAction struct { } type LadderTimelineRunPage struct { - Runs []LadderTimelineRun `json:"runs"` - NextCursor *LadderTimelineCursor `json:"next_cursor,omitempty"` - TotalCount int64 `json:"total_count"` + Runs []LadderTimelineRun `json:"runs"` + TotalCount int64 `json:"total_count"` } type LadderTimelineStore interface { diff --git a/internal/config/store_postgres_ladder_timeline.go b/internal/config/store_postgres_ladder_timeline.go index a095d39c5..792ba0b40 100644 --- a/internal/config/store_postgres_ladder_timeline.go +++ b/internal/config/store_postgres_ladder_timeline.go @@ -6,8 +6,18 @@ import ( "fmt" ) +// The timeline reads return the whole account scope from ONE statement, so +// the counts, totals and rows all come from a single snapshot: a concurrent +// amendment, run or tranche insert lands either wholly before or wholly after +// the read, never between two pages of it (#668). Every page of the earlier +// paginated read re-aggregated the full scope, so loading n rows cost about +// n/100 statements each scanning n rows; this is one statement scanning n. +// +// $3/$4 is an optional (created_at, id) lower bound kept for a client that +// still follows a next_cursor from before this change: it narrows the rows +// returned, never the totals, and no next cursor is ever issued. const ladderTimelineQuery = ` -WITH scoped AS MATERIALIZED ( +WITH scoped AS ( SELECT t.id, t.run_id, t.config_id, t.layer_type, t.term, t.payment_option, t.status, r.status AS run_status, t.execution_id, t.revision, t.amount_usd_hr::text AS amount_usd_hr, t.amount_basis, t.commitment_rate, @@ -17,18 +27,15 @@ WITH scoped AS MATERIALIZED ( JOIN ladder_runs r ON r.id = t.run_id AND r.config_id = c.id JOIN cloud_accounts a ON a.id = c.cloud_account_id AND a.provider = c.provider WHERE c.cloud_account_id = $1 AND c.provider = $2 -), page AS ( - SELECT * FROM scoped - WHERE $3::timestamptz IS NULL OR (created_at, id) > ($3, $4::uuid) - ORDER BY created_at, id LIMIT $5 ) SELECT COUNT(*), COALESCE(SUM(amount_usd_hr::numeric), 0::numeric(20,6))::text, - COALESCE((SELECT jsonb_agg(to_jsonb(page) ORDER BY created_at, id) FROM page), '[]'::jsonb), + COALESCE(jsonb_agg(to_jsonb(s) ORDER BY created_at, id) + FILTER (WHERE $3::timestamptz IS NULL OR (created_at, id) > ($3, $4::uuid)), '[]'::jsonb), COALESCE(jsonb_agg(jsonb_build_array(amount_usd_hr, amount_basis, commitment_rate)), '[]'::jsonb) -FROM scoped` +FROM scoped s` const ladderTimelineRunsQuery = ` -WITH scoped AS MATERIALIZED ( +WITH scoped AS ( SELECT r.id, r.config_id, r.started_at, r.created_at, r.status, r.baseline_usd_hr::text AS baseline_usd_hr, r.target_usd_hr::text AS target_usd_hr, r.existing_usd_hr::text AS existing_usd_hr, r.gap_usd_hr::text AS gap_usd_hr, @@ -42,20 +49,17 @@ WITH scoped AS MATERIALIZED ( JOIN ladder_configs c ON c.id = r.config_id JOIN cloud_accounts a ON a.id = c.cloud_account_id AND a.provider = c.provider WHERE c.cloud_account_id = $1 AND c.provider = $2 -), page AS ( - SELECT * FROM scoped - WHERE $3::timestamptz IS NULL OR (created_at, id) > ($3, $4::uuid) - ORDER BY created_at, id LIMIT $5 ) SELECT COUNT(*), - COALESCE((SELECT jsonb_agg(to_jsonb(page) ORDER BY created_at, id) FROM page), '[]'::jsonb) -FROM scoped` + COALESCE(jsonb_agg(to_jsonb(s) ORDER BY created_at, id) + FILTER (WHERE $3::timestamptz IS NULL OR (created_at, id) > ($3, $4::uuid)), '[]'::jsonb) +FROM scoped s` func ladderTimelineArgs(accountID, provider string, cursor *LadderTimelineCursor) []any { if cursor == nil { - return []any{accountID, provider, nil, nil, LadderTimelinePageSize + 1} + return []any{accountID, provider, nil, nil} } - return []any{accountID, provider, cursor.CreatedAt, cursor.ID, LadderTimelinePageSize + 1} + return []any{accountID, provider, cursor.CreatedAt, cursor.ID} } func (s *PostgresStore) ListLadderTimeline(ctx context.Context, accountID, provider string, cursor *LadderTimelineCursor) (*LadderTimelinePage, error) { @@ -71,11 +75,6 @@ func (s *PostgresStore) ListLadderTimeline(ctx context.Context, accountID, provi if err := fillLadderTimelineOnDemand(page, scopeRaw); err != nil { return nil, err } - if len(page.Events) > LadderTimelinePageSize { - page.Events = page.Events[:LadderTimelinePageSize] - last := page.Events[len(page.Events)-1] - page.NextCursor = &LadderTimelineCursor{CreatedAt: last.CreatedAt, ID: last.ID} - } return page, nil } @@ -89,10 +88,5 @@ func (s *PostgresStore) ListLadderTimelineRuns(ctx context.Context, accountID, p if err := json.Unmarshal(raw, &page.Runs); err != nil { return nil, fmt.Errorf("decode ladder runs: %w", err) } - if len(page.Runs) > LadderTimelinePageSize { - page.Runs = page.Runs[:LadderTimelinePageSize] - last := page.Runs[len(page.Runs)-1] - page.NextCursor = &LadderTimelineCursor{CreatedAt: last.CreatedAt, ID: last.ID} - } return page, nil } diff --git a/internal/config/store_postgres_ladder_timeline_integration_test.go b/internal/config/store_postgres_ladder_timeline_integration_test.go index 8f5b076e8..1302e5af2 100644 --- a/internal/config/store_postgres_ladder_timeline_integration_test.go +++ b/internal/config/store_postgres_ladder_timeline_integration_test.go @@ -26,7 +26,7 @@ func TestLadderTimelineScopedPagesAndSnapshots(t *testing.T) { Plan: []byte(`{"secret":"never-serialize-this","actions":[{"action":"Hold","layer":"compute-sp","rationale":"Baseline already covered","data_sources":["aws-ce"],"secret":"never-serialize-this"}]}`), }) require.NoError(t, err) - tranches := make([]LadderTrancheDB, LadderTimelinePageSize+1) + tranches := make([]LadderTrancheDB, 101) for i := range tranches { tranches[i] = LadderTrancheDB{ ID: uuid.NewString(), ConfigID: &cfgID, RunID: &run.ID, @@ -37,16 +37,18 @@ func TestLadderTimelineScopedPagesAndSnapshots(t *testing.T) { require.NoError(t, store.SaveLadderTranches(ctx, tranches)) first, err := store.ListLadderTimeline(ctx, accountID, "aws", nil) require.NoError(t, err) - require.Len(t, first.Events, 100) + require.Len(t, first.Events, 101, "one read returns the whole scope") require.Equal(t, int64(101), first.TotalCount) require.Equal(t, "12.469056", first.TotalUSDHr) - require.NotNil(t, first.NextCursor) - second, err := store.ListLadderTimeline(ctx, accountID, "aws", first.NextCursor) + // A cursor from the earlier paginated API narrows the rows, never the totals. + legacy, err := store.ListLadderTimeline(ctx, accountID, "aws", &LadderTimelineCursor{CreatedAt: first.Events[99].CreatedAt, ID: first.Events[99].ID}) require.NoError(t, err) - require.Len(t, second.Events, 1) - require.Nil(t, second.NextCursor) + require.Len(t, legacy.Events, 1) + require.Equal(t, first.Events[100].ID, legacy.Events[0].ID) + require.Equal(t, int64(101), legacy.TotalCount) + require.Equal(t, "12.469056", legacy.TotalUSDHr) ids := make(map[string]bool) - for _, event := range append(first.Events, second.Events...) { + for _, event := range first.Events { require.False(t, ids[event.ID], "cursor must never duplicate an event") ids[event.ID] = true require.Zero(t, event.Revision, "untouched tranches report revision 0") @@ -82,13 +84,12 @@ func TestLadderTimelineScopedPagesAndSnapshots(t *testing.T) { runs, err = store.ListLadderTimelineRuns(ctx, accountID, "aws", nil) require.NoError(t, err) require.Equal(t, int64(101), runs.TotalCount) - require.Len(t, runs.Runs, 100) - lastRunPage, err := store.ListLadderTimelineRuns(ctx, accountID, "aws", runs.NextCursor) + require.Len(t, runs.Runs, 101) + lastRunPage, err := store.ListLadderTimelineRuns(ctx, accountID, "aws", &LadderTimelineCursor{CreatedAt: runs.Runs[99].CreatedAt, ID: runs.Runs[99].ID}) require.NoError(t, err) require.Len(t, lastRunPage.Runs, 1) - for _, item := range runs.Runs { - require.NotEqual(t, lastRunPage.Runs[0].ID, item.ID) - } + require.Equal(t, runs.Runs[100].ID, lastRunPage.Runs[0].ID) + require.Equal(t, int64(101), lastRunPage.TotalCount) otherConfig := seedLadderConfigWithExtID(ctx, t, store, "222222222222") otherRun, err := store.SaveLadderRun(ctx, &LadderRunDB{ConfigID: &otherConfig, Status: ladder.RunStatusPlanned, StartedAt: time.Now().UTC(), Plan: []byte(`{}`)}) require.NoError(t, err) diff --git a/internal/config/store_postgres_ladder_timeline_snapshot_integration_test.go b/internal/config/store_postgres_ladder_timeline_snapshot_integration_test.go new file mode 100644 index 000000000..bd2a0e447 --- /dev/null +++ b/internal/config/store_postgres_ladder_timeline_snapshot_integration_test.go @@ -0,0 +1,309 @@ +//go:build integration + +package config + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "math/big" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +// errLadderTimelineChanged is the failure the web client's completeTimeline +// (frontend/src/api/ladder.ts) raises when the pages it stitches together +// disagree: "The ladder timeline changed while loading. Reload to try again." +var errLadderTimelineChanged = errors.New("the ladder timeline changed while loading") + +// clientTimelinePage is the wire shape completeTimeline reads. +type clientTimelinePage struct { + TotalUSDHr *string `json:"total_usd_hr"` + TotalODUSDHr *string `json:"total_od_usd_hr"` + NextCursor *LadderTimelineCursor `json:"next_cursor"` + Events []clientTimelineRow `json:"events"` + Runs []clientTimelineRow `json:"runs"` + TotalCount int64 `json:"total_count"` +} + +type clientTimelineRow struct { + ID string `json:"id"` + AmountUSDHr string `json:"amount_usd_hr"` + ODUSDHr string `json:"od_usd_hr"` +} + +type clientTimelineLoad struct { + ids []string + total string + rounds int +} + +func fetchClientTimelinePage(ctx context.Context, store *PostgresStore, accountID string, runs bool, cursor *LadderTimelineCursor) (*clientTimelinePage, error) { + var page any + var err error + if runs { + page, err = store.ListLadderTimelineRuns(ctx, accountID, "aws", cursor) + } else { + page, err = store.ListLadderTimeline(ctx, accountID, "aws", cursor) + } + if err != nil { + return nil, err + } + raw, err := json.Marshal(page) + if err != nil { + return nil, err + } + var wire clientTimelinePage + return &wire, json.Unmarshal(raw, &wire) +} + +func sameTimelineTotals(a, b *clientTimelinePage) bool { + same := func(x, y *string) bool { return (x == nil) == (y == nil) && (x == nil || *x == *y) } + return a.TotalCount == b.TotalCount && same(a.TotalUSDHr, b.TotalUSDHr) && same(a.TotalODUSDHr, b.TotalODUSDHr) +} + +func addLadderMicro(sum *big.Int, amount string) error { + r, ok := new(big.Rat).SetString(amount) + if !ok { + return fmt.Errorf("unparseable amount %q", amount) + } + micro := new(big.Rat).Mul(r, new(big.Rat).SetInt(ladderMicro)) + if !micro.IsInt() { + return fmt.Errorf("amount %q is finer than a micro-dollar", amount) + } + sum.Add(sum, micro.Num()) + return nil +} + +// loadLadderTimelineLikeClient follows next_cursor exactly as completeTimeline +// does and applies the same cross-page checks: one total_count, total_usd_hr +// and total_od_usd_hr across pages, no repeated ID, and rows that add up to +// the totals. between runs before each follow-up page request. +func loadLadderTimelineLikeClient(ctx context.Context, store *PostgresStore, accountID string, runs bool, between func()) (clientTimelineLoad, error) { + var load clientTimelineLoad + var first *clientTimelinePage + var cursor *LadderTimelineCursor + seen := map[string]bool{} + amount, od := new(big.Int), new(big.Int) + for { + page, err := fetchClientTimelinePage(ctx, store, accountID, runs, cursor) + if err != nil { + return load, err + } + load.rounds++ + if first == nil { + first = page + } else if !sameTimelineTotals(first, page) { + return load, fmt.Errorf("%w: totals moved on page %d", errLadderTimelineChanged, load.rounds) + } + for _, row := range append(page.Events, page.Runs...) { + if seen[row.ID] { + return load, fmt.Errorf("%w: %s repeated", errLadderTimelineChanged, row.ID) + } + seen[row.ID] = true + load.ids = append(load.ids, row.ID) + if !runs && (addLadderMicro(amount, row.AmountUSDHr) != nil || addLadderMicro(od, row.ODUSDHr) != nil) { + return load, fmt.Errorf("unparseable amounts on %s", row.ID) + } + } + if page.NextCursor == nil { + break + } + if between != nil { + between() + } + cursor = page.NextCursor + } + return load, checkClientTimelineLoad(&load, first, runs, amount, od) +} + +func checkClientTimelineLoad(load *clientTimelineLoad, first *clientTimelinePage, runs bool, amount, od *big.Int) error { + if int64(len(load.ids)) != first.TotalCount { + return fmt.Errorf("%w: %d rows for total_count %d", errLadderTimelineChanged, len(load.ids), first.TotalCount) + } + if runs { + return nil + } + load.total = *first.TotalUSDHr + if formatLadderMicro(amount) != *first.TotalUSDHr || formatLadderMicro(od) != *first.TotalODUSDHr { + return fmt.Errorf("%w: rows add up to %s/%s, totals say %s/%s", errLadderTimelineChanged, + formatLadderMicro(amount), formatLadderMicro(od), *first.TotalUSDHr, *first.TotalODUSDHr) + } + return nil +} + +type ladderTimelineFixture struct { + store *PostgresStore + cfgID string + accountID string + runID string + amendID string + date time.Time +} + +// seedLadderTimelineFixture saves one planned run whose n scheduled tranches +// are all editable, plus n-1 more planned runs, so both endpoints span +// several of the earlier 100-row pages. +func seedLadderTimelineFixture(ctx context.Context, t *testing.T, n int) *ladderTimelineFixture { + t.Helper() + store := setupLadderStore(ctx, t) + f := &ladderTimelineFixture{store: store, cfgID: seedLadderConfig(ctx, t, store), date: time.Now().UTC().Add(48 * time.Hour)} + require.NoError(t, store.db.QueryRow(ctx, "SELECT cloud_account_id FROM ladder_configs WHERE id=$1", f.cfgID).Scan(&f.accountID)) + run, err := store.SaveLadderRun(ctx, &LadderRunDB{ConfigID: &f.cfgID, Status: ladder.RunStatusPlanned, StartedAt: time.Now().UTC(), TotalHourlyCommit: float64(n), Plan: []byte(`{}`)}) + require.NoError(t, err) + f.runID = run.ID + tranches := make([]LadderTrancheDB, n) + for i := range tranches { + tranches[i] = f.tranche(float64(i%7+1) / 8) + } + require.NoError(t, store.SaveLadderTranches(ctx, tranches)) + f.amendID = tranches[0].ID + _, err = store.db.Exec(ctx, `INSERT INTO ladder_runs(config_id,status,created_at) + SELECT $1,'planned',NOW() FROM generate_series(1,$2::int)`, f.cfgID, n-1) + require.NoError(t, err) + return f +} + +func (f *ladderTimelineFixture) tranche(amount float64) LadderTrancheDB { + return LadderTrancheDB{ + ID: uuid.NewString(), ConfigID: &f.cfgID, RunID: &f.runID, + LayerType: ladder.LayerComputeSP, Term: ladder.Term1Year, PaymentOption: ladder.PaymentNoUpfront, + Status: ladder.TrancheStatusScheduled, AmountUSDHr: amount, ScheduledDate: f.date, + } +} + +// amend lowers the fixture's editable tranche through the real amend path, +// which always allows a reduction. +func (f *ladderTimelineFixture) amend(ctx context.Context, revision int64, amount string) error { + _, err := f.store.AmendLadderTranche(ctx, f.amendID, f.accountID, "aws", "timeline-test", LadderAmendment{ + ExpectedRevision: &revision, ScheduledDate: f.date, AmountUSDHr: amount, + }) + return err +} + +// TestLadderTimelineLoadsWholeScopeInOneRead pins the read scaling half of +// #668: a client load is one statement whatever the history length (it was +// ceil(n/100) statements, each aggregating all n rows), and the rows and +// totals it gets are exactly the scope's. +func TestLadderTimelineLoadsWholeScopeInOneRead(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + f := seedLadderTimelineFixture(ctx, t, 250) + events, err := loadLadderTimelineLikeClient(ctx, f.store, f.accountID, false, nil) + require.NoError(t, err) + require.Equal(t, 1, events.rounds, "the whole tranche timeline must come back in one read") + var wantIDs []string + var wantTotal string + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT array_agg(id::text ORDER BY created_at, id), SUM(amount_usd_hr)::text + FROM ladder_tranches WHERE config_id=$1`, f.cfgID).Scan(&wantIDs, &wantTotal)) + require.Equal(t, wantIDs, events.ids, "same rows in the same (created_at, id) order") + require.Equal(t, wantTotal, events.total) + runs, err := loadLadderTimelineLikeClient(ctx, f.store, f.accountID, true, nil) + require.NoError(t, err) + require.Equal(t, 1, runs.rounds, "the whole run timeline must come back in one read") + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT array_agg(id::text ORDER BY created_at, id) FROM ladder_runs WHERE config_id=$1`, f.cfgID).Scan(&wantIDs)) + require.Equal(t, wantIDs, runs.ids) +} + +// TestLadderTimelineLoadToleratesMidLoadChanges replays what used to fail the +// whole timeline: an amendment, a new tranche or a new run committed while +// the client was between two page requests. +func TestLadderTimelineLoadToleratesMidLoadChanges(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + f := seedLadderTimelineFixture(ctx, t, 250) + for _, tc := range []struct { + change func() error + name string + runs bool + }{ + {name: "amendment", change: func() error { return f.amend(ctx, 0, "0.100000") }}, + {name: "new tranche", change: func() error { return f.store.SaveLadderTranches(ctx, []LadderTrancheDB{f.tranche(2)}) }}, + {name: "new run", runs: true, change: func() error { + _, err := f.store.SaveLadderRun(ctx, &LadderRunDB{ConfigID: &f.cfgID, Status: ladder.RunStatusPlanned, StartedAt: time.Now().UTC(), Plan: []byte(`{}`)}) + return err + }}, + } { + t.Run(tc.name, func(t *testing.T) { + before, err := loadLadderTimelineLikeClient(ctx, f.store, f.accountID, tc.runs, nil) + require.NoError(t, err) + changed := false + load, err := loadLadderTimelineLikeClient(ctx, f.store, f.accountID, tc.runs, func() { + if !changed { + changed = true + require.NoError(t, tc.change()) + } + }) + require.NoError(t, err, "a change between page requests must not fail the load") + if !changed { // nothing between requests: the load must still see a full scope + require.NoError(t, tc.change()) + } + after, err := loadLadderTimelineLikeClient(ctx, f.store, f.accountID, tc.runs, nil) + require.NoError(t, err) + require.True(t, len(after.ids) != len(before.ids) || after.total != before.total, "the change must be visible to the next load") + require.GreaterOrEqual(t, len(load.ids), len(before.ids)) + }) + } +} + +// TestLadderTimelineLoadIsSnapshotConsistentUnderConcurrentWrites runs client +// loads while another connection keeps amending the timeline and adding +// tranches and runs. Every load must be internally consistent, and the loads +// must have observed the writes (the totals move between loads). +func TestLadderTimelineLoadIsSnapshotConsistentUnderConcurrentWrites(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + f := seedLadderTimelineFixture(ctx, t, 450) + stop := make(chan struct{}) + writes := make(chan error, 1) + var writer sync.WaitGroup + writer.Add(1) + go func() { + defer writer.Done() + writes <- f.writeUntil(ctx, stop) + }() + totals := map[string]bool{} + var loadErr error + for i := 0; i < 40 && loadErr == nil; i++ { + for _, runs := range []bool{false, true} { + load, err := loadLadderTimelineLikeClient(ctx, f.store, f.accountID, runs, nil) + if err != nil { + loadErr = err + break + } + totals[fmt.Sprintf("%t/%d/%s", runs, len(load.ids), load.total)] = true + } + } + close(stop) + writer.Wait() + require.NoError(t, <-writes) + require.NoError(t, loadErr, "every load must come from one snapshot") + require.Greater(t, len(totals), 4, "the writer must have changed the timeline between loads") +} + +// writeUntil keeps lowering the editable tranche by one micro-dollar and +// adding a tranche and a run until stop closes. +func (f *ladderTimelineFixture) writeUntil(ctx context.Context, stop <-chan struct{}) error { + for revision := int64(0); ; revision++ { + select { + case <-stop: + return nil + default: + } + if err := f.amend(ctx, revision, fmt.Sprintf("0.%06d", 124999-revision)); err != nil { + return fmt.Errorf("amend %d: %w", revision, err) + } + if err := f.store.SaveLadderTranches(ctx, []LadderTrancheDB{f.tranche(0.25)}); err != nil { + return err + } + if _, err := f.store.SaveLadderRun(ctx, &LadderRunDB{ConfigID: &f.cfgID, Status: ladder.RunStatusPlanned, StartedAt: time.Now().UTC(), Plan: []byte(`{}`)}); err != nil { + return err + } + } +} diff --git a/internal/config/store_postgres_ladder_timeline_test.go b/internal/config/store_postgres_ladder_timeline_test.go index cbb3dd722..94dbb6920 100644 --- a/internal/config/store_postgres_ladder_timeline_test.go +++ b/internal/config/store_postgres_ladder_timeline_test.go @@ -13,13 +13,17 @@ import ( "github.com/stretchr/testify/require" ) -func TestLadderTimelinePagination(t *testing.T) { +// ladderTimelineWholeScope is more rows than the 100-row pages the timeline +// used to return, so a read that still paged would truncate it. +const ladderTimelineWholeScope = 250 + +func TestLadderTimelineReturnsWholeScopeFromOneStatement(t *testing.T) { pool, err := pgxmock.NewPool() require.NoError(t, err) defer pool.Close() store := &PostgresStore{db: pool} stamp := time.Now().UTC() - events := make([]LadderTimelineEvent, LadderTimelinePageSize+1) + events := make([]LadderTimelineEvent, ladderTimelineWholeScope) for i := range events { events[i] = LadderTimelineEvent{ID: uuid.NewString(), CreatedAt: stamp, AmountUSDHr: "0.123456", Revision: int64(i)} } @@ -27,17 +31,19 @@ func TestLadderTimelinePagination(t *testing.T) { require.NoError(t, err) scope := []byte(`[["0.123456",null,null]]`) pool.ExpectQuery(regexp.QuoteMeta(ladderTimelineQuery)). - WithArgs("account", "aws", nil, nil, LadderTimelinePageSize+1). - WillReturnRows(pgxmock.NewRows([]string{"count", "sum", "page", "scope"}).AddRow(int64(101), "12.469056", raw, scope)) + WithArgs("account", "aws", nil, nil). + WillReturnRows(pgxmock.NewRows([]string{"count", "sum", "page", "scope"}).AddRow(int64(ladderTimelineWholeScope), "30.864000", raw, scope)) page, err := store.ListLadderTimeline(context.Background(), "account", "aws", nil) require.NoError(t, err) - require.Len(t, page.Events, LadderTimelinePageSize) - require.Equal(t, int64(101), page.TotalCount) - require.Equal(t, "12.469056", page.TotalUSDHr) - require.Equal(t, "0.123456", page.Events[0].AmountUSDHr) + require.Len(t, page.Events, ladderTimelineWholeScope, "every scoped row comes back from the one statement") + require.Equal(t, int64(ladderTimelineWholeScope), page.TotalCount) + require.Equal(t, "30.864000", page.TotalUSDHr) + require.Equal(t, events[ladderTimelineWholeScope-1].ID, page.Events[ladderTimelineWholeScope-1].ID) require.Equal(t, int64(1), page.Events[1].Revision) - require.Equal(t, events[99].ID, page.NextCursor.ID) - require.NoError(t, pool.ExpectationsWereMet()) + wire, err := json.Marshal(page) + require.NoError(t, err) + require.NotContains(t, string(wire), "next_cursor", "the client must never be told to fetch a second page") + require.NoError(t, pool.ExpectationsWereMet(), "exactly one statement") } func TestLadderTimelineQueryFailureIsNotEmpty(t *testing.T) { @@ -46,7 +52,7 @@ func TestLadderTimelineQueryFailureIsNotEmpty(t *testing.T) { defer pool.Close() store := &PostgresStore{db: pool} pool.ExpectQuery(regexp.QuoteMeta(ladderTimelineQuery)). - WithArgs("account", "aws", nil, nil, LadderTimelinePageSize+1). + WithArgs("account", "aws", nil, nil). WillReturnError(errors.New("database unavailable")) page, err := store.ListLadderTimeline(context.Background(), "account", "aws", nil) require.ErrorContains(t, err, "database unavailable") @@ -54,29 +60,33 @@ func TestLadderTimelineQueryFailureIsNotEmpty(t *testing.T) { require.NoError(t, pool.ExpectationsWereMet()) } -func TestLadderTimelineRunsPaginationAndFailure(t *testing.T) { +func TestLadderTimelineRunsWholeScopeCursorAndFailure(t *testing.T) { pool, err := pgxmock.NewPool() require.NoError(t, err) defer pool.Close() store := &PostgresStore{db: pool} stamp := time.Now().UTC() - runs := make([]LadderTimelineRun, LadderTimelinePageSize+1) + runs := make([]LadderTimelineRun, ladderTimelineWholeScope) for i := range runs { runs[i] = LadderTimelineRun{ID: uuid.NewString(), CreatedAt: stamp} } raw, err := json.Marshal(runs) require.NoError(t, err) pool.ExpectQuery(regexp.QuoteMeta(ladderTimelineRunsQuery)). - WithArgs("account", "aws", nil, nil, LadderTimelinePageSize+1). - WillReturnRows(pgxmock.NewRows([]string{"count", "page"}).AddRow(int64(101), raw)) + WithArgs("account", "aws", nil, nil). + WillReturnRows(pgxmock.NewRows([]string{"count", "page"}).AddRow(int64(ladderTimelineWholeScope), raw)) page, err := store.ListLadderTimelineRuns(context.Background(), "account", "aws", nil) require.NoError(t, err) - require.Len(t, page.Runs, 100) - require.Equal(t, runs[99].ID, page.NextCursor.ID) + require.Len(t, page.Runs, ladderTimelineWholeScope) + wire, err := json.Marshal(page) + require.NoError(t, err) + require.NotContains(t, string(wire), "next_cursor") + // A cursor from the earlier paginated API is still passed through as the bound. + cursor := &LadderTimelineCursor{CreatedAt: stamp, ID: runs[99].ID} pool.ExpectQuery(regexp.QuoteMeta(ladderTimelineRunsQuery)). - WithArgs("account", "aws", stamp, runs[99].ID, LadderTimelinePageSize+1). + WithArgs("account", "aws", stamp, runs[99].ID). WillReturnError(errors.New("database unavailable")) - page, err = store.ListLadderTimelineRuns(context.Background(), "account", "aws", page.NextCursor) + page, err = store.ListLadderTimelineRuns(context.Background(), "account", "aws", cursor) require.ErrorContains(t, err, "database unavailable") require.Nil(t, page) require.NoError(t, pool.ExpectationsWereMet()) @@ -98,7 +108,7 @@ func TestLadderTimelineOnDemandFromOneStatement(t *testing.T) { // Two rows of exactly 1.5000015 on-demand: per-row ceilings sum to 3.000004, not the rounded exact sum 3.000003. scope := []byte(`[["1.000001","commitment","2/3"],["1.000001","commitment","2/3"],["4.000000",null,null]]`) pool.ExpectQuery(regexp.QuoteMeta(ladderTimelineQuery)). - WithArgs("account", "aws", nil, nil, LadderTimelinePageSize+1). + WithArgs("account", "aws", nil, nil). WillReturnRows(pgxmock.NewRows([]string{"count", "sum", "page", "scope"}).AddRow(int64(3), "6.000002", raw, scope)) got, err := store.ListLadderTimeline(context.Background(), "account", "aws", nil) require.NoError(t, err) @@ -123,7 +133,7 @@ func TestLadderTimelineMalformedRateFailsLoud(t *testing.T) { defer pool.Close() store := &PostgresStore{db: pool} pool.ExpectQuery(regexp.QuoteMeta(ladderTimelineQuery)). - WithArgs("account", "aws", nil, nil, LadderTimelinePageSize+1). + WithArgs("account", "aws", nil, nil). WillReturnRows(pgxmock.NewRows([]string{"count", "sum", "page", "scope"}).AddRow(int64(1), "1.000000", []byte(`[]`), []byte(scope))) got, err := store.ListLadderTimeline(context.Background(), "account", "aws", nil) require.ErrorIs(t, err, ErrLadderTimelineRate)