From d879dbed000b2fc0deb790497ba17ee7cce68ce2 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 9 Oct 2026 18:31:37 +0200 Subject: [PATCH 1/5] fix(purchase): scope plan fan-out recommendations to each account Each plan account now buys only its own recommendations and its row carries its own totals. A step with no recommendations, or a rec not attributed to one of the plan's accounts, fails loudly instead of completing or being dropped. Refs #631. Co-Authored-By: Claude Sonnet 5.5 --- .../handler_purchases_retry_fanout_test.go | 14 +++- internal/purchase/account_scope.go | 63 +++++++++++++++ internal/purchase/account_scope_test.go | 79 +++++++++++++++++++ internal/purchase/execution.go | 17 +++- internal/purchase/execution_test.go | 16 +++- .../purchase/money_path_regression_test.go | 12 ++- .../ramp_step_progress_integration_test.go | 15 +++- 7 files changed, 204 insertions(+), 12 deletions(-) create mode 100644 internal/purchase/account_scope.go create mode 100644 internal/purchase/account_scope_test.go diff --git a/internal/api/handler_purchases_retry_fanout_test.go b/internal/api/handler_purchases_retry_fanout_test.go index bd584086..35cdd884 100644 --- a/internal/api/handler_purchases_retry_fanout_test.go +++ b/internal/api/handler_purchases_retry_fanout_test.go @@ -63,6 +63,18 @@ func fanoutRecs() []config.RecommendationRecord { } } +// 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 { + var recs []config.RecommendationRecord + for _, id := range []string{fanoutAcctA, fanoutAcctB, fanoutAcctC} { + rec := fanoutRecs()[0] + rec.CloudAccountID = strPtr(id) + recs = append(recs, rec) + } + return recs +} + // fanoutCredStore resolves a static AWS access-key blob so per-account // credential resolution succeeds without touching the cloud. type fanoutCredStore struct { @@ -329,7 +341,7 @@ func TestRetryOfFailedRootRowStillFansOutToEveryAccount(t *testing.T) { CreatedByUserID: &creator, CapacityPercent: 100, Source: common.PurchaseSourceWeb, - Recommendations: fanoutRecs(), + Recommendations: fanoutRootRecs(), } successor := retryFailedRow(t, failedRoot) diff --git a/internal/purchase/account_scope.go b/internal/purchase/account_scope.go new file mode 100644 index 00000000..d2bcbe65 --- /dev/null +++ b/internal/purchase/account_scope.go @@ -0,0 +1,63 @@ +package purchase + +import ( + "fmt" + "strings" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// scopeExecutionsByAccount splits a plan step's recommendations across the +// plan's accounts so each account buys only its own (platform#631). The +// returned executions are per-account copies of base whose Recommendations +// and money totals cover only that account's recs; accounts with no recs of +// their own are omitted. +// +// It fails loudly rather than dropping anything: a rec with no account, or an +// account outside the plan, would otherwise be silently unbought (or, before +// this scoping existed, bought by every account). An empty step returns +// ErrPlanStepNoRecommendations so it cannot complete and advance the ramp. +func scopeExecutionsByAccount(base *config.PurchaseExecution, accounts []config.CloudAccount) (map[string]*config.PurchaseExecution, error) { + if len(base.Recommendations) == 0 { + return nil, fmt.Errorf("execution %s: %w", base.ExecutionID, ErrPlanStepNoRecommendations) + } + planAccounts := make(map[string]bool, len(accounts)) + for i := range accounts { + planAccounts[accounts[i].ID] = true + } + byAccount := make(map[string][]config.RecommendationRecord, len(accounts)) + var stray []string + for i := range base.Recommendations { + rec := base.Recommendations[i] + if rec.CloudAccountID == nil || !planAccounts[*rec.CloudAccountID] { + stray = append(stray, describeScopeRec(&rec)) + continue + } + byAccount[*rec.CloudAccountID] = append(byAccount[*rec.CloudAccountID], rec) + } + if len(stray) > 0 { + return nil, fmt.Errorf("execution %s: %d recommendation(s) are not attributed to one of the plan's accounts and cannot be bought: %s", + base.ExecutionID, len(stray), strings.Join(stray, ", ")) + } + + scoped := make(map[string]*config.PurchaseExecution, len(byAccount)) + for id, recs := range byAccount { + e := *base + e.Recommendations = recs + e.TotalUpfrontCost, e.EstimatedSavings = 0, 0 + for i := range recs { + e.TotalUpfrontCost += recs[i].UpfrontCost + e.EstimatedSavings += recs[i].Savings + } + scoped[id] = &e + } + return scoped, nil +} + +func describeScopeRec(rec *config.RecommendationRecord) string { + acct := "no account" + if rec.CloudAccountID != nil { + acct = "account " + *rec.CloudAccountID + } + return fmt.Sprintf("%s/%s %s (%s)", rec.Provider, rec.Service, rec.ResourceType, acct) +} diff --git a/internal/purchase/account_scope_test.go b/internal/purchase/account_scope_test.go new file mode 100644 index 00000000..b719695b --- /dev/null +++ b/internal/purchase/account_scope_test.go @@ -0,0 +1,79 @@ +package purchase + +import ( + "context" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func withAccount(rec config.RecommendationRecord, id string) config.RecommendationRecord { + rec.CloudAccountID = &id + return rec +} + +func scopedTestRec(accountID string) config.RecommendationRecord { + return withAccount(config.RecommendationRecord{ + Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", + Count: 1, UpfrontCost: 300, Savings: 75, Selected: true, + }, accountID) +} + +func scopeFixture() (*config.PurchaseExecution, []config.CloudAccount) { + a, b := scopedTestRec("acct-A"), scopedTestRec("acct-B") + b.ResourceType, b.UpfrontCost, b.Savings = "c5.large", 100, 20 + return &config.PurchaseExecution{ + ExecutionID: "root", PlanID: "plan-x", StepNumber: 1, + TotalUpfrontCost: 400, EstimatedSavings: 95, + Recommendations: []config.RecommendationRecord{a, b}, + }, []config.CloudAccount{{ID: "acct-A"}, {ID: "acct-B"}} +} + +// Each account gets only its own recs, and its row carries its own totals +// (not the step's, which would show N times the money in History). +func TestScopeExecutionsByAccount_OwnRecsAndTotals(t *testing.T) { + base, accounts := scopeFixture() + scoped, err := scopeExecutionsByAccount(base, accounts) + require.NoError(t, err) + require.Len(t, scoped, 2) + assert.Len(t, scoped["acct-A"].Recommendations, 1) + assert.Equal(t, "m5.large", scoped["acct-A"].Recommendations[0].ResourceType) + assert.Equal(t, 300.0, scoped["acct-A"].TotalUpfrontCost) + assert.Equal(t, 75.0, scoped["acct-A"].EstimatedSavings) + assert.Equal(t, 100.0, scoped["acct-B"].TotalUpfrontCost) + assert.Equal(t, 20.0, scoped["acct-B"].EstimatedSavings) + assert.Len(t, base.Recommendations, 2, "base execution must be untouched") +} + +// A rec with no account, or one outside the plan, fails the step loudly. +func TestScopeExecutionsByAccount_StrayRecFailsLoudly(t *testing.T) { + for name, stray := range map[string]config.RecommendationRecord{ + "nil account": {Provider: "azure", Service: "compute", ResourceType: "D2"}, + "foreign account": scopedTestRec("acct-other"), + } { + t.Run(name, func(t *testing.T) { + base, accounts := scopeFixture() + base.Recommendations = append(base.Recommendations, stray) + scoped, err := scopeExecutionsByAccount(base, accounts) + require.Error(t, err) + assert.Nil(t, scoped) + assert.Contains(t, err.Error(), "not attributed to one of the plan's accounts") + }) + } +} + +// executeMultiAccount: an empty step must fail with ErrPlanStepNoRecommendations +// (nil would be recorded as completed and advance the ramp), buying nothing. +func TestExecuteMultiAccount_EmptyStepFailsAndBuysNothing(t *testing.T) { + base, accounts := scopeFixture() + base.Recommendations = nil + store := new(MockConfigStore) + m := &Manager{config: store} + + err := m.executeMultiAccount(context.Background(), base, &config.PurchasePlan{ID: "plan-x"}, accounts) + + require.ErrorIs(t, err, ErrPlanStepNoRecommendations) + store.AssertNotCalled(t, "SavePurchaseExecution") +} diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 20d6f6b9..f5d0d44f 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -191,8 +191,21 @@ var errAllAccountsFailed = errors.New("multi-account execution: all accounts fai // callers ACK rather than redeliver, issue #1014); and errAllAccountsFailed // (wrapping the per-account errors) when no account committed anything. func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.PurchaseExecution, plan *config.PurchasePlan, accounts []config.CloudAccount) error { - results := execution.RunForAccountsWithConcurrency(ctx, accounts, func(ctx context.Context, account config.CloudAccount) (bool, error) { - return m.executeForAccount(ctx, baseExec, plan, account) + // Checked before any account runs: every account would otherwise buy the + // whole step, and a step with nothing to buy would return nil below and + // be recorded as completed, advancing the ramp (platform#631). + scoped, scopeErr := scopeExecutionsByAccount(baseExec, accounts) + if scopeErr != nil { + return scopeErr + } + buying := make([]config.CloudAccount, 0, len(scoped)) + for i := range accounts { + if _, ok := scoped[accounts[i].ID]; ok { + buying = append(buying, accounts[i]) + } + } + results := execution.RunForAccountsWithConcurrency(ctx, buying, func(ctx context.Context, account config.CloudAccount) (bool, error) { + return m.executeForAccount(ctx, scoped[account.ID], plan, account) }, getMaxAccountParallelism()) committed := 0 diff --git a/internal/purchase/execution_test.go b/internal/purchase/execution_test.go index de4cd133..7a13ee8b 100644 --- a/internal/purchase/execution_test.go +++ b/internal/purchase/execution_test.go @@ -743,7 +743,7 @@ func TestManager_ExecutePurchase_MultiAccount(t *testing.T) { ExecutionID: "exec-multi", PlanID: "plan-multi", Status: "pending", - Recommendations: []config.RecommendationRecord{rec}, + Recommendations: []config.RecommendationRecord{withAccount(rec, "aaaaaaaa-0000-0000-0000-000000000001"), withAccount(rec, "aaaaaaaa-0000-0000-0000-000000000002")}, } accounts := []config.CloudAccount{ @@ -797,6 +797,14 @@ func TestManager_ExecutePurchase_MultiAccount(t *testing.T) { "aaaaaaaa-0000-0000-0000-000000000002", }, accountIDs) + // platform#631: each account buys only its own rec (2 calls, not 4) and + // its row carries its own totals, not the whole step's. + mockServiceClient.AssertNumberOfCalls(t, "PurchaseCommitment", 2) + for _, e := range savedExecs { + require.Len(t, e.Recommendations, 1) + assert.Equal(t, *e.CloudAccountID, *e.Recommendations[0].CloudAccountID) + } + mockStore.AssertExpectations(t) mockEmail.AssertExpectations(t) mockFactory.AssertExpectations(t) @@ -871,7 +879,7 @@ func TestExecuteForAccount_CredentialFailure_MarksFailed(t *testing.T) { ExecutionID: "exec-credfail", PlanID: "plan-credfail", Status: "pending", - Recommendations: []config.RecommendationRecord{rec}, + Recommendations: []config.RecommendationRecord{withAccount(rec, tt.account.ID)}, } mockStore.On("GetPurchasePlan", ctx, "plan-credfail").Return(plan, nil) @@ -1032,7 +1040,7 @@ func TestExecuteMultiAccount_PartialFailure_IsolatesAccounts(t *testing.T) { ExecutionID: "exec-partial", PlanID: "plan-partial", Status: "pending", - Recommendations: []config.RecommendationRecord{rec}, + Recommendations: []config.RecommendationRecord{withAccount(rec, accountVID), withAccount(rec, accountIID)}, } plan := &config.PurchasePlan{ @@ -1203,7 +1211,7 @@ func TestExecuteMultiAccount_RunsAccountsInParallel(t *testing.T) { ExecutionID: "exec-parallel", PlanID: "plan-parallel", Status: "pending", - Recommendations: []config.RecommendationRecord{rec}, + Recommendations: []config.RecommendationRecord{withAccount(rec, accountAID), withAccount(rec, accountBID)}, } plan := &config.PurchasePlan{ diff --git a/internal/purchase/money_path_regression_test.go b/internal/purchase/money_path_regression_test.go index 043ebcca..6e990919 100644 --- a/internal/purchase/money_path_regression_test.go +++ b/internal/purchase/money_path_regression_test.go @@ -240,7 +240,8 @@ func TestMultiAccountSeedsStablePerAccountKey(t *testing.T) { PlanID: "plan-x", Source: common.PurchaseSourceWeb, Recommendations: []config.RecommendationRecord{ - {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true}, + scopedTestRec("acct-A"), + scopedTestRec("acct-B"), }, } plan := &config.PurchasePlan{ID: "plan-x", Name: "Plan X"} @@ -385,7 +386,8 @@ func TestMultiAccountPartialSuccessIsAcked(t *testing.T) { Status: "pending", PlanID: "plan-x", Recommendations: []config.RecommendationRecord{ - {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true}, + scopedTestRec("acct-ok"), + scopedTestRec("acct-bad"), }, } plan := &config.PurchasePlan{ID: "plan-x", Name: "Plan X", AutoPurchase: true} @@ -484,7 +486,8 @@ func TestExecuteForAccount_CommittedButUnsavedIsAcked(t *testing.T) { Status: "pending", PlanID: "plan-y", Recommendations: []config.RecommendationRecord{ - {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true}, + scopedTestRec("acct-bad"), + scopedTestRec("acct-unsaved"), }, } plan := &config.PurchasePlan{ID: "plan-y", Name: "Plan Y", AutoPurchase: true} @@ -786,7 +789,8 @@ func TestRootKeyStillFansOutThroughExecutePurchase(t *testing.T) { PlanID: "plan-x", Source: common.PurchaseSourceWeb, Recommendations: []config.RecommendationRecord{ - {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true}, + scopedTestRec("acct-A"), + scopedTestRec("acct-B"), }, } require.NoError(t, manager.executePurchase(ctx, root), diff --git a/internal/purchase/ramp_step_progress_integration_test.go b/internal/purchase/ramp_step_progress_integration_test.go index 742f48f6..5537869f 100644 --- a/internal/purchase/ramp_step_progress_integration_test.go +++ b/internal/purchase/ramp_step_progress_integration_test.go @@ -199,6 +199,19 @@ func rampStepRecommendation() []config.RecommendationRecord { } } +// rampStepRecommendationsPerAccount is the step's rec set: one rec per plan +// account, since the fan-out buys each account's own recs only (platform#631). +func (f *rampStepFixture) rampStepRecommendationsPerAccount() []config.RecommendationRecord { + var recs []config.RecommendationRecord + for i := range f.accounts { + id := f.accounts[i].ID + rec := rampStepRecommendation()[0] + rec.CloudAccountID = &id + recs = append(recs, rec) + } + return recs +} + // saveRootExecution persists the plan-scheduled root row for ramp step // stepNumber: no cloud_account_id, so the executor fans it out. func (f *rampStepFixture) saveRootExecution(ctx context.Context, t *testing.T, stepNumber int) *config.PurchaseExecution { @@ -210,7 +223,7 @@ func (f *rampStepFixture) saveRootExecution(ctx context.Context, t *testing.T, s Status: "pending", StepNumber: stepNumber, ScheduledDate: time.Now(), - Recommendations: rampStepRecommendation(), + Recommendations: f.rampStepRecommendationsPerAccount(), } require.NoError(t, f.store.SavePurchaseExecution(ctx, exec)) return exec From e6ad5725c51d46b9fb8f44da40196a1b91b1ef7a Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 9 Oct 2026 18:39:37 +0200 Subject: [PATCH 2/5] test(purchase): preallocate per-account rec slices flagged by prealloc Co-Authored-By: Claude Sonnet 5.5 --- internal/api/handler_purchases_retry_fanout_test.go | 2 +- internal/purchase/ramp_step_progress_integration_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/api/handler_purchases_retry_fanout_test.go b/internal/api/handler_purchases_retry_fanout_test.go index 9509b027..2daa6c66 100644 --- a/internal/api/handler_purchases_retry_fanout_test.go +++ b/internal/api/handler_purchases_retry_fanout_test.go @@ -66,7 +66,7 @@ func fanoutRecs() []config.RecommendationRecord { // 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 { - var recs []config.RecommendationRecord + recs := make([]config.RecommendationRecord, 0, 3) for _, id := range []string{fanoutAcctA, fanoutAcctB, fanoutAcctC} { rec := fanoutRecs()[0] rec.CloudAccountID = strPtr(id) diff --git a/internal/purchase/ramp_step_progress_integration_test.go b/internal/purchase/ramp_step_progress_integration_test.go index f82a761e..0e821c91 100644 --- a/internal/purchase/ramp_step_progress_integration_test.go +++ b/internal/purchase/ramp_step_progress_integration_test.go @@ -202,7 +202,7 @@ func rampStepRecommendation() []config.RecommendationRecord { // rampStepRecommendationsPerAccount is the step's rec set: one rec per plan // account, since the fan-out buys each account's own recs only (platform#631). func (f *rampStepFixture) rampStepRecommendationsPerAccount() []config.RecommendationRecord { - var recs []config.RecommendationRecord + recs := make([]config.RecommendationRecord, 0, len(f.accounts)) for i := range f.accounts { id := f.accounts[i].ID rec := rampStepRecommendation()[0] From ad1534dd7051c9bdf3a601564f54241bfc78899f Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Sat, 10 Oct 2026 04:34:59 +0200 Subject: [PATCH 3/5] feat(purchase): atomic plan-step get-or-create, shared suppressions and account checks Refs #631 Co-Authored-By: Claude Sonnet 5.5 --- internal/analytics/collector_test.go | 12 ++ internal/api/handler_history.go | 12 +- ...dler_history_actionable_starvation_test.go | 4 +- internal/api/handler_history_test.go | 45 ++-- internal/api/handler_purchases.go | 66 +----- ..._purchases_pause_claim_integration_test.go | 2 +- .../handler_purchases_retry_fanout_test.go | 10 +- ...dler_purchases_run_now_integration_test.go | 3 + internal/config/interfaces.go | 14 ++ internal/config/store_postgres_ramp_step.go | 54 +++++ internal/config/suppressions_build.go | 65 ++++++ .../suppressions_build_test.go} | 21 +- internal/mocks/stores.go | 28 +++ internal/purchase/account_scope.go | 21 ++ internal/purchase/attach_recommendations.go | 58 +++++ internal/purchase/execution.go | 3 + internal/purchase/manager.go | 19 +- .../purchase/money_path_regression_test.go | 2 +- internal/purchase/notifications.go | 156 ++++++++++---- internal/purchase/notifications_test.go | 37 ++-- .../plan_step_atomic_integration_test.go | 172 +++++++++++++++ internal/purchase/plan_step_atomic_test.go | 202 ++++++++++++++++++ internal/purchase/plan_step_no_recs_test.go | 10 +- .../ramp_step_progress_integration_test.go | 2 +- internal/purchase/step_rows_test.go | 13 ++ internal/server/test_helpers_test.go | 12 ++ 26 files changed, 868 insertions(+), 175 deletions(-) create mode 100644 internal/config/suppressions_build.go rename internal/{api/build_suppressions_test.go => config/suppressions_build_test.go} (81%) create mode 100644 internal/purchase/attach_recommendations.go create mode 100644 internal/purchase/plan_step_atomic_integration_test.go create mode 100644 internal/purchase/plan_step_atomic_test.go create mode 100644 internal/purchase/step_rows_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 26e840d9..8b5e0017 100644 --- a/internal/analytics/collector_test.go +++ b/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 +} diff --git a/internal/api/handler_history.go b/internal/api/handler_history.go index 864938cd..e2a54ac4 100644 --- a/internal/api/handler_history.go +++ b/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) } diff --git a/internal/api/handler_history_actionable_starvation_test.go b/internal/api/handler_history_actionable_starvation_test.go index 5a9e7a02..15efcfbb 100644 --- a/internal/api/handler_history_actionable_starvation_test.go +++ b/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) } diff --git a/internal/api/handler_history_test.go b/internal/api/handler_history_test.go index e679f665..b09bb831 100644 --- a/internal/api/handler_history_test.go +++ b/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,5 @@ 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 } diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go index 232f14ca..9e8ec54d 100644 --- a/internal/api/handler_purchases.go +++ b/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 diff --git a/internal/api/handler_purchases_pause_claim_integration_test.go b/internal/api/handler_purchases_pause_claim_integration_test.go index a1c7844f..3a69ab83 100644 --- a/internal/api/handler_purchases_pause_claim_integration_test.go +++ b/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) diff --git a/internal/api/handler_purchases_retry_fanout_test.go b/internal/api/handler_purchases_retry_fanout_test.go index 2daa6c66..dfe371e2 100644 --- a/internal/api/handler_purchases_retry_fanout_test.go +++ b/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), } } diff --git a/internal/api/handler_purchases_run_now_integration_test.go b/internal/api/handler_purchases_run_now_integration_test.go index 7d4cd568..2c236d58 100644 --- a/internal/api/handler_purchases_run_now_integration_test.go +++ b/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} diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 0fcf8bf3..7f6b64fd 100644 --- a/internal/config/interfaces.go +++ b/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 diff --git a/internal/config/store_postgres_ramp_step.go b/internal/config/store_postgres_ramp_step.go index 87770a6a..d97dd8e6 100644 --- a/internal/config/store_postgres_ramp_step.go +++ b/internal/config/store_postgres_ramp_step.go @@ -2,6 +2,7 @@ package config import ( "context" + "encoding/json" "fmt" "github.com/jackc/pgx/v5" @@ -319,3 +320,56 @@ func (s *PostgresStore) GetStuckRampSteps(ctx context.Context) (map[string]RampS } return blocks, nil } + +// ListExecutionsForPlanStepTx implements StoreInterface. +func (s *PostgresStore) ListExecutionsForPlanStepTx(ctx context.Context, tx pgx.Tx, planID string, stepNumber int) ([]PurchaseExecution, error) { + rows, err := tx.Query(ctx, ` + 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, canceled_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 plan_id = $1 AND step_number = $2 + ORDER BY retry_attempt_n, execution_id`, planID, stepNumber) + if err != nil { + return nil, fmt.Errorf("failed to query executions of plan %s step %d: %w", planID, stepNumber, err) + } + defer rows.Close() + return scanExecutionRows(rows) +} + +// SetExecutionRecommendationsIfEmptyTx implements StoreInterface. +func (s *PostgresStore) SetExecutionRecommendationsIfEmptyTx(ctx context.Context, tx pgx.Tx, executionID string, recs []RecommendationRecord, upfront, savings float64) (bool, error) { + if len(recs) == 0 { + return false, fmt.Errorf("execution %s: refusing to attach an empty recommendation set", executionID) + } + payload, err := json.Marshal(recs) + if err != nil { + return false, fmt.Errorf("failed to marshal recommendations: %w", err) + } + tag, err := tx.Exec(ctx, ` + UPDATE purchase_executions + SET recommendations = $2, total_upfront_cost = $3, estimated_savings = $4, updated_at = NOW() + WHERE execution_id = $1 + AND status IN ('pending', 'notified') + AND (recommendations IS NULL OR recommendations IN ('[]'::jsonb, 'null'::jsonb))`, executionID, payload, upfront, savings) + if err != nil { + return false, fmt.Errorf("failed to attach recommendations to execution %s: %w", executionID, err) + } + return tag.RowsAffected() == 1, nil +} + +// ExpireExecutionAtomic implements StoreInterface. +func (s *PostgresStore) ExpireExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string) (bool, error) { + tag, err := tx.Exec(ctx, ` + UPDATE purchase_executions SET status = $2, updated_at = NOW() + WHERE execution_id = $1 AND status IN ('pending', 'notified')`, executionID, StatusExpired) + if err != nil { + return false, fmt.Errorf("failed to expire execution %s: %w", executionID, err) + } + return tag.RowsAffected() == 1, nil +} diff --git a/internal/config/suppressions_build.go b/internal/config/suppressions_build.go new file mode 100644 index 00000000..2b91de73 --- /dev/null +++ b/internal/config/suppressions_build.go @@ -0,0 +1,65 @@ +package config + +import "time" + +// 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 []RecommendationRecord, executionID string, cfg *GlobalConfig, now time.Time) []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([]PurchaseSuppression, 0, len(order)) + for _, k := range order { + graceDays := DefaultGracePeriodDays + if cfg != nil { + graceDays = cfg.GracePeriodFor(k.provider) + } + if graceDays <= 0 { + continue // feature disabled for this provider + } + out = append(out, 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 +} diff --git a/internal/api/build_suppressions_test.go b/internal/config/suppressions_build_test.go similarity index 81% rename from internal/api/build_suppressions_test.go rename to internal/config/suppressions_build_test.go index ba894112..9ff0a167 100644 --- a/internal/api/build_suppressions_test.go +++ b/internal/config/suppressions_build_test.go @@ -1,21 +1,18 @@ -package api +package config import ( "testing" "time" - "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -func strPtr(s string) *string { return &s } - func TestBuildSuppressions_CreatesOneRowPerTuple(t *testing.T) { now := time.Date(2026, 4, 24, 0, 0, 0, 0, time.UTC) - gp := &config.GlobalConfig{GracePeriodDays: map[string]int{"aws": 7, "azure": 7, "gcp": 7}} + gp := &GlobalConfig{GracePeriodDays: map[string]int{"aws": 7, "azure": 7, "gcp": 7}} - sups := buildSuppressions([]config.RecommendationRecord{ + sups := BuildSuppressions([]RecommendationRecord{ {Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "t4g.nano", Count: 3, CloudAccountID: strPtr("acct-1")}, {Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "t4g.nano", Count: 2, CloudAccountID: strPtr("acct-1")}, {Provider: "aws", Service: "rds", Region: "us-east-1", ResourceType: "db.t4g.micro", Engine: "postgres", Count: 1, CloudAccountID: strPtr("acct-1")}, @@ -34,9 +31,9 @@ func TestBuildSuppressions_CreatesOneRowPerTuple(t *testing.T) { func TestBuildSuppressions_SkipsProvidersWithGrace0(t *testing.T) { now := time.Now() - gp := &config.GlobalConfig{GracePeriodDays: map[string]int{"aws": 7, "azure": 0, "gcp": 14}} + gp := &GlobalConfig{GracePeriodDays: map[string]int{"aws": 7, "azure": 0, "gcp": 14}} - sups := buildSuppressions([]config.RecommendationRecord{ + sups := BuildSuppressions([]RecommendationRecord{ {Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "t4g.nano", Count: 1, CloudAccountID: strPtr("acct-1")}, {Provider: "azure", Service: "compute", Region: "eastus", ResourceType: "Standard_D2d_v5", Count: 2, CloudAccountID: strPtr("acct-2")}, {Provider: "gcp", Service: "compute", Region: "us-central1", ResourceType: "n1-standard-1", Count: 3, CloudAccountID: strPtr("acct-3")}, @@ -51,9 +48,9 @@ func TestBuildSuppressions_SkipsProvidersWithGrace0(t *testing.T) { func TestBuildSuppressions_NilAccountIDNormalised(t *testing.T) { now := time.Now() - gp := &config.GlobalConfig{} + gp := &GlobalConfig{} - sups := buildSuppressions([]config.RecommendationRecord{ + sups := BuildSuppressions([]RecommendationRecord{ {Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "t4g.nano", Count: 1, CloudAccountID: nil}, }, "exec-1", gp, now) @@ -63,9 +60,9 @@ func TestBuildSuppressions_NilAccountIDNormalised(t *testing.T) { func TestBuildSuppressions_SkipsZeroCountRecs(t *testing.T) { now := time.Now() - gp := &config.GlobalConfig{} + gp := &GlobalConfig{} - sups := buildSuppressions([]config.RecommendationRecord{ + sups := BuildSuppressions([]RecommendationRecord{ {Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "t4g.nano", Count: 0, CloudAccountID: strPtr("acct-1")}, }, "exec-1", gp, now) diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 5adc426b..1296446a 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -251,6 +251,34 @@ func (m *MockConfigStore) LockPurchasePlanTx(ctx context.Context, tx pgx.Tx, pla // probe keep exercising the create path: no occupied steps in range is the // shape of a plan whose next steps are still unscheduled, which is what those // tests set up. +func (m *MockConfigStore) ListExecutionsForPlanStepTx(ctx context.Context, tx pgx.Tx, planID string, stepNumber int) ([]config.PurchaseExecution, error) { + m.record("ListExecutionsForPlanStepTx", ctx, tx, planID, stepNumber) + if !isExpected(&m.Mock, "ListExecutionsForPlanStepTx") { + return nil, nil + } + args := m.Called(ctx, tx, planID, stepNumber) + 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) SetExecutionRecommendationsIfEmptyTx(ctx context.Context, tx pgx.Tx, executionID string, recs []config.RecommendationRecord, upfront, savings float64) (bool, error) { + m.record("SetExecutionRecommendationsIfEmptyTx", ctx, tx, executionID, recs, upfront, savings) + args := m.Called(ctx, tx, executionID, recs, upfront, savings) + return args.Bool(0), args.Error(1) +} + +func (m *MockConfigStore) ExpireExecutionAtomic(ctx context.Context, tx pgx.Tx, executionID string) (bool, error) { + m.record("ExpireExecutionAtomic", ctx, tx, executionID) + args := m.Called(ctx, tx, executionID) + return args.Bool(0), args.Error(1) +} + func (m *MockConfigStore) OccupiedRampStepsInRangeTx(ctx context.Context, tx pgx.Tx, planID string, from, to int) ([]int, error) { m.record("OccupiedRampStepsInRangeTx", ctx, tx, planID, from, to) if !isExpected(&m.Mock, "OccupiedRampStepsInRangeTx") { diff --git a/internal/purchase/account_scope.go b/internal/purchase/account_scope.go index d2bcbe65..9419c1fe 100644 --- a/internal/purchase/account_scope.go +++ b/internal/purchase/account_scope.go @@ -61,3 +61,24 @@ func describeScopeRec(rec *config.RecommendationRecord) string { } return fmt.Sprintf("%s/%s %s (%s)", rec.Provider, rec.Service, rec.ResourceType, acct) } + +// requireRecsMatchRowAccount is the single-account path's version of the +// fan-out scoping: a plan row bound to an account may only buy recs attributed +// to that same account (strict equality; a rec with no account fails). Direct +// executes (no plan) and unscoped plan rows keep their existing resolution. +func requireRecsMatchRowAccount(exec *config.PurchaseExecution) error { + if exec.PlanID == "" || exec.CloudAccountID == nil { + return nil + } + for i := range exec.Recommendations { + rec := exec.Recommendations[i] + if !rec.Selected { + continue + } + if rec.CloudAccountID == nil || *rec.CloudAccountID != *exec.CloudAccountID { + return fmt.Errorf("execution %s is bound to account %s but recommendation %s is not attributed to it", + exec.ExecutionID, *exec.CloudAccountID, describeScopeRec(&rec)) + } + } + return nil +} diff --git a/internal/purchase/attach_recommendations.go b/internal/purchase/attach_recommendations.go new file mode 100644 index 00000000..75c19a74 --- /dev/null +++ b/internal/purchase/attach_recommendations.go @@ -0,0 +1,58 @@ +package purchase + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/jackc/pgx/v5" +) + +// AttachResolvedRecommendations persists the recommendations a plan step +// resolved onto its still-empty pending/notified execution, together with the +// suppressions that stop Opportunities (and the next step's resolution) from +// offering the same recs again. Both writes happen in one transaction: a crash +// leaves the row empty with no suppressions, or populated with them. +// +// Returns false (and writes nothing) when another resolver attached first or +// the row left pending/notified; the caller re-reads the row instead of +// overwriting it. On success exec is updated in place. +func (m *Manager) AttachResolvedRecommendations(ctx context.Context, exec *config.PurchaseExecution, recs []config.RecommendationRecord) (bool, error) { + if len(recs) == 0 { + return false, errors.New("refusing to attach an empty recommendation set") + } + var upfront, savings float64 + for i := range recs { + upfront += recs[i].UpfrontCost + savings += recs[i].Savings + } + globalCfg, err := m.config.GetGlobalConfig(ctx) + if err != nil { + return false, fmt.Errorf("failed to read grace periods for suppressions: %w", err) + } + sups := config.BuildSuppressions(recs, exec.ExecutionID, globalCfg, time.Now()) + + attached := false + err = m.config.WithTx(ctx, func(tx pgx.Tx) error { + ok, setErr := m.config.SetExecutionRecommendationsIfEmptyTx(ctx, tx, exec.ExecutionID, recs, upfront, savings) + if setErr != nil || !ok { + return setErr + } + for i := range sups { + if supErr := m.config.CreateSuppressionTx(ctx, tx, &sups[i]); supErr != nil { + return fmt.Errorf("failed to record suppressions for execution %s: %w", exec.ExecutionID, supErr) + } + } + attached = true + return nil + }) + if err != nil { + return false, err + } + if attached { + exec.Recommendations, exec.TotalUpfrontCost, exec.EstimatedSavings = recs, upfront, savings + } + return attached, nil +} diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index a3ac8001..82fa2164 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -95,6 +95,9 @@ func (m *Manager) executePurchase(ctx context.Context, exec *config.PurchaseExec // plain error when nothing committed. Split out of executePurchase to keep that // function under the gocyclo budget. func (m *Manager) executeSingleAccount(ctx context.Context, exec *config.PurchaseExecution, plan *config.PurchasePlan) error { + if err := requireRecsMatchRowAccount(exec); err != nil { + return err + } provCfg, targetAccountID, err := m.resolveSingleAccountProvider(ctx, exec) if err != nil { return err diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 9d48f57a..46318986 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -16,6 +16,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/oidc" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/sts" + "github.com/jackc/pgx/v5" ) // STSClient interface for AWS STS operations. @@ -231,7 +232,7 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE if execErr != nil { logging.Errorf("Failed to execute purchase %s: %v", exec.ExecutionID, execErr) } - if err := m.config.SavePurchaseExecution(ctx, exec); err != nil { + if err := m.saveTerminalExecution(ctx, exec); err != nil { logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err) // Wrap with ErrAuditLoss regardless of whether executePurchase itself // failed. When execErr != nil (provider/partial error), finalizeExecution @@ -255,6 +256,22 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE return execErr } +// saveTerminalExecution persists the finalized row. A plan row that ended +// failed having bought nothing also releases its suppressions in the same +// transaction: the recommendations were never purchased, so they must come +// back on Opportunities and for the next resolution (platform#631). +func (m *Manager) saveTerminalExecution(ctx context.Context, exec *config.PurchaseExecution) error { + if exec.PlanID == "" || exec.Status != "failed" || anyRecPurchased(exec.Recommendations) { + return m.config.SavePurchaseExecution(ctx, exec) + } + return m.config.WithTx(ctx, func(tx pgx.Tx) error { + if err := m.config.SavePurchaseExecutionTx(ctx, tx, exec); err != nil { + return err + } + return m.config.DeleteSuppressionsByExecutionTx(ctx, tx, exec.ExecutionID) + }) +} + // recordRampAdvanceRefusal reports a ramp advance that did not happen, stamping // it onto the execution row that just completed when the outcome is permanent // so the decision outlives the log retention window. diff --git a/internal/purchase/money_path_regression_test.go b/internal/purchase/money_path_regression_test.go index 6e990919..9f0a46a2 100644 --- a/internal/purchase/money_path_regression_test.go +++ b/internal/purchase/money_path_regression_test.go @@ -650,7 +650,7 @@ func scopelessPerAccountRow(accountID string) *config.PurchaseExecution { PlanID: "plan-x", Source: common.PurchaseSourceWeb, Recommendations: []config.RecommendationRecord{ - {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true}, + scopedTestRec(accountID), }, } } diff --git a/internal/purchase/notifications.go b/internal/purchase/notifications.go index fee0b0e5..1b4507d6 100644 --- a/internal/purchase/notifications.go +++ b/internal/purchase/notifications.go @@ -11,6 +11,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/LeanerCloud/cloud-commitments-platform/internal/email" "github.com/google/uuid" + "github.com/jackc/pgx/v5" ) // SendUpcomingPurchaseNotifications sends notifications for upcoming automated purchases. @@ -141,52 +142,107 @@ func (m *Manager) globalNotificationEmail(ctx context.Context) string { // RotatePendingApprovalToken once the email has been sent. A new row is // saved with its hash up front, since nothing was emailed for it yet. func (m *Manager) getOrCreateExecution(ctx context.Context, plan *config.PurchasePlan) (execution *config.PurchaseExecution, rawToken string, rotationPending bool, err error) { - // Check for existing execution for this date to prevent duplicates. - // GetExecutionByPlanAndDate wraps ErrNotFound on zero rows; any other - // error is a real store failure and must propagate. - existing, err := m.config.GetExecutionByPlanAndDate(ctx, plan.ID, *plan.NextExecutionDate) - switch { - case err == nil && existing != nil: - logging.Debugf("Found existing execution %s for plan %s on %s", existing.ExecutionID, plan.ID, plan.NextExecutionDate) - // GetExecutionByPlanAndDate filters only on plan_id + scheduled_date, - // not status, so this row can already be approved/completed/canceled - // by the time a later notification tick re-runs for the same date. - // Rotating then would kill whichever link (approval or post-approve - // revoke) is currently live. - if existing.Status != "pending" && existing.Status != "notified" { - return nil, "", false, fmt.Errorf("%w: %s is %s", errExecutionNotNotifiable, existing.ExecutionID, existing.Status) - } - if len(existing.Recommendations) == 0 { - return nil, "", false, m.failExistingBarePlanStep(ctx, existing) - } - tok, genErr := common.GenerateApprovalToken() - if genErr != nil { - return nil, "", false, fmt.Errorf("failed to generate approval token: %w", genErr) - } - expiry := time.Now().Add(config.ApprovalTokenTTL) - existing.ApprovalToken = config.HashApprovalToken(tok) - existing.ApprovalTokenExpiresAt = &expiry - return existing, tok, true, nil - case err != nil && !errors.Is(err, config.ErrNotFound): - return nil, "", false, fmt.Errorf("failed to check for existing execution: %w", err) - } - // ErrNotFound (or nil error with nil row): no existing execution for this - // plan+date. Nothing attaches recommendations to a plan step yet - // (platform#609), so the row is recorded as failed instead of emailing a $0 - // approval link. The failed row also makes every later tick for this date - // return errExecutionNotNotifiable quietly rather than minting another. - // - // step_number names the step this row will COMPLETE, not the count already - // completed, matching api.createPurchaseExecutionsTx (CurrentStep + 1). The - // ramp advance is keyed on this value since issue #1669. - execution = &config.PurchaseExecution{ + // The whole decision runs under the plan's ramp lock (platform#631): two + // ticks (SQS redelivery, overlapping cron, scheduler retries) otherwise + // both read "nothing there" and both write a row for the same step. The + // step, not the date, is the key, because rows created ahead of time carry + // their own scheduled_date. + var step stepClaim + txErr := m.config.WithTx(ctx, func(tx pgx.Tx) error { + var claimErr error + step, claimErr = m.claimPlanStep(ctx, tx, plan) + return claimErr + }) + if txErr != nil { + return nil, "", false, txErr + } + if step.failedNew { + return nil, "", false, bareStepFailure(plan.ID, step.number) + } + if step.blocked != nil { + return nil, "", false, step.blocked + } + existing := step.existing + if len(existing.Recommendations) == 0 { + return nil, "", false, m.failExistingBarePlanStep(ctx, existing) + } + tok, genErr := common.GenerateApprovalToken() + if genErr != nil { + return nil, "", false, fmt.Errorf("failed to generate approval token: %w", genErr) + } + expiry := time.Now().Add(config.ApprovalTokenTTL) + existing.ApprovalToken = config.HashApprovalToken(tok) + existing.ApprovalTokenExpiresAt = &expiry + return existing, tok, true, nil +} + +// stepClaim is the outcome of claimPlanStep: a freshly recorded failed row, a +// quiet skip error, or the pending root row to notify. +type stepClaim struct { + number int + existing *config.PurchaseExecution + blocked error + failedNew bool +} + +// claimPlanStep runs inside the tick's transaction: lock the plan, look at +// every row of the step, and either record the step as failed (nothing to +// buy yet) or hand back the one row that may be notified. +func (m *Manager) claimPlanStep(ctx context.Context, tx pgx.Tx, plan *config.PurchasePlan) (stepClaim, error) { + locked, lockErr := m.config.LockPurchasePlanTx(ctx, tx, plan.ID) + if lockErr != nil { + return stepClaim{}, fmt.Errorf("failed to lock plan %s: %w", plan.ID, lockErr) + } + if locked == nil { + return stepClaim{}, fmt.Errorf("plan not found: %s", plan.ID) + } + // step_number names the step this row will COMPLETE, not the count + // already completed, matching api.createPurchaseExecutionsTx + // (CurrentStep + 1). The ramp advance is keyed on it since #1669. + claim := stepClaim{number: locked.RampSchedule.CurrentStep + 1} + rows, listErr := m.config.ListExecutionsForPlanStepTx(ctx, tx, plan.ID, claim.number) + if listErr != nil { + return stepClaim{}, fmt.Errorf("failed to check for existing execution: %w", listErr) + } + if len(rows) > 0 { + claim.existing, claim.blocked = adoptableRootRow(rows) + return claim, nil + } + // Nothing attaches recommendations to a plan step yet (platform#609), so + // the row is recorded as failed instead of emailing a $0 approval link. + // The failed row also makes every later tick for this step return + // errExecutionNotNotifiable. + fresh := &config.PurchaseExecution{ PlanID: plan.ID, ExecutionID: uuid.New().String(), Status: "pending", - StepNumber: plan.RampSchedule.CurrentStep + 1, + StepNumber: claim.number, ScheduledDate: *plan.NextExecutionDate, } - return nil, "", false, m.recordFailedBarePlanStep(ctx, execution) + stampFailedBarePlanStep(fresh) + if saveErr := m.config.SavePurchaseExecutionTx(ctx, tx, fresh); saveErr != nil { + return stepClaim{}, fmt.Errorf("failed to record plan step %d of plan %s as failed: %w", claim.number, plan.ID, saveErr) + } + claim.failedNew = true + return claim, nil +} + +// adoptableRootRow picks the row a notification tick may notify for a step +// that already has rows. Only a pending/notified ROOT row (no account, first +// attempt) qualifies: child and retry rows share the step number, and a step +// whose root is failed, partially_completed, expired, canceled or in flight +// must not get a second root, because approving it would buy the step again +// (E1/E2). The second return value is the quiet skip error when none qualifies. +func adoptableRootRow(rows []config.PurchaseExecution) (*config.PurchaseExecution, error) { + for i := range rows { + r := rows[i] + if r.CloudAccountID == nil && r.RetryAttemptN == 0 && (r.Status == "pending" || r.Status == "notified") { + return &r, nil + } + } + first := rows[0] + return nil, fmt.Errorf("%w: step %d of plan %s already has execution %s (%s)", + errExecutionNotNotifiable, first.StepNumber, first.PlanID, first.ExecutionID, first.Status) } // failExistingBarePlanStep fails a pending/notified plan-step row that carries @@ -210,14 +266,26 @@ func (m *Manager) failExistingBarePlanStep(ctx context.Context, exec *config.Pur // or a new row), saves it, logs it once, and returns errExecutionNotNotifiable // so the notification tick sends nothing for it. func (m *Manager) recordFailedBarePlanStep(ctx context.Context, exec *config.PurchaseExecution) error { - exec.Status = "failed" - exec.Error = ErrPlanStepNoRecommendations.Error() + stampFailedBarePlanStep(exec) if err := m.config.SavePurchaseExecution(ctx, exec); err != nil { return fmt.Errorf("failed to record plan step %d of plan %s as failed: %w", exec.StepNumber, exec.PlanID, err) } + return bareStepFailure(exec.PlanID, exec.StepNumber) +} + +// stampFailedBarePlanStep marks exec failed with the no-recommendations reason +// and logs it once. +func stampFailedBarePlanStep(exec *config.PurchaseExecution) { + exec.Status = "failed" + exec.Error = ErrPlanStepNoRecommendations.Error() logging.Errorf("purchase[%s]: plan %s step %d has no recommendations and was marked failed: %v", exec.ExecutionID, exec.PlanID, exec.StepNumber, ErrPlanStepNoRecommendations) - return fmt.Errorf("%w: %s has no recommendations", errExecutionNotNotifiable, exec.ExecutionID) +} + +// bareStepFailure is the quiet skip error a tick returns after failing a bare +// step, so it sends nothing for it. +func bareStepFailure(planID string, step int) error { + return fmt.Errorf("%w: step %d of plan %s has no recommendations", errExecutionNotNotifiable, step, planID) } // buildNotificationData creates notification data from plan and execution. diff --git a/internal/purchase/notifications_test.go b/internal/purchase/notifications_test.go index 40edc2d2..27037320 100644 --- a/internal/purchase/notifications_test.go +++ b/internal/purchase/notifications_test.go @@ -3,7 +3,6 @@ package purchase import ( "context" "errors" - "fmt" "testing" "time" @@ -189,7 +188,7 @@ func TestManager_GetOrCreateExecution(t *testing.T) { // No existing execution found var saved *config.PurchaseExecution - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(nil, nil) + expectStepRows(mockStore, plan) mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")). Run(func(args mock.Arguments) { saved = args.Get(1).(*config.PurchaseExecution) }).Return(nil) @@ -243,7 +242,7 @@ func TestManager_GetOrCreateExecution_ExistingExecution(t *testing.T) { // the returned copy (issue #103 -- a stored hash can never be re-emailed // raw) with the full ApprovalTokenTTL, but persists nothing: no // SavePurchaseExecution expectation, so a write here panics the mock. - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(existingExec, nil) + expectStepRows(mockStore, plan, *existingExec) manager := &Manager{ config: mockStore, @@ -284,7 +283,7 @@ func TestManager_GetOrCreateExecution_ExistingBareRowIsFailedNotRotated(t *testi StepNumber: 1, ScheduledDate: nextExec, } - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(bare, nil) + expectStepRows(mockStore, plan, *bare) failed := *bare failed.Status = "failed" mockStore.On("TransitionExecutionStatus", ctx, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil)).Return(&failed, nil) @@ -313,7 +312,7 @@ func TestManager_GetOrCreateExecution_ExistingBareRowApprovedMeanwhileIsNotOverw stale := &config.PurchaseExecution{ ExecutionID: "bare-exec-id", PlanID: "plan-123", Status: "pending", StepNumber: 1, ScheduledDate: nextExec, } - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(stale, nil) + expectStepRows(mockStore, plan, *stale) mockStore.On("TransitionExecutionStatus", ctx, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil)). Return(nil, config.ErrExecutionNotInExpectedStatus) // No SavePurchaseExecution expectation: an upsert here would clobber the approval. @@ -358,7 +357,7 @@ func TestManager_GetOrCreateExecution_ExistingCompletedNotRotated(t *testing.T) ApprovalToken: config.HashApprovalToken("live-revocation-token"), ScheduledDate: nextExec, } - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(completedExec, nil) + expectStepRows(mockStore, plan, *completedExec) // No SavePurchaseExecution / GetExecutionByID expectation: a completed // row must never be written to by the reminder-notification path. @@ -393,7 +392,7 @@ func TestManager_GetOrCreateExecution_SaveError(t *testing.T) { } // No existing execution found - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(nil, nil) + expectStepRows(mockStore, plan) mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(errors.New("save failed")) manager := &Manager{ @@ -423,7 +422,8 @@ func TestManager_GetOrCreateExecution_LookupError(t *testing.T) { } // Error looking up existing execution - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(nil, errors.New("db error")) + mockStore.On("LockPurchasePlanTx", mock.Anything, mock.Anything, plan.ID).Return(plan, nil) + mockStore.On("ListExecutionsForPlanStepTx", mock.Anything, mock.Anything, plan.ID, plan.RampSchedule.CurrentStep+1).Return(nil, errors.New("db error")) manager := &Manager{ config: mockStore, @@ -457,8 +457,7 @@ func TestManager_GetOrCreateExecution_CreatesOnErrNotFound(t *testing.T) { } // Store returns ErrNotFound (wrapped), matching the post-fix store behavior. - notFoundErr := fmt.Errorf("%w: plan plan-f2 at %v", config.ErrNotFound, nextExec) - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-f2", nextExec).Return(nil, notFoundErr) + expectStepRows(mockStore, plan) var saved *config.PurchaseExecution mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")). Run(func(args mock.Arguments) { saved = args.Get(1).(*config.PurchaseExecution) }).Return(nil) @@ -510,7 +509,7 @@ func TestManager_SendUpcomingPurchaseNotifications_WithNotification(t *testing.T ScheduledDate: nextExec, Recommendations: []config.RecommendationRecord{{Provider: "aws", Service: "ec2", Count: 1}}, } - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(pendingExec, nil) + expectStepRows(mockStore, &plans[0], *pendingExec) mockStore.On("RotatePendingApprovalToken", ctx, "pending-exec-id", mock.Anything, mock.Anything).Return(true, nil) mockStore.On("GetGlobalConfig", ctx).Return(globalCfg, nil) mockEmail.On("SendScheduledPurchaseNotification", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil) @@ -664,7 +663,7 @@ func TestManager_SendUpcomingPurchaseNotifications_EmailFails(t *testing.T) { ScheduledDate: nextExec, Recommendations: []config.RecommendationRecord{{Provider: "aws", Service: "ec2", Count: 1}}, } - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-123", nextExec).Return(pendingExec, nil) + expectStepRows(mockStore, &plans[0], *pendingExec) mockStore.On("GetGlobalConfig", ctx).Return(globalCfg, nil) mockEmail.On("SendScheduledPurchaseNotification", ctx, mock.AnythingOfType("email.NotificationData")).Return(errors.New("email failed")) @@ -687,7 +686,7 @@ func TestManager_SendUpcomingPurchaseNotifications_EmailFails(t *testing.T) { // pendingExecForNotification sets up a plan due for a notification whose // execution row already exists as "pending" with the hash of the token the // previous notification emailed. -func pendingExecForNotification(ctx context.Context, mockStore *MockConfigStore) (*config.PurchasePlan, *config.PurchaseExecution) { +func pendingExecForNotification(mockStore *MockConfigStore) (*config.PurchasePlan, *config.PurchaseExecution) { nextExec := time.Now().Add(3 * 24 * time.Hour) plan := &config.PurchasePlan{ID: "plan-rot", Name: "Rotation Plan", NextExecutionDate: &nextExec} existing := &config.PurchaseExecution{ @@ -698,7 +697,9 @@ func pendingExecForNotification(ctx context.Context, mockStore *MockConfigStore) ScheduledDate: nextExec, Recommendations: []config.RecommendationRecord{{Provider: "aws", Service: "ec2", Count: 1}}, } - mockStore.On("GetExecutionByPlanAndDate", ctx, "plan-rot", nextExec).Return(existing, nil).Maybe() + mockStore.On("LockPurchasePlanTx", mock.Anything, mock.Anything, plan.ID).Return(plan, nil).Maybe() + mockStore.On("ListExecutionsForPlanStepTx", mock.Anything, mock.Anything, plan.ID, plan.RampSchedule.CurrentStep+1). + Return([]config.PurchaseExecution{*existing}, nil).Maybe() return plan, existing } @@ -710,7 +711,7 @@ func TestManager_SendPlanNotification_NoRecipientLeavesLiveTokenAlone(t *testing ctx := context.Background() mockStore := new(MockConfigStore) mockEmail := new(MockEmailSender) - plan, existing := pendingExecForNotification(ctx, mockStore) + plan, existing := pendingExecForNotification(mockStore) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{}, nil) manager := &Manager{config: mockStore, email: mockEmail, dashboardURL: "https://dashboard.example.com"} @@ -727,7 +728,7 @@ func TestManager_SendPlanNotification_SendFailureLeavesLiveTokenAlone(t *testing ctx := context.Background() mockStore := new(MockConfigStore) mockEmail := new(MockEmailSender) - plan, _ := pendingExecForNotification(ctx, mockStore) + plan, _ := pendingExecForNotification(mockStore) notify := "notify@example.com" mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: ¬ify}, nil) mockEmail.On("SendScheduledPurchaseNotification", ctx, mock.AnythingOfType("email.NotificationData")).Return(errors.New("ses down")) @@ -748,7 +749,7 @@ func TestManager_SendPlanNotification_PersistsEmailedTokenHashAfterSend(t *testi ctx := context.Background() mockStore := new(MockConfigStore) mockEmail := new(MockEmailSender) - plan, _ := pendingExecForNotification(ctx, mockStore) + plan, _ := pendingExecForNotification(mockStore) notify := "notify@example.com" mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: ¬ify}, nil) @@ -778,7 +779,7 @@ func TestManager_SendPlanNotification_StampFailure(t *testing.T) { ctx := context.Background() store := new(MockConfigStore) sender := new(MockEmailSender) - plan, _ := pendingExecForNotification(ctx, store) + plan, _ := pendingExecForNotification(store) recipient := "fixture@example.invalid" store.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &recipient}, nil) sender.On("SendScheduledPurchaseNotification", ctx, mock.Anything).Return(nil) diff --git a/internal/purchase/plan_step_atomic_integration_test.go b/internal/purchase/plan_step_atomic_integration_test.go new file mode 100644 index 00000000..fb5ef494 --- /dev/null +++ b/internal/purchase/plan_step_atomic_integration_test.go @@ -0,0 +1,172 @@ +//go:build integration + +package purchase + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func atomicStepPlan(ctx context.Context, t *testing.T, store *config.PostgresStore) *config.PurchasePlan { + t.Helper() + next := time.Now().Add(24 * time.Hour).Truncate(time.Microsecond) + plan := &config.PurchasePlan{ + Name: "atomic step", Enabled: true, AutoPurchase: true, NotificationDaysBefore: 30, + Services: map[string]config.ServiceConfig{"ec2": {Provider: "aws", Service: "ec2"}}, + RampSchedule: config.RampSchedule{Type: "weekly", PercentPerStep: 25, StepIntervalDays: 7, + CurrentStep: 0, TotalSteps: 4, StartDate: next.AddDate(0, 0, -7)}, + NextExecutionDate: &next, + } + require.NoError(t, plan.Validate()) + require.NoError(t, store.CreatePurchasePlan(ctx, plan)) + return plan +} + +func stepRowsInDB(ctx context.Context, t *testing.T, store *config.PostgresStore, planID string, step int) []config.PurchaseExecution { + t.Helper() + var rows []config.PurchaseExecution + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + var err error + rows, err = store.ListExecutionsForPlanStepTx(ctx, tx, planID, step) + return err + })) + return rows +} + +// A4: concurrent ticks (SQS redelivery, cron overlap, scheduler retries) must +// leave exactly one row for the step. Without the plan lock every tick reads +// "nothing there" and inserts its own. +func TestGetOrCreateExecution_ConcurrentTicksMintOneRow(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + plan := atomicStepPlan(ctx, t, store) + m := &Manager{config: store} + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, _, _, _ = m.getOrCreateExecution(ctx, plan) + }() + } + wg.Wait() + + rows := stepRowsInDB(ctx, t, store, plan.ID, 1) + require.Len(t, rows, 1, "concurrent ticks must mint exactly one row for the step") + assert.Equal(t, "failed", rows[0].Status) +} + +// E1: a settled root (failed here; partially_completed is the same branch) +// blocks creation, so no second root can be bought. +func TestGetOrCreateExecution_SettledRootBlocksNewRoot(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + plan := atomicStepPlan(ctx, t, store) + for _, status := range []string{"failed", "partially_completed"} { + p := atomicStepPlan(ctx, t, store) + root := &config.PurchaseExecution{PlanID: p.ID, ExecutionID: uuid.New().String(), Status: status, StepNumber: 1, + ScheduledDate: *p.NextExecutionDate, Recommendations: []config.RecommendationRecord{scopedTestRec("acct-A")}} + require.NoError(t, store.SavePurchaseExecution(ctx, root)) + + _, _, _, err := (&Manager{config: store}).getOrCreateExecution(ctx, p) + + require.ErrorIs(t, err, errExecutionNotNotifiable, status) + assert.Len(t, stepRowsInDB(ctx, t, store, p.ID, 1), 1, status) + } + _ = plan +} + +// A5/CAS: racing resolvers attach exactly once, with exactly one suppression +// set written in the same transaction. Skipping the suppressions fails this. +func TestAttachResolvedRecommendations_ConcurrentAttachesOnce(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + plan := atomicStepPlan(ctx, t, store) + rootID := uuid.New().String() + require.NoError(t, store.SavePurchaseExecution(ctx, &config.PurchaseExecution{ + PlanID: plan.ID, ExecutionID: rootID, Status: "pending", StepNumber: 1, ScheduledDate: *plan.NextExecutionDate})) + rec := scopedTestRec("") + rec.CloudAccountID = nil + recs := []config.RecommendationRecord{rec} + m := &Manager{config: store} + + var wg sync.WaitGroup + var mu sync.Mutex + wins := 0 + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ok, err := m.AttachResolvedRecommendations(ctx, &config.PurchaseExecution{ExecutionID: rootID}, recs) + assert.NoError(t, err) + if ok { + mu.Lock() + wins++ + mu.Unlock() + } + }() + } + wg.Wait() + + assert.Equal(t, 1, wins, "exactly one resolver may attach") + got, err := store.GetExecutionByID(ctx, rootID) + require.NoError(t, err) + assert.Len(t, got.Recommendations, 1) + assert.Equal(t, 300.0, got.TotalUpfrontCost) + sups, err := store.ListActiveSuppressions(ctx) + require.NoError(t, err) + require.Len(t, sups, 1, "the winner's suppressions must exist, and only once") + assert.Equal(t, rootID, sups[0].ExecutionID) + + // A row that already has recs is never overwritten. + ok, err := m.AttachResolvedRecommendations(ctx, &config.PurchaseExecution{ExecutionID: rootID}, recs) + require.NoError(t, err) + assert.False(t, ok) +} + +// E3: expiry flips the status and drops the suppressions in one tx; a row that +// already left pending/notified is left alone. +func TestExpireExecutionAtomic_ReleasesSuppressions(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + plan := atomicStepPlan(ctx, t, store) + rootID := uuid.New().String() + require.NoError(t, store.SavePurchaseExecution(ctx, &config.PurchaseExecution{ + PlanID: plan.ID, ExecutionID: rootID, Status: "pending", StepNumber: 1, ScheduledDate: *plan.NextExecutionDate})) + rec := scopedTestRec("") + rec.CloudAccountID = nil + m := &Manager{config: store} + ok, err := m.AttachResolvedRecommendations(ctx, &config.PurchaseExecution{ExecutionID: rootID}, []config.RecommendationRecord{rec}) + require.NoError(t, err) + require.True(t, ok) + + expire := func() bool { + var expired bool + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + var e error + expired, e = store.ExpireExecutionAtomic(ctx, tx, rootID) + if e != nil || !expired { + return e + } + return store.DeleteSuppressionsByExecutionTx(ctx, tx, rootID) + })) + return expired + } + require.True(t, expire()) + got, err := store.GetExecutionByID(ctx, rootID) + require.NoError(t, err) + assert.Equal(t, "expired", got.Status) + sups, err := store.ListActiveSuppressions(ctx) + require.NoError(t, err) + assert.Empty(t, sups) + assert.False(t, expire(), "an already-expired row must not be flipped again") +} diff --git a/internal/purchase/plan_step_atomic_test.go b/internal/purchase/plan_step_atomic_test.go new file mode 100644 index 00000000..b6ada8f5 --- /dev/null +++ b/internal/purchase/plan_step_atomic_test.go @@ -0,0 +1,202 @@ +package purchase + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func stepTestPlan() *config.PurchasePlan { + next := time.Now().Add(24 * time.Hour) + return &config.PurchasePlan{ID: "plan-s", RampSchedule: config.RampSchedule{CurrentStep: 1}, NextExecutionDate: &next} +} + +func stepRow(status string, mutate func(*config.PurchaseExecution)) config.PurchaseExecution { + e := config.PurchaseExecution{ + ExecutionID: "row-1", PlanID: "plan-s", StepNumber: 2, Status: status, + Recommendations: []config.RecommendationRecord{scopedTestRec("acct-A")}, + } + if mutate != nil { + mutate(&e) + } + return e +} + +// E1/E2: any existing row that is not a pending/notified ROOT blocks creation +// quietly, and nothing is written. Mutating adoptableRootRow to a "live status" +// filter makes the failed and partially_completed cases mint a second root. +func TestGetOrCreateExecution_ExistingNonAdoptableRowBlocks(t *testing.T) { + acct := "acct-A" + cases := map[string]config.PurchaseExecution{ + "failed root": stepRow("failed", nil), + "partially_completed root": stepRow("partially_completed", nil), + "expired root": stepRow("expired", nil), + "canceled root": stepRow("canceled", nil), + "completed root": stepRow("completed", nil), + "approved root": stepRow("approved", nil), + "pending child": stepRow("pending", func(e *config.PurchaseExecution) { e.CloudAccountID = &acct }), + "pending retry": stepRow("pending", func(e *config.PurchaseExecution) { e.RetryAttemptN = 1 }), + } + for name, row := range cases { + t.Run(name, func(t *testing.T) { + store := new(MockConfigStore) + plan := stepTestPlan() + expectStepRows(store, plan, row) + m := &Manager{config: store} + + exec, _, _, err := m.getOrCreateExecution(context.Background(), plan) + + require.ErrorIs(t, err, errExecutionNotNotifiable) + assert.Nil(t, exec) + store.AssertNotCalled(t, "SavePurchaseExecution") + store.AssertNotCalled(t, "SavePurchaseExecutionTx") + store.AssertNotCalled(t, "TransitionExecutionStatus") + }) + } +} + +// A pending root with recommendations is adopted (not duplicated) and gets a +// fresh token to email. +func TestGetOrCreateExecution_AdoptsPendingRootOnly(t *testing.T) { + acct := "acct-A" + store := new(MockConfigStore) + plan := stepTestPlan() + child := stepRow("pending", func(e *config.PurchaseExecution) { e.ExecutionID = "child"; e.CloudAccountID = &acct }) + root := stepRow("notified", func(e *config.PurchaseExecution) { e.ExecutionID = "root" }) + expectStepRows(store, plan, child, root) + m := &Manager{config: store} + + exec, tok, rotate, err := m.getOrCreateExecution(context.Background(), plan) + + require.NoError(t, err) + assert.Equal(t, "root", exec.ExecutionID) + assert.NotEmpty(t, tok) + assert.True(t, rotate) +} + +// The step is keyed on the locked plan's CurrentStep, not on the date. +func TestGetOrCreateExecution_KeysOnStepNotDate(t *testing.T) { + store := new(MockConfigStore) + plan := stepTestPlan() + row := stepRow("pending", nil) + row.ScheduledDate = time.Now().AddDate(0, 0, -30) // differs from NextExecutionDate + expectStepRows(store, plan, row) + m := &Manager{config: store} + + exec, _, _, err := m.getOrCreateExecution(context.Background(), plan) + + require.NoError(t, err) + assert.Equal(t, "row-1", exec.ExecutionID, "a pre-created row with another date must be adopted, not duplicated") +} + +func TestAttachResolvedRecommendations(t *testing.T) { + recs := []config.RecommendationRecord{scopedTestRec("acct-A"), scopedTestRec("acct-B")} + t.Run("attaches recs and writes one suppression per tuple in the same tx", func(t *testing.T) { + store := new(MockConfigStore) + store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil) + store.On("SetExecutionRecommendationsIfEmptyTx", mock.Anything, mock.Anything, "row-1", recs, 600.0, 150.0).Return(true, nil) + store.On("CreateSuppressionTx", mock.Anything, mock.Anything, mock.Anything).Return(nil).Twice() + exec := &config.PurchaseExecution{ExecutionID: "row-1"} + + ok, err := (&Manager{config: store}).AttachResolvedRecommendations(context.Background(), exec, recs) + + require.NoError(t, err) + assert.True(t, ok) + assert.Equal(t, 600.0, exec.TotalUpfrontCost) + assert.Len(t, exec.Recommendations, 2) + store.AssertNumberOfCalls(t, "CreateSuppressionTx", 2) + }) + t.Run("lost race writes no suppressions", func(t *testing.T) { + store := new(MockConfigStore) + store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil) + store.On("SetExecutionRecommendationsIfEmptyTx", mock.Anything, mock.Anything, "row-1", recs, 600.0, 150.0).Return(false, nil) + exec := &config.PurchaseExecution{ExecutionID: "row-1"} + + ok, err := (&Manager{config: store}).AttachResolvedRecommendations(context.Background(), exec, recs) + + require.NoError(t, err) + assert.False(t, ok) + assert.Empty(t, exec.Recommendations) + store.AssertNotCalled(t, "CreateSuppressionTx") + }) + t.Run("suppression failure surfaces", func(t *testing.T) { + store := new(MockConfigStore) + store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil) + store.On("SetExecutionRecommendationsIfEmptyTx", mock.Anything, mock.Anything, "row-1", recs, 600.0, 150.0).Return(true, nil) + store.On("CreateSuppressionTx", mock.Anything, mock.Anything, mock.Anything).Return(errors.New("boom")) + exec := &config.PurchaseExecution{ExecutionID: "row-1"} + + ok, err := (&Manager{config: store}).AttachResolvedRecommendations(context.Background(), exec, recs) + + require.Error(t, err) + assert.False(t, ok) + assert.Empty(t, exec.Recommendations, "exec must stay untouched when the tx fails") + }) + t.Run("empty set refused", func(t *testing.T) { + _, err := (&Manager{config: new(MockConfigStore)}).AttachResolvedRecommendations(context.Background(), &config.PurchaseExecution{}, nil) + require.Error(t, err) + }) +} + +// E5: a plan row bound to an account refuses recs of another account, or with +// none, before any credential or provider work. +func TestExecuteSingleAccount_PlanRowRecsMustMatchRowAccount(t *testing.T) { + rowAcct := "acct-A" + for name, rec := range map[string]config.RecommendationRecord{ + "other account": scopedTestRec("acct-B"), + "no account": {Provider: "aws", Service: "ec2", ResourceType: "m5.large", Selected: true}, + } { + t.Run(name, func(t *testing.T) { + exec := &config.PurchaseExecution{ExecutionID: "x", PlanID: "plan-s", CloudAccountID: &rowAcct, + Recommendations: []config.RecommendationRecord{rec}} + err := (&Manager{config: new(MockConfigStore)}).executeSingleAccount(context.Background(), exec, &config.PurchasePlan{ID: "plan-s"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "is not attributed to it") + }) + } + t.Run("direct execute unchanged", func(t *testing.T) { + exec := &config.PurchaseExecution{ExecutionID: "x", CloudAccountID: &rowAcct, + Recommendations: []config.RecommendationRecord{scopedTestRec("acct-B")}} + require.NoError(t, requireRecsMatchRowAccount(exec)) + }) +} + +// E3: a plan row that ends failed with nothing bought releases its +// suppressions in the same tx as the terminal save; a partial or direct row +// does not. +func TestSaveTerminalExecution_ReleasesSuppressionsOnCleanFailure(t *testing.T) { + bought := scopedTestRec("acct-A") + bought.Purchased = true + cases := []struct { + name string + exec config.PurchaseExecution + release bool + }{ + {"plan row failed, nothing bought", config.PurchaseExecution{ExecutionID: "e", PlanID: "p", Status: "failed", Recommendations: []config.RecommendationRecord{scopedTestRec("acct-A")}}, true}, + {"plan row failed but a rec bought", config.PurchaseExecution{ExecutionID: "e", PlanID: "p", Status: "failed", Recommendations: []config.RecommendationRecord{bought}}, false}, + {"plan row completed", config.PurchaseExecution{ExecutionID: "e", PlanID: "p", Status: "completed"}, false}, + {"direct execute failed", config.PurchaseExecution{ExecutionID: "e", Status: "failed"}, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + store := new(MockConfigStore) + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil) + store.On("DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e").Return(nil).Maybe() + exec := c.exec + + require.NoError(t, (&Manager{config: store}).saveTerminalExecution(context.Background(), &exec)) + + if c.release { + store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e") + } else { + store.AssertNotCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e") + } + }) + } +} diff --git a/internal/purchase/plan_step_no_recs_test.go b/internal/purchase/plan_step_no_recs_test.go index c5f3688a..856c96b4 100644 --- a/internal/purchase/plan_step_no_recs_test.go +++ b/internal/purchase/plan_step_no_recs_test.go @@ -129,13 +129,15 @@ func TestNotificationTickOnBarePlanStepCreatesOneFailedRowAndSendsNoEmail(t *tes // Tick 1 finds no row for the date; tick 2 finds whatever tick 1 saved. var stored config.PurchaseExecution var saves []config.PurchaseExecution - mockStore.On("GetExecutionByPlanAndDate", ctx, plan.ID, *plan.NextExecutionDate). - Return(nil, config.ErrNotFound).Once() - mockStore.On("GetExecutionByPlanAndDate", ctx, plan.ID, *plan.NextExecutionDate). - Return(&stored, nil) + mockStore.On("LockPurchasePlanTx", mock.Anything, mock.Anything, plan.ID).Return(&plan, nil) + mockStore.On("ListExecutionsForPlanStepTx", mock.Anything, mock.Anything, plan.ID, plan.RampSchedule.CurrentStep+1). + Return(nil, nil).Once() mockStore.SavePurchaseExecutionFn = func(_ context.Context, exec *config.PurchaseExecution) error { stored = *exec saves = append(saves, *exec) + // From now on the step has a row, as the real store would report. + mockStore.On("ListExecutionsForPlanStepTx", mock.Anything, mock.Anything, plan.ID, plan.RampSchedule.CurrentStep+1). + Return([]config.PurchaseExecution{stored}, nil) return nil } // A $0 approval email must never go out; Maybe() keeps a violation a clean diff --git a/internal/purchase/ramp_step_progress_integration_test.go b/internal/purchase/ramp_step_progress_integration_test.go index 0e821c91..3a0882a7 100644 --- a/internal/purchase/ramp_step_progress_integration_test.go +++ b/internal/purchase/ramp_step_progress_integration_test.go @@ -268,7 +268,7 @@ func (f *rampStepFixture) saveRetryExecution(ctx context.Context, t *testing.T, Status: "approved", StepNumber: predecessor.StepNumber, ScheduledDate: time.Now(), - Recommendations: rampStepRecommendation(), + Recommendations: []config.RecommendationRecord{withAccount(rampStepRecommendation()[0], accountID)}, Source: common.PurchaseSourceWeb, RetryAttemptN: predecessor.RetryAttemptN + 1, } diff --git a/internal/purchase/step_rows_test.go b/internal/purchase/step_rows_test.go new file mode 100644 index 00000000..c5331bff --- /dev/null +++ b/internal/purchase/step_rows_test.go @@ -0,0 +1,13 @@ +package purchase + +import ( + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/mock" +) + +// expectStepRows registers the plan lock and the per-step row listing that +// getOrCreateExecution reads under the ramp lock. +func expectStepRows(m *MockConfigStore, plan *config.PurchasePlan, rows ...config.PurchaseExecution) { + m.On("LockPurchasePlanTx", mock.Anything, mock.Anything, plan.ID).Return(plan, nil) + m.On("ListExecutionsForPlanStepTx", mock.Anything, mock.Anything, plan.ID, plan.RampSchedule.CurrentStep+1).Return(rows, nil) +} diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 5a878922..13335aea 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -440,3 +440,15 @@ func (m *mockConfigStoreForHealth) UpsertNotificationMute(_ context.Context, _, func (m *mockConfigStoreForHealth) IsNotificationMuted(_ context.Context, _, _ string) (bool, error) { return false, nil } + +func (m *mockConfigStoreForHealth) ListExecutionsForPlanStepTx(_ context.Context, _ pgx.Tx, _ string, _ int) ([]config.PurchaseExecution, error) { + return nil, nil +} + +func (m *mockConfigStoreForHealth) SetExecutionRecommendationsIfEmptyTx(_ context.Context, _ pgx.Tx, _ string, _ []config.RecommendationRecord, _, _ float64) (bool, error) { + return false, nil +} + +func (m *mockConfigStoreForHealth) ExpireExecutionAtomic(_ context.Context, _ pgx.Tx, _ string) (bool, error) { + return false, nil +} From 44c5c48545c883a3883f6cf6971849b2eaa869dd Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Sat, 10 Oct 2026 05:02:54 +0200 Subject: [PATCH 4/5] fix(purchase): release plan-step suppressions only when no purchase call was submitted Outcome-unknown and refused re-drive failures keep their suppressions. Adds sweep and compare-and-set tests. Refs #631. Co-Authored-By: Claude Sonnet 5.5 --- internal/api/handler_history_test.go | 21 +++++++ internal/config/store_postgres_ramp_step.go | 2 +- internal/purchase/execution.go | 12 ++-- internal/purchase/manager.go | 35 ++++++++--- .../plan_step_atomic_integration_test.go | 18 ++++++ internal/purchase/plan_step_atomic_test.go | 58 ++++++++++++++++--- 6 files changed, 122 insertions(+), 24 deletions(-) diff --git a/internal/api/handler_history_test.go b/internal/api/handler_history_test.go index b09bb831..f16767cd 100644 --- a/internal/api/handler_history_test.go +++ b/internal/api/handler_history_test.go @@ -2093,3 +2093,24 @@ func TestSummarizePurchaseHistory_RevokedExcludedFromKPIs(t *testing.T) { } 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") + } + }) + } +} diff --git a/internal/config/store_postgres_ramp_step.go b/internal/config/store_postgres_ramp_step.go index d97dd8e6..f9d39330 100644 --- a/internal/config/store_postgres_ramp_step.go +++ b/internal/config/store_postgres_ramp_step.go @@ -356,7 +356,7 @@ func (s *PostgresStore) SetExecutionRecommendationsIfEmptyTx(ctx context.Context SET recommendations = $2, total_upfront_cost = $3, estimated_savings = $4, updated_at = NOW() WHERE execution_id = $1 AND status IN ('pending', 'notified') - AND (recommendations IS NULL OR recommendations IN ('[]'::jsonb, 'null'::jsonb))`, executionID, payload, upfront, savings) + AND recommendations IN ('[]'::jsonb, 'null'::jsonb)`, executionID, payload, upfront, savings) if err != nil { return false, fmt.Errorf("failed to attach recommendations to execution %s: %w", executionID, err) } diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 82fa2164..72b2e61d 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -59,17 +59,17 @@ func (m *Manager) executePurchase(ctx context.Context, exec *config.PurchaseExec } else { plan, err = m.config.GetPurchasePlan(ctx, exec.PlanID) if err != nil { - return fmt.Errorf("failed to get plan: %w", err) + return beforeSubmission(fmt.Errorf("failed to get plan: %w", err)) } if plan == nil { - return fmt.Errorf("plan not found: %s", exec.PlanID) + return beforeSubmission(fmt.Errorf("plan not found: %s", exec.PlanID)) } // Checked once here, before the per-account fan-out: past this point // every account would write its own failed row and strand its own ramp // unit, and credential resolution would mask the real cause. if len(exec.Recommendations) == 0 { - return fmt.Errorf("execution %s: %w", exec.ExecutionID, ErrPlanStepNoRecommendations) + return beforeSubmission(fmt.Errorf("execution %s: %w", exec.ExecutionID, ErrPlanStepNoRecommendations)) } // Fan out across plan accounts when accounts are configured. @@ -96,11 +96,11 @@ func (m *Manager) executePurchase(ctx context.Context, exec *config.PurchaseExec // function under the gocyclo budget. func (m *Manager) executeSingleAccount(ctx context.Context, exec *config.PurchaseExecution, plan *config.PurchasePlan) error { if err := requireRecsMatchRowAccount(exec); err != nil { - return err + return beforeSubmission(err) } provCfg, targetAccountID, err := m.resolveSingleAccountProvider(ctx, exec) if err != nil { - return err + return beforeSubmission(err) } // #646: stamp the resolved target account on history, not the ambient // AWS host account. resolveSingleAccountProvider returns the target's @@ -199,7 +199,7 @@ func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.Purc // be recorded as completed, advancing the ramp (platform#631). scoped, scopeErr := scopeExecutionsByAccount(baseExec, accounts) if scopeErr != nil { - return scopeErr + return beforeSubmission(scopeErr) } buying := make([]config.CloudAccount, 0, len(scoped)) for i := range accounts { diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 46318986..5fa97f43 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -223,7 +223,9 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE // entry point funnels through here, so one check covers all of them. execErr := armedRedriveRefusal(exec) if execErr == nil { - execErr = m.staleAzurePricingRefusal(ctx, exec) + if refusal := m.staleAzurePricingRefusal(ctx, exec); refusal != nil { + execErr = beforeSubmission(refusal) + } } if execErr == nil { execErr = m.executePurchase(ctx, exec) @@ -232,7 +234,7 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE if execErr != nil { logging.Errorf("Failed to execute purchase %s: %v", exec.ExecutionID, execErr) } - if err := m.saveTerminalExecution(ctx, exec); err != nil { + if err := m.saveTerminalExecution(ctx, exec, execErr); err != nil { logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err) // Wrap with ErrAuditLoss regardless of whether executePurchase itself // failed. When execErr != nil (provider/partial error), finalizeExecution @@ -256,12 +258,29 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE return execErr } -// saveTerminalExecution persists the finalized row. A plan row that ended -// failed having bought nothing also releases its suppressions in the same -// transaction: the recommendations were never purchased, so they must come -// back on Opportunities and for the next resolution (platform#631). -func (m *Manager) saveTerminalExecution(ctx context.Context, exec *config.PurchaseExecution) error { - if exec.PlanID == "" || exec.Status != "failed" || anyRecPurchased(exec.Recommendations) { +// beforeSubmissionError marks a failure that happened before any cloud +// purchase call was made for the execution: the code path itself proves no +// commitment can exist, independent of the error text or of provider +// sentinels (the go outcome-unknown sentinel is not pinned yet). It keeps the +// wrapped error's message unchanged. +type beforeSubmissionError struct{ err error } + +func (e *beforeSubmissionError) Error() string { return e.err.Error() } +func (e *beforeSubmissionError) Unwrap() error { return e.err } + +func beforeSubmission(err error) error { return &beforeSubmissionError{err: err} } + +// saveTerminalExecution persists the finalized row. A plan row releases its +// suppressions in the same transaction ONLY when it ended failed and execErr +// proves no purchase call was submitted for any rec (beforeSubmissionError). +// Every other failure keeps them: a lost response ("outcome unknown"), a +// refused armed re-drive or a partial run may still have created a +// commitment, and releasing would let the next step buy it again. Kept +// suppressions expire with the grace period or are released on explicit +// cancel or revoke (platform#631). +func (m *Manager) saveTerminalExecution(ctx context.Context, exec *config.PurchaseExecution, execErr error) error { + var notSubmitted *beforeSubmissionError + if exec.PlanID == "" || exec.Status != "failed" || anyRecPurchased(exec.Recommendations) || !errors.As(execErr, ¬Submitted) { return m.config.SavePurchaseExecution(ctx, exec) } return m.config.WithTx(ctx, func(tx pgx.Tx) error { diff --git a/internal/purchase/plan_step_atomic_integration_test.go b/internal/purchase/plan_step_atomic_integration_test.go index fb5ef494..c1d0062f 100644 --- a/internal/purchase/plan_step_atomic_integration_test.go +++ b/internal/purchase/plan_step_atomic_integration_test.go @@ -170,3 +170,21 @@ func TestExpireExecutionAtomic_ReleasesSuppressions(t *testing.T) { assert.Empty(t, sups) assert.False(t, expire(), "an already-expired row must not be flipped again") } + +// The compare-and-set also matches a row stored with an empty JSON array (not +// only the null that a nil slice marshals to). +func TestSetExecutionRecommendationsIfEmpty_MatchesEmptyArray(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + plan := atomicStepPlan(ctx, t, store) + rootID := uuid.New().String() + require.NoError(t, store.SavePurchaseExecution(ctx, &config.PurchaseExecution{ + PlanID: plan.ID, ExecutionID: rootID, Status: "pending", StepNumber: 1, ScheduledDate: *plan.NextExecutionDate, + Recommendations: []config.RecommendationRecord{}})) + m := &Manager{config: store} + + ok, err := m.AttachResolvedRecommendations(ctx, &config.PurchaseExecution{ExecutionID: rootID}, []config.RecommendationRecord{scopedTestRec("")}) + + require.NoError(t, err) + assert.True(t, ok) +} diff --git a/internal/purchase/plan_step_atomic_test.go b/internal/purchase/plan_step_atomic_test.go index b6ada8f5..9b286927 100644 --- a/internal/purchase/plan_step_atomic_test.go +++ b/internal/purchase/plan_step_atomic_test.go @@ -167,21 +167,33 @@ func TestExecuteSingleAccount_PlanRowRecsMustMatchRowAccount(t *testing.T) { }) } -// E3: a plan row that ends failed with nothing bought releases its -// suppressions in the same tx as the terminal save; a partial or direct row -// does not. -func TestSaveTerminalExecution_ReleasesSuppressionsOnCleanFailure(t *testing.T) { +// E3: a plan row releases its suppressions in the same tx as the terminal +// save ONLY when it failed before any purchase call was submitted. Outcome +// unknown, a refused armed re-drive, partial runs and direct executes keep them. +func TestSaveTerminalExecution_ReleasesSuppressionsOnlyBeforeSubmission(t *testing.T) { + failedRow := func() config.PurchaseExecution { + return config.PurchaseExecution{ExecutionID: "e", PlanID: "p", Status: "failed", Recommendations: []config.RecommendationRecord{scopedTestRec("acct-A")}} + } bought := scopedTestRec("acct-A") bought.Purchased = true + boughtRow := failedRow() + boughtRow.Recommendations = []config.RecommendationRecord{bought} + direct := failedRow() + direct.PlanID = "" + completed := failedRow() + completed.Status = "completed" cases := []struct { name string exec config.PurchaseExecution + err error release bool }{ - {"plan row failed, nothing bought", config.PurchaseExecution{ExecutionID: "e", PlanID: "p", Status: "failed", Recommendations: []config.RecommendationRecord{scopedTestRec("acct-A")}}, true}, - {"plan row failed but a rec bought", config.PurchaseExecution{ExecutionID: "e", PlanID: "p", Status: "failed", Recommendations: []config.RecommendationRecord{bought}}, false}, - {"plan row completed", config.PurchaseExecution{ExecutionID: "e", PlanID: "p", Status: "completed"}, false}, - {"direct execute failed", config.PurchaseExecution{ExecutionID: "e", Status: "failed"}, false}, + {"failed before any provider call", failedRow(), beforeSubmission(errors.New("credential resolution failed")), true}, + {"outcome unknown (lost response) stays failed", failedRow(), errors.New("purchase outcome unknown: response lost"), false}, + {"refused armed re-drive", failedRow(), errors.New("refusing to execute retry attempt 1: unsafe to re-drive"), false}, + {"pre-submission error but a rec bought", boughtRow, beforeSubmission(errors.New("x")), false}, + {"direct execute", direct, beforeSubmission(errors.New("x")), false}, + {"completed", completed, nil, false}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -190,7 +202,7 @@ func TestSaveTerminalExecution_ReleasesSuppressionsOnCleanFailure(t *testing.T) store.On("DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e").Return(nil).Maybe() exec := c.exec - require.NoError(t, (&Manager{config: store}).saveTerminalExecution(context.Background(), &exec)) + require.NoError(t, (&Manager{config: store}).saveTerminalExecution(context.Background(), &exec, c.err)) if c.release { store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e") @@ -200,3 +212,31 @@ func TestSaveTerminalExecution_ReleasesSuppressionsOnCleanFailure(t *testing.T) }) } } + +// End to end through executeAndFinalize: a rec not attributed to the row's +// account is refused before the provider (released); a provider failure keeps +// the suppressions. +func TestExecuteAndFinalize_SuppressionReleaseFollowsSubmission(t *testing.T) { + acct := "acct-A" + run := func(rec config.RecommendationRecord) (*MockConfigStore, error) { + store := new(MockConfigStore) + store.On("GetPurchasePlan", mock.Anything, "p").Return(&config.PurchasePlan{ID: "p"}, nil) + store.On("GetPlanAccounts", mock.Anything, "p").Return([]config.CloudAccount{}, nil).Maybe() + store.On("GetCloudAccount", mock.Anything, acct).Return(nil, errors.New("account lookup failed")).Maybe() + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil) + store.On("DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e").Return(nil).Maybe() + exec := &config.PurchaseExecution{ExecutionID: "e", PlanID: "p", CloudAccountID: &acct, Status: "running", + Recommendations: []config.RecommendationRecord{rec}} + return store, (&Manager{config: store}).executeAndFinalize(context.Background(), exec) + } + t.Run("rec for another account is refused before any call", func(t *testing.T) { + store, err := run(scopedTestRec("acct-B")) + require.Error(t, err) + store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e") + }) + t.Run("credential failure before any call", func(t *testing.T) { + store, err := run(scopedTestRec(acct)) + require.Error(t, err) + store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e") + }) +} From 21ecfef3db35fe59de7e563a72926b80d249065d Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Sat, 10 Oct 2026 05:18:28 +0200 Subject: [PATCH 5/5] test(purchase): pin that post-submission failures keep plan-step suppressions Refs #631 Co-Authored-By: Claude Sonnet 5.5 --- internal/purchase/plan_step_atomic_test.go | 56 ++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/internal/purchase/plan_step_atomic_test.go b/internal/purchase/plan_step_atomic_test.go index 9b286927..8f1361bd 100644 --- a/internal/purchase/plan_step_atomic_test.go +++ b/internal/purchase/plan_step_atomic_test.go @@ -6,7 +6,10 @@ import ( "testing" "time" + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sts" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -240,3 +243,56 @@ func TestExecuteAndFinalize_SuppressionReleaseFollowsSubmission(t *testing.T) { store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e") }) } + +// Post-submission failures driven through the real executeAndFinalize path +// must KEEP the suppressions: the provider may have created a commitment. +// Marking any post-call error as before-submission makes these fail. +func TestExecuteAndFinalize_PostSubmissionFailureKeepsSuppressions(t *testing.T) { + mk := func(types ...string) []config.RecommendationRecord { + recs := make([]config.RecommendationRecord, 0, len(types)) + for _, rt := range types { + recs = append(recs, config.RecommendationRecord{Provider: "aws", Service: "ec2", ResourceType: rt, + Region: "us-east-1", Count: 1, UpfrontCost: 100, Selected: true}) + } + return recs + } + cases := []struct { + name string + recs []config.RecommendationRecord + results []error // PurchaseCommitment outcome per call, nil = success + }{ + {"provider failure", mk("m5.large"), []error{errors.New("provider rejected the purchase")}}, + {"outcome unknown (lost response)", mk("m5.large"), []error{errors.New("purchase outcome unknown: response lost after submission")}}, + {"partial failure", mk("m5.large", "c5.large"), []error{nil, errors.New("provider rejected the purchase")}}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + store := new(MockConfigStore) + factory := new(MockProviderFactory) + prov := new(MockProvider) + svc := new(MockServiceClient) + stsMock := new(MockSTSClient) + store.On("GetPurchasePlan", mock.Anything, "p").Return(&config.PurchasePlan{ID: "p"}, nil) + store.On("GetPlanAccounts", mock.Anything, "p").Return([]config.CloudAccount{}, nil) + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil) + store.On("SavePurchaseHistory", mock.Anything, mock.Anything).Return(nil).Maybe() + store.On("DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e").Return(nil).Maybe() + email := new(MockEmailSender) + email.On("SendPurchaseConfirmation", mock.Anything, mock.Anything).Return(nil).Maybe() + stsMock.On("GetCallerIdentity", mock.Anything, mock.Anything).Return(&sts.GetCallerIdentityOutput{Account: aws.String("123456789012")}, nil).Maybe() + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil) + prov.On("GetServiceClient", mock.Anything, mock.Anything, "us-east-1").Return(svc, nil) + for _, e := range c.results { + res := common.PurchaseResult{Success: e == nil, CommitmentID: "ri-1"} + svc.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything).Return(res, e).Once() + } + exec := &config.PurchaseExecution{ExecutionID: "e", PlanID: "p", Status: "running", Recommendations: c.recs} + m := &Manager{config: store, email: email, stsClient: stsMock, providerFactory: factory} + + _ = m.executeAndFinalize(context.Background(), exec) + + store.AssertNotCalled(t, "DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "e") + svc.AssertExpectations(t) + }) + } +}