diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index edfa2805f..26e840d92 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -813,6 +813,9 @@ func (m *mockConfigStore) ListActiveSuppressions(_ context.Context) ([]config.Pu func (m *mockConfigStore) SavePurchaseExecutionTx(ctx context.Context, _ pgx.Tx, e *config.PurchaseExecution) error { return m.SavePurchaseExecution(ctx, e) } +func (m *mockConfigStore) ListRecentSubmitsTx(ctx context.Context, _ pgx.Tx, _ string, _ time.Time) ([]config.PurchaseExecution, error) { + return m.GetPendingExecutions(ctx) +} func (m *mockConfigStore) GetPendingExecutionsTx(ctx context.Context, _ pgx.Tx) ([]config.PurchaseExecution, error) { return m.GetPendingExecutions(ctx) } diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go index 760df4a86..64977f029 100644 --- a/internal/api/handler_purchases.go +++ b/internal/api/handler_purchases.go @@ -2808,7 +2808,7 @@ func fourEyesActorIdentity(session *Session) string { } // executePurchase handles direct purchase execution from recommendations -// matchDuplicateInList scans a slice of pending executions for one that +// matchDuplicateInList scans a slice of recent executions for one that // matches creatorID + idempotencyKey within the idempotency window. // Returns the first match, or nil when there is no duplicate. // Extracted from persistExecutionAndSuppressions to keep cyclomatic @@ -2833,9 +2833,9 @@ func matchDuplicateInList(pending []config.PurchaseExecution, creatorID, idempot // persistExecutionAndSuppressions saves the execution + its suppression // records in a single transaction. It also performs the duplicate-execution -// check inside the same transaction (using SELECT ... FOR UPDATE) so that the -// read and the insert are atomic — closing the TOCTOU race that the pre-tx -// duplicatePurchaseResponse call could not prevent (#643). +// check inside the same transaction, behind a per-creator advisory lock, so +// that the read and the insert are atomic: a plain SELECT ... FOR UPDATE +// cannot see a concurrent transaction's uncommitted insert (#643, MON-01). // // Return values: // - (nil, nil) — no duplicate found; execution was inserted. @@ -2849,13 +2849,15 @@ func (h *Handler) persistExecutionAndSuppressions( creatorID, idempotencyKey string, ) (dup *config.PurchaseExecution, err error) { txErr := h.config.WithTx(ctx, func(tx pgx.Tx) error { - // Duplicate check inside the tx (SELECT FOR UPDATE) — atomic with - // the insert below. - pending, err := h.config.GetPendingExecutionsTx(ctx, tx) + // Duplicate check inside the tx, after taking the creator's submit + // lock: a concurrent identical submit waits for this tx to commit, + // then finds the row. Atomic with the insert below. + now := time.Now() + recent, err := h.config.ListRecentSubmitsTx(ctx, tx, creatorID, now.Add(-purchaseIdempotencyWindow)) if err != nil { return err } - if dup = matchDuplicateInList(pending, creatorID, idempotencyKey, time.Now()); dup != nil { + if dup = matchDuplicateInList(recent, creatorID, idempotencyKey, now); dup != nil { return nil // found duplicate — skip insert, commit tx (read-only) } @@ -2949,16 +2951,22 @@ func (h *Handler) findDuplicatePendingExecution(ctx context.Context, creatorID, // instead of a freshly-minted duplicate. duplicate=true lets the UI explain // why no new approval email was sent. func buildDuplicatePurchaseResponse(ex *config.PurchaseExecution) map[string]any { - return map[string]any{ + resp := map[string]any{ "execution_id": ex.ExecutionID, "status": ex.Status, "recommendation_count": len(ex.Recommendations), "total_upfront_cost": ex.TotalUpfrontCost, "estimated_savings": ex.EstimatedSavings, - "email_sent": ex.NotificationSent != nil, "duplicate": true, - "message": "Duplicate submission collapsed onto the existing pending execution; no new approval request was created.", + "message": "Duplicate submission collapsed onto the existing execution (see status); no new execution or approval request was created.", } + // An approval email only exists for a row still awaiting approval. A row + // that already moved on (direct execute, approved, completed) never had + // one, and email_sent=false would make the client report a failure. + if ex.Status == "pending" || ex.Status == "notified" { + resp["email_sent"] = ex.NotificationSent != nil + } + return resp } // newPendingExecution builds a fresh pending PurchaseExecution with a @@ -3004,7 +3012,7 @@ func (h *Handler) executePurchase(ctx context.Context, req *events.LambdaFunctio // an identical actor + scaled rec set + capacity within a short window must // resolve to the original pending execution rather than minting a second // approvable row (double-spend). The atomic guard lives inside - // persistExecutionAndSuppressions (SELECT FOR UPDATE + INSERT in one tx) + // persistExecutionAndSuppressions (per-creator lock + read + INSERT in one tx) // which closes the TOCTOU race that a pre-tx read alone cannot prevent. creator := resolveCreatorUserID(session) creatorID := derefStringOrEmpty(creator) diff --git a/internal/api/handler_purchases_guards_test.go b/internal/api/handler_purchases_guards_test.go index 804332217..8684fdb1e 100644 --- a/internal/api/handler_purchases_guards_test.go +++ b/internal/api/handler_purchases_guards_test.go @@ -596,6 +596,14 @@ func TestBuildDuplicatePurchaseResponse(t *testing.T) { ex.NotificationSent = nil assert.Equal(t, false, buildDuplicatePurchaseResponse(ex)["email_sent"]) + + // A row that already left the approval queue never had an approval email. + for _, status := range []string{"approved", "running", "completed", "partially_completed"} { + ex.Status = status + resp = buildDuplicatePurchaseResponse(ex) + assert.NotContains(t, resp, "email_sent", status) + assert.Equal(t, status, resp["status"]) + } } // --- #647: capacity_percent consistency with scaled rec counts --- diff --git a/internal/api/handler_purchases_submit_dedupe_integration_test.go b/internal/api/handler_purchases_submit_dedupe_integration_test.go new file mode 100644 index 000000000..d921aa26e --- /dev/null +++ b/internal/api/handler_purchases_submit_dedupe_integration_test.go @@ -0,0 +1,184 @@ +//go:build integration + +package api + +import ( + "context" + "sync" + "testing" + "time" + + "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/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// insertBarrierStore holds each submit transaction at its INSERT until every +// participant has reached it (or the wait times out). That forces the +// interleaving MON-01 describes: both transactions have already run their +// duplicate check before either inserts. A guard that serializes the check +// (the per-creator lock) keeps the second transaction out of the barrier, so +// the first one times out and inserts alone. +type insertBarrierStore struct { + *config.PostgresStore + arrived chan struct{} + parties int + wait time.Duration +} + +func (s *insertBarrierStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, e *config.PurchaseExecution) error { + s.arrived <- struct{}{} + deadline := time.After(s.wait) + for len(s.arrived) < s.parties { + select { + case <-deadline: + return s.PostgresStore.SavePurchaseExecutionTx(ctx, tx, e) + case <-time.After(10 * time.Millisecond): + } + } + return s.PostgresStore.SavePurchaseExecutionTx(ctx, tx, e) +} + +func submitFixture(t *testing.T) ([]config.RecommendationRecord, string) { + t.Helper() + recs := []config.RecommendationRecord{{ + Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.large", + Count: 2, Term: 1, Payment: "no-upfront", + }} + return recs, purchaseIdempotencyKey("", recs, 100) +} + +func newSubmitExecution(t *testing.T, recs []config.RecommendationRecord) *config.PurchaseExecution { + t.Helper() + exec, _, err := newPendingExecution(&ExecutePurchaseRequest{Recommendations: recs, CapacityPercent: 100}, 0, 0) + require.NoError(t, err) + return exec +} + +func countSubmitRows(t *testing.T, pg *testhelpers.PostgresContainer) int { + t.Helper() + var n int + require.NoError(t, pg.DB.Pool().QueryRow(context.Background(), `SELECT count(*) FROM purchase_executions`).Scan(&n)) + return n +} + +func setupSubmitDB(t *testing.T) (*testhelpers.PostgresContainer, *config.PostgresStore) { + t.Helper() + ctx := context.Background() + pg, err := testhelpers.SetupPostgresContainer(ctx, t) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, pg.Cleanup(ctx)) }) + require.NoError(t, migrations.RunMigrations(ctx, pg.DB.Pool(), "../database/postgres/migrations", "", "")) + return pg, config.NewPostgresStore(pg.DB) +} + +// TestPersistExecution_ConcurrentIdenticalSubmitsCreateOneRow is MON-01 case 1: +// two identical submits (double click, client retry) must not both insert. +func TestPersistExecution_ConcurrentIdenticalSubmitsCreateOneRow(t *testing.T) { + pg, store := setupSubmitDB(t) + recs, key := submitFixture(t) + barrier := &insertBarrierStore{PostgresStore: store, arrived: make(chan struct{}, 2), parties: 2, wait: time.Second} + h := &Handler{config: barrier} + + var wg sync.WaitGroup + dups := make([]*config.PurchaseExecution, 2) + errs := make([]error, 2) + for i := range dups { + wg.Add(1) + go func() { + defer wg.Done() + dups[i], errs[i] = h.persistExecutionAndSuppressions(context.Background(), newSubmitExecution(t, recs), nil, "", key) + }() + } + wg.Wait() + + require.NoError(t, errs[0]) + require.NoError(t, errs[1]) + assert.Equal(t, 1, countSubmitRows(t, pg), "identical concurrent submits must create exactly one execution") + collapsed := 0 + for _, d := range dups { + if d != nil { + collapsed++ + } + } + assert.Equal(t, 1, collapsed, "exactly one submit must report the other as its duplicate") +} + +// TestPersistExecution_RetryAfterRowLeftPending is MON-01 case 2: a direct +// execute moves its row out of pending within seconds, and a retry arriving +// then must still collapse onto it instead of buying again. A row that bought +// nothing (failed, canceled) must not absorb a resubmit. +func TestPersistExecution_RetryAfterRowLeftPending(t *testing.T) { + pg, store := setupSubmitDB(t) + for _, tc := range []struct { + status string + wantDup bool + }{ + {"pending", true}, {"notified", true}, {"approved", true}, {"running", true}, + {"completed", true}, {"partially_completed", true}, + {config.StatusFailed, false}, {config.StatusCanceled, false}, {config.LegacyStatusCanceled, false}, {config.StatusExpired, false}, + } { + t.Run(tc.status, func(t *testing.T) { + recs, key := submitFixture(t) + h := &Handler{config: store} + ctx := context.Background() + _, err := pg.DB.Pool().Exec(ctx, `TRUNCATE purchase_executions CASCADE`) + require.NoError(t, err) + + first := newSubmitExecution(t, recs) + dup, err := h.persistExecutionAndSuppressions(ctx, first, nil, "", key) + require.NoError(t, err) + require.Nil(t, dup) + _, err = pg.DB.Pool().Exec(ctx, `UPDATE purchase_executions SET status = $1 WHERE execution_id = $2`, tc.status, first.ExecutionID) + require.NoError(t, err) + + dup, err = h.persistExecutionAndSuppressions(ctx, newSubmitExecution(t, recs), nil, "", key) + 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)) + } + }) + } +} + +// TestPersistExecution_DedupeIsPerCreator pins the creator filter in the +// store query: two users submitting identical recommendations each get their +// own execution, while the same user's resubmit collapses. +func TestPersistExecution_DedupeIsPerCreator(t *testing.T) { + pg, store := setupSubmitDB(t) + ctx := context.Background() + recs, _ := submitFixture(t) + h := &Handler{config: store} + + 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 + } + submit := func(creator string) *config.PurchaseExecution { + exec := newSubmitExecution(t, recs) + exec.CreatedByUserID = &creator + dup, err := h.persistExecutionAndSuppressions(ctx, exec, nil, creator, purchaseIdempotencyKey(creator, recs, 100)) + require.NoError(t, err) + return dup + } + alice, bob := newUser("alice@example.com"), newUser("bob@example.com") + + assert.Nil(t, submit(alice)) + assert.Nil(t, submit(bob), "another user's identical submit is not a duplicate") + assert.NotNil(t, submit(alice), "the same user's resubmit collapses") + assert.Equal(t, 2, countSubmitRows(t, pg)) +} diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 4043bf627..0fcf8bf3f 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -504,11 +504,18 @@ type StoreInterface interface { SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, execution *PurchaseExecution) error // GetPendingExecutionsTx is the tx-accepting variant of - // GetPendingExecutions. Used inside the executePurchase WithTx block - // so the duplicate-detection read and the new-execution insert are - // atomic under the same transaction, eliminating the TOCTOU race (#643). + // GetPendingExecutions. It row-locks the pending rows it returns; it does + // not serialize concurrent inserts (see ListRecentSubmitsTx). GetPendingExecutionsTx(ctx context.Context, tx pgx.Tx) ([]PurchaseExecution, error) + // ListRecentSubmitsTx takes a per-creator transaction-scoped lock and + // returns the creator's web-submitted executions scheduled at or after + // since that are pending, in flight or bought. executePurchase's + // duplicate guard reads it inside the WithTx block that inserts the new + // execution, so concurrent identical submits serialize and a retry + // that arrives after a direct execute left "pending" is still seen. + ListRecentSubmitsTx(ctx context.Context, tx pgx.Tx, creatorID string, since time.Time) ([]PurchaseExecution, error) + // WithTx opens a pgx transaction, runs fn, and commits on success or // rolls back on error. fn can call any *Tx method on the store to // participate in the transaction. Nested transactions are not diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 085c8be24..bb26dd37c 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -1630,6 +1630,54 @@ func (s *PostgresStore) GetPendingExecutionsTx(ctx context.Context, tx pgx.Tx) ( return scanExecutionRows(rows) } +// submitDedupeStatuses are the execution states that still stand for a +// commitment the submitter asked for: awaiting approval, being bought, or +// bought. A failed, expired or canceled row bought nothing, so a resubmit +// after one of those must create a fresh execution. +const submitDedupeStatuses = "'pending', 'notified', 'approved', 'running', 'completed', 'partially_completed'" + +// ListRecentSubmitsTx serializes concurrent submits by creatorID and returns +// that creator's web-submitted executions scheduled at or after since, in +// every state in submitDedupeStatuses. The transaction-scoped advisory lock is +// held until tx ends, so a second submit by the same creator waits here, then +// reads the first one's committed row and can collapse onto it. SELECT ... +// FOR UPDATE could not do this: it locks only rows that already exist, so two +// concurrent submits both saw nothing and both inserted. The status set is +// wider than pending/notified because a direct execute moves its row to +// approved/running/completed within seconds, and a retry arriving then must +// still be recognized. An empty creatorID matches rows with no creator. +func (s *PostgresStore) ListRecentSubmitsTx(ctx context.Context, tx pgx.Tx, creatorID string, since time.Time) ([]PurchaseExecution, error) { + if _, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", "purchase-submit:"+creatorID); err != nil { + return nil, fmt.Errorf("failed to lock purchase submits: %w", err) + } + var creator *string + if creatorID != "" { + creator = &creatorID + } + query := ` + SELECT plan_id, execution_id, status, step_number, scheduled_date, + notification_sent, approval_token_hash, recommendations, + total_upfront_cost, estimated_savings, completed_at, error, expires_at, + cloud_account_id, source, approved_by, COALESCE(canceled_by, cancelled_by) AS cancelled_by, capacity_percent, + created_by_user_id, retry_execution_id, retry_attempt_n, + approval_token_expires_at, + executed_by_user_id, executed_at, pre_approval_skip_reason, + idempotency_key, scheduled_execution_at + FROM purchase_executions + WHERE source = $3 + AND status IN (` + submitDedupeStatuses + `) + AND scheduled_date >= $1 + AND created_by_user_id IS NOT DISTINCT FROM $2::uuid + ORDER BY scheduled_date ASC + ` + rows, err := tx.Query(ctx, query, since, creator, common.PurchaseSourceWeb) + if err != nil { + return nil, fmt.Errorf("failed to query recent submits in tx: %w", err) + } + defer rows.Close() + return scanExecutionRows(rows) +} + // GetExecutionByID retrieves a purchase execution by execution ID. // Returns an error wrapping ErrNotFound when no row matches executionID so // callers can cleanly distinguish "not found" (errors.Is(err, ErrNotFound)) diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index c95ee0ebb..5adc426b7 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1532,6 +1532,26 @@ func (m *MockConfigStore) ListActiveSuppressions(ctx context.Context) ([]config. return v, args.Error(1) } +// ListRecentSubmitsTx mocks the ListRecentSubmitsTx operation. It falls back +// to GetPendingExecutions when no expectation is registered (as +// GetPendingExecutionsTx does) so tests that exercise the WithTx path keep +// receiving the pending list; the caller filters by creator and window. +func (m *MockConfigStore) ListRecentSubmitsTx(ctx context.Context, tx pgx.Tx, creatorID string, since time.Time) ([]config.PurchaseExecution, error) { + m.record("ListRecentSubmitsTx", ctx, tx, creatorID, since) + if !isExpected(&m.Mock, "ListRecentSubmitsTx") { + return m.GetPendingExecutions(ctx) + } + args := m.Called(ctx, tx, creatorID, since) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).([]config.PurchaseExecution) + if !ok { + panic(fmt.Sprintf("mock: expected []config.PurchaseExecution, got %T", args.Get(0))) + } + return v, args.Error(1) +} + // GetPendingExecutionsTx mocks the GetPendingExecutionsTx operation. // Falls back to GetPendingExecutions when no explicit expectation is registered // (same pattern as SavePurchaseExecutionTx) so existing tests that exercise diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 7308396ac..5a8789228 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -342,6 +342,9 @@ func (m *mockConfigStoreForHealth) ListActiveSuppressions(_ context.Context) ([] func (m *mockConfigStoreForHealth) SavePurchaseExecutionTx(ctx context.Context, _ pgx.Tx, e *config.PurchaseExecution) error { return m.SavePurchaseExecution(ctx, e) } +func (m *mockConfigStoreForHealth) ListRecentSubmitsTx(ctx context.Context, _ pgx.Tx, _ string, _ time.Time) ([]config.PurchaseExecution, error) { + return m.GetPendingExecutions(ctx) +} func (m *mockConfigStoreForHealth) GetPendingExecutionsTx(ctx context.Context, _ pgx.Tx) ([]config.PurchaseExecution, error) { return m.GetPendingExecutions(ctx) }