fix(purchases): atomic CAS cancel to close TOCTOU with concurrent approve (closes #671)

internal/analytics/collector_test.go

@@ -212,6 +212,10 @@ func (m *mockConfigStore) TransitionExecutionStatus(ctx context.Context, executi
 	return nil, nil
 }
 
+func (m *mockConfigStore) CancelExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string, cancelledBy *string) (bool, string, error) {
+	return false, "", nil
+}
+
 func (m *mockConfigStore) SaveRIExchangeRecord(ctx context.Context, record *config.RIExchangeRecord) error {
 	return nil
 }

internal/api/handler_purchases.go

@@ -520,24 +520,34 @@ func (h *Handler) cancelPurchaseViaSession(ctx context.Context, req *events.Lamb
 		return nil, err
 	}
 
-	// Flip status + clear suppressions + stamp CancelledBy in one tx.
-	// An optimistic-locking guard inside the tx (status IN
-	// ('pending','notified')) prevents a concurrent approval from
-	// landing on top of us — if the status drifted, the UPDATE 0-rows
-	// the row count and we 409 cleanly without rolling back the entire
-	// flow into an inconsistent state.
-	execution.Status = "cancelled"
+	// Atomically flip status from pending/notified to cancelled + clear
+	// suppressions in one tx. CancelExecutionAtomic issues a conditional
+	// UPDATE WHERE status IN ('pending','notified'), so a concurrent approve
+	// that has already transitioned the row to 'approved' causes zero rows
+	// to be affected and we return a 409 with the current status rather
+	// than silently overwriting an approved purchase.
+	var cancelledBy *string
 	if session.Email != "" {
-		actor := session.Email
-		execution.CancelledBy = &actor
+		e := session.Email
+		cancelledBy = &e
 	}
+	var cancelled bool
+	var currentStatus string
 	if err := h.config.WithTx(ctx, func(tx pgx.Tx) error {
-		if err := h.config.SavePurchaseExecutionTx(ctx, tx, execution); err != nil {
+		var err error
+		cancelled, currentStatus, err = h.config.CancelExecutionAtomic(ctx, tx, execution.ExecutionID, cancelledBy)
+		if err != nil {
 			return err
 		}
+		if !cancelled {
+			return nil
+		}
 		return h.config.DeleteSuppressionsByExecutionTx(ctx, tx, execution.ExecutionID)
 	}); err != nil {
-		return nil, NewClientError(409, fmt.Sprintf("execution %s cannot be cancelled: %v", execution.ExecutionID, err))
+		return nil, fmt.Errorf("cancel execution %s: %w", execution.ExecutionID, err)
+	}
+	if !cancelled {
+		return nil, NewClientError(409, fmt.Sprintf("execution %s cannot be cancelled: a concurrent operation already transitioned it to %q", execution.ExecutionID, currentStatus))
 	}
 
 	return map[string]string{"status": "cancelled"}, nil

internal/api/handler_purchases_test.go

@@ -1588,47 +1588,48 @@ func sessionCancelReq() *events.LambdaFunctionURLRequest {
 
 // runSessionCancelAllowed asserts the success path of the session-authed
 // branch given a permission-matrix cell that should be allowed. The
-// cancel commits in a single tx (SavePurchaseExecutionTx +
-// DeleteSuppressionsByExecutionTx via WithTx); the mock store's WithTx
-// default forwards fn(nil) and SavePurchaseExecutionTx default routes
-// through SavePurchaseExecution, which we wire here. The suppression
-// delete returns nil by default so we don't need to register it.
+// cancel commits in a single tx via CancelExecutionAtomic +
+// DeleteSuppressionsByExecutionTx; the mock store's WithTx default
+// forwards fn(nil) and CancelExecutionAtomic default returns
+// (true, "cancelled", nil) when no explicit expectation is registered.
 //
-// Captures the saved execution so the caller can assert the audit-stamp
-// invariants — primarily that CancelledBy is set to session.Email when
-// the session has a non-empty email. cancelPurchase relies on this stamp
-// for History UI attribution; if SavePurchaseExecution stops being
-// called with the email-bearing copy the matrix tests would otherwise
-// silently regress.
+// Asserts the audit-stamp invariant: when session.Email is non-empty
+// the cancelledBy pointer passed to CancelExecutionAtomic must carry
+// that email so the DB column is stamped correctly for History UI
+// attribution.
 func runSessionCancelAllowed(t *testing.T, exec *config.PurchaseExecution, session *Session, hasAny, hasOwn bool) {
 	t.Helper()
 	handler, mockConfig, mockAuth := buildSessionCancelHandler(exec, session, hasAny, hasOwn)
-	var saved *config.PurchaseExecution
-	mockConfig.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")).
+
+	// Capture the cancelledBy pointer passed to CancelExecutionAtomic
+	// so we can assert attribution was stamped correctly.
+	var capturedCancelledBy *string
+	mockConfig.On("CancelExecutionAtomic", mock.Anything, mock.Anything, cancelExecID, mock.Anything).
 		Run(func(args mock.Arguments) {
-			saved = args.Get(1).(*config.PurchaseExecution)
+			if v, ok := args.Get(3).(*string); ok {
+				capturedCancelledBy = v
+			}
 		}).
+		Return(true, "cancelled", nil)
+	// When cancel succeeds the transaction must also clean up suppressions.
+	mockConfig.On("DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, cancelExecID).
 		Return(nil)
 
 	result, err := handler.cancelPurchase(context.Background(), sessionCancelReq(), cancelExecID, "")
 	require.NoError(t, err)
 	assert.Equal(t, "cancelled", result.(map[string]string)["status"])
-	// Status flip + suppression cleanup are paired in one tx — the mock
-	// only sees the un-tx variants because of how MockConfigStore wires
-	// SavePurchaseExecutionTx → SavePurchaseExecution. Asserting the
-	// un-tx call ran is enough for the matrix tests; the atomicity
-	// itself is exercised by the live integration tests.
-	mockConfig.AssertCalled(t, "SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution"))
-	require.NotNil(t, saved, "SavePurchaseExecution should have captured the execution")
-	assert.Equal(t, "cancelled", saved.Status)
+	// Verify the atomic cancel was called — this is the primary guard against
+	// regressions that skip the conditional UPDATE.
+	mockConfig.AssertCalled(t, "CancelExecutionAtomic", mock.Anything, mock.Anything, cancelExecID, mock.Anything)
+	// Verify suppression cleanup ran within the same transaction.
+	mockConfig.AssertCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, cancelExecID)
 	if session != nil && session.Email != "" {
-		require.NotNil(t, saved.CancelledBy, "CancelledBy must be stamped when session has an email")
-		assert.Equal(t, session.Email, *saved.CancelledBy, "CancelledBy must equal session.Email for audit attribution")
+		require.NotNil(t, capturedCancelledBy, "cancelledBy must be stamped when session has an email")
+		assert.Equal(t, session.Email, *capturedCancelledBy, "cancelledBy must equal session.Email for audit attribution")
 	}
 	// Verify the session-auth boundary actually fired — without this a
 	// regression that bypassed ValidateSession (or stopped consulting
-	// HasPermissionAPI for non-admins) would silently still pass the
-	// status/audit assertions above.
+	// HasPermissionAPI for non-admins) would silently still pass.
 	mockAuth.AssertExpectations(t)
 }
 
@@ -1776,6 +1777,40 @@ func TestHandler_cancelPurchase_Session_AllowsEachCancelableStatus(t *testing.T)
 	}
 }
 
+// TestHandler_cancelPurchase_Session_RaceWithApprove is the regression
+// guard for issue #671 on the session-authed cancel path. When a concurrent
+// approve transitions the execution out of pending/notified before the
+// conditional UPDATE runs, CancelExecutionAtomic returns
+// (false, "approved", nil) and the handler must 409 with the racing status
+// rather than silently overwriting the approved row.
+func TestHandler_cancelPurchase_Session_RaceWithApprove(t *testing.T) {
+	creator := cancelCallerID
+	exec := &config.PurchaseExecution{
+		ExecutionID:     cancelExecID,
+		Status:          "pending", // status at fetch time
+		CreatedByUserID: &creator,
+	}
+	session := &Session{UserID: cancelCallerID, Role: "admin", Email: "admin@example.com"}
+
+	handler, mockConfig, mockAuth := buildSessionCancelHandler(exec, session, false, false)
+	// Simulate concurrent approve winning between IsCancelable check and
+	// the conditional UPDATE inside the tx.
+	mockConfig.On("CancelExecutionAtomic", mock.Anything, mock.Anything, cancelExecID, mock.Anything).
+		Return(false, "approved", nil)
+
+	_, err := handler.cancelPurchase(context.Background(), sessionCancelReq(), cancelExecID, "")
+	require.Error(t, err)
+	var ce *clientError
+	require.ErrorAs(t, err, &ce)
+	assert.Equal(t, 409, ce.code)
+	assert.Contains(t, ce.message, "approved", "409 body must surface the racing status")
+	assert.Contains(t, ce.message, "concurrent", "409 body must mention the concurrent operation")
+	// Suppression cleanup must NOT have been called because the atomic
+	// UPDATE returned zero rows — the approve path owns the execution now.
+	mockConfig.AssertNotCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, mock.Anything)
+	mockAuth.AssertExpectations(t)
+}
+
 func TestHandler_cancelPurchase_Session_LegacyNullCreator_NonAdminRejected(t *testing.T) {
 	// Pre-migration row: created_by_user_id is NULL. cancel-own can't
 	// match a NULL creator, so a non-admin must be rejected. The email
@@ -1848,12 +1883,17 @@ func TestHandler_cancelPurchase_DeepLink_AdminBypassesContactEmailGate(t *testin
 	session := &Session{UserID: cancelCallerID, Role: "admin", Email: "admin@example.com"}
 
 	handler, mockConfig, mockAuth := buildSessionCancelHandler(exec, session, false, false)
-	var saved *config.PurchaseExecution
-	mockConfig.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")).
+
+	// Capture cancelledBy to verify the audit-stamp is passed to the
+	// atomic UPDATE.
+	var capturedCancelledBy *string
+	mockConfig.On("CancelExecutionAtomic", mock.Anything, mock.Anything, cancelExecID, mock.Anything).
 		Run(func(args mock.Arguments) {
-			saved = args.Get(1).(*config.PurchaseExecution)
+			if v, ok := args.Get(3).(*string); ok {
+				capturedCancelledBy = v
+			}
 		}).
-		Return(nil)
+		Return(true, "cancelled", nil)
 
 	// Token IS present in the URL — the deep-link flow always sends one.
 	// The fix's whole point is that the admin session takes the
@@ -1862,13 +1902,11 @@ func TestHandler_cancelPurchase_DeepLink_AdminBypassesContactEmailGate(t *testin
 	require.NoError(t, err, "admin clicking Cancel from notification email must succeed even when no contact_email is configured")
 	assert.Equal(t, "cancelled", result.(map[string]string)["status"])
 
-	require.NotNil(t, saved, "session-authed branch must commit the status flip")
-	assert.Equal(t, "cancelled", saved.Status)
-	require.NotNil(t, saved.CancelledBy, "session-authed branch must stamp CancelledBy")
-	assert.Equal(t, session.Email, *saved.CancelledBy)
+	require.NotNil(t, capturedCancelledBy, "session-authed branch must stamp cancelledBy")
+	assert.Equal(t, session.Email, *capturedCancelledBy)
 
 	// Critical security assertion: the token branch's contact_email gate
-	// (authorizeApprovalAction → GetGlobalConfig → resolveApprovalRecipients)
+	// (authorizeApprovalAction -> GetGlobalConfig -> resolveApprovalRecipients)
 	// was NOT consulted. If a regression re-routed admins through the
 	// token path, GetGlobalConfig would fire because the gate fetches
 	// the global notification email; asserting it didn't is the cleanest
@@ -1895,7 +1933,9 @@ func TestHandler_cancelPurchase_DeepLink_CancelOwnBypassesContactEmailGate(t *te
 	session := &Session{UserID: cancelCallerID, Role: "user", Email: "u1@example.com"}
 
 	handler, mockConfig, mockAuth := buildSessionCancelHandler(exec, session, false /*hasAny*/, true /*hasOwn*/)
-	mockConfig.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil)
+	// CancelExecutionAtomic is called by the session-authed branch.
+	mockConfig.On("CancelExecutionAtomic", mock.Anything, mock.Anything, cancelExecID, mock.Anything).
+		Return(true, "cancelled", nil)
 
 	result, err := handler.cancelPurchase(context.Background(), sessionCancelReq(), cancelExecID, "deep-link-token")
 	require.NoError(t, err)

internal/api/mocks_test.go

@@ -548,6 +548,17 @@ func (m *MockConfigStore) ListActiveSuppressions(ctx context.Context) ([]config.
 	return args.Get(0).([]config.PurchaseSuppression), args.Error(1)
 }
 
+func (m *MockConfigStore) CancelExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string, cancelledBy *string) (bool, string, error) {
+	if !m.isExpected("CancelExecutionAtomic") {
+		// Default: succeed, returning "cancelled". Tests that exercise the
+		// race (zero-rows) path register an explicit expectation that
+		// returns (false, <racing status>, nil).
+		return true, "cancelled", nil
+	}
+	args := m.Called(ctx, tx, executionID, cancelledBy)
+	return args.Bool(0), args.String(1), args.Error(2)
+}
+
 func (m *MockConfigStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, execution *config.PurchaseExecution) error {
 	if !m.isExpected("SavePurchaseExecutionTx") {
 		// Default to calling SavePurchaseExecution so tests that only

internal/config/interfaces.go

@@ -65,6 +65,13 @@ type StoreInterface interface {
 	ListPendingExecutionIDsForAccount(ctx context.Context, accountID string) ([]string, error)
 	CleanupOldExecutions(ctx context.Context, retentionDays int) (int64, error)
 	TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string) (*PurchaseExecution, error)
+	// CancelExecutionAtomic atomically flips status from pending/notified to
+	// cancelled, setting cancelled_by. Returns (true, "cancelled", nil) on
+	// success and (false, currentStatus, nil) when zero rows were affected
+	// (the execution had already been approved or otherwise transitioned).
+	// Must be called inside a WithTx block so the suppression cleanup and
+	// the status flip commit atomically.
+	CancelExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string, cancelledBy *string) (cancelled bool, currentStatus string, err error)
 
 	// Purchase history
 	SavePurchaseHistory(ctx context.Context, record *PurchaseHistoryRecord) error

internal/config/store_postgres.go

@@ -802,6 +802,63 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution
 	return &records[0], nil
 }
 
+// CancelExecutionAtomic atomically transitions an execution from
+// pending or notified to cancelled, setting cancelled_by to the supplied
+// actor (NULL when actor is nil). The UPDATE is conditional on
+// status IN ('pending','notified') so a concurrent approve that has
+// already transitioned the row to 'approved' causes zero rows to be
+// affected and the method returns (false, currentStatus, nil) with the
+// live status fetched via a follow-up SELECT. Returns (true, "cancelled",
+// nil) on success and (false, "", err) on a real DB error.
+//
+// Callers must run the suppression cleanup in the same transaction; use
+// the WithTx + DeleteSuppressionsByExecutionTx pairing at the call site
+// exactly as the old SavePurchaseExecutionTx path did, except now the
+// status guard is inside the UPDATE rather than checked optimistically
+// before entering the tx.
+func (s *PostgresStore) CancelExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string, cancelledBy *string) (cancelled bool, currentStatus string, err error) {
+	q := `
+		UPDATE purchase_executions
+		   SET status       = 'cancelled',
+		       cancelled_by = $2,
+		       updated_at   = NOW()
+		 WHERE execution_id = $1
+		   AND status IN ('pending', 'notified')
+		RETURNING status
+	`
+	rows, err := tx.Query(ctx, q, executionID, cancelledBy)
+	if err != nil {
+		return false, "", fmt.Errorf("failed to cancel execution: %w", err)
+	}
+	defer rows.Close()
+
+	if rows.Next() {
+		var st string
+		if scanErr := rows.Scan(&st); scanErr != nil {
+			return false, "", fmt.Errorf("failed to scan cancel result: %w", scanErr)
+		}
+		if rowsErr := rows.Err(); rowsErr != nil {
+			return false, "", fmt.Errorf("failed to iterate cancel result: %w", rowsErr)
+		}
+		return true, st, nil
+	}
+	if rowsErr := rows.Err(); rowsErr != nil {
+		return false, "", fmt.Errorf("failed to iterate cancel result: %w", rowsErr)
+	}
+
+	// Zero rows affected: execution either does not exist or has already
+	// transitioned out of pending/notified. Surface the current status so
+	// callers can return a meaningful 409 body.
+	existing, existErr := s.GetExecutionByID(ctx, executionID)
+	if existErr != nil {
+		return false, "", fmt.Errorf("execution not found or db error: %w", existErr)
+	}
+	if existing == nil {
+		return false, "", fmt.Errorf("execution not found: %s", executionID)
+	}
+	return false, existing.Status, nil
+}
+
 // GetExecutionsByStatuses returns executions whose Status is any of the
 // supplied values, newest-first, capped at `limit`. Used by the History
 // handler to merge pending/failed/expired rows alongside completed purchases

internal/mocks/stores.go

@@ -117,6 +117,12 @@ func (m *MockConfigStore) TransitionExecutionStatus(ctx context.Context, executi
 	return args.Get(0).(*config.PurchaseExecution), args.Error(1)
 }
 
+// CancelExecutionAtomic mocks the CancelExecutionAtomic operation.
+func (m *MockConfigStore) CancelExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string, cancelledBy *string) (bool, string, error) {
+	args := m.Called(ctx, tx, executionID, cancelledBy)
+	return args.Bool(0), args.String(1), args.Error(2)
+}
+
 // GetPendingExecutions mocks the GetPendingExecutions operation
 func (m *MockConfigStore) GetPendingExecutions(ctx context.Context) ([]config.PurchaseExecution, error) {
 	args := m.Called(ctx)

internal/purchase/approvals.go

@@ -133,31 +133,53 @@ func (m *Manager) ApproveAndExecute(ctx context.Context, executionID, actor stri
 // caller (HTTP path: authorizeApprovalAction; SQS path:
 // verifyAsyncApprovalActor) before reaching here. Same empty-actor
 // rationale as ApproveExecution.
+//
+// Concurrency: CancelExecutionAtomic uses a conditional UPDATE WHERE
+// status IN ('pending','notified') so a concurrent approve that wins
+// the race causes zero rows to be affected and the caller receives a
+// clean error with the current status rather than silently overwriting
+// the approved row. This is the token/email-link cancel analogue of
+// the atomic guard TransitionExecutionStatus provides for ApproveAndExecute.
 func (m *Manager) CancelExecution(ctx context.Context, executionID, token, actor string) error {
 	logging.Infof("Cancelling execution: %s", executionID)
 
-	execution, err := m.loadCancelableExecution(ctx, executionID, token)
-	if err != nil {
+	if _, err := m.loadCancelableExecution(ctx, executionID, token); err != nil {
 		return err
 	}
 
-	// Update status + attribution — see ApproveExecution for the empty-actor
-	// nil-vs-empty-string rationale. Paired with DeleteSuppressionsByExecution
-	// in the same transaction so the status flip and the un-suppression
-	// commit atomically — a crash between the two would otherwise leave the
-	// rec-list hiding capacity the user already cancelled.
-	execution.Status = "cancelled"
+	// Build the nullable cancelled_by pointer — see ApproveExecution for
+	// the nil-vs-empty-string rationale.
+	var cancelledBy *string
 	if actor != "" {
 		a := actor
-		execution.CancelledBy = &a
+		cancelledBy = &a
 	}
+
+	// Atomic conditional UPDATE + suppression cleanup in one transaction.
+	// CancelExecutionAtomic flips status only when status IN
+	// ('pending','notified') so a concurrent approve that has already
+	// transitioned the row causes zero rows affected and we surface a 409.
+	var cancelled bool
+	var currentStatus string
 	if err := m.config.WithTx(ctx, func(tx pgx.Tx) error {
-		if err := m.config.SavePurchaseExecutionTx(ctx, tx, execution); err != nil {
+		var err error
+		cancelled, currentStatus, err = m.config.CancelExecutionAtomic(ctx, tx, executionID, cancelledBy)
+		if err != nil {
 			return err
 		}
+		if !cancelled {
+			// Row already transitioned (concurrent approve/cancel won the
+			// race). Return early without touching suppressions — the other
+			// operation owns the execution state now.
+			return nil
+		}
 		return m.config.DeleteSuppressionsByExecutionTx(ctx, tx, executionID)
 	}); err != nil {
-		return fmt.Errorf("failed to save execution: %w", err)
+		return fmt.Errorf("failed to cancel execution: %w", err)
+	}
+
+	if !cancelled {
+		return fmt.Errorf("execution %s cannot be cancelled: concurrent operation already transitioned it to %q", executionID, currentStatus)
 	}
 
 	logging.Infof("Execution %s cancelled", executionID)