feat(purchase): atomic plan-step claim, shared suppressions and account checks

internal/analytics/collector_test.go

@@ -859,3 +859,15 @@ func (m *mockConfigStore) UpdateGlobalConfigAtomic(_ context.Context, apply func
 	}
 	return cfg, nil
 }
+
+func (m *mockConfigStore) ListExecutionsForPlanStepTx(_ context.Context, _ pgx.Tx, _ string, _ int) ([]config.PurchaseExecution, error) {
+	return nil, nil
+}
+
+func (m *mockConfigStore) SetExecutionRecommendationsIfEmptyTx(_ context.Context, _ pgx.Tx, _ string, _ []config.RecommendationRecord, _, _ float64) (bool, error) {
+	return false, nil
+}
+
+func (m *mockConfigStore) ExpireExecutionAtomic(_ context.Context, _ pgx.Tx, _ string) (bool, error) {
+	return false, nil
+}

internal/api/handler_history.go

@@ -13,6 +13,7 @@ import (
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/runtime"
 	"github.com/aws/aws-lambda-go/events"
+	"github.com/jackc/pgx/v5"
 )
 
 // History handlers.
@@ -287,7 +288,16 @@ func (h *Handler) expireStaleExecutions(staleExecs []config.PurchaseExecution) {
 func (h *Handler) expireStaleExecutionsSweep(staleExecs []config.PurchaseExecution) {
 	ctx := context.Background()
 	for i := range staleExecs {
-		_, err := h.config.TransitionExecutionStatus(ctx, staleExecs[i].ExecutionID, []string{"pending", "notified"}, "expired", nil)
+		// The suppressions go in the same transaction as the status flip: an
+		// expired step bought nothing, and leaving its rows would keep hiding
+		// recommendations nobody purchased (platform#631).
+		err := h.config.WithTx(ctx, func(tx pgx.Tx) error {
+			expired, expErr := h.config.ExpireExecutionAtomic(ctx, tx, staleExecs[i].ExecutionID)
+			if expErr != nil || !expired {
+				return expErr
+			}
+			return h.config.DeleteSuppressionsByExecutionTx(ctx, tx, staleExecs[i].ExecutionID)
+		})
 		if err != nil {
 			logging.Warnf("history: expire sweep for execution %s failed: %v", staleExecs[i].ExecutionID, err)
 		}

internal/api/handler_history_actionable_starvation_test.go

@@ -163,9 +163,9 @@ func TestHandler_getHistory_StaleSweepBoundedToDefaultListLimit(t *testing.T) {
 	stale := executionsWithStatus(150, "pending", time.Now().Add(-8*24*time.Hour), "stale")
 	store := newStarvationStore(ctx)
 	store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(stale, nil)
-	store.On("TransitionExecutionStatus", mock.Anything, mock.Anything, mock.Anything, "expired", mock.Anything).Return(nil, nil)
+	store.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, mock.Anything).Return(false, nil)
 
 	runStarvationHistory(t, ctx, store)
 
-	store.AssertNumberOfCalls(t, "TransitionExecutionStatus", config.DefaultListLimit)
+	store.AssertNumberOfCalls(t, "ExpireExecutionAtomic", config.DefaultListLimit)
 }

internal/api/handler_history_test.go

@@ -244,9 +244,9 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) {
 		// The goroutine uses context.Background(); context.Background() == ctx in
 		// this test, so the matcher fires correctly. The trailing mock.Anything
 		// matches the actor *string (nil for the system-initiated async expire).
-		mockStore.On("TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired", mock.Anything).
+		mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, staleID).
 			Run(waitForCall(done)).
-			Return(&expired, nil).Once()
+			Return(true, nil).Once()
 
 		mockAuth, req := adminHistoryReq(ctx)
 		handler := &Handler{auth: mockAuth, config: mockStore}
@@ -263,8 +263,8 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) {
 		}
 
 		// Exactly one Transition call, only for the stale row.
-		mockStore.AssertNumberOfCalls(t, "TransitionExecutionStatus", 1)
-		mockStore.AssertCalled(t, "TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired", mock.Anything)
+		mockStore.AssertNumberOfCalls(t, "ExpireExecutionAtomic", 1)
+		mockStore.AssertCalled(t, "ExpireExecutionAtomic", mock.Anything, mock.Anything, staleID)
 
 		historyResp := result.(HistoryResponse)
 		require.Len(t, historyResp.Purchases, 2, "both executions must render as history rows")
@@ -301,9 +301,9 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) {
 		mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).
 			Return([]config.PurchaseExecution{staleExec("notified")}, nil)
 		mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil)
-		mockStore.On("TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired", mock.Anything).
+		mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, staleID).
 			Run(waitForCall(done)).
-			Return(&expired, nil).Once()
+			Return(true, nil).Once()
 
 		mockAuth, req := adminHistoryReq(ctx)
 		handler := &Handler{auth: mockAuth, config: mockStore}
@@ -316,7 +316,7 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) {
 		case <-time.After(5 * time.Second):
 			t.Fatal("TransitionExecutionStatus goroutine did not fire within 5s")
 		}
-		mockStore.AssertNumberOfCalls(t, "TransitionExecutionStatus", 1)
+		mockStore.AssertNumberOfCalls(t, "ExpireExecutionAtomic", 1)
 
 		historyResp := result.(HistoryResponse)
 		require.Len(t, historyResp.Purchases, 1)
@@ -338,9 +338,9 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) {
 		mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).
 			Return([]config.PurchaseExecution{staleExec("pending")}, nil)
 		mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil)
-		mockStore.On("TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired", mock.Anything).
+		mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, staleID).
 			Run(waitForCall(done)).
-			Return(nil, errors.New("simulated store failure")).Once()
+			Return(false, errors.New("simulated store failure")).Once()
 
 		mockAuth, req := adminHistoryReq(ctx)
 		handler := &Handler{auth: mockAuth, config: mockStore}
@@ -356,7 +356,7 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) {
 		case <-time.After(5 * time.Second):
 			t.Fatal("TransitionExecutionStatus goroutine did not fire within 5s")
 		}
-		mockStore.AssertNumberOfCalls(t, "TransitionExecutionStatus", 1)
+		mockStore.AssertNumberOfCalls(t, "ExpireExecutionAtomic", 1)
 
 		historyResp := result.(HistoryResponse)
 		require.Len(t, historyResp.Purchases, 1, "transition error must not drop the row")
@@ -432,12 +432,12 @@ func TestHandler_getHistory_GetIsReadOnly(t *testing.T) {
 	mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil)
 	mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{stale}, nil)
 	mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil)
-	mockStore.On("TransitionExecutionStatus", mock.Anything, "stale-ro-exec", []string{"pending", "notified"}, "expired", mock.Anything).
+	mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, "stale-ro-exec").
 		Run(func(_ mock.Arguments) {
 			close(transitionCalled) // signal that the goroutine reached the transition
 			<-gate                  // block until the test releases it
 		}).
-		Return(nil, nil).Maybe()
+		Return(false, nil).Maybe()
 
 	mockAuth, req := adminHistoryReq(ctx)
 	handler := &Handler{auth: mockAuth, config: mockStore}
@@ -623,7 +623,6 @@ func TestHandler_expireStaleExecutionsAsync_SystemActorIsNil(t *testing.T) {
 	t.Cleanup(func() { mockStore.AssertExpectations(t) })
 
 	staleID := "actor-nil-stale-exec"
-	expired := config.PurchaseExecution{ExecutionID: staleID, Status: "expired"}
 	done := make(chan struct{})
 
 	mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil)
@@ -637,9 +636,7 @@ func TestHandler_expireStaleExecutionsAsync_SystemActorIsNil(t *testing.T) {
 	// System path: the async expire must pass nil actor so transitioned_by =
 	// NULL. The (*string)(nil) literal is the contract under test. The
 	// goroutine uses context.Background(); use mock.Anything for ctx.
-	mockStore.On("TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired",
-		(*string)(nil),
-	).Run(func(_ mock.Arguments) { close(done) }).Return(&expired, nil).Once()
+	mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, staleID).Run(func(_ mock.Arguments) { close(done) }).Return(true, nil).Once()
 
 	mockAuth, req := adminHistoryReq(ctx)
 	handler := &Handler{auth: mockAuth, config: mockStore}
@@ -1979,9 +1976,7 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
 		mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).
 			Return([]config.PurchaseExecution{staleExec()}, nil)
 		mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil)
-		mockStore.On("TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired",
-			(*string)(nil),
-		).Return(&expired, nil).Once()
+		mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, staleID).Return(true, nil).Once()
 
 		mockAuth, req := adminHistoryReq(ctx)
 		handler := &Handler{auth: mockAuth, config: mockStore}
@@ -1992,7 +1987,7 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
 		// No channel wait: on Lambda the transition must already have fired
 		// by the time getHistory returns. Pre-fix this fails because the
 		// sweep ran in a goroutine the frozen sandbox never resumes.
-		mockStore.AssertNumberOfCalls(t, "TransitionExecutionStatus", 1)
+		mockStore.AssertNumberOfCalls(t, "ExpireExecutionAtomic", 1)
 
 		historyResp := result.(HistoryResponse)
 		require.Len(t, historyResp.Purchases, 1)
@@ -2020,12 +2015,10 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
 		mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).
 			Return([]config.PurchaseExecution{staleExec()}, nil)
 		mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil)
-		mockStore.On("TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired",
-			(*string)(nil),
-		).Run(func(_ mock.Arguments) {
+		mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, staleID).Run(func(_ mock.Arguments) {
 			<-release
 			close(swept)
-		}).Return(&expired, nil).Once()
+		}).Return(true, nil).Once()
 
 		mockAuth, req := adminHistoryReq(ctx)
 		handler := &Handler{auth: mockAuth, config: mockStore}
@@ -2054,7 +2047,7 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
 		}
 
 		require.NoError(t, err)
-		mockStore.AssertNumberOfCalls(t, "TransitionExecutionStatus", 1)
+		mockStore.AssertNumberOfCalls(t, "ExpireExecutionAtomic", 1)
 		mockStore.AssertExpectations(t)
 		historyResp := result.(HistoryResponse)
 		require.Len(t, historyResp.Purchases, 1)
@@ -2098,3 +2091,26 @@ func TestSummarizePurchaseHistory_RevokedExcludedFromKPIs(t *testing.T) {
 	assert.InDelta(t, 120.0, summary.TotalAnnualSavings, 0.001,
 		"TotalAnnualSavings must exclude revoked rows")
 }
+
+func strPtr(s string) *string { return &s }
+
+// E3: the expiry sweep releases an expired step's suppressions in the same tx
+// as the status flip, and only when the flip won.
+func TestExpireStaleExecutionsSweep_ReleasesSuppressionsOnlyWhenExpired(t *testing.T) {
+	for name, won := range map[string]bool{"flip won": true, "row already moved": false} {
+		t.Run(name, func(t *testing.T) {
+			store := new(MockConfigStore)
+			store.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, "stale-1").Return(won, nil)
+			store.On("DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "stale-1").Return(nil).Maybe()
+			h := &Handler{config: store}
+
+			h.expireStaleExecutionsSweep([]config.PurchaseExecution{{ExecutionID: "stale-1", Status: "pending"}})
+
+			if won {
+				store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "stale-1")
+			} else {
+				store.AssertNotCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "stale-1")
+			}
+		})
+	}
+}

internal/api/handler_purchases.go

@@ -23,68 +23,6 @@ import (
 	"github.com/jackc/pgx/v5"
 )
 
-// buildSuppressions converts the recs on an executePurchase request into
-// purchase_suppressions rows, one per unique 6-tuple (account, provider,
-// service, region, resource_type, engine). Counts for duplicate tuples
-// within the same request are summed (defensive — the UI should collapse
-// them client-side, but the backend doesn't trust that). Providers with
-// a grace period of 0 (feature disabled) contribute no rows.
-func buildSuppressions(recs []config.RecommendationRecord, executionID string, cfg *config.GlobalConfig, now time.Time) []config.PurchaseSuppression {
-	// Aggregate count per 6-tuple before writing suppression rows so
-	// the UNIQUE(execution_id, ...6-tuple) constraint can't fire from
-	// a request that repeated the same tuple.
-	type key struct {
-		accountID, provider, service, region, resourceType, engine string
-	}
-	agg := map[key]int{}
-	order := []key{}
-	for _rvc := range recs {
-		rec := recs[_rvc]
-		if rec.Count <= 0 {
-			continue
-		}
-		accountID := ""
-		if rec.CloudAccountID != nil {
-			accountID = *rec.CloudAccountID
-		}
-		k := key{
-			accountID:    accountID,
-			provider:     rec.Provider,
-			service:      rec.Service,
-			region:       rec.Region,
-			resourceType: rec.ResourceType,
-			engine:       rec.Engine,
-		}
-		if _, seen := agg[k]; !seen {
-			order = append(order, k)
-		}
-		agg[k] += rec.Count
-	}
-
-	out := make([]config.PurchaseSuppression, 0, len(order))
-	for _, k := range order {
-		graceDays := config.DefaultGracePeriodDays
-		if cfg != nil {
-			graceDays = cfg.GracePeriodFor(k.provider)
-		}
-		if graceDays <= 0 {
-			continue // feature disabled for this provider
-		}
-		out = append(out, config.PurchaseSuppression{
-			ExecutionID:     executionID,
-			AccountID:       k.accountID,
-			Provider:        k.provider,
-			Service:         k.service,
-			Region:          k.region,
-			ResourceType:    k.resourceType,
-			Engine:          k.engine,
-			SuppressedCount: agg[k],
-			ExpiresAt:       now.Add(time.Duration(graceDays) * 24 * time.Hour),
-		})
-	}
-	return out
-}
-
 // plannedListStatuses are the execution statuses surfaced in the Scheduled
 // (Planned) Purchases list. It deliberately includes "paused" so a paused
 // execution stays VISIBLE with its badge instead of silently dropping out of
@@ -2197,7 +2135,7 @@ func (h *Handler) persistRetryExecution(ctx context.Context, failedExec *config.
 	if g, err := h.config.GetGlobalConfig(ctx); err == nil {
 		gracePeriodCfg = g
 	}
-	suppressions := buildSuppressions(failedExec.Recommendations, newExecutionID, gracePeriodCfg, time.Now())
+	suppressions := config.BuildSuppressions(failedExec.Recommendations, newExecutionID, gracePeriodCfg, time.Now())
 
 	if err := h.persistRetrySuccessorTx(ctx, failedExec, newExecution, suppressions); err != nil {
 		return nil, "", err
@@ -3048,7 +2986,7 @@ func (h *Handler) executePurchase(ctx context.Context, req *events.LambdaFunctio
 		gracePeriodCfg = g
 	}
 
-	suppressions := buildSuppressions(execReq.Recommendations, executionID, gracePeriodCfg, time.Now())
+	suppressions := config.BuildSuppressions(execReq.Recommendations, executionID, gracePeriodCfg, time.Now())
 	dupExec, err := h.persistExecutionAndSuppressions(ctx, execution, suppressions, creatorID, idempotencyKey)
 	if err != nil {
 		return nil, err

internal/api/handler_purchases_pause_claim_integration_test.go

@@ -101,7 +101,7 @@ func newPauseClaimFixture(t *testing.T) *pauseClaimFixture {
 	f.execution = &config.PurchaseExecution{
 		ExecutionID: uuid.NewString(), PlanID: plan.ID, CloudAccountID: &account.ID,
 		Status: "pending", StepNumber: 3, ScheduledDate: time.Now(), IdempotencyKey: uuid.NewString(),
-		CreatedByUserID: &creator, Recommendations: fanoutRecs(),
+		CreatedByUserID: &creator, Recommendations: fanoutRecsForAccount(account.ID),
 	}
 	require.NoError(t, f.store.SavePurchaseExecution(ctx, f.execution))
 	row, err := f.store.GetExecutionByID(ctx, f.execution.ExecutionID)

internal/api/handler_purchases_retry_fanout_test.go

@@ -63,6 +63,14 @@ func fanoutRecs() []config.RecommendationRecord {
 	}
 }
 
+// fanoutRecsForAccount is a per-account row's rec set, attributed to its own
+// account as the fan-out leaves it (platform#631).
+func fanoutRecsForAccount(accountID string) []config.RecommendationRecord {
+	recs := fanoutRecs()
+	recs[0].CloudAccountID = strPtr(accountID)
+	return recs
+}
+
 // fanoutRootRecs is a root step's rec set: one rec per plan account, because
 // the fan-out buys each account's own recs only (platform#631).
 func fanoutRootRecs() []config.RecommendationRecord {
@@ -251,7 +259,7 @@ func failedPerAccountRow(execID, accountID string) *config.PurchaseExecution {
 		CreatedByUserID: &creator,
 		CapacityPercent: 100,
 		Source:          common.PurchaseSourceWeb,
-		Recommendations: fanoutRecs(),
+		Recommendations: fanoutRecsForAccount(accountID),
 	}
 }
 

internal/api/handler_purchases_run_now_integration_test.go

@@ -54,6 +54,9 @@ func (f runNowFactory) CreateAndValidateProvider(context.Context, string, *provi
 func newRunNowFixture(t *testing.T, failOn int32, recs []config.RecommendationRecord) (*pauseClaimFixture, *runNowService) {
 	t.Helper()
 	f := newPauseClaimFixture(t)
+	for i := range recs {
+		recs[i].CloudAccountID = f.execution.CloudAccountID
+	}
 	f.execution.Recommendations = recs
 	require.NoError(t, f.store.SavePurchaseExecution(f.ctx, f.execution))
 	service := &runNowService{failOn: failOn}

internal/config/interfaces.go

@@ -81,6 +81,20 @@ type StoreInterface interface {
 	// transaction; without it the answer is advisory and two concurrent callers
 	// both see the step free.
 	OccupiedRampStepsInRangeTx(ctx context.Context, tx pgx.Tx, planID string, from, to int) ([]int, error)
+	// ListExecutionsForPlanStepTx returns EVERY execution row (any status,
+	// account or retry attempt) recorded for stepNumber of planID. The
+	// get-or-create of a plan step reads it under LockPurchasePlanTx so any
+	// existing row, settled or not, blocks minting a second root (platform#631).
+	ListExecutionsForPlanStepTx(ctx context.Context, tx pgx.Tx, planID string, stepNumber int) ([]PurchaseExecution, error)
+	// SetExecutionRecommendationsIfEmptyTx attaches resolved recommendations
+	// and their totals to a pending/notified execution that has none yet.
+	// Returns false when another resolver already attached them (or the row
+	// left pending/notified); the caller re-reads instead of overwriting.
+	SetExecutionRecommendationsIfEmptyTx(ctx context.Context, tx pgx.Tx, executionID string, recs []RecommendationRecord, upfront, savings float64) (bool, error)
+	// ExpireExecutionAtomic flips a pending/notified execution to expired in
+	// tx so its suppressions can be deleted in the same transaction. Returns
+	// false when the row had already left pending/notified.
+	ExpireExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string) (bool, error)
 	// UpdatePurchasePlanTx checks UpdatedAt and replaces it with the returned version.
 	// The caller must discard the transaction-local plan if the transaction rolls back.
 	UpdatePurchasePlanTx(ctx context.Context, tx pgx.Tx, plan *PurchasePlan) error