fix(api): make the purchase submit duplicate guard atomic

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)
 }

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)

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 ---

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))
+}

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

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))

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