From f30184eb27bc2dc62ae7384fec6331dae12016e0 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Tue, 20 Oct 2026 15:35:04 +0100 Subject: [PATCH] chore(api): remove dead submit-dedupe code and lock the retry-successor path persistRetrySuccessorTx inserts a web-source pending row for the retrying user and the failed row's recommendations, the same fingerprint an identical fresh submit carries, but it took no per-creator submit lock and ran no duplicate check. A retry racing (or following) an identical submit by the same user left two approvable pending rows. The retry transaction now starts with ListRecentSubmitsTx, the same per-creator advisory lock at the same point as persistExecutionAndSuppressions, and after the link CAS refuses with a 409 naming the existing execution when the creator already has a live submit with the successor's fingerprint inside the idempotency window. The lock comes before the ladder tranche lock; the submit path takes only this lock and the reconciler never takes it, so no new lock-order cycle. Also removes findDuplicatePendingExecution and GetPendingExecutionsTx (interface, Postgres store, MockConfigStore and the server/analytics fakes), which have had no production caller since #761, and moves the remaining cases of TestFindDuplicatePendingExecution onto matchDuplicateInList and persistExecutionAndSuppressions. Closes #760 Co-Authored-By: Claude Opus 5.5 --- internal/analytics/collector_test.go | 3 - internal/api/handler_purchases.go | 151 ++++++++++-------- internal/api/handler_purchases_dedupe_test.go | 34 ++++ internal/api/handler_purchases_guards_test.go | 91 ----------- .../handler_purchases_retry_dedupe_test.go | 107 +++++++++++++ ...ases_retry_submit_race_integration_test.go | 149 +++++++++++++++++ internal/config/interfaces.go | 5 - internal/config/store_postgres.go | 31 +--- ...gres_created_by_system_integration_test.go | 4 +- internal/mocks/stores.go | 32 +--- internal/server/test_helpers_test.go | 3 - 11 files changed, 383 insertions(+), 227 deletions(-) create mode 100644 internal/api/handler_purchases_retry_dedupe_test.go create mode 100644 internal/api/handler_purchases_retry_submit_race_integration_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 2400b1ad9..acbb11fc0 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -873,9 +873,6 @@ func (m *mockConfigStore) RecoverReapedExecutionTx(ctx context.Context, _ pgx.Tx 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) -} func (m *mockConfigStore) WithTx(_ context.Context, fn func(tx pgx.Tx) error) error { return fn(nil) } func (m *mockConfigStore) GetLadderConfigs(_ context.Context) ([]config.LadderConfigDB, error) { diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go index d90c7282b..2a731f3ce 100644 --- a/internal/api/handler_purchases.go +++ b/internal/api/handler_purchases.go @@ -2524,47 +2524,102 @@ func (h *Handler) persistRetryExecution(ctx context.Context, failedExec *config. // matches zero rows and the tx rolls back (issue #220) instead of both // persisting an approvable successor and reverting whatever the other // writer changed on the original row. +// +// The successor is a web-source pending row for the retrying user, so it +// also goes through the submit dedupe (issue #760): see writeRetrySuccessorTx. func (h *Handler) persistRetrySuccessorTx(ctx context.Context, failedExec, newExecution *config.PurchaseExecution, suppressions []config.PurchaseSuppression) error { - var lostRace, settledTranche bool err := h.config.WithTx(ctx, func(tx pgx.Tx) error { - if err := h.guardLadderTrancheForRetry(ctx, tx, failedExec.IdempotencyKey); err != nil { - settledTranche = errors.Is(err, errRetryLadderTrancheSettled) - return err - } - if err := h.config.SavePurchaseExecutionTx(ctx, tx, newExecution); err != nil { - return err - } - linked, linkErr := h.config.LinkRetryExecutionAtomic(ctx, tx, failedExec.ExecutionID, newExecution.ExecutionID) - if linkErr != nil { - return linkErr - } - if !linked { - lostRace = true - return errRetryLinkRaceLost - } - for i := range suppressions { - if err := h.config.CreateSuppressionTx(ctx, tx, &suppressions[i]); err != nil { - return err - } - } - return nil + return h.writeRetrySuccessorTx(ctx, tx, failedExec, newExecution, suppressions) }) if err == nil { return nil } - if settledTranche { + var dup *retryDuplicateSubmitError + switch { + case errors.Is(err, errRetryLadderTrancheSettled): return NewClientErrorWithDetails(409, fmt.Sprintf("execution %s belongs to a ladder tranche that is already settled; retrying it could buy the same commitment twice", failedExec.ExecutionID), map[string]any{}) - } - if lostRace { + case errors.Is(err, errRetryLinkRaceLost): return NewClientErrorWithDetails(409, fmt.Sprintf("execution %s was already retried; act on its descendant instead", failedExec.ExecutionID), map[string]any{}) + case errors.As(err, &dup): + return NewClientErrorWithDetails(409, + fmt.Sprintf("an identical purchase was submitted moments ago as execution %s; act on it instead of retrying %s", dup.existingID, failedExec.ExecutionID), + map[string]any{"duplicate_execution_id": dup.existingID}) } return fmt.Errorf("failed to save retry execution: %w", err) } +// writeRetrySuccessorTx is persistRetrySuccessorTx's transaction body. +// +// Its first statement takes the retrying user's per-creator submit lock +// (ListRecentSubmitsTx), the same lock at the same point as +// persistExecutionAndSuppressions, so a retry and an identical fresh submit +// by that user serialize instead of both inserting a pending row. Taking it +// before the ladder tranche lock keeps one order everywhere: the submit path +// takes only this lock, and the reconciler never takes it. The rows read +// under the lock are checked for a duplicate of the successor only after the +// link CAS, so a lost retry race still reports "already retried". +func (h *Handler) writeRetrySuccessorTx(ctx context.Context, tx pgx.Tx, failedExec, newExecution *config.PurchaseExecution, suppressions []config.PurchaseSuppression) error { + creatorID := derefStringOrEmpty(newExecution.CreatedByUserID) + now := time.Now() + recent, listErr := h.config.ListRecentSubmitsTx(ctx, tx, creatorID, now.Add(-purchaseIdempotencyWindow)) + if listErr != nil { + return listErr + } + if err := h.guardLadderTrancheForRetry(ctx, tx, failedExec.IdempotencyKey); err != nil { + return err + } + if err := h.config.SavePurchaseExecutionTx(ctx, tx, newExecution); err != nil { + return err + } + linked, linkErr := h.config.LinkRetryExecutionAtomic(ctx, tx, failedExec.ExecutionID, newExecution.ExecutionID) + if linkErr != nil { + return linkErr + } + if !linked { + return errRetryLinkRaceLost + } + if err := checkRetryDuplicateSubmit(recent, creatorID, newExecution, now); err != nil { + return err + } + for i := range suppressions { + if err := h.config.CreateSuppressionTx(ctx, tx, &suppressions[i]); err != nil { + return err + } + } + return nil +} + +// checkRetryDuplicateSubmit returns a *retryDuplicateSubmitError when recent +// (the creator's live web submits, read under the submit lock) already holds +// an execution with the successor's submit fingerprint. The successor's own +// row is not in recent: it was read before the successor was inserted. A +// fingerprint that cannot be derived fails the retry rather than skipping +// the check, since a skipped check is a possible double purchase. +func checkRetryDuplicateSubmit(recent []config.PurchaseExecution, creatorID string, successor *config.PurchaseExecution, now time.Time) error { + key, err := purchaseIdempotencyKey(creatorID, successor.Recommendations, successor.CapacityPercent) + if err != nil { + return fmt.Errorf("derive retry idempotency key: %w", err) + } + if dup := matchDuplicateInList(recent, creatorID, key, now); dup != nil { + return &retryDuplicateSubmitError{existingID: dup.ExecutionID} + } + return nil +} + +// retryDuplicateSubmitError rolls back persistRetrySuccessorTx's tx when the +// retrying user already has an identical live submit inside the idempotency +// window, so the retry would mint a second approvable row for the same +// purchase. persistRetrySuccessorTx translates it into a 409. +type retryDuplicateSubmitError struct{ existingID string } + +func (e *retryDuplicateSubmitError) Error() string { + return "identical purchase already submitted as execution " + e.existingID +} + // errRetryLinkRaceLost is returned from persistRetrySuccessorTx's tx // closure to force a rollback (via WithTx's error path) when // LinkRetryExecutionAtomic matches zero rows -- a concurrent retry already @@ -2574,8 +2629,9 @@ func (h *Handler) persistRetrySuccessorTx(ctx context.Context, failedExec, newEx var errRetryLinkRaceLost = errors.New("retry link race lost") // guardLadderTrancheForRetry locks the ladder tranche named by the execution's -// idempotency key (if any) before anything else in the retry tx, the same order -// the reconciler uses (tranche, then execution rows), and refuses when the +// idempotency key (if any) before any execution row in the retry tx (only the +// per-creator submit advisory lock comes first), the same order the reconciler +// uses (tranche, then execution rows), and refuses when the // tranche is settled. A failed tranche no longer nets, so an approvable // successor would buy the same slice a second time. func (h *Handler) guardLadderTrancheForRetry(ctx context.Context, tx pgx.Tx, idempotencyKey string) error { @@ -3278,47 +3334,6 @@ func recDetailsAvailabilityZone(r *config.RecommendationRecord) (string, error) return "", nil } -// findDuplicatePendingExecution returns an existing pending/notified execution -// whose idempotency fingerprint matches key and whose creation falls inside -// purchaseIdempotencyWindow of now, or (nil, nil) when there is no duplicate. -// It recomputes each candidate's key from its persisted recs + creator + -// capacity rather than relying on a stored column, so no schema change is -// needed. Restricted to web-sourced executions to avoid colliding with -// scheduler/CLI rows. A lookup error is non-fatal to the caller's decision -// (returned so the caller can log and proceed with a fresh execution). -func (h *Handler) findDuplicatePendingExecution(ctx context.Context, creatorID, key string, now time.Time) (*config.PurchaseExecution, error) { - pending, err := h.config.GetPendingExecutions(ctx) - if err != nil { - return nil, err - } - cutoff := now.Add(-purchaseIdempotencyWindow) - for i := range pending { - ex := &pending[i] - if ex.Source != common.PurchaseSourceWeb { - continue - } - if ex.ScheduledDate.Before(cutoff) { - continue - } - exCreator := "" - if ex.CreatedByUserID != nil { - exCreator = *ex.CreatedByUserID - } - if exCreator != creatorID { - continue - } - exKey, err := purchaseIdempotencyKey(exCreator, ex.Recommendations, ex.CapacityPercent) - if err != nil { - logging.Errorf("purchase idempotency: skipping pending execution %s, cannot derive its key: %v", ex.ExecutionID, err) - continue - } - if exKey == key { - return ex, nil - } - } - return nil, nil -} - // buildDuplicatePurchaseResponse returns the executePurchase response body for // a submit that collapsed onto an existing pending execution (#644). It points // at the original row so the client lands on the same approvable execution diff --git a/internal/api/handler_purchases_dedupe_test.go b/internal/api/handler_purchases_dedupe_test.go index 33b1a74cf..e8dba8f56 100644 --- a/internal/api/handler_purchases_dedupe_test.go +++ b/internal/api/handler_purchases_dedupe_test.go @@ -1,6 +1,7 @@ package api import ( + "encoding/json" "testing" "time" @@ -64,3 +65,36 @@ func TestMatchDuplicateInList(t *testing.T) { }) } } + +// TestMatchDuplicateInList_ReturnsTheMatchingRow covers what the removed +// findDuplicatePendingExecution test pinned beyond the clause table: the scan +// returns the matching row itself (not just "some row") from a mixed list, +// and a candidate whose key cannot be derived is skipped without ending the +// scan (#760). +func TestMatchDuplicateInList_ReturnsTheMatchingRow(t *testing.T) { + t.Parallel() + now := time.Now() + creator := "11111111-1111-1111-1111-111111111111" + recs := []config.RecommendationRecord{validRec()} + key := mustPurchaseIdempotencyKey(creator, recs, 100) + cand := func(id string, mut func(e *config.PurchaseExecution)) config.PurchaseExecution { + e := config.PurchaseExecution{ + ExecutionID: id, Status: "pending", Source: common.PurchaseSourceWeb, + ScheduledDate: now.Add(-10 * time.Second), CreatedByUserID: &creator, + Recommendations: append([]config.RecommendationRecord(nil), recs...), CapacityPercent: 100, + } + mut(&e) + return e + } + list := []config.PurchaseExecution{ + cand("other-count", func(e *config.PurchaseExecution) { e.Recommendations[0].Count = 99 }), + cand("undecodable", func(e *config.PurchaseExecution) { e.Recommendations[0].Details = json.RawMessage(`{"not":`) }), + cand("cli", func(e *config.PurchaseExecution) { e.Source = common.PurchaseSourceCLI }), + cand("the-dup", func(*config.PurchaseExecution) {}), + } + got := matchDuplicateInList(list, creator, key, now) + if assert.NotNil(t, got) { + assert.Equal(t, "the-dup", got.ExecutionID) + } + assert.Nil(t, matchDuplicateInList(list[:3], creator, key, now), "no candidate matches without the-dup") +} diff --git a/internal/api/handler_purchases_guards_test.go b/internal/api/handler_purchases_guards_test.go index 36e1e3954..1d6e8a8dd 100644 --- a/internal/api/handler_purchases_guards_test.go +++ b/internal/api/handler_purchases_guards_test.go @@ -2,7 +2,6 @@ package api import ( "context" - "errors" "testing" "time" @@ -11,7 +10,6 @@ import ( "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" - "github.com/LeanerCloud/cloud-commitments-go/pkg/common" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" ) @@ -487,95 +485,6 @@ func TestPurchaseIdempotencyKey_StableAndDiscriminating(t *testing.T) { assert.NotEqual(t, mustPurchaseIdempotencyKey("user-1", acctA, 100), mustPurchaseIdempotencyKey("user-1", acctB, 100), "account") } -func TestFindDuplicatePendingExecution(t *testing.T) { - ctx := context.Background() - now := time.Now() - creator := "11111111-1111-1111-1111-111111111111" - recs := []config.RecommendationRecord{validRec()} - key := mustPurchaseIdempotencyKey(creator, recs, 100) - - makeExec := func(id string, age time.Duration, src string, c *string, capacity int) config.PurchaseExecution { - // Copy recs so a subtest that mutates exec.Recommendations does not - // corrupt the shared slice used to compute `key` above. - recsCopy := append([]config.RecommendationRecord(nil), recs...) - return config.PurchaseExecution{ - ExecutionID: id, - Status: "pending", - Source: src, - ScheduledDate: now.Add(-age), - Recommendations: recsCopy, - CreatedByUserID: c, - CapacityPercent: capacity, - } - } - - t.Run("matching recent web execution is a duplicate", func(t *testing.T) { - store := new(MockConfigStore) - store.On("GetPendingExecutions", ctx).Return([]config.PurchaseExecution{ - makeExec("exec-dup", 30*time.Second, common.PurchaseSourceWeb, &creator, 100), - }, nil) - h := &Handler{config: store} - dup, err := h.findDuplicatePendingExecution(ctx, creator, key, now) - require.NoError(t, err) - require.NotNil(t, dup) - assert.Equal(t, "exec-dup", dup.ExecutionID) - }) - - t.Run("outside the window is not a duplicate", func(t *testing.T) { - store := new(MockConfigStore) - store.On("GetPendingExecutions", ctx).Return([]config.PurchaseExecution{ - makeExec("exec-old", purchaseIdempotencyWindow+time.Minute, common.PurchaseSourceWeb, &creator, 100), - }, nil) - h := &Handler{config: store} - dup, err := h.findDuplicatePendingExecution(ctx, creator, key, now) - require.NoError(t, err) - assert.Nil(t, dup) - }) - - t.Run("different creator is not a duplicate", func(t *testing.T) { - other := "22222222-2222-2222-2222-222222222222" - store := new(MockConfigStore) - store.On("GetPendingExecutions", ctx).Return([]config.PurchaseExecution{ - makeExec("exec-other", 10*time.Second, common.PurchaseSourceWeb, &other, 100), - }, nil) - h := &Handler{config: store} - dup, err := h.findDuplicatePendingExecution(ctx, creator, key, now) - require.NoError(t, err) - assert.Nil(t, dup) - }) - - t.Run("non-web source is skipped", func(t *testing.T) { - store := new(MockConfigStore) - store.On("GetPendingExecutions", ctx).Return([]config.PurchaseExecution{ - makeExec("exec-cli", 10*time.Second, "cudly-cli", &creator, 100), - }, nil) - h := &Handler{config: store} - dup, err := h.findDuplicatePendingExecution(ctx, creator, key, now) - require.NoError(t, err) - assert.Nil(t, dup) - }) - - t.Run("distinct rec set is not a duplicate", func(t *testing.T) { - differing := makeExec("exec-diff", 10*time.Second, common.PurchaseSourceWeb, &creator, 100) - differing.Recommendations[0].Count = 99 - store := new(MockConfigStore) - store.On("GetPendingExecutions", ctx).Return([]config.PurchaseExecution{differing}, nil) - h := &Handler{config: store} - dup, err := h.findDuplicatePendingExecution(ctx, creator, key, now) - require.NoError(t, err) - assert.Nil(t, dup) - }) - - t.Run("lookup error is surfaced, not swallowed", func(t *testing.T) { - store := new(MockConfigStore) - store.On("GetPendingExecutions", ctx).Return(nil, errors.New("db down")) - h := &Handler{config: store} - dup, err := h.findDuplicatePendingExecution(ctx, creator, key, now) - require.Error(t, err) - assert.Nil(t, dup) - }) -} - func TestBuildDuplicatePurchaseResponse(t *testing.T) { t.Parallel() sent := time.Now() diff --git a/internal/api/handler_purchases_retry_dedupe_test.go b/internal/api/handler_purchases_retry_dedupe_test.go new file mode 100644 index 000000000..6bc58e7cd --- /dev/null +++ b/internal/api/handler_purchases_retry_dedupe_test.go @@ -0,0 +1,107 @@ +package api + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// retryDedupeFixture is a failed execution owned by the retrying caller plus +// a handler whose store accepts the successor insert and the link CAS. +func retryDedupeFixture(t *testing.T) (*config.PurchaseExecution, *Handler, *MockConfigStore) { + t.Helper() + creator := retryCallerID + failed := &config.PurchaseExecution{ + ExecutionID: retryExecID, Status: "failed", CreatedByUserID: &creator, CapacityPercent: 100, + Recommendations: []config.RecommendationRecord{{Provider: "aws", Service: "rds", Term: 1, Count: 1, UpfrontCost: 100}}, + } + handler, mockConfig, _ := buildSessionRetryHandler(failed, &Session{UserID: retryCallerID}, false, true) + mockConfig.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil) + mockConfig.On("LinkRetryExecutionAtomic", mock.Anything, mock.Anything, failed.ExecutionID, mock.AnythingOfType("string")).Return(true, nil) + return failed, handler, mockConfig +} + +// liveSubmit is a web-source row the retrying caller submitted seconds ago +// with the given recommendations. +func liveSubmit(id string, recs []config.RecommendationRecord) config.PurchaseExecution { + creator := retryCallerID + return config.PurchaseExecution{ + ExecutionID: id, Status: "pending", Source: common.PurchaseSourceWeb, CreatedByUserID: &creator, + ScheduledDate: time.Now().Add(-10 * time.Second), CapacityPercent: 100, + Recommendations: append([]config.RecommendationRecord(nil), recs...), + } +} + +// TestHandler_retryPurchase_RefusesIdenticalLiveSubmit (issue #760): the retry +// reads the caller's recent submits under the per-creator submit lock and +// refuses with a 409 naming the existing row when one has the successor's +// fingerprint, writing no suppressions and sending no approval email. The +// real lock and race are covered by TestRetrySuccessorRacesFreshSubmit. +func TestHandler_retryPurchase_RefusesIdenticalLiveSubmit(t *testing.T) { + failed, handler, mockConfig := retryDedupeFixture(t) + notifier := &recordingEmailNotifier{} + handler.emailNotifier = notifier + mockConfig.On("ListRecentSubmitsTx", mock.Anything, mock.Anything, retryCallerID, mock.AnythingOfType("time.Time")). + Return([]config.PurchaseExecution{liveSubmit("fresh-submit", failed.Recommendations)}, nil) + + result, err := handler.retryPurchase(context.Background(), sessionRetryReq(), retryExecID) + + require.Error(t, err) + assert.Nil(t, result) + ce, ok := IsClientError(err) + require.True(t, ok, "a duplicate must surface as a client error, not a 500: %v", err) + assert.Equal(t, 409, ce.code) + assert.Contains(t, ce.message, "identical purchase") + assert.Equal(t, "fresh-submit", ce.Details()["duplicate_execution_id"]) + mockConfig.AssertNotCalled(t, "CreateSuppressionTx") + assert.Empty(t, notifier.captured.ApprovalToken, "no approval email for a refused retry") +} + +// A recent submit with a different fingerprint does not block the retry. +func TestHandler_retryPurchase_AllowsDifferentLiveSubmit(t *testing.T) { + failed, handler, mockConfig := retryDedupeFixture(t) + other := append([]config.RecommendationRecord(nil), failed.Recommendations...) + other[0].Count = 7 + mockConfig.On("ListRecentSubmitsTx", mock.Anything, mock.Anything, retryCallerID, mock.AnythingOfType("time.Time")). + Return([]config.PurchaseExecution{liveSubmit("other-submit", other)}, nil) + + _, err := handler.retryPurchase(context.Background(), sessionRetryReq(), retryExecID) + require.NoError(t, err) +} + +// A failure to read the recent submits fails the retry before any write. +func TestHandler_retryPurchase_RecentSubmitsErrorFailsRetry(t *testing.T) { + _, handler, mockConfig := retryDedupeFixture(t) + mockConfig.On("ListRecentSubmitsTx", mock.Anything, mock.Anything, retryCallerID, mock.AnythingOfType("time.Time")). + Return(nil, errors.New("db down")) + + _, err := handler.retryPurchase(context.Background(), sessionRetryReq(), retryExecID) + require.ErrorContains(t, err, "db down") + _, isClient := IsClientError(err) + assert.False(t, isClient) + mockConfig.AssertNotCalled(t, "SavePurchaseExecution") + mockConfig.AssertNotCalled(t, "LinkRetryExecutionAtomic") +} + +// TestPersistExecution_RecentSubmitsErrorIsSurfaced moves the "lookup error is +// surfaced, not swallowed" case of the removed findDuplicatePendingExecution +// test onto the live submit path (#760): a failed dedupe read must fail the +// submit, not fall through to an unchecked insert. +func TestPersistExecution_RecentSubmitsErrorIsSurfaced(t *testing.T) { + mockConfig := new(MockConfigStore) + mockConfig.On("ListRecentSubmitsTx", mock.Anything, mock.Anything, "", mock.AnythingOfType("time.Time")). + Return(nil, errors.New("db down")) + h := &Handler{config: mockConfig} + + dup, err := h.persistExecutionAndSuppressions(context.Background(), &config.PurchaseExecution{ExecutionID: "new"}, nil, "", "key") + require.ErrorContains(t, err, "db down") + assert.Nil(t, dup) + mockConfig.AssertNotCalled(t, "SavePurchaseExecution") +} diff --git a/internal/api/handler_purchases_retry_submit_race_integration_test.go b/internal/api/handler_purchases_retry_submit_race_integration_test.go new file mode 100644 index 000000000..07711112d --- /dev/null +++ b/internal/api/handler_purchases_retry_submit_race_integration_test.go @@ -0,0 +1,149 @@ +//go:build integration + +package api + +import ( + "context" + "errors" + "math/rand/v2" + "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/testhelpers" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// retrySubmitRound seeds one failed web execution owned by creator, then runs +// a retry of it and a fresh identical submit by the same creator +// concurrently. The side named by retryLeads starts first and the other one +// starts lag later. It returns both results once both have finished: the +// submit's duplicate (nil when it inserted), the retry error and the submit +// error. +func retrySubmitRound(t *testing.T, pg *testhelpers.PostgresContainer, h *Handler, creator string, retryLeads bool, lag time.Duration) (submitDup *config.PurchaseExecution, retryErr, submitErr error) { + t.Helper() + ctx := context.Background() + _, err := pg.DB.Pool().Exec(ctx, `TRUNCATE purchase_executions CASCADE`) + require.NoError(t, err) + recs, _ := submitFixture(t) + failed := &config.PurchaseExecution{ + ExecutionID: uuid.NewString(), IdempotencyKey: uuid.NewString(), Status: config.StatusFailed, + ScheduledDate: time.Now().Add(-time.Hour), Recommendations: recs, Source: common.PurchaseSourceWeb, + CapacityPercent: 100, CreatedByUserID: &creator, + } + require.NoError(t, h.config.SavePurchaseExecution(ctx, failed)) + + fresh := newSubmitExecution(t, recs) + fresh.CreatedByUserID = &creator + key := mustPurchaseIdempotencyKey(creator, recs, 100) + + retryLag, submitLag := lag, time.Duration(0) + if retryLeads { + retryLag, submitLag = 0, lag + } + start := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + <-start + time.Sleep(retryLag) + _, _, retryErr = h.persistRetryExecution(ctx, failed, &Session{UserID: creator}, 0, 0) + }() + go func() { + defer wg.Done() + <-start + time.Sleep(submitLag) + submitDup, submitErr = h.persistExecutionAndSuppressions(ctx, fresh, nil, creator, key) + }() + close(start) + wg.Wait() + return submitDup, retryErr, submitErr +} + +func requireNoDeadlock(t *testing.T, err error) { + t.Helper() + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + require.NotEqual(t, "40P01", pgErr.Code, "deadlock between the retry and the submit: %v", err) + } +} + +func countPendingRows(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 WHERE status = 'pending'`).Scan(&n)) + return n +} + +// TestRetrySuccessorRacesFreshSubmit is issue #760: a retry of a failed +// execution writes a web-source pending row for the retrying user and the +// same recommendations, so it must serialize with that user's identical +// submits on the same per-creator lock and dedupe against them. Whichever +// takes the lock first wins; the other collapses (submit) or is refused with +// a 409 (retry). Every round must end with exactly one pending row and no +// deadlock (40P01). Rounds alternate which side starts first. +// +// The forced case holds each transaction at its INSERT until both have +// arrived (or 300ms pass), so without the lock both pass their checks before +// either inserts; its 50ms head start makes each side win half the rounds. +// The free case races with a sub-millisecond random lag. +func TestRetrySuccessorRacesFreshSubmit(t *testing.T) { + pg, store := setupSubmitDB(t) + ctx := context.Background() + creator := uuid.NewString() + _, err := pg.DB.Pool().Exec(ctx, ` + INSERT INTO users (id, email, password_hash, salt, active, group_ids, created_at, updated_at) + SELECT $1, 'retrier@example.com', 'x', 'y', true, ARRAY[g.id], now(), now() + FROM groups g ORDER BY g.name LIMIT 1`, creator) + require.NoError(t, err) + + for _, tc := range []struct { + name string + h *Handler + rounds int + lag func() time.Duration + bothOrders bool + }{ + {"forced interleaving", &Handler{config: &insertBarrierStore{PostgresStore: store, arrived: make(chan struct{}, 2), parties: 2, wait: 300 * time.Millisecond}}, + 8, func() time.Duration { return 50 * time.Millisecond }, true}, + {"free race", &Handler{config: store}, 100, func() time.Duration { return time.Duration(rand.IntN(500)) * time.Microsecond }, false}, + } { + t.Run(tc.name, func(t *testing.T) { + retryWon := 0 + for round := 0; round < tc.rounds; round++ { + if b, ok := tc.h.config.(*insertBarrierStore); ok { + for len(b.arrived) > 0 { + <-b.arrived + } + } + submitDup, retryErr, submitErr := retrySubmitRound(t, pg, tc.h, creator, round%2 == 0, tc.lag()) + requireNoDeadlock(t, retryErr) + requireNoDeadlock(t, submitErr) + require.NoError(t, submitErr, "round %d", round) + require.Equal(t, 1, countPendingRows(t, pg), "round %d: a retry racing an identical submit must leave exactly one pending row", round) + if retryErr == nil { + require.NotNil(t, submitDup, "round %d: the retry won, so the submit must collapse onto its successor", round) + retryWon++ + continue + } + ce, ok := IsClientError(retryErr) + require.True(t, ok, "round %d: unexpected retry error %v", round, retryErr) + assert.Equal(t, 409, ce.code, "round %d", round) + assert.Contains(t, ce.message, "identical purchase", "round %d", round) + assert.Nil(t, submitDup, "round %d: the submit won, so it must not report a duplicate", round) + } + t.Logf("%d rounds: retry won %d, submit won %d", tc.rounds, retryWon, tc.rounds-retryWon) + if tc.bothOrders { + assert.Positive(t, retryWon, "no round exercised the retry-first order") + assert.Less(t, retryWon, tc.rounds, "no round exercised the submit-first order") + } + }) + } +} diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index f2e627890..5db2641bb 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -618,11 +618,6 @@ type StoreInterface interface { SaveExecutionIfStatusTx(ctx context.Context, tx pgx.Tx, execution *PurchaseExecution, expected []string) error RecoverReapedExecutionTx(ctx context.Context, tx pgx.Tx, execution *PurchaseExecution) error - // GetPendingExecutionsTx is the tx-accepting variant of - // 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 diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index c90c71ab5..6334e53af 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -1753,35 +1753,6 @@ func (s *PostgresStore) GetPendingExecutions(ctx context.Context) ([]PurchaseExe return s.queryExecutions(ctx, query) } -// GetPendingExecutionsTx is the tx-accepting variant of GetPendingExecutions. -// Running the read inside the same transaction as the subsequent insert makes -// duplicate-detection and execution creation atomic, closing the TOCTOU race -// in executePurchase (issue #643). -func (s *PostgresStore) GetPendingExecutionsTx(ctx context.Context, tx pgx.Tx) ([]PurchaseExecution, error) { - const 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, no_submission, created_by_system - FROM purchase_executions - WHERE status IN ('pending', 'notified') - AND (expires_at IS NULL OR expires_at > NOW()) - ORDER BY scheduled_date ASC - LIMIT 1000 - FOR UPDATE - ` - rows, err := tx.Query(ctx, query) - if err != nil { - return nil, fmt.Errorf("failed to query pending executions in tx: %w", err) - } - defer rows.Close() - return scanExecutionRows(rows) -} - // planStepProviderLockPrefix namespaces the per-provider plan-step advisory lock. const planStepProviderLockPrefix = "cudly:planstep:" @@ -2051,7 +2022,7 @@ func applyNullTimesToExecution(exec *PurchaseExecution, notifSent, completedAt, } // scanExecutionRows scans a pgx.Rows cursor into a slice of PurchaseExecution. -// It is used by queryExecutions (pool query) and GetPendingExecutionsTx (tx +// It is used by queryExecutions (pool query) and ListRecentSubmitsTx (tx // query) so the scan logic lives in one place. func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) { executions := make([]PurchaseExecution, 0) diff --git a/internal/config/store_postgres_created_by_system_integration_test.go b/internal/config/store_postgres_created_by_system_integration_test.go index 0324f2cd0..87df52c86 100644 --- a/internal/config/store_postgres_created_by_system_integration_test.go +++ b/internal/config/store_postgres_created_by_system_integration_test.go @@ -55,9 +55,7 @@ func TestIntegration_PurchaseExecution_CreatedBySystem(t *testing.T) { got, err = store.GetExecutionsByStatuses(ctx, []string{"pending"}, 100) assertFlag("GetExecutionsByStatuses", got, err) require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { - rows, txErr := store.GetPendingExecutionsTx(ctx, tx) - assertFlag("GetPendingExecutionsTx", rows, txErr) - rows, txErr = store.ListExecutionsForPlanStepTx(ctx, tx, plan.ID, 1) + rows, txErr := store.ListExecutionsForPlanStepTx(ctx, tx, plan.ID, 1) assertFlag("ListExecutionsForPlanStepTx", rows, txErr) _, txErr = store.ListPlanCommittedTx(ctx, tx, plan.ID) require.NoError(t, txErr, "ListPlanCommittedTx") diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 38b64862f..2c762e417 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1791,12 +1791,17 @@ func (m *MockConfigStore) ListActiveSuppressions(ctx context.Context) ([]config. } // 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. +// to GetPendingExecutions when no expectation is registered so tests that +// exercise the WithTx path keep receiving the pending list; the caller +// filters by creator and window. With neither registered it returns no rows: +// retry tests (which now pass through the submit dedupe, issue #760) need no +// stub unless they exercise a duplicate. 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") { + if !isExpected(&m.Mock, "GetPendingExecutions") { + return nil, nil + } return m.GetPendingExecutions(ctx) } args := m.Called(ctx, tx, creatorID, since) @@ -1810,27 +1815,6 @@ func (m *MockConfigStore) ListRecentSubmitsTx(ctx context.Context, tx pgx.Tx, cr 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 -// the WithTx path transparently get the same pending-execution list. -func (m *MockConfigStore) GetPendingExecutionsTx(ctx context.Context, tx pgx.Tx) ([]config.PurchaseExecution, error) { - m.record("GetPendingExecutionsTx", ctx, tx) - if !isExpected(&m.Mock, "GetPendingExecutionsTx") { - // Recorded under both names; see UpdatePurchasePlanTx. - return m.GetPendingExecutions(ctx) - } - args := m.Called(ctx, tx) - 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) -} - func (m *MockConfigStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, execution *config.PurchaseExecution) error { m.record("SavePurchaseExecutionTx", ctx, tx, execution) if !isExpected(&m.Mock, "SavePurchaseExecutionTx") { diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index f58bee1df..1ee15ec31 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -402,9 +402,6 @@ func (m *mockConfigStoreForHealth) RecoverReapedExecutionTx(ctx context.Context, 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) -} func (m *mockConfigStoreForHealth) WithTx(_ context.Context, fn func(tx pgx.Tx) error) error { return fn(nil) }