From 9e34d25ce27b22d247595eb5472b3ed6bec13528 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 22:37:23 +0100 Subject: [PATCH] test(api): pin the submit dedupe as a predicate, not a 1000-row page Issue #266 reported that persistExecutionAndSuppressions found a duplicate submit by scanning one page of pending rows (GetPendingExecutionsTx: ORDER BY scheduled_date ASC LIMIT 1000 FOR UPDATE), so more than 1000 earlier pending rows hid the first submit and its resubmit bought twice. #761 already replaced that read with ListRecentSubmitsTx, an unbounded predicate on source, dedupe statuses, window start and creator behind a per-creator advisory lock. This adds the regression coverage that pins it: - real Postgres: 1,600 earlier pending rows (1,100 plan-style, 500 same creator in window with a different key), then an identical resubmit collapses onto the first submit - real Postgres: each idempotency-key component, capacity, source and window changed alone is not a duplicate; provider/payment case is - real Postgres: ListRecentSubmitsTx returns exactly the rows matching each clause (source, status, inclusive window start, creator, NULL) - unit: matchDuplicateInList clause table Closes #266 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_purchases_dedupe_test.go | 66 +++++++ ...urchases_submit_dedupe_integration_test.go | 185 ++++++++++++++++++ 2 files changed, 251 insertions(+) create mode 100644 internal/api/handler_purchases_dedupe_test.go diff --git a/internal/api/handler_purchases_dedupe_test.go b/internal/api/handler_purchases_dedupe_test.go new file mode 100644 index 00000000..33b1a74c --- /dev/null +++ b/internal/api/handler_purchases_dedupe_test.go @@ -0,0 +1,66 @@ +package api + +import ( + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" +) + +// TestMatchDuplicateInList pins each clause of the Go-side duplicate check +// that persistExecutionAndSuppressions runs over ListRecentSubmitsTx's rows: +// web source, window start (inclusive), creator (nil creator equals ""), +// and the recomputed idempotency key (which also hashes the creator, so the +// creator cases hold through either check). The SQL predicate applies the same +// source, window and creator filters; this keeps the Go side honest if the +// rows it is given ever widen (#266). +func TestMatchDuplicateInList(t *testing.T) { + t.Parallel() + now := time.Now() + creator := "11111111-1111-1111-1111-111111111111" + other := "22222222-2222-2222-2222-222222222222" + recs := []config.RecommendationRecord{validRec()} + key := mustPurchaseIdempotencyKey(creator, recs, 100) + exec := func(mut func(e *config.PurchaseExecution)) config.PurchaseExecution { + e := config.PurchaseExecution{ + ExecutionID: "cand", Status: "pending", Source: common.PurchaseSourceWeb, + ScheduledDate: now.Add(-30 * time.Second), CreatedByUserID: &creator, + Recommendations: append([]config.RecommendationRecord(nil), recs...), CapacityPercent: 100, + } + if mut != nil { + mut(&e) + } + return e + } + for _, tc := range []struct { + name string + cand config.PurchaseExecution + creator string + key string + wantDup bool + }{ + {"exact match", exec(nil), creator, key, true}, + {"window start is inclusive", exec(func(e *config.PurchaseExecution) { e.ScheduledDate = now.Add(-purchaseIdempotencyWindow) }), creator, key, true}, + {"before the window", exec(func(e *config.PurchaseExecution) { + e.ScheduledDate = now.Add(-purchaseIdempotencyWindow - time.Second) + }), creator, key, false}, + {"not web-submitted", exec(func(e *config.PurchaseExecution) { e.Source = common.PurchaseSourceCLI }), creator, key, false}, + {"other creator", exec(func(e *config.PurchaseExecution) { e.CreatedByUserID = &other }), creator, key, false}, + {"no creator vs a creator", exec(func(e *config.PurchaseExecution) { e.CreatedByUserID = nil }), creator, key, false}, + {"no creator on both sides", exec(func(e *config.PurchaseExecution) { e.CreatedByUserID = nil }), "", mustPurchaseIdempotencyKey("", recs, 100), true}, + {"different capacity", exec(func(e *config.PurchaseExecution) { e.CapacityPercent = 50 }), creator, key, false}, + {"different rec set", exec(func(e *config.PurchaseExecution) { e.Recommendations[0].Count++ }), creator, key, false}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := matchDuplicateInList([]config.PurchaseExecution{tc.cand}, tc.creator, tc.key, now) + if tc.wantDup { + assert.NotNil(t, got) + } else { + assert.Nil(t, got) + } + }) + } +} diff --git a/internal/api/handler_purchases_submit_dedupe_integration_test.go b/internal/api/handler_purchases_submit_dedupe_integration_test.go index 22ea060f..5f89a65d 100644 --- a/internal/api/handler_purchases_submit_dedupe_integration_test.go +++ b/internal/api/handler_purchases_submit_dedupe_integration_test.go @@ -4,10 +4,12 @@ package api import ( "context" + "encoding/json" "sync" "testing" "time" + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" "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" @@ -182,3 +184,186 @@ func TestPersistExecution_DedupeIsPerCreator(t *testing.T) { assert.NotNil(t, submit(alice), "the same user's resubmit collapses") assert.Equal(t, 2, countSubmitRows(t, pg)) } + +// seedSubmitRows bulk-inserts n pending executions scheduled at `at` with the +// given source, creator (nil stores NULL) and recommendations. +func seedSubmitRows(t *testing.T, pg *testhelpers.PostgresContainer, n int, at time.Time, source string, creator *string, recs []config.RecommendationRecord) { + t.Helper() + recsJSON, err := json.Marshal(recs) + require.NoError(t, err) + _, err = pg.DB.Pool().Exec(context.Background(), ` + INSERT INTO purchase_executions (status, scheduled_date, recommendations, source, created_by_user_id, capacity_percent, error) + SELECT 'pending', $1, $2::jsonb, $3, $4::uuid, 100, '' FROM generate_series(1, $5)`, + at, recsJSON, source, creator, n) + require.NoError(t, err) +} + +// TestPersistExecution_DedupeSeesPastThousandPendingRows pins #266: the +// duplicate check must not depend on how many unrelated rows are pending. +// The old check read one page of pending rows (ORDER BY scheduled_date ASC +// LIMIT 1000), so 1,600 earlier-scheduled pending rows pushed the first +// submit off the page and its identical resubmit was inserted as a second +// execution. +func TestPersistExecution_DedupeSeesPastThousandPendingRows(t *testing.T) { + pg, store := setupSubmitDB(t) + ctx := context.Background() + recs, key := submitFixture(t) + h := &Handler{config: store} + now := time.Now() + + // Plan/ramp-style rows: not web-submitted, scheduled an hour ago. + seedSubmitRows(t, pg, 1100, now.Add(-time.Hour), "", nil, recs) + // The same creator's earlier web submits of a different rec set, inside + // the window: read by the dedupe query but with a different key. + otherRecs := append([]config.RecommendationRecord(nil), recs...) + otherRecs[0].Count = 7 + seedSubmitRows(t, pg, 500, now.Add(-time.Minute), common.PurchaseSourceWeb, nil, otherRecs) + + first := newSubmitExecution(t, recs) + dup, err := h.persistExecutionAndSuppressions(ctx, first, nil, "", key) + require.NoError(t, err) + require.Nil(t, dup, "no earlier row carries this submit's key") + + dup, err = h.persistExecutionAndSuppressions(ctx, newSubmitExecution(t, recs), nil, "", key) + require.NoError(t, err) + require.NotNil(t, dup, "the resubmit must collapse onto the first submit despite 1,600 earlier pending rows") + assert.Equal(t, first.ExecutionID, dup.ExecutionID) + assert.Equal(t, 1601, countSubmitRows(t, pg), "1,600 seeded rows plus the first submit") +} + +// TestPersistExecution_DedupeKeySemantics round-trips the first submit +// through Postgres and changes exactly one component of the duplicate key in +// the resubmit: every component of purchaseIdempotencyKey, the capacity, the +// web source and the window. Only an exact match (provider and payment are +// compared case- and whitespace-insensitively) collapses. +func TestPersistExecution_DedupeKeySemantics(t *testing.T) { + pg, store := setupSubmitDB(t) + ctx := context.Background() + h := &Handler{config: store} + acct := "acct-b" + base := func() config.RecommendationRecord { + return config.RecommendationRecord{ + Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.large", Engine: "", + Count: 2, Term: 1, Payment: "no-upfront", Details: zonalEC2Details("us-east-1a"), + } + } + for _, tc := range []struct { + name string + mutate func(r *config.RecommendationRecord) + capacity int + afterSQL string // applied to the first row before the resubmit + wantDup bool + }{ + {name: "identical", wantDup: true}, + {name: "provider case and spaces", mutate: func(r *config.RecommendationRecord) { r.Provider = " AWS " }, wantDup: true}, + {name: "payment case", mutate: func(r *config.RecommendationRecord) { r.Payment = "No-Upfront" }, wantDup: true}, + {name: "provider", mutate: func(r *config.RecommendationRecord) { r.Provider = "azure" }}, + {name: "service", mutate: func(r *config.RecommendationRecord) { r.Service = "rds"; r.Details = nil }}, + {name: "region", mutate: func(r *config.RecommendationRecord) { r.Region = "us-west-2" }}, + {name: "resource type", mutate: func(r *config.RecommendationRecord) { r.ResourceType = "m5.xlarge" }}, + {name: "engine", mutate: func(r *config.RecommendationRecord) { r.Engine = "postgres" }}, + {name: "account", mutate: func(r *config.RecommendationRecord) { r.CloudAccountID = &acct }}, + {name: "count", mutate: func(r *config.RecommendationRecord) { r.Count = 3 }}, + {name: "term", mutate: func(r *config.RecommendationRecord) { r.Term = 3 }}, + {name: "payment", mutate: func(r *config.RecommendationRecord) { r.Payment = "all-upfront" }}, + {name: "availability zone", mutate: func(r *config.RecommendationRecord) { r.Details = zonalEC2Details("us-east-1b") }}, + {name: "platform", mutate: func(r *config.RecommendationRecord) { + r.Details = json.RawMessage(`{"instance_type":"m5.large","platform":"Windows","tenancy":"default","scope":"Availability Zone","availability_zone":"us-east-1a"}`) + }}, + {name: "tenancy", mutate: func(r *config.RecommendationRecord) { + r.Details = json.RawMessage(`{"instance_type":"m5.large","platform":"Linux/UNIX","tenancy":"dedicated","scope":"Availability Zone","availability_zone":"us-east-1a"}`) + }}, + {name: "capacity", capacity: 50}, + {name: "first row not web-submitted", afterSQL: `UPDATE purchase_executions SET source = 'cudly-cli'`}, + {name: "first row outside the window", afterSQL: `UPDATE purchase_executions SET scheduled_date = now() - interval '3 minutes'`}, + } { + t.Run(tc.name, func(t *testing.T) { + _, err := pg.DB.Pool().Exec(ctx, `TRUNCATE purchase_executions CASCADE`) + require.NoError(t, err) + firstRecs := []config.RecommendationRecord{base()} + first := newSubmitExecution(t, firstRecs) + dup, err := h.persistExecutionAndSuppressions(ctx, first, nil, "", mustPurchaseIdempotencyKey("", firstRecs, 100)) + require.NoError(t, err) + require.Nil(t, dup) + if tc.afterSQL != "" { + _, err = pg.DB.Pool().Exec(ctx, tc.afterSQL) + require.NoError(t, err) + } + + r := base() + if tc.mutate != nil { + tc.mutate(&r) + } + capacity := 100 + if tc.capacity != 0 { + capacity = tc.capacity + } + resub := []config.RecommendationRecord{r} + exec, _, err := newPendingExecution(&ExecutePurchaseRequest{Recommendations: resub, CapacityPercent: capacity}, 0, 0) + require.NoError(t, err) + dup, err = h.persistExecutionAndSuppressions(ctx, exec, nil, "", mustPurchaseIdempotencyKey("", resub, capacity)) + require.NoError(t, err) + if tc.wantDup { + require.NotNil(t, dup) + assert.Equal(t, first.ExecutionID, dup.ExecutionID) + assert.Equal(t, 1, countSubmitRows(t, pg)) + } else { + assert.Nil(t, dup) + assert.Equal(t, 2, countSubmitRows(t, pg)) + } + }) + } +} + +// TestListRecentSubmitsTx_Predicate pins each clause of the dedupe query in +// SQL (source, status, window start inclusive, creator, NULL creator), so the +// read stays bounded by matching rows rather than relying on a page size. +func TestListRecentSubmitsTx_Predicate(t *testing.T) { + pg, store := setupSubmitDB(t) + ctx := context.Background() + recs, _ := submitFixture(t) + newUser := func(email string) *string { + id := uuid.New().String() + _, err := pg.DB.Pool().Exec(ctx, ` + INSERT INTO users (id, email, password_hash, salt, active, group_ids, created_at, updated_at) + SELECT $1, $2, 'x', 'y', true, ARRAY[g.id], now(), now() + FROM groups g ORDER BY g.name LIMIT 1`, id, email) + require.NoError(t, err) + return &id + } + alice, bob := newUser("alice@example.com"), newUser("bob@example.com") + since := time.Now().Add(-purchaseIdempotencyWindow).Truncate(time.Microsecond) + + insert := func(source, status string, at time.Time, creator *string) string { + recsJSON, err := json.Marshal(recs) + require.NoError(t, err) + var id string + require.NoError(t, pg.DB.Pool().QueryRow(ctx, ` + INSERT INTO purchase_executions (status, scheduled_date, recommendations, source, created_by_user_id, capacity_percent, error) + VALUES ($1, $2, $3::jsonb, $4, $5::uuid, 100, '') RETURNING execution_id::text`, + status, at, recsJSON, source, creator).Scan(&id)) + return id + } + web, inWindow := common.PurchaseSourceWeb, since.Add(time.Minute) + match := insert(web, "pending", inWindow, alice) + atBoundary := insert(web, "completed", since, alice) + insert(common.PurchaseSourceCLI, "pending", inWindow, alice) + insert(web, config.StatusFailed, inWindow, alice) + insert(web, "pending", since.Add(-time.Microsecond), alice) + insert(web, "pending", inWindow, bob) + noCreator := insert(web, "notified", inWindow, nil) + + list := func(creatorID string) []string { + var ids []string + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + rows, err := store.ListRecentSubmitsTx(ctx, tx, creatorID, since) + for i := range rows { + ids = append(ids, rows[i].ExecutionID) + } + return err + })) + return ids + } + assert.ElementsMatch(t, []string{match, atBoundary}, list(*alice)) + assert.ElementsMatch(t, []string{noCreator}, list("")) +}