From d80428d5eb28474ad254b700bc66b15d79de1fd6 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 9 Oct 2026 11:46:15 +0200 Subject: [PATCH 1/2] fix(api): make the purchase submit duplicate guard atomic The duplicate check in persistExecutionAndSuppressions used SELECT ... FOR UPDATE over pending/notified rows. FOR UPDATE cannot see another transaction's uncommitted insert, so two concurrent identical submits both inserted; and a retry that arrived after a direct execute moved its row to approved/running/completed saw nothing and bought the commitment again. Take a per-creator pg_advisory_xact_lock as the first statement of the submit transaction (ListRecentSubmitsTx), then read the creator's recent web executions in pending, notified, approved, running, completed and partially_completed, and insert in the same transaction. A concurrent identical submit waits, then collapses onto the committed row. Failed, cancelled and expired rows do not absorb a resubmit. A collapsed response no longer reports email_sent for a row that already left the approval queue, and its message no longer says "pending". Integration tests (real Postgres) force both submits through their duplicate check before either inserts, and cover every status. Refs #758 --- internal/analytics/collector_test.go | 3 + internal/api/handler_purchases.go | 32 +-- internal/api/handler_purchases_guards_test.go | 8 + ...urchases_submit_dedupe_integration_test.go | 184 ++++++++++++++++++ internal/config/interfaces.go | 13 +- internal/config/store_postgres.go | 48 +++++ internal/mocks/stores.go | 20 ++ internal/server/test_helpers_test.go | 3 + 8 files changed, 296 insertions(+), 15 deletions(-) create mode 100644 internal/api/handler_purchases_submit_dedupe_integration_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index cd3e89fc1..6c0393e4e 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -809,6 +809,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..2e4cc0a6e --- /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}, + {"failed", false}, {"cancelled", false}, {"expired", 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 009d662f5..3f60ec05d 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -500,11 +500,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 68f0bde0b..0482122fb 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 recognised. 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 a7caa7a01..59ea010bf 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1524,6 +1524,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 5944314b5..5b117af5e 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -338,6 +338,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) } From 4d134602971bcc088261b35a22c883d9e61c6662 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 9 Oct 2026 12:10:59 +0200 Subject: [PATCH 2/2] test(api): use the canonical canceled status in the submit dedupe test CI lint (misspell) rejected the literal legacy spelling. Use the status constants, which keeps both canceled and the legacy value covered, and fix "recognised" in a comment. Refs #758 --- .../api/handler_purchases_submit_dedupe_integration_test.go | 2 +- internal/config/store_postgres.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/api/handler_purchases_submit_dedupe_integration_test.go b/internal/api/handler_purchases_submit_dedupe_integration_test.go index 2e4cc0a6e..d921aa26e 100644 --- a/internal/api/handler_purchases_submit_dedupe_integration_test.go +++ b/internal/api/handler_purchases_submit_dedupe_integration_test.go @@ -120,7 +120,7 @@ func TestPersistExecution_RetryAfterRowLeftPending(t *testing.T) { }{ {"pending", true}, {"notified", true}, {"approved", true}, {"running", true}, {"completed", true}, {"partially_completed", true}, - {"failed", false}, {"cancelled", false}, {"expired", false}, + {config.StatusFailed, false}, {config.StatusCanceled, false}, {config.LegacyStatusCanceled, false}, {config.StatusExpired, false}, } { t.Run(tc.status, func(t *testing.T) { recs, key := submitFixture(t) diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index fe037c583..bb26dd37c 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -1645,7 +1645,7 @@ const submitDedupeStatuses = "'pending', 'notified', 'approved', 'running', 'com // 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 recognised. An empty creatorID matches rows with no creator. +// 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)