chore(api): remove dead submit-dedupe code and lock the retry-successor path

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

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

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

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