From b1f4aaa3bc3993cbce30beb995dd2fd0d6f0b299 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 5 Nov 2026 00:55:15 +0100 Subject: [PATCH 1/7] fix(purchase): record fan-out accounts skipped before they started When a per-account fan-out run had a committed account and its context ended before other accounts started, the run was acked as a partial success (#1014), but the never-started accounts had no row. Nothing retried them and nothing recorded the skip, so their purchases silently never happened. Each such account now gets its own failed row, with the stable per-account idempotency key, its own recommendations, NoSubmission set and an error saying nothing was submitted, so the existing retry can re-run it. The rows are saved on a context detached from the ended run. Only the acked partial path writes them: a flat failure is redelivered and re-runs those accounts, where a failed row would invite a second, undeduped retry. Closes #1672 Co-Authored-By: Claude Opus 5.5 --- internal/purchase/execution.go | 76 +++++-- .../execution_skipped_accounts_test.go | 195 ++++++++++++++++++ 2 files changed, 255 insertions(+), 16 deletions(-) create mode 100644 internal/purchase/execution_skipped_accounts_test.go diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index ecf287ae7..758a77168 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -7,6 +7,7 @@ import ( "slices" "strconv" "strings" + "sync" "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" @@ -14,6 +15,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-go/pkg/provider" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/LeanerCloud/cloud-commitments-platform/internal/credentials" + "github.com/LeanerCloud/cloud-commitments-platform/internal/ctxutil" "github.com/LeanerCloud/cloud-commitments-platform/internal/email" "github.com/LeanerCloud/cloud-commitments-platform/internal/execution" "github.com/LeanerCloud/cloud-commitments-platform/internal/providerfactory" @@ -245,7 +247,9 @@ func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.Purc buying = append(buying, accounts[i]) } } + var started sync.Map results := execution.RunForAccountsWithConcurrency(ctx, buying, func(ctx context.Context, account config.CloudAccount) (bool, error) { + started.Store(account.ID, true) return m.executeForAccount(ctx, scoped[account.ID], plan, account) }, getMaxAccountParallelism()) @@ -268,6 +272,7 @@ func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.Purc // Some accounts committed real purchases; the per-account rows already // recorded the truth. Surface a partial sentinel so the entry points // ack and never mislabel the run as a flat failure (issue #1014). + errs = append(errs, m.recordSkippedAccounts(ctx, scoped, results, &started)...) return &multiAccountPartialError{committed: committed, errors: errs} } return fmt.Errorf("%w: %s", errAllAccountsFailed, strings.Join(errs, "; ")) @@ -295,22 +300,7 @@ func (m *Manager) saveExecutionStatusBestEffort(ctx context.Context, exec *confi func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.PurchaseExecution, plan *config.PurchasePlan, account config.CloudAccount) (committed bool, err error) { // Create a per-account copy of the execution record with an independent // Recommendations slice so concurrent goroutines don't race on writes. - acctID := account.ID - acctExec := *baseExec - // ExecutionID is the per-account row identity (still a fresh UUID), but the - // per-rec provider idempotency token must NOT derive from it: a re-drive of - // the root execution would mint new UUIDs and lose the provider dedupe, - // double-buying every account (issue #1012 / H1). Seed a STABLE per-account - // idempotency key from the root's lineage key + the account ID so a re-drive - // reproduces the identical token per account. The account ID is the cloud - // account's stable config ID, not the (also stable) ExternalID — either is - // durable, but account.ID is the value that uniquely keys the fan-out unit. - acctExec.ExecutionID = uuid.New().String() - acctExec.IdempotencyKey = idempotencyLineageKey(baseExec) + ":" + account.ID - acctExec.CloudAccountID = &acctID - recs := make([]config.RecommendationRecord, len(baseExec.Recommendations)) - copy(recs, baseExec.Recommendations) - acctExec.Recommendations = recs + acctExec := newAccountExecution(baseExec, account.ID) // Runs on every return path below, once the row's terminal status is // stamped; a no-op unless it ended failed or partially_completed (#714). defer m.notifyExecutionFailure(ctx, &acctExec) @@ -372,6 +362,60 @@ func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.Purcha return committed, nil } +// newAccountExecution is the per-account copy of baseExec, with an independent +// Recommendations slice so concurrent goroutines don't race on writes. +// +// ExecutionID is the per-account row identity (still a fresh UUID), but the +// per-rec provider idempotency token must NOT derive from it: a re-drive of +// the root execution would mint new UUIDs and lose the provider dedupe, +// double-buying every account (issue #1012 / H1). Seed a STABLE per-account +// idempotency key from the root's lineage key + the account ID so a re-drive +// reproduces the identical token per account. The account ID is the cloud +// account's stable config ID, not the (also stable) ExternalID — either is +// durable, but account.ID is the value that uniquely keys the fan-out unit. +func newAccountExecution(baseExec *config.PurchaseExecution, accountID string) config.PurchaseExecution { + acctExec := *baseExec + acctExec.ExecutionID = uuid.New().String() + acctExec.IdempotencyKey = idempotencyLineageKey(baseExec) + ":" + accountID + acctExec.CloudAccountID = &accountID + recs := make([]config.RecommendationRecord, len(baseExec.Recommendations)) + copy(recs, baseExec.Recommendations) + acctExec.Recommendations = recs + return acctExec +} + +// recordSkippedAccounts writes a failed row for each account the fan-out +// skipped before executeForAccount ran (the run context ended first), so the +// acked partial run does not silently drop its purchases (issue #1672). +// Nothing was submitted for it, so the row carries NoSubmission and the +// ordinary retry can re-run it. Only the acked partial path calls this: a flat +// failure is redelivered and re-runs these accounts under their stable keys, +// where a failed row would invite a second, undeduped retry beside it. The +// context has ended, so the saves run detached. It returns the save failures. +func (m *Manager) recordSkippedAccounts(ctx context.Context, scoped map[string]*config.PurchaseExecution, results []execution.Result[bool], started *sync.Map) []string { + var lost []string + for _, r := range results { + if r.Err == nil { + continue + } + if _, ran := started.Load(r.AccountID); ran { + continue + } + row := newAccountExecution(scoped[r.AccountID], r.AccountID) + row.Status = "failed" + row.NoSubmission = true + row.Error = fmt.Sprintf("the run ended before account %s started, so nothing was submitted for it; retry to buy it: %v", r.AccountID, r.Err) + saveCtx, cancel := ctxutil.DetachedWithTimeout(ctx, m.budgets.terminalSaveTimeout()) + if err := m.config.SavePurchaseExecution(saveCtx, &row); err != nil { + lost = append(lost, fmt.Sprintf("AUDIT LOSS: failed to save the skipped-account record for account %s: %v", r.AccountID, err)) + } else { + m.notifyExecutionFailure(saveCtx, &row) + } + cancel() + } + return lost +} + // persistFailedAccountExecution writes a per-account execution row that the // caller has already stamped with Status="failed" + Error, and joins any save // failure with the originating cause via errors.Join so neither error is diff --git a/internal/purchase/execution_skipped_accounts_test.go b/internal/purchase/execution_skipped_accounts_test.go new file mode 100644 index 000000000..b1568ae81 --- /dev/null +++ b/internal/purchase/execution_skipped_accounts_test.go @@ -0,0 +1,195 @@ +package purchase + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// skipFixture runs a three-account fan-out one account at a time. Account A +// (valid credentials) buys; when buyCancels is set, its purchase call cancels +// the run context, so B and C never start (issue #1672). B and C have no +// stored credentials, so if either did start it would leave a credential +// error instead of the skip. +type skipFixture struct { + saved []config.PurchaseExecution + mu sync.Mutex + saveHook func(context.Context) + // cancelAtB moves the cancel from A's purchase to B's credential load, so B + // starts and fails on its own (a started account, not a skip). + cancelAtB bool +} + +const ( + skipAcctA = "aaaaaaaa-0000-0000-0000-000000000001" + skipAcctB = "bbbbbbbb-0000-0000-0000-000000000002" + skipAcctC = "cccccccc-0000-0000-0000-000000000003" +) + +func (f *skipFixture) run(t *testing.T, ctx context.Context, cancel context.CancelFunc, aBuys bool) error { + t.Helper() + t.Setenv("CUDLY_MAX_ACCOUNT_PARALLELISM", "1") + store := new(MockConfigStore) + mailer := allowFailedNotification(new(MockEmailSender)) + factory := new(MockProviderFactory) + prov := new(MockProvider) + client := new(MockServiceClient) + + accounts := []config.CloudAccount{ + {ID: skipAcctA, Name: "A", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys"}, + {ID: skipAcctB, Name: "B", Provider: "aws", ExternalID: "222222222222", AWSAuthMode: "access_keys"}, + {ID: skipAcctC, Name: "C", Provider: "aws", ExternalID: "333333333333", AWSAuthMode: "access_keys"}, + } + rec := config.RecommendationRecord{Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", + Count: 1, Savings: 75, UpfrontCost: 300, Selected: true} + exec := &config.PurchaseExecution{ + ExecutionID: "exec-skip", PlanID: "plan-skip", Status: "pending", IdempotencyKey: "lineage-skip", + Recommendations: []config.RecommendationRecord{withAccount(rec, skipAcctA), withAccount(rec, skipAcctB), withAccount(rec, skipAcctC)}, + } + plan := &config.PurchasePlan{ID: "plan-skip", Name: "Skip", RampSchedule: config.RampSchedule{TotalSteps: 1}} + + store.SavePurchaseExecutionFn = func(saveCtx context.Context, e *config.PurchaseExecution) error { + if f.saveHook != nil { + f.saveHook(saveCtx) + } + f.mu.Lock() + defer f.mu.Unlock() + f.saved = append(f.saved, *e) + return nil + } + store.On("SavePurchaseHistory", mock.Anything, mock.Anything).Return(nil).Maybe() + mailer.On("SendPurchaseConfirmation", mock.Anything, mock.Anything).Return(nil).Maybe() + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil).Maybe() + prov.On("GetServiceClient", mock.Anything, common.ServiceEC2, "us-east-1").Return(client, nil).Maybe() + result := common.PurchaseResult{Success: true, CommitmentID: "ri-a"} + var buyErr error + if !aBuys { + result, buyErr = common.PurchaseResult{}, errors.New("insufficient capacity") + } + client.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything). + Run(func(mock.Arguments) { + if !f.cancelAtB { + cancel() + } + }).Return(result, buyErr).Once() + + m := &Manager{config: store, email: mailer, providerFactory: factory, credStore: &MockCredentialStore{ + LoadRawFn: func(_ context.Context, accountID, _ string) ([]byte, error) { + if accountID == skipAcctA { + return []byte(`{"access_key_id":"AKIAIOSFODNN7EXAMPLE","secret_access_key":"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"}`), nil + } + if accountID == skipAcctB && f.cancelAtB { + cancel() + } + return nil, nil + }, + }} + return m.executeMultiAccount(ctx, exec, plan, accounts) +} + +func (f *skipFixture) rowsFor(id string) []config.PurchaseExecution { + f.mu.Lock() + defer f.mu.Unlock() + var out []config.PurchaseExecution + for _, e := range f.saved { + if e.CloudAccountID != nil && *e.CloudAccountID == id { + out = append(out, e) + } + } + return out +} + +// A cancel that lands mid-fan-out, after one account committed: the run is +// acked (#1014), so every account that never started must get its own failed +// row proving nothing was submitted, or its purchases silently never happen. +func TestExecuteMultiAccount_RecordsAccountsSkippedAfterACommit(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f := &skipFixture{} + err := f.run(t, ctx, cancel, true) + + var partial *multiAccountPartialError + require.ErrorAs(t, err, &partial, "a committed account keeps the partial ack") + require.Len(t, f.rowsFor(skipAcctA), 1) + assert.Equal(t, "completed", f.rowsFor(skipAcctA)[0].Status) + for _, id := range []string{skipAcctB, skipAcctC} { + rows := f.rowsFor(id) + require.Len(t, rows, 1, "account %s must get exactly one row", id) + r := rows[0] + assert.Equal(t, "failed", r.Status, id) + assert.True(t, r.NoSubmission, "%s: the skip proves nothing was submitted", id) + assert.Contains(t, r.Error, "nothing was submitted", id) + assert.Equal(t, "lineage-skip:"+id, r.IdempotencyKey, "%s: the stable per-account key", id) + assert.NotEqual(t, "exec-skip", r.ExecutionID, id) + require.Len(t, r.Recommendations, 1, "%s: the row keeps only its own account's recs, so a retry buys exactly them", id) + require.NotNil(t, r.Recommendations[0].CloudAccountID) + assert.Equal(t, id, *r.Recommendations[0].CloudAccountID) + assert.Nil(t, r.CompletedAt, id) + } +} + +// Without a commit the run is a flat failure that is redelivered, which re-runs +// the skipped accounts under their stable keys. A failed row for them would +// invite a manual retry beside that re-run (a new approval lineage, so no +// provider dedupe), so none is written. +func TestExecuteMultiAccount_NoSkipRowsWhenNothingCommitted(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f := &skipFixture{} + err := f.run(t, ctx, cancel, false) + + require.ErrorIs(t, err, errAllAccountsFailed) + assert.Len(t, f.rowsFor(skipAcctA), 1) + assert.Empty(t, f.rowsFor(skipAcctB)) + assert.Empty(t, f.rowsFor(skipAcctC)) +} + +// The skip rows are saved after the run context ended, so the save must not +// use it. +func TestExecuteMultiAccount_SkipRowSaveOutlivesTheRunContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f := &skipFixture{} + var live []bool + var mu sync.Mutex + f.saveHook = func(saveCtx context.Context) { + mu.Lock() + defer mu.Unlock() + live = append(live, saveCtx.Err() == nil) + if dl, ok := saveCtx.Deadline(); ok { + assert.WithinDuration(t, time.Now(), dl, 10*time.Minute, "a bounded save budget") + } + } + err := f.run(t, ctx, cancel, true) + require.Error(t, err) + require.Len(t, f.rowsFor(skipAcctB), 1) + mu.Lock() + defer mu.Unlock() + assert.Equal(t, []bool{true, true}, live[len(live)-2:], "the two skip-row saves run on a live context") +} + +// An account that started and failed already wrote its own row; only the +// accounts that never started get a skip row. +func TestExecuteMultiAccount_StartedAccountGetsNoSkipRow(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f := &skipFixture{cancelAtB: true} + err := f.run(t, ctx, cancel, true) + + var partial *multiAccountPartialError + require.ErrorAs(t, err, &partial) + b := f.rowsFor(skipAcctB) + require.Len(t, b, 1, "B started: its own credential-failure row, no skip row beside it") + assert.NotContains(t, b[0].Error, "nothing was submitted") + c := f.rowsFor(skipAcctC) + require.Len(t, c, 1) + assert.Contains(t, c[0].Error, "nothing was submitted") +} From 503d4d11882be21dd7ab6c39b40c44bd21f18228 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 14:46:09 +0100 Subject: [PATCH 2/7] fix(purchase): word a skipped account by whether Retry accepts it Retry refuses EC2, Redshift and Azure savings-plan rows whatever NoSubmission says, so the skip row no longer promises a retry for them; it points at starting a new purchase. Both notes are allowlisted so History shows them instead of the generic provider error, and the run-context error moves to the log. Comments on the flat-failure path are corrected. Co-Authored-By: Claude Opus 5.5 --- internal/purchase/errnotes.go | 2 ++ internal/purchase/errnotes_test.go | 2 ++ internal/purchase/execution.go | 28 +++++++++++++++---- .../execution_skipped_accounts_test.go | 27 ++++++++++++++---- 4 files changed, 48 insertions(+), 11 deletions(-) diff --git a/internal/purchase/errnotes.go b/internal/purchase/errnotes.go index 263d35fa4..8bbf57ef8 100644 --- a/internal/purchase/errnotes.go +++ b/internal/purchase/errnotes.go @@ -31,6 +31,8 @@ var serverNotePatterns = []string{ `target already met: target \d+, committed \d+(?:; ` + skippedNotePattern + `)?; \d+ failed row\(s\) of unknown outcome counted as committed`, `selected recommendations do not resolve to a single cloud account: (?:\d+ selected recommendation\(s\) target \d+ cloud accounts \(` + idListPattern + `\); submit one purchase per account|\d+ selected recommendation\(s\) carry no cloud_account_id while \d+ target account ` + idPattern + `)`, `purchase was not successful`, + regexp.QuoteMeta(skippedAccountRetryNote), + regexp.QuoteMeta(skippedAccountNoRetryNote), `EC2 recommendation for \S+ has zonal scope but no AvailabilityZone: refusing to buy a zonal RI in an arbitrary AZ; re-fetch the recommendation`, `zonal RI offering needs an availability zone; re-fetch the recommendation`, `email notifier not configured for this deployment`, diff --git a/internal/purchase/errnotes_test.go b/internal/purchase/errnotes_test.go index cccc1e40e..a867a393a 100644 --- a/internal/purchase/errnotes_test.go +++ b/internal/purchase/errnotes_test.go @@ -50,6 +50,8 @@ func TestSanitizeStoredError_AllowlistedNotesPassThroughVerbatim(t *testing.T) { fmt.Errorf("%w: 2 selected recommendation(s) target 2 cloud accounts (a, b); submit one purchase per account", errAmbiguousAccountScope).Error(), fmt.Errorf("%w: 1 selected recommendation(s) carry no cloud_account_id while 2 target account a", errAmbiguousAccountScope).Error(), "purchase was not successful", + skippedAccountRetryNote, + skippedAccountNoRetryNote, "EC2 recommendation for m5.large has zonal scope but no AvailabilityZone: refusing to buy a zonal RI in an arbitrary AZ; re-fetch the recommendation", "zonal RI offering needs an availability zone; re-fetch the recommendation", "email notifier not configured for this deployment", diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 758a77168..03edf73ba 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -384,14 +384,29 @@ func newAccountExecution(baseExec *config.PurchaseExecution, accountID string) c return acctExec } +// Skip-row notes, shared with the stored-error allowlist (errnotes.go). +const ( + skippedAccountRetryNote = "the run ended before this account's purchase started, so nothing was submitted; retry it to buy it" + skippedAccountNoRetryNote = "the run ended before this account's purchase started, so nothing was submitted; this purchase type cannot be retried automatically, so start a new purchase for this account if it is still wanted" +) + +// skippedAccountNote words the skip row by whether Retry accepts it: +// RedriveRefusalReason refuses EC2, Redshift and Azure savings plans whatever +// NoSubmission says, so those rows point at a new purchase instead. +func skippedAccountNote(row *config.PurchaseExecution) string { + if RedriveRefusalReason(row) != "" { + return skippedAccountNoRetryNote + } + return skippedAccountRetryNote +} + // recordSkippedAccounts writes a failed row for each account the fan-out // skipped before executeForAccount ran (the run context ended first), so the // acked partial run does not silently drop its purchases (issue #1672). -// Nothing was submitted for it, so the row carries NoSubmission and the -// ordinary retry can re-run it. Only the acked partial path calls this: a flat -// failure is redelivered and re-runs these accounts under their stable keys, -// where a failed row would invite a second, undeduped retry beside it. The -// context has ended, so the saves run detached. It returns the save failures. +// Nothing was submitted for it, so the row carries NoSubmission. Only the +// acked partial path calls this: a flat failure finalizes the root as failed, +// and a root Retry re-runs every account. The context has ended, so the saves +// run detached. It returns the save failures. func (m *Manager) recordSkippedAccounts(ctx context.Context, scoped map[string]*config.PurchaseExecution, results []execution.Result[bool], started *sync.Map) []string { var lost []string for _, r := range results { @@ -404,7 +419,8 @@ func (m *Manager) recordSkippedAccounts(ctx context.Context, scoped map[string]* row := newAccountExecution(scoped[r.AccountID], r.AccountID) row.Status = "failed" row.NoSubmission = true - row.Error = fmt.Sprintf("the run ended before account %s started, so nothing was submitted for it; retry to buy it: %v", r.AccountID, r.Err) + row.Error = skippedAccountNote(&row) + logging.Warnf("execution %s: account %s skipped before it started: %v", row.ExecutionID, r.AccountID, r.Err) saveCtx, cancel := ctxutil.DetachedWithTimeout(ctx, m.budgets.terminalSaveTimeout()) if err := m.config.SavePurchaseExecution(saveCtx, &row); err != nil { lost = append(lost, fmt.Sprintf("AUDIT LOSS: failed to save the skipped-account record for account %s: %v", r.AccountID, err)) diff --git a/internal/purchase/execution_skipped_accounts_test.go b/internal/purchase/execution_skipped_accounts_test.go index b1568ae81..e9bd46e8b 100644 --- a/internal/purchase/execution_skipped_accounts_test.go +++ b/internal/purchase/execution_skipped_accounts_test.go @@ -126,7 +126,7 @@ func TestExecuteMultiAccount_RecordsAccountsSkippedAfterACommit(t *testing.T) { r := rows[0] assert.Equal(t, "failed", r.Status, id) assert.True(t, r.NoSubmission, "%s: the skip proves nothing was submitted", id) - assert.Contains(t, r.Error, "nothing was submitted", id) + assert.Equal(t, skippedAccountNoRetryNote, r.Error, "%s: Retry refuses EC2, so the row must not promise one", id) assert.Equal(t, "lineage-skip:"+id, r.IdempotencyKey, "%s: the stable per-account key", id) assert.NotEqual(t, "exec-skip", r.ExecutionID, id) require.Len(t, r.Recommendations, 1, "%s: the row keeps only its own account's recs, so a retry buys exactly them", id) @@ -136,10 +136,27 @@ func TestExecuteMultiAccount_RecordsAccountsSkippedAfterACommit(t *testing.T) { } } -// Without a commit the run is a flat failure that is redelivered, which re-runs -// the skipped accounts under their stable keys. A failed row for them would -// invite a manual retry beside that re-run (a new approval lineage, so no -// provider dedupe), so none is written. +// The skip row promises a retry only where Retry accepts the row. +func TestSkippedAccountNote_MatchesRetryEligibility(t *testing.T) { + cases := []struct { + provider, service string + want string + }{ + {"aws", "ec2", skippedAccountNoRetryNote}, + {"aws", "redshift", skippedAccountNoRetryNote}, + {"azure", "savingsplans", skippedAccountNoRetryNote}, + {"aws", "rds", skippedAccountRetryNote}, + {"gcp", "compute", skippedAccountRetryNote}, + } + for _, c := range cases { + row := &config.PurchaseExecution{Recommendations: []config.RecommendationRecord{{Provider: c.provider, Service: c.service}}} + assert.Equal(t, c.want, skippedAccountNote(row), c.provider+"/"+c.service) + assert.Equal(t, RedriveRefusalReason(row) == "", c.want == skippedAccountRetryNote, c.provider+"/"+c.service) + } +} + +// Without a commit the run is a flat failure: the root is finalized failed and +// a root Retry re-runs every account, so no per-account skip rows are written. func TestExecuteMultiAccount_NoSkipRowsWhenNothingCommitted(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() From f18ec4f57654b69122dfb3d2c41959e5c08fbbbb Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 19:20:06 +0100 Subject: [PATCH 3/7] fix(purchase): save account and skipped rows on live contexts A started account's terminal row (credential, pin, processing-error and final saves) now uses a detached, bounded context, so a run context that ended mid-account still leaves its History row and Retry. Skipped-account rows are built by the fan-out but saved by executeAndFinalize after the root, on the root's single terminal-save deadline, with one batched failure email inside the notify window. Rows past the deadline are logged as AUDIT LOSS. The 270s budget is unchanged. Part of #1672 Co-Authored-By: Claude Sonnet 5.5 --- internal/purchase/execution.go | 51 ++++--- .../execution_skipped_accounts_test.go | 131 +++++++++++++++++- internal/purchase/failure_notify.go | 20 +++ internal/purchase/manager.go | 38 ++++- 4 files changed, 215 insertions(+), 25 deletions(-) diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 03edf73ba..483ac203c 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -215,6 +215,10 @@ func anyRecPurchased(recs []config.RecommendationRecord) bool { type multiAccountPartialError struct { errors []string committed int + // skipped are the failed rows for accounts that never started. They are + // saved and announced by executeAndFinalize, inside the root's terminal + // windows, not here (issue #1672). + skipped []*config.PurchaseExecution } func (e *multiAccountPartialError) Error() string { @@ -272,8 +276,7 @@ func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.Purc // Some accounts committed real purchases; the per-account rows already // recorded the truth. Surface a partial sentinel so the entry points // ack and never mislabel the run as a flat failure (issue #1014). - errs = append(errs, m.recordSkippedAccounts(ctx, scoped, results, &started)...) - return &multiAccountPartialError{committed: committed, errors: errs} + return &multiAccountPartialError{committed: committed, errors: errs, skipped: skippedAccountRows(scoped, results, &started)} } return fmt.Errorf("%w: %s", errAllAccountsFailed, strings.Join(errs, "; ")) } @@ -282,11 +285,21 @@ func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.Purc // Used in error paths where we need to persist the failure status but cannot // propagate the save error (the original error is already being returned). func (m *Manager) saveExecutionStatusBestEffort(ctx context.Context, exec *config.PurchaseExecution) { - if err := m.config.SavePurchaseExecution(ctx, exec); err != nil { + if err := m.saveAccountRow(ctx, exec); err != nil { logging.Warnf("execution: failed to persist error status for account %v: %v", exec.CloudAccountID, err) } } +// saveAccountRow saves a started account's terminal row on a context detached +// from the run's cancellation and bounded by the terminal-save budget, so a run +// context that ended mid-account still leaves the History row and its Retry +// (issue #1672). Same pattern as the root's terminal save. +func (m *Manager) saveAccountRow(ctx context.Context, exec *config.PurchaseExecution) error { + saveCtx, cancel := ctxutil.DetachedWithTimeout(ctx, m.budgets.terminalSaveTimeout()) + defer cancel() + return m.config.SavePurchaseExecution(saveCtx, exec) +} + // executeForAccount runs a single plan execution for one cloud account. // It creates a new PurchaseExecution record tagged with cloud_account_id, resolves // per-account credentials, executes purchases, and saves the result. @@ -337,7 +350,7 @@ func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.Purcha // (#642), and learn whether any rec committed (#1014). _, committed = applyAccountOutcome(&acctExec, purchaseErrors) - if saveErr := m.config.SavePurchaseExecution(ctx, &acctExec); saveErr != nil { + if saveErr := m.saveAccountRow(ctx, &acctExec); saveErr != nil { return committed, fmt.Errorf("AUDIT LOSS: failed to save execution record for account %s: %w", account.ID, saveErr) } @@ -400,15 +413,15 @@ func skippedAccountNote(row *config.PurchaseExecution) string { return skippedAccountRetryNote } -// recordSkippedAccounts writes a failed row for each account the fan-out -// skipped before executeForAccount ran (the run context ended first), so the -// acked partial run does not silently drop its purchases (issue #1672). -// Nothing was submitted for it, so the row carries NoSubmission. Only the -// acked partial path calls this: a flat failure finalizes the root as failed, -// and a root Retry re-runs every account. The context has ended, so the saves -// run detached. It returns the save failures. -func (m *Manager) recordSkippedAccounts(ctx context.Context, scoped map[string]*config.PurchaseExecution, results []execution.Result[bool], started *sync.Map) []string { - var lost []string +// skippedAccountRows builds a failed row for each account the fan-out skipped +// before executeForAccount ran (the run context ended first), so the acked +// partial run does not silently drop its purchases (issue #1672). Nothing was +// submitted for it, so the row carries NoSubmission. Only the acked partial +// path calls this: a flat failure finalizes the root as failed, and a root +// Retry re-runs every account. It only builds the rows; executeAndFinalize +// saves them after the root, on the root's terminal-save deadline. +func skippedAccountRows(scoped map[string]*config.PurchaseExecution, results []execution.Result[bool], started *sync.Map) []*config.PurchaseExecution { + var rows []*config.PurchaseExecution for _, r := range results { if r.Err == nil { continue @@ -421,15 +434,9 @@ func (m *Manager) recordSkippedAccounts(ctx context.Context, scoped map[string]* row.NoSubmission = true row.Error = skippedAccountNote(&row) logging.Warnf("execution %s: account %s skipped before it started: %v", row.ExecutionID, r.AccountID, r.Err) - saveCtx, cancel := ctxutil.DetachedWithTimeout(ctx, m.budgets.terminalSaveTimeout()) - if err := m.config.SavePurchaseExecution(saveCtx, &row); err != nil { - lost = append(lost, fmt.Sprintf("AUDIT LOSS: failed to save the skipped-account record for account %s: %v", r.AccountID, err)) - } else { - m.notifyExecutionFailure(saveCtx, &row) - } - cancel() + rows = append(rows, &row) } - return lost + return rows } // persistFailedAccountExecution writes a per-account execution row that the @@ -439,7 +446,7 @@ func (m *Manager) recordSkippedAccounts(ctx context.Context, scoped map[string]* // #1184 / COR-10). Extracted from executeForAccount so the new branch keeps // that function under the gocyclo:10 budget. func (m *Manager) persistFailedAccountExecution(ctx context.Context, acctExec *config.PurchaseExecution, account config.CloudAccount, cause error) error { - if saveErr := m.config.SavePurchaseExecution(ctx, acctExec); saveErr != nil { + if saveErr := m.saveAccountRow(ctx, acctExec); saveErr != nil { return errors.Join( fmt.Errorf("AUDIT LOSS: failed to save execution record for account %s: %w", account.ID, saveErr), cause, diff --git a/internal/purchase/execution_skipped_accounts_test.go b/internal/purchase/execution_skipped_accounts_test.go index e9bd46e8b..1b803c6a1 100644 --- a/internal/purchase/execution_skipped_accounts_test.go +++ b/internal/purchase/execution_skipped_accounts_test.go @@ -3,12 +3,15 @@ package purchase import ( "context" "errors" + "fmt" "sync" "testing" "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/ctxutil" + "github.com/LeanerCloud/cloud-commitments-platform/internal/email" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -21,6 +24,7 @@ import ( // error instead of the skip. type skipFixture struct { saved []config.PurchaseExecution + live []bool // live[i]: saved[i]'s save context was not done mu sync.Mutex saveHook func(context.Context) // cancelAtB moves the cancel from A's purchase to B's credential load, so B @@ -63,6 +67,7 @@ func (f *skipFixture) run(t *testing.T, ctx context.Context, cancel context.Canc f.mu.Lock() defer f.mu.Unlock() f.saved = append(f.saved, *e) + f.live = append(f.live, saveCtx.Err() == nil) return nil } store.On("SavePurchaseHistory", mock.Anything, mock.Anything).Return(nil).Maybe() @@ -92,7 +97,16 @@ func (f *skipFixture) run(t *testing.T, ctx context.Context, cancel context.Canc return nil, nil }, }} - return m.executeMultiAccount(ctx, exec, plan, accounts) + err := m.executeMultiAccount(ctx, exec, plan, accounts) + // executeAndFinalize saves the skip rows after the root, on the root's + // terminal-save deadline; here only that step is replayed. + var partial *multiAccountPartialError + if errors.As(err, &partial) { + saveCtx, cancelSave := ctxutil.DetachedWithTimeout(ctx, m.budgets.terminalSaveTimeout()) + defer cancelSave() + m.saveSkippedRows(saveCtx, exec, partial.skipped) + } + return err } func (f *skipFixture) rowsFor(id string) []config.PurchaseExecution { @@ -210,3 +224,118 @@ func TestExecuteMultiAccount_StartedAccountGetsNoSkipRow(t *testing.T) { require.Len(t, c, 1) assert.Contains(t, c[0].Error, "nothing was submitted") } + +// B started, then the run context ended at its credential load. B's failed row +// is its only History row and Retry, so it must be saved on a live context. +func TestExecuteMultiAccount_StartedAccountRowSavedOnLiveContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f := &skipFixture{cancelAtB: true} + require.Error(t, f.run(t, ctx, cancel, true)) + + f.mu.Lock() + defer f.mu.Unlock() + found := false + for i, row := range f.saved { + if row.CloudAccountID != nil && *row.CloudAccountID == skipAcctB { + found = true + assert.Equal(t, "failed", row.Status) + assert.True(t, f.live[i], "B's failed row must not be saved on the ended run context") + } + } + require.True(t, found, "B's failed row was saved") +} + +// skipBatchHarness drives saveTerminalAndSkipped with n skip rows. Every save +// is recorded in order; skip-row saves sleep perSkipSave. +type skipBatchHarness struct { + m *Manager + mu sync.Mutex + order []string // "root" or the skip row's account id + sent []email.NotificationData + lastSkip time.Time +} + +func newSkipBatchHarness(t *testing.T, terminalSave, perSkipSave time.Duration) *skipBatchHarness { + t.Helper() + h := &skipBatchHarness{} + store := new(MockConfigStore) + mailer := new(MockEmailSender) + store.SavePurchaseExecutionFn = func(_ context.Context, e *config.PurchaseExecution) error { + time.Sleep(perSkipSave) + h.mu.Lock() + defer h.mu.Unlock() + h.order = append(h.order, *e.CloudAccountID) + h.lastSkip = time.Now() + return nil + } + store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, mock.Anything). + Run(func(mock.Arguments) { + h.mu.Lock() + defer h.mu.Unlock() + h.order = append(h.order, "root") + }).Return(nil) + store.On("GetGlobalConfig", mock.Anything).Return(failureNotifyCfg(), nil).Maybe() + mailer.On("SendPurchaseFailedNotification", mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + h.mu.Lock() + defer h.mu.Unlock() + h.sent = append(h.sent, args.Get(1).(email.NotificationData)) + }).Return(nil).Maybe() + h.m = &Manager{config: store, email: mailer, budgets: purchaseBudgets{terminalSave: terminalSave}} + return h +} + +func skipBatchRows(n int) []*config.PurchaseExecution { + var rows []*config.PurchaseExecution + for i := 0; i < n; i++ { + id := fmt.Sprintf("acct-%d", i) + rows = append(rows, &config.PurchaseExecution{ + ExecutionID: "skip-" + id, Status: "failed", NoSubmission: true, CloudAccountID: &id, + Recommendations: []config.RecommendationRecord{{Service: "rds", ResourceType: "db.m5.large", Region: "us-east-1", Count: 1, Selected: true}}, + }) + } + return rows +} + +// The root saves first, the skip rows share its one deadline, and one email +// covers every saved row (issue #1672). +func TestSaveTerminalAndSkipped_RootFirstOneEmail(t *testing.T) { + h := newSkipBatchHarness(t, time.Second, 0) + root := &config.PurchaseExecution{ExecutionID: "root", PlanID: "plan", Status: "partially_completed"} + partial := &multiAccountPartialError{committed: 1, skipped: skipBatchRows(3)} + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.NoError(t, h.m.saveTerminalAndSkipped(ctx, root, partial)) + + assert.Equal(t, []string{"root", "acct-0", "acct-1", "acct-2"}, h.order, "root first, then the skip rows") + require.Len(t, h.sent, 1, "one batched email, not one per row") + assert.Equal(t, "root", h.sent[0].ExecutionID) + assert.Len(t, h.sent[0].Recommendations, 3) +} + +// Rows that do not fit in the root's terminal-save window are AUDIT LOSS and +// never extend it or fail the root. +func TestSaveTerminalAndSkipped_SharedDeadline(t *testing.T) { + h := newSkipBatchHarness(t, 50*time.Millisecond, 30*time.Millisecond) + root := &config.PurchaseExecution{ExecutionID: "root", PlanID: "plan", Status: "partially_completed"} + partial := &multiAccountPartialError{committed: 1, skipped: skipBatchRows(5)} + + ctx, cancel := ctxutil.DetachedWithTimeout(context.Background(), h.m.budgets.terminalSaveTimeout()) + defer cancel() + start := time.Now() + require.NoError(t, h.m.saveTerminalAndSkipped(ctx, root, partial), "a lost skip row never fails the root") + elapsed := time.Since(start) + + require.NotEmpty(t, h.order) + assert.Equal(t, "root", h.order[0]) + saved := len(h.order) - 1 + assert.Less(t, saved, 5, "rows past the deadline are not saved") + // Deadline plus at most one in-flight save and scheduling slack. + assert.Less(t, elapsed, 50*time.Millisecond+30*time.Millisecond+100*time.Millisecond, "one shared window, not one per row") + if saved > 0 { + require.Len(t, h.sent, 1) + assert.Len(t, h.sent[0].Recommendations, saved, "the email covers only the saved rows") + } +} diff --git a/internal/purchase/failure_notify.go b/internal/purchase/failure_notify.go index 211b5bb31..aa991f14d 100644 --- a/internal/purchase/failure_notify.go +++ b/internal/purchase/failure_notify.go @@ -79,6 +79,26 @@ func (m *Manager) sendFailureNotice(ctx context.Context, exec *config.PurchaseEx } } +// notifySkippedAccounts logs the marker for each saved skip row and sends ONE +// purchase-failed email listing all their recs, so a run that skipped many +// accounts costs one failureNotifyTimeout window, not one per row (#1672). +func (m *Manager) notifySkippedAccounts(ctx context.Context, root *config.PurchaseExecution, rows []*config.PurchaseExecution) { + batch := &config.PurchaseExecution{ExecutionID: root.ExecutionID, PlanID: root.PlanID} + for _, row := range rows { + if logExecutionFailure(row) { + batch.Recommendations = append(batch.Recommendations, row.Recommendations...) + } + } + if len(batch.Recommendations) == 0 || m.email == nil { + return + } + ctx, cancel := ctxutil.DetachedWithTimeout(ctx, failureNotifyTimeout) + defer cancel() + if err := m.email.SendPurchaseFailedNotification(ctx, m.buildPurchaseFailedData(ctx, batch)); err != nil { + logging.Errorf("Failed to send skipped-accounts notification for execution %s: %v", root.ExecutionID, err) + } +} + // buildPurchaseFailedData lists the selected recommendations that did not // purchase (or, for an outcome-unknown row, whose purchase is unconfirmed) and // addresses the email to the global notification address (the diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index b719366bc..dbbd4c93f 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -271,7 +271,10 @@ func isMultiAccountAckable(execErr error) bool { const ( // The two budgets run back to back, and for a failed or partial row the // failure alert follows them with its own failureNotifyTimeout (30s), so a - // run can take 3m30s + 30s + 30s = 270s. On AWS Lambda the function timeout + // run can take 3m30s + 30s + 30s = 270s. The failed rows of accounts a + // fan-out skipped (#1672) add no window: they are saved after the root, on + // the root's terminal-save deadline, and announced in one email inside the + // failure-alert window. On AWS Lambda the function timeout // ends the invocation whatever these deadlines say, so the Lambda timeout // must cover that sum plus a 30s margin (lambda_timeout >= 300, enforced by // Terraform validation and TestLambdaTimeoutCoversPurchaseBudget, #913, @@ -342,7 +345,7 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE // when the terminal save below fails. A fan-out root logs the alarm // marker without an email (issue #1195; see notifyRootFailure). defer m.notifyRootFailure(saveCtx, exec, execErr) - if err := m.saveOrRecoverTerminal(saveCtx, exec, execErr); err != nil { + if err := m.saveTerminalAndSkipped(saveCtx, 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 @@ -366,6 +369,37 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE return execErr } +// saveTerminalAndSkipped saves the root row, then the rows of accounts a +// fan-out skipped, all on ctx's single terminal-save deadline (issue #1672). +// The root goes first and its result is the return value: a skip row that does +// not fit in the deadline is logged as AUDIT LOSS and never fails the root. +func (m *Manager) saveTerminalAndSkipped(ctx context.Context, exec *config.PurchaseExecution, execErr error) error { + err := m.saveOrRecoverTerminal(ctx, exec, execErr) + var partial *multiAccountPartialError + if errors.As(execErr, &partial) { + m.saveSkippedRows(ctx, exec, partial.skipped) + } + return err +} + +// saveSkippedRows saves each skip row until ctx ends and sends one email for +// the rows it saved. +func (m *Manager) saveSkippedRows(ctx context.Context, root *config.PurchaseExecution, rows []*config.PurchaseExecution) { + var saved []*config.PurchaseExecution + for _, row := range rows { + err := ctx.Err() + if err == nil { + err = m.config.SavePurchaseExecution(ctx, row) + } + if err != nil { + logging.Errorf("AUDIT LOSS: failed to save the skipped-account record for account %v of execution %s: %v", row.CloudAccountID, root.ExecutionID, err) + continue + } + saved = append(saved, row) + } + m.notifySkippedAccounts(ctx, root, saved) +} + // preExecutionRefusal runs every check that must pass before any provider call. // All refusals except the armed re-drive one are beforeSubmission errors, so // the row ends failed with no_submission stamped. From 7adf1b0c628cb46fd2647c3231219f0cc7f0bbe7 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 19:57:58 +0100 Subject: [PATCH 4/7] fix(purchase): anchor account row saves on the run deadline Account rows save on a context detached from the run, with a deadline of the run deadline plus half the terminal-save budget, and the root save ends at the run deadline plus the full budget, so accounts in flight when the run ends cannot starve the root. A row that misses its slice is logged as AUDIT LOSS with the account and execution IDs. Part of #1672 Co-Authored-By: Claude Sonnet 5.5 --- internal/ctxutil/detached.go | 10 ++ internal/ctxutil/detached_test.go | 26 ++++ .../purchase/account_row_deadline_test.go | 111 ++++++++++++++++++ internal/purchase/execute_detached_test.go | 7 +- internal/purchase/execution.go | 30 ++++- internal/purchase/manager.go | 30 +++-- 6 files changed, 198 insertions(+), 16 deletions(-) create mode 100644 internal/purchase/account_row_deadline_test.go diff --git a/internal/ctxutil/detached.go b/internal/ctxutil/detached.go index 67daf3662..fd009c9f1 100644 --- a/internal/ctxutil/detached.go +++ b/internal/ctxutil/detached.go @@ -17,3 +17,13 @@ import ( func DetachedWithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { return context.WithTimeout(context.WithoutCancel(ctx), d) } + +// DetachedWithDeadline is DetachedWithTimeout with an absolute deadline: ctx's +// values are kept, its cancellation and deadline are dropped, and the result +// ends at deadline. Use it when several detached phases must share one +// schedule anchored on an earlier deadline instead of each starting a fresh +// timeout from "now". A deadline already in the past yields a context that is +// already done. Always defer the returned CancelFunc. +func DetachedWithDeadline(ctx context.Context, deadline time.Time) (context.Context, context.CancelFunc) { + return context.WithDeadline(context.WithoutCancel(ctx), deadline) +} diff --git a/internal/ctxutil/detached_test.go b/internal/ctxutil/detached_test.go index ffeaacf9c..c2bac805c 100644 --- a/internal/ctxutil/detached_test.go +++ b/internal/ctxutil/detached_test.go @@ -35,3 +35,29 @@ func TestDetachedWithTimeout(t *testing.T) { t.Fatal("the returned CancelFunc must cancel the detached context") } } + +func TestDetachedWithDeadline(t *testing.T) { + parent, cancelParent := context.WithTimeout(context.WithValue(context.Background(), ctxKey{}, "req-2"), time.Millisecond) + cancelParent() + + want := time.Now().Add(time.Hour) + ctx, cancel := DetachedWithDeadline(parent, want) + defer cancel() + + if err := ctx.Err(); err != nil { + t.Fatalf("parent cancellation must not reach the detached context, got %v", err) + } + got, ok := ctx.Deadline() + if !ok || !got.Equal(want) { + t.Fatalf("deadline must be exactly the absolute one given: got %v ok=%v want %v", got, ok, want) + } + if v := ctx.Value(ctxKey{}); v != "req-2" { + t.Fatalf("detached context must keep parent values, got %v", v) + } + + past, cancelPast := DetachedWithDeadline(context.Background(), time.Now().Add(-time.Second)) + defer cancelPast() + if past.Err() == nil { + t.Fatal("a deadline in the past must give an already-done context") + } +} diff --git a/internal/purchase/account_row_deadline_test.go b/internal/purchase/account_row_deadline_test.go new file mode 100644 index 000000000..c00ba627d --- /dev/null +++ b/internal/purchase/account_row_deadline_test.go @@ -0,0 +1,111 @@ +package purchase + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +const lateTerminalT = 200 * time.Millisecond // T; the account slice T/2 is 100ms + +// accountSaveCase triggers the run context ending at one of the three places +// executeForAccount saves its row. +type accountSaveCase struct { + name string + trigger func(f *lateFixtureOneAccount) + status string +} + +type lateFixtureOneAccount struct { + cancel context.CancelFunc + store *MockConfigStore + client *MockServiceClient + loadRaw func(context.Context, string, string) ([]byte, error) +} + +// Each of executeForAccount's three row saves must run on a live context whose +// deadline is the run's deadline plus T/2, however the run context ended. +func TestExecuteForAccount_RowSavesAreLiveAndAnchoredOnTheRunDeadline(t *testing.T) { + keys := func(context.Context, string, string) ([]byte, error) { + return []byte(`{"access_key_id":"AKIAIOSFODNN7EXAMPLE","secret_access_key":"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"}`), nil + } + cases := []accountSaveCase{ + {"processing error", func(f *lateFixtureOneAccount) { + f.store.On("GetGlobalConfig", mock.Anything).Run(func(mock.Arguments) { f.cancel() }).Return(nil, errors.New("db down")) + }, "failed"}, + {"outcome", func(f *lateFixtureOneAccount) { + f.client.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything). + Run(func(mock.Arguments) { f.cancel() }).Return(common.PurchaseResult{Success: true, CommitmentID: "ri-1"}, nil) + }, "completed"}, + {"credential failure", func(f *lateFixtureOneAccount) { + f.loadRaw = func(context.Context, string, string) ([]byte, error) { f.cancel(); return nil, nil } + }, "failed"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + store := new(MockConfigStore) + mailer := allowFailedNotification(new(MockEmailSender)) + factory := new(MockProviderFactory) + prov := new(MockProvider) + client := new(MockServiceClient) + runCtx, runCancel := context.WithTimeout(context.Background(), time.Hour) + defer runCancel() + f := &lateFixtureOneAccount{cancel: runCancel, store: store, client: client, loadRaw: keys} + + var saved []config.PurchaseExecution + var live []bool + var deadlines []time.Time + store.SavePurchaseExecutionFn = func(ctx context.Context, e *config.PurchaseExecution) error { + saved = append(saved, *e) + live = append(live, ctx.Err() == nil) + dl, _ := ctx.Deadline() + deadlines = append(deadlines, dl) + return nil + } + store.On("SavePurchaseHistory", mock.Anything, mock.Anything).Return(nil).Maybe() + mailer.On("SendPurchaseConfirmation", mock.Anything, mock.Anything).Return(nil).Maybe() + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil).Maybe() + prov.On("GetServiceClient", mock.Anything, common.ServiceEC2, "us-east-1").Return(client, nil).Maybe() + tc.trigger(f) + + m := &Manager{config: store, email: mailer, providerFactory: factory, + budgets: purchaseBudgets{terminalSave: lateTerminalT}, + credStore: &MockCredentialStore{LoadRawFn: func(c context.Context, a, k string) ([]byte, error) { + return f.loadRaw(c, a, k) + }}} + account := config.CloudAccount{ID: skipAcctA, Name: "A", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys"} + base := &config.PurchaseExecution{ExecutionID: "exec-pin", PlanID: "plan-pin", IdempotencyKey: "lineage-pin", + Recommendations: []config.RecommendationRecord{withAccount(config.RecommendationRecord{Provider: "aws", Service: "ec2", + ResourceType: "m5.large", Region: "us-east-1", Count: 1, Selected: true}, skipAcctA)}} + runDeadline, _ := runCtx.Deadline() + + _, _ = m.executeForAccount(runCtx, base, &config.PurchasePlan{ID: "plan-pin", Name: "pin"}, account) + + require.Len(t, saved, 1) + assert.Equal(t, tc.status, saved[0].Status) + assert.True(t, live[0], "the row is saved on a context the run's end did not cancel") + assert.True(t, deadlines[0].Equal(runDeadline.Add(lateTerminalT/2)), "deadline is E+T/2, got %v want %v", deadlines[0], runDeadline.Add(lateTerminalT/2)) + }) + } +} + +// Without a run deadline the account row falls back to a fresh full budget. +func TestSaveAccountRow_NoRunDeadlineFallsBackToFullBudget(t *testing.T) { + store := new(MockConfigStore) + var deadline time.Time + store.SavePurchaseExecutionFn = func(ctx context.Context, _ *config.PurchaseExecution) error { + deadline, _ = ctx.Deadline() + return nil + } + m := &Manager{config: store, budgets: purchaseBudgets{terminalSave: time.Second}} + start := time.Now() + require.NoError(t, m.saveAccountRow(context.Background(), &config.PurchaseExecution{})) + assert.WithinDuration(t, start.Add(time.Second), deadline, 100*time.Millisecond) +} diff --git a/internal/purchase/execute_detached_test.go b/internal/purchase/execute_detached_test.go index 260875735..9ca64d53b 100644 --- a/internal/purchase/execute_detached_test.go +++ b/internal/purchase/execute_detached_test.go @@ -90,10 +90,13 @@ func TestExecuteAndFinalize_SlowRunGetsFreshTerminalSaveBudget(t *testing.T) { const ( budget = 2 * time.Second // terminal save and per-rec cap callTime = 1200 * time.Millisecond // each call fits its cap; two exceed the save budget + // The root save deadline is execDeadline+budget (#1672), so the + // execution budget is scaled down too; the run ends well inside it. + execTimeout = 10 * time.Second ) t.Setenv("CUDLY_MAX_ACCOUNT_PARALLELISM", "1") f := newDetachFixture(t) - f.manager.budgets = purchaseBudgets{terminalSave: budget, perRec: budget} + f.manager.budgets = purchaseBudgets{execTimeout: execTimeout, terminalSave: budget, perRec: budget} f.exec.Recommendations = append(f.exec.Recommendations, f.exec.Recommendations[0]) f.exec.Recommendations[1].ResourceType = "m5.xlarge" var active atomic.Int32 @@ -124,7 +127,7 @@ func TestExecuteAndFinalize_SlowRunGetsFreshTerminalSaveBudget(t *testing.T) { require.NoError(t, err) assert.Equal(t, []string{"completed"}, f.saved) assert.Greater(t, f.saveRemaining, budget-time.Second) - assert.LessOrEqual(t, f.saveRemaining, budget) + assert.LessOrEqual(t, f.saveRemaining, execTimeout+budget) } var stsOut = sts.GetCallerIdentityOutput{Account: aws.String("123456789012")} diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 483ac203c..ceefda7b6 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -291,13 +291,33 @@ func (m *Manager) saveExecutionStatusBestEffort(ctx context.Context, exec *confi } // saveAccountRow saves a started account's terminal row on a context detached -// from the run's cancellation and bounded by the terminal-save budget, so a run -// context that ended mid-account still leaves the History row and its Retry -// (issue #1672). Same pattern as the root's terminal save. +// from the run's cancellation, so a run context that ended mid-account still +// leaves the History row and its Retry (issue #1672). The deadline is half the +// terminal-save budget past the run's own deadline (execDeadline+T/2), so +// every account in flight when the run ends shares one slice and the root keeps +// the other half. A row that misses it is logged as AUDIT LOSS. func (m *Manager) saveAccountRow(ctx context.Context, exec *config.PurchaseExecution) error { - saveCtx, cancel := ctxutil.DetachedWithTimeout(ctx, m.budgets.terminalSaveTimeout()) + deadline := time.Now().Add(m.budgets.terminalSaveTimeout()) + if runDeadline, ok := ctx.Deadline(); ok { + // ctx is the run context. Without a deadline (tests, callers outside + // executeAndFinalize) the fallback is a fresh full budget. + deadline = runDeadline.Add(m.budgets.terminalSaveTimeout() / 2) + } + saveCtx, cancel := ctxutil.DetachedWithDeadline(ctx, deadline) defer cancel() - return m.config.SavePurchaseExecution(saveCtx, exec) + err := m.config.SavePurchaseExecution(saveCtx, exec) + if err != nil { + logging.Errorf("AUDIT LOSS: failed to save the record of execution %s for account %s: %v", exec.ExecutionID, accountIDOf(exec), err) + } + return err +} + +// accountIDOf is the row's cloud account ID for log lines ("" for a root row). +func accountIDOf(row *config.PurchaseExecution) string { + if row.CloudAccountID == nil { + return "" + } + return *row.CloudAccountID } // executeForAccount runs a single plan execution for one cloud account. diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index dbbd4c93f..609a0bbb9 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -269,12 +269,12 @@ func isMultiAccountAckable(execErr error) bool { } const ( - // The two budgets run back to back, and for a failed or partial row the - // failure alert follows them with its own failureNotifyTimeout (30s), so a - // run can take 3m30s + 30s + 30s = 270s. The failed rows of accounts a - // fan-out skipped (#1672) add no window: they are saved after the root, on - // the root's terminal-save deadline, and announced in one email inside the - // failure-alert window. On AWS Lambda the function timeout + // The budgets are anchored on the execution deadline E (3m30s), not on + // when the previous phase ended: account rows save by E+T/2, the root and + // the skip rows by E+T, and one batched email for the skipped and late + // accounts goes out by E+T+N, so the worst case is 3m30s + 30s + 30s = + // 270s however many accounts are in flight at E (#1672). A late account + // sends no email of its own (executeForAccountLate). On AWS Lambda the function timeout // ends the invocation whatever these deadlines say, so the Lambda timeout // must cover that sum plus a 30s margin (lambda_timeout >= 300, enforced by // Terraform validation and TestLambdaTimeoutCoversPurchaseBudget, #913, @@ -292,10 +292,18 @@ const ( // purchaseBudgets overrides deadlines for tests. A zero field means the // production default, so a Manager built by NewManager uses the constants. type purchaseBudgets struct { + execTimeout time.Duration terminalSave time.Duration perRec time.Duration } +func (b purchaseBudgets) executionTimeout() time.Duration { + if b.execTimeout > 0 { + return b.execTimeout + } + return purchaseExecutionTimeout +} + func (b purchaseBudgets) terminalSaveTimeout() time.Duration { if b.terminalSave > 0 { return b.terminalSave @@ -324,8 +332,9 @@ func (b purchaseBudgets) perRecTimeout() time.Duration { func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseExecution) error { // A disconnected caller must not cancel a purchase after the claim. // Its terminal outcome must survive an exhausted execution budget too. - execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, purchaseExecutionTimeout) + execCtx, cancelExec := ctxutil.DetachedWithTimeout(ctx, m.budgets.executionTimeout()) defer cancelExec() + execDeadline, _ := execCtx.Deadline() // always set: DetachedWithTimeout // Last line of defense before money moves (issue #1718). Every executor // entry point funnels through here, so one check covers all of them. execErr := m.preExecutionRefusal(execCtx, exec) @@ -339,7 +348,10 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE if execErr != nil { logging.Errorf("Failed to execute purchase %s: %v", exec.ExecutionID, execErr) } - saveCtx, cancelSave := ctxutil.DetachedWithTimeout(ctx, m.budgets.terminalSaveTimeout()) + // Anchored on the execution deadline, not "now": the late accounts' row + // saves end at execDeadline+T/2 (saveAccountRow), so the root always keeps + // the rest of its slice and the whole run ends by execDeadline+T+N (#1672). + saveCtx, cancelSave := ctxutil.DetachedWithDeadline(ctx, execDeadline.Add(m.budgets.terminalSaveTimeout())) defer cancelSave() // Operators hear about every failed or partial row (issue #714), even // when the terminal save below fails. A fan-out root logs the alarm @@ -392,7 +404,7 @@ func (m *Manager) saveSkippedRows(ctx context.Context, root *config.PurchaseExec err = m.config.SavePurchaseExecution(ctx, row) } if err != nil { - logging.Errorf("AUDIT LOSS: failed to save the skipped-account record for account %v of execution %s: %v", row.CloudAccountID, root.ExecutionID, err) + logging.Errorf("AUDIT LOSS: failed to save the skipped-account record for account %s of execution %s: %v", accountIDOf(row), root.ExecutionID, err) continue } saved = append(saved, row) From f6c1c74261c949deedce6b0c9169b9fb0900dfe6 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 19:58:49 +0100 Subject: [PATCH 5/7] fix(purchase): batch late account failures into one outcome-unknown email An account that finishes after the run context ended sends no email of its own; its row joins the skipped accounts in the single email sent after the root save, so the worst case stays inside the 270s run budget. A provider call cut off by the run context is wrapped with ErrOutcomeUnknown, and the batched email uses the outcome-unknown wording when any late row carries it. Part of #1672 Co-Authored-By: Claude Sonnet 5.5 --- internal/purchase/execution.go | 43 +++- internal/purchase/failure_notify.go | 24 ++- internal/purchase/late_accounts.go | 79 +++++++ internal/purchase/late_accounts_test.go | 266 ++++++++++++++++++++++++ internal/purchase/manager.go | 16 +- 5 files changed, 410 insertions(+), 18 deletions(-) create mode 100644 internal/purchase/late_accounts.go create mode 100644 internal/purchase/late_accounts_test.go diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index ceefda7b6..3138cd116 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -252,11 +252,18 @@ func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.Purc } } var started sync.Map + late := &lateRows{} results := execution.RunForAccountsWithConcurrency(ctx, buying, func(ctx context.Context, account config.CloudAccount) (bool, error) { started.Store(account.ID, true) - return m.executeForAccount(ctx, scoped[account.ID], plan, account) + return m.executeForAccountLate(ctx, scoped[account.ID], plan, account, late) }, getMaxAccountParallelism()) + return withLateRows(summarizeFanOut(scoped, results, &started), late.take()) +} +// summarizeFanOut turns the per-account results into the fan-out's error (see +// executeMultiAccount for the contract). Split out to keep both under the +// gocyclo budget. +func summarizeFanOut(scoped map[string]*config.PurchaseExecution, results []execution.Result[bool], started *sync.Map) error { committed := 0 var errs []string for _, r := range results { @@ -276,7 +283,7 @@ func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.Purc // Some accounts committed real purchases; the per-account rows already // recorded the truth. Surface a partial sentinel so the entry points // ack and never mislabel the run as a flat failure (issue #1014). - return &multiAccountPartialError{committed: committed, errors: errs, skipped: skippedAccountRows(scoped, results, &started)} + return &multiAccountPartialError{committed: committed, errors: errs, skipped: skippedAccountRows(scoped, results, started)} } return fmt.Errorf("%w: %s", errAllAccountsFailed, strings.Join(errs, "; ")) } @@ -331,12 +338,20 @@ func accountIDOf(row *config.PurchaseExecution) string { // failed, but the authoritative per-account row has already been saved with the // correct status (partially_completed / failed) before it surfaces. func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.PurchaseExecution, plan *config.PurchasePlan, account config.CloudAccount) (committed bool, err error) { + return m.executeForAccountLate(ctx, baseExec, plan, account, nil) +} + +// executeForAccountLate is executeForAccount for the fan-out: when the account +// finishes after ctx ended it hands its failed row to late instead of sending +// its own email, and executeAndFinalize sends one batched email after the root +// save (issue #1672). A nil late keeps the per-account email. +func (m *Manager) executeForAccountLate(ctx context.Context, baseExec *config.PurchaseExecution, plan *config.PurchasePlan, account config.CloudAccount, late *lateRows) (committed bool, err error) { // Create a per-account copy of the execution record with an independent // Recommendations slice so concurrent goroutines don't race on writes. acctExec := newAccountExecution(baseExec, account.ID) // Runs on every return path below, once the row's terminal status is // stamped; a no-op unless it ended failed or partially_completed (#714). - defer m.notifyExecutionFailure(ctx, &acctExec) + defer m.notifyAccountOutcome(ctx, &acctExec, late) provCfg, err := m.resolveAccountProvider(ctx, account) if err != nil { @@ -395,6 +410,16 @@ func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.Purcha return committed, nil } +// notifyAccountOutcome is executeForAccountLate's failure alert: the account's +// own email, unless the run context already ended, in which case a failed row +// goes to late for the batched email. +func (m *Manager) notifyAccountOutcome(ctx context.Context, row *config.PurchaseExecution, late *lateRows) { + if ctx.Err() != nil && late.add(row) { + return + } + m.notifyExecutionFailure(ctx, row) +} + // newAccountExecution is the per-account copy of baseExec, with an independent // Recommendations slice so concurrent goroutines don't race on writes. // @@ -1320,7 +1345,7 @@ func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.Recommen logRecCtxErr(opts.ExecutionID, recTuple, elapsed, recCtx.Err()) logging.Errorf("purchase[%s]: %s/%s/%s/%s PurchaseCommitment failed after %s: %v", opts.ExecutionID, rec.Provider, rec.Service, rec.Region, rec.ResourceType, elapsed, err) - return result, fmt.Errorf("purchase failed: %w", err) + return result, purchaseCallError(ctx, err) } if !result.Success { @@ -1343,6 +1368,16 @@ func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.Recommen return result, nil } +// purchaseCallError wraps a failed PurchaseCommitment. When the run context +// ended mid-call the request may have landed, so the error also carries +// ErrOutcomeUnknown: the row records an unknown outcome, never a plain failure. +func purchaseCallError(ctx context.Context, err error) error { + if ctx.Err() != nil && !errors.Is(err, common.ErrOutcomeUnknown) { + return fmt.Errorf("purchase failed: %w: %w", common.ErrOutcomeUnknown, err) + } + return fmt.Errorf("purchase failed: %w", err) +} + // mapServiceType maps a service string to common.ServiceType. Both the // canonical hyphenated slugs (compute, relational-db, cache, search, // data-warehouse) and the legacy AWS-only slugs (ec2, rds, elasticache, diff --git a/internal/purchase/failure_notify.go b/internal/purchase/failure_notify.go index aa991f14d..f69980e45 100644 --- a/internal/purchase/failure_notify.go +++ b/internal/purchase/failure_notify.go @@ -79,23 +79,33 @@ func (m *Manager) sendFailureNotice(ctx context.Context, exec *config.PurchaseEx } } -// notifySkippedAccounts logs the marker for each saved skip row and sends ONE -// purchase-failed email listing all their recs, so a run that skipped many -// accounts costs one failureNotifyTimeout window, not one per row (#1672). -func (m *Manager) notifySkippedAccounts(ctx context.Context, root *config.PurchaseExecution, rows []*config.PurchaseExecution) { +// notifyBatchedAccounts logs the marker for each skipped and late row and sends +// ONE purchase-failed email listing all their recs, so a run that skipped many +// accounts costs one failureNotifyTimeout window, not one per row (#1672). The +// late rows are accounts that finished after the run context ended; they were +// cut off mid-call, so the email says "outcome unknown" when any of them +// carries an outcome-unknown rec error. The window starts after the root save, +// which ends by execDeadline+T, so the email ends by execDeadline+T+N. +func (m *Manager) notifyBatchedAccounts(ctx context.Context, root *config.PurchaseExecution, skipped, late []*config.PurchaseExecution) { batch := &config.PurchaseExecution{ExecutionID: root.ExecutionID, PlanID: root.PlanID} - for _, row := range rows { + outcomeUnknown := false + for _, row := range append(append([]*config.PurchaseExecution{}, skipped...), late...) { if logExecutionFailure(row) { batch.Recommendations = append(batch.Recommendations, row.Recommendations...) } } + for _, row := range late { + outcomeUnknown = outcomeUnknown || hasOutcomeUnknownRec(row) + } if len(batch.Recommendations) == 0 || m.email == nil { return } ctx, cancel := ctxutil.DetachedWithTimeout(ctx, failureNotifyTimeout) defer cancel() - if err := m.email.SendPurchaseFailedNotification(ctx, m.buildPurchaseFailedData(ctx, batch)); err != nil { - logging.Errorf("Failed to send skipped-accounts notification for execution %s: %v", root.ExecutionID, err) + data := m.buildPurchaseFailedData(ctx, batch) + data.OutcomeUnknown = outcomeUnknown + if err := m.email.SendPurchaseFailedNotification(ctx, data); err != nil { + logging.Errorf("Failed to send batched account notification for execution %s: %v", root.ExecutionID, err) } } diff --git a/internal/purchase/late_accounts.go b/internal/purchase/late_accounts.go new file mode 100644 index 000000000..a223cbcdb --- /dev/null +++ b/internal/purchase/late_accounts.go @@ -0,0 +1,79 @@ +package purchase + +import ( + "errors" + "strings" + "sync" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// A late account is one that finished after the run context ended (issue +// #1672). It already saved its own row, on its slice of the terminal window, +// but it sends no email of its own: the run's single batched email goes out +// after the root save, so the failure alerts of a fan-out cannot stack up past +// the Lambda budget. The row travels back to executeAndFinalize on the error +// executeMultiAccount returns. + +// lateRows collects the late accounts' rows from the fan-out goroutines. +type lateRows struct { + mu sync.Mutex + rows []*config.PurchaseExecution +} + +// add keeps a copy of row when it ended failed or partially_completed, the only +// statuses that email, and reports whether it did. A nil collector keeps nothing. +func (l *lateRows) add(row *config.PurchaseExecution) bool { + if l == nil || (row.Status != "failed" && row.Status != "partially_completed") { + return false + } + cp := *row + l.mu.Lock() + defer l.mu.Unlock() + l.rows = append(l.rows, &cp) + return true +} + +func (l *lateRows) take() []*config.PurchaseExecution { + l.mu.Lock() + defer l.mu.Unlock() + return l.rows +} + +// lateAccountsError carries the late rows alongside the fan-out's own error. +// It is transparent: same message, and errors.Is/As reach the wrapped error. +type lateAccountsError struct { + err error + rows []*config.PurchaseExecution +} + +func (e *lateAccountsError) Error() string { return e.err.Error() } +func (e *lateAccountsError) Unwrap() error { return e.err } + +func withLateRows(err error, rows []*config.PurchaseExecution) error { + if err == nil || len(rows) == 0 { + return err + } + return &lateAccountsError{err: err, rows: rows} +} + +func lateRowsOf(err error) []*config.PurchaseExecution { + var late *lateAccountsError + if errors.As(err, &late) { + return late.rows + } + return nil +} + +// hasOutcomeUnknownRec reports whether a rec of row failed with the go +// library's ErrOutcomeUnknown: its purchase call may have reached the cloud, so +// the commitment may exist. The rec stores only the error text. +func hasOutcomeUnknownRec(row *config.PurchaseExecution) bool { + for i := range row.Recommendations { + if strings.Contains(row.Recommendations[i].Error, common.ErrOutcomeUnknown.Error()) { + return true + } + } + return false +} diff --git a/internal/purchase/late_accounts_test.go b/internal/purchase/late_accounts_test.go new file mode 100644 index 000000000..3a44704df --- /dev/null +++ b/internal/purchase/late_accounts_test.go @@ -0,0 +1,266 @@ +package purchase + +import ( + "bytes" + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/email" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +const ( + lateExecTimeout = 300 * time.Millisecond // E + lateMailDelay = 100 * time.Millisecond + lateAcctD = "dddddddd-0000-0000-0000-000000000004" +) + +type lateSave struct { + exec config.PurchaseExecution + deadline time.Time + done time.Time +} + +// lateLogBuf is a goroutine-safe log sink. +type lateLogBuf struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lateLogBuf) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *lateLogBuf) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// lateFixture drives executeAndFinalize over a four-account fan-out, one +// account at a time (or two): A buys at once, B is in flight when the run +// deadline passes, C and D never start (issue #1672). +type lateFixture struct { + m *Manager + exec *config.PurchaseExecution + mu sync.Mutex + saves []lateSave + mails []email.NotificationData + logs *lateLogBuf +} + +// newLateFixture builds the fan-out. acctSave handles the saves of started +// accounts (rows that are neither the root nor a skip row); nil saves at once. +func newLateFixture(t *testing.T, parallelism string, acctSave func(context.Context, *config.PurchaseExecution) error) *lateFixture { + t.Helper() + t.Setenv("CUDLY_MAX_ACCOUNT_PARALLELISM", parallelism) + f := &lateFixture{logs: &lateLogBuf{}} + prev := logging.SetOutput(f.logs) + t.Cleanup(func() { logging.SetOutput(prev) }) + + store := new(MockConfigStore) + mailer := new(MockEmailSender) + factory := new(MockProviderFactory) + prov := new(MockProvider) + client := new(MockServiceClient) + + ids := []string{skipAcctA, skipAcctB, skipAcctC, lateAcctD} + types := []string{"a.large", "b.large", "c.large", "d.large"} + var accounts []config.CloudAccount + var recs []config.RecommendationRecord + for i, id := range ids { + accounts = append(accounts, config.CloudAccount{ID: id, Name: id[:1], Provider: "aws", ExternalID: strings.Repeat(string(rune('1'+i)), 12), AWSAuthMode: "access_keys"}) + recs = append(recs, withAccount(config.RecommendationRecord{Provider: "aws", Service: "ec2", ResourceType: types[i], Region: "us-east-1", + Count: 1, Savings: 75, UpfrontCost: 300, Selected: true}, id)) + } + f.exec = &config.PurchaseExecution{ExecutionID: "exec-late", PlanID: "plan-late", Status: "running", IdempotencyKey: "lineage-late", Recommendations: recs} + + store.GetPlanAccountsFn = func(context.Context, string) ([]config.CloudAccount, error) { return accounts, nil } + store.SavePurchaseExecutionFn = func(ctx context.Context, e *config.PurchaseExecution) error { + s := lateSave{exec: *e} + s.deadline, _ = ctx.Deadline() + var err error + if e.CloudAccountID != nil && !e.NoSubmission && acctSave != nil { + err = acctSave(ctx, e) + } + s.done = time.Now() + f.mu.Lock() + defer f.mu.Unlock() + f.saves = append(f.saves, s) + return err + } + store.On("SavePurchaseHistory", mock.Anything, mock.Anything).Return(nil).Maybe() + mailer.On("SendPurchaseConfirmation", mock.Anything, mock.Anything).Return(nil).Maybe() + mailer.On("SendPurchaseFailedNotification", mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + time.Sleep(lateMailDelay) + f.mu.Lock() + defer f.mu.Unlock() + f.mails = append(f.mails, args.Get(1).(email.NotificationData)) + }).Return(nil).Maybe() + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil).Maybe() + prov.On("GetServiceClient", mock.Anything, common.ServiceEC2, "us-east-1").Return(client, nil).Maybe() + byType := func(rt string) interface{} { + return mock.MatchedBy(func(r common.Recommendation) bool { return r.ResourceType == rt }) + } + client.On("PurchaseCommitment", mock.Anything, byType("a.large"), mock.Anything). + Return(common.PurchaseResult{Success: true, CommitmentID: "ri-a"}, nil).Maybe() + // B's provider call hangs until the run context ends: it may have landed. + client.On("PurchaseCommitment", mock.Anything, byType("b.large"), mock.Anything). + Run(func(args mock.Arguments) { <-args.Get(0).(context.Context).Done() }). + Return(common.PurchaseResult{}, context.DeadlineExceeded).Maybe() + + f.m = &Manager{config: store, email: mailer, providerFactory: factory, + budgets: purchaseBudgets{execTimeout: lateExecTimeout, terminalSave: lateTerminalT, perRec: 5 * time.Second}, + credStore: &MockCredentialStore{LoadRawFn: func(context.Context, string, string) ([]byte, error) { + return []byte(`{"access_key_id":"AKIAIOSFODNN7EXAMPLE","secret_access_key":"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"}`), nil + }}} + return f +} + +func (f *lateFixture) save(account string, skip bool) (lateSave, bool) { + f.mu.Lock() + defer f.mu.Unlock() + for _, s := range f.saves { + isRoot := s.exec.CloudAccountID == nil + switch { + case account == "" && isRoot: + return s, true + case !isRoot && *s.exec.CloudAccountID == account && s.exec.NoSubmission == skip: + return s, true + } + } + return lateSave{}, false +} + +// An account in flight when the run deadline passes saves its row on its +// slice, sends no email of its own, and joins the run's single email after the +// root save. Its provider call was cut off, so the row records an unknown +// outcome and the email says so, never "failed". The run ends by E+T plus the +// one email. +func TestExecuteAndFinalize_LateAccountBatchedIntoOneOutcomeUnknownEmail(t *testing.T) { + f := newLateFixture(t, "1", func(context.Context, *config.PurchaseExecution) error { + time.Sleep(30 * time.Millisecond) // a slow store, inside the T/2 slice + return nil + }) + start := time.Now() + err := f.m.executeAndFinalize(context.Background(), f.exec) + elapsed := time.Since(start) + + var partial *multiAccountPartialError + require.ErrorAs(t, err, &partial, "A committed, so the run is an acked partial") + assert.Equal(t, "partially_completed", f.exec.Status) + + root, ok := f.save("", false) + require.True(t, ok, "the root must be saved") + assert.Equal(t, "partially_completed", root.exec.Status, "the root is saved WITH its terminal status") + assert.WithinDuration(t, start.Add(lateExecTimeout+lateTerminalT), root.deadline, 80*time.Millisecond, "root deadline is E+T") + + b, ok := f.save(skipAcctB, false) + require.True(t, ok, "the late account saves its own row") + assert.Equal(t, "failed", b.exec.Status) + assert.False(t, b.exec.NoSubmission, "nothing proves B submitted nothing") + assert.Contains(t, b.exec.Error, common.ErrOutcomeUnknown.Error(), "B's call was cut off: outcome unknown, not a retryable failure") + assert.WithinDuration(t, start.Add(lateExecTimeout+lateTerminalT/2), b.deadline, 80*time.Millisecond, "account slice is E+T/2") + assert.True(t, b.done.Before(root.done), "the root saves after the late account it summarises") + + for _, id := range []string{skipAcctC, lateAcctD} { + s, ok := f.save(id, true) + require.True(t, ok, "skip row for %s", id) + assert.True(t, s.done.After(root.done), "skip rows follow the root") + assert.WithinDuration(t, start.Add(lateExecTimeout+lateTerminalT), s.deadline, 80*time.Millisecond) + } + + require.Len(t, f.mails, 1, "one email for the late account and the skipped ones, none per account") + assert.True(t, f.mails[0].OutcomeUnknown, "the late account was cut off mid-call") + assert.Len(t, f.mails[0].Recommendations, 3, "B, C and D") + assert.Less(t, elapsed, lateExecTimeout+lateTerminalT+lateMailDelay+150*time.Millisecond, "inside E+T+N") +} + +// Every account save outruns its T/2 slice. Each lost row is logged as AUDIT +// LOSS with its IDs, the root still saves with its status inside its own +// slice, and the skip rows and the email still follow. +func TestExecuteAndFinalize_AccountSavesStarveNotTheRoot(t *testing.T) { + var mu sync.Mutex + ended := map[string]time.Time{} + f := newLateFixture(t, "2", func(ctx context.Context, e *config.PurchaseExecution) error { + <-ctx.Done() // never finishes inside its slice + mu.Lock() + ended[*e.CloudAccountID] = time.Now() + mu.Unlock() + return ctx.Err() + }) + start := time.Now() + err := f.m.executeAndFinalize(context.Background(), f.exec) + elapsed := time.Since(start) + + var partial *multiAccountPartialError + require.ErrorAs(t, err, &partial) + root, ok := f.save("", false) + require.True(t, ok, "the root is saved although every account save timed out") + assert.Equal(t, "partially_completed", root.exec.Status) + + slice := start.Add(lateExecTimeout + lateTerminalT/2) + logs := f.logs.String() + mu.Lock() + defer mu.Unlock() + for _, id := range []string{skipAcctA, skipAcctB} { + require.Contains(t, ended, id, "account %s save started", id) + assert.WithinDuration(t, slice, ended[id], 80*time.Millisecond, "%s save is cut at E+T/2", id) + row, _ := f.save(id, false) + assert.Contains(t, logs, "AUDIT LOSS: failed to save the record of execution "+row.exec.ExecutionID+" for account "+id) + } + for _, id := range []string{skipAcctC, lateAcctD} { + _, ok := f.save(id, true) + assert.True(t, ok, "skip row for %s still saved after the root", id) + } + assert.Len(t, f.mails, 1) + assert.Less(t, elapsed, lateExecTimeout+lateTerminalT+lateMailDelay+150*time.Millisecond) +} + +// A cut-off call is outcome unknown only when the run context ended: a +// provider error on a live context stays a plain failure. +func TestExecuteSinglePurchase_OnlyACutOffCallIsOutcomeUnknown(t *testing.T) { + for _, cut := range []bool{false, true} { + store := new(MockConfigStore) + factory := new(MockProviderFactory) + prov := new(MockProvider) + client := new(MockServiceClient) + ctx, cancel := context.WithCancel(context.Background()) + client.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything). + Run(func(mock.Arguments) { + if cut { + cancel() + } + }).Return(common.PurchaseResult{}, errors.New("connection reset")) + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil) + prov.On("GetServiceClient", mock.Anything, common.ServiceEC2, "us-east-1").Return(client, nil) + m := &Manager{config: store, providerFactory: factory} + _, err := m.executeSinglePurchase(ctx, config.RecommendationRecord{Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1}, + nil, common.PurchaseOptions{}) + require.Error(t, err) + assert.Equal(t, cut, errors.Is(err, common.ErrOutcomeUnknown), "cut=%v: %v", cut, err) + cancel() + } +} + +// The cut-off text must read in History exactly as the plain failure it +// replaced, and must never count as a proven refusal that unlocks a retry. +func TestCutOffCallText_HistoryAndRefusalUnchanged(t *testing.T) { + const old = "purchase failed: context deadline exceeded" + cut := "purchase failed: " + common.ErrOutcomeUnknown.Error() + ": context deadline exceeded" + assert.Equal(t, SanitizeStoredError(old), SanitizeStoredError(cut)) + recs := []config.RecommendationRecord{{Provider: "aws", Service: "ec2"}} + assert.False(t, IsDefiniteAWSRefusal(recs, "ec2 m5.large: "+cut)) +} diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 609a0bbb9..784c3b75a 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -382,21 +382,23 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE } // saveTerminalAndSkipped saves the root row, then the rows of accounts a -// fan-out skipped, all on ctx's single terminal-save deadline (issue #1672). -// The root goes first and its result is the return value: a skip row that does +// fan-out skipped, all on ctx's single terminal-save deadline, then sends one +// email for the skipped rows and the late accounts' rows (issue #1672). The +// root goes first and its result is the return value: a skip row that does // not fit in the deadline is logged as AUDIT LOSS and never fails the root. func (m *Manager) saveTerminalAndSkipped(ctx context.Context, exec *config.PurchaseExecution, execErr error) error { err := m.saveOrRecoverTerminal(ctx, exec, execErr) var partial *multiAccountPartialError + var saved []*config.PurchaseExecution if errors.As(execErr, &partial) { - m.saveSkippedRows(ctx, exec, partial.skipped) + saved = m.saveSkippedRows(ctx, exec, partial.skipped) } + m.notifyBatchedAccounts(ctx, exec, saved, lateRowsOf(execErr)) return err } -// saveSkippedRows saves each skip row until ctx ends and sends one email for -// the rows it saved. -func (m *Manager) saveSkippedRows(ctx context.Context, root *config.PurchaseExecution, rows []*config.PurchaseExecution) { +// saveSkippedRows saves each skip row until ctx ends and returns the saved ones. +func (m *Manager) saveSkippedRows(ctx context.Context, root *config.PurchaseExecution, rows []*config.PurchaseExecution) []*config.PurchaseExecution { var saved []*config.PurchaseExecution for _, row := range rows { err := ctx.Err() @@ -409,7 +411,7 @@ func (m *Manager) saveSkippedRows(ctx context.Context, root *config.PurchaseExec } saved = append(saved, row) } - m.notifySkippedAccounts(ctx, root, saved) + return saved } // preExecutionRefusal runs every check that must pass before any provider call. From 1358d7a740006b6bc3a58a9d4fa191535e16ce8a Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 20:04:23 +0100 Subject: [PATCH 6/7] test(purchase): fit the root-save deadline test and lint to the anchored budgets The root save now ends at the run deadline plus the save budget, so the expired-caller test expects the long remaining window. Also fixes two prealloc findings and a misspelling in the new tests. Part of #1672 Co-Authored-By: Claude Sonnet 5.5 --- internal/purchase/execute_detached_test.go | 6 ++++-- internal/purchase/late_accounts_test.go | 6 +++--- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/internal/purchase/execute_detached_test.go b/internal/purchase/execute_detached_test.go index 9ca64d53b..11bb73317 100644 --- a/internal/purchase/execute_detached_test.go +++ b/internal/purchase/execute_detached_test.go @@ -174,8 +174,10 @@ func TestExecuteAndFinalize_ExpiredCallerDeadline_UsesOwnBudgets(t *testing.T) { require.True(t, ok, "terminal save needs a deadline") // Literal durations on purpose: comparing against the constants would // pass whatever their values are. - assert.LessOrEqual(t, f.saveRemaining, 30*time.Second) - assert.Greater(t, f.saveRemaining, 29*time.Second) + // The root save ends at the run deadline plus the 30s save budget (#1672), + // so a run that finishes at once has almost 3m30s + 30s left. + assert.LessOrEqual(t, f.saveRemaining, 4*time.Minute) + assert.Greater(t, f.saveRemaining, 3*time.Minute+59*time.Second) // SavePurchaseHistory runs inside executePurchase, so it sees the run ctx. _, ok = f.histCtx.Deadline() require.True(t, ok, "run ctx needs a deadline") diff --git a/internal/purchase/late_accounts_test.go b/internal/purchase/late_accounts_test.go index 3a44704df..187db3545 100644 --- a/internal/purchase/late_accounts_test.go +++ b/internal/purchase/late_accounts_test.go @@ -77,8 +77,8 @@ func newLateFixture(t *testing.T, parallelism string, acctSave func(context.Cont ids := []string{skipAcctA, skipAcctB, skipAcctC, lateAcctD} types := []string{"a.large", "b.large", "c.large", "d.large"} - var accounts []config.CloudAccount - var recs []config.RecommendationRecord + accounts := make([]config.CloudAccount, 0, len(ids)) + recs := make([]config.RecommendationRecord, 0, len(ids)) for i, id := range ids { accounts = append(accounts, config.CloudAccount{ID: id, Name: id[:1], Provider: "aws", ExternalID: strings.Repeat(string(rune('1'+i)), 12), AWSAuthMode: "access_keys"}) recs = append(recs, withAccount(config.RecommendationRecord{Provider: "aws", Service: "ec2", ResourceType: types[i], Region: "us-east-1", @@ -173,7 +173,7 @@ func TestExecuteAndFinalize_LateAccountBatchedIntoOneOutcomeUnknownEmail(t *test assert.False(t, b.exec.NoSubmission, "nothing proves B submitted nothing") assert.Contains(t, b.exec.Error, common.ErrOutcomeUnknown.Error(), "B's call was cut off: outcome unknown, not a retryable failure") assert.WithinDuration(t, start.Add(lateExecTimeout+lateTerminalT/2), b.deadline, 80*time.Millisecond, "account slice is E+T/2") - assert.True(t, b.done.Before(root.done), "the root saves after the late account it summarises") + assert.True(t, b.done.Before(root.done), "the root saves after the late account it summarizes") for _, id := range []string{skipAcctC, lateAcctD} { s, ok := f.save(id, true) From b3dc23e516a3b30c8e664d5e631d3b3cd53d3ff5 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 6 Nov 2026 20:43:09 +0100 Subject: [PATCH 7/7] fix(purchase): bound a fan-out account failure email by the account slice An account that failed just before the run deadline sent its own email on a fresh 30s window, keeping the fan-out busy past the root save slice. The email now shares the account slice (run deadline plus half the save budget). Part of #1672 Co-Authored-By: Claude Sonnet 5.5 --- internal/purchase/execution.go | 2 +- internal/purchase/failure_notify.go | 21 ++++++++++- internal/purchase/late_accounts_test.go | 49 ++++++++++++++++++++++++- 3 files changed, 68 insertions(+), 4 deletions(-) diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 3138cd116..272b94f6d 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -417,7 +417,7 @@ func (m *Manager) notifyAccountOutcome(ctx context.Context, row *config.Purchase if ctx.Err() != nil && late.add(row) { return } - m.notifyExecutionFailure(ctx, row) + m.notifyFanOutFailure(ctx, row) } // newAccountExecution is the per-account copy of baseExec, with an independent diff --git a/internal/purchase/failure_notify.go b/internal/purchase/failure_notify.go index f69980e45..61f532c48 100644 --- a/internal/purchase/failure_notify.go +++ b/internal/purchase/failure_notify.go @@ -67,10 +67,29 @@ func (m *Manager) notifyOutcomeUnknown(ctx context.Context, exec *config.Purchas // sendFailureNotice logs the marker and sends the purchase-failed email, // flagged outcome-unknown when outcomeUnknown is set. func (m *Manager) sendFailureNotice(ctx context.Context, exec *config.PurchaseExecution, outcomeUnknown bool) { + m.sendFailureNoticeUntil(ctx, exec, outcomeUnknown, time.Now().Add(failureNotifyTimeout)) +} + +// notifyFanOutFailure is notifyExecutionFailure for a fan-out account's own +// row. The email shares the account slice (run deadline + T/2) instead of +// opening a fresh failureNotifyTimeout window, so an account that fails just +// before the run deadline cannot keep the fan-out busy past the root's save +// slice (#1672). Without a run deadline it keeps the fresh window. +func (m *Manager) notifyFanOutFailure(ctx context.Context, exec *config.PurchaseExecution) { + until := time.Now().Add(failureNotifyTimeout) + if runDeadline, ok := ctx.Deadline(); ok { + if slice := runDeadline.Add(m.budgets.terminalSaveTimeout() / 2); slice.Before(until) { + until = slice + } + } + m.sendFailureNoticeUntil(ctx, exec, false, until) +} + +func (m *Manager) sendFailureNoticeUntil(ctx context.Context, exec *config.PurchaseExecution, outcomeUnknown bool, until time.Time) { if !logExecutionFailure(exec) || m.email == nil { return } - ctx, cancel := ctxutil.DetachedWithTimeout(ctx, failureNotifyTimeout) + ctx, cancel := ctxutil.DetachedWithDeadline(ctx, until) defer cancel() data := m.buildPurchaseFailedData(ctx, exec) data.OutcomeUnknown = outcomeUnknown diff --git a/internal/purchase/late_accounts_test.go b/internal/purchase/late_accounts_test.go index 187db3545..3d333b939 100644 --- a/internal/purchase/late_accounts_test.go +++ b/internal/purchase/late_accounts_test.go @@ -58,6 +58,12 @@ type lateFixture struct { saves []lateSave mails []email.NotificationData logs *lateLogBuf + + client *MockServiceClient + // blockCMail makes the email about account C alone wait for its context + // (at most 2s) and records when it ended in cMailEnd. + blockCMail bool + cMailEnd time.Time } // newLateFixture builds the fan-out. acctSave handles the saves of started @@ -74,6 +80,7 @@ func newLateFixture(t *testing.T, parallelism string, acctSave func(context.Cont factory := new(MockProviderFactory) prov := new(MockProvider) client := new(MockServiceClient) + f.client = client ids := []string{skipAcctA, skipAcctB, skipAcctC, lateAcctD} types := []string{"a.large", "b.large", "c.large", "d.large"} @@ -104,10 +111,21 @@ func newLateFixture(t *testing.T, parallelism string, acctSave func(context.Cont mailer.On("SendPurchaseConfirmation", mock.Anything, mock.Anything).Return(nil).Maybe() mailer.On("SendPurchaseFailedNotification", mock.Anything, mock.Anything). Run(func(args mock.Arguments) { - time.Sleep(lateMailDelay) + d := args.Get(1).(email.NotificationData) + if f.blockCMail && len(d.Recommendations) == 1 && d.Recommendations[0].ResourceType == "c.large" { + select { + case <-time.After(2 * time.Second): + case <-args.Get(0).(context.Context).Done(): + } + f.mu.Lock() + f.cMailEnd = time.Now() + f.mu.Unlock() + } else { + time.Sleep(lateMailDelay) + } f.mu.Lock() defer f.mu.Unlock() - f.mails = append(f.mails, args.Get(1).(email.NotificationData)) + f.mails = append(f.mails, d) }).Return(nil).Maybe() factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil).Maybe() prov.On("GetServiceClient", mock.Anything, common.ServiceEC2, "us-east-1").Return(client, nil).Maybe() @@ -264,3 +282,30 @@ func TestCutOffCallText_HistoryAndRefusalUnchanged(t *testing.T) { recs := []config.RecommendationRecord{{Provider: "aws", Service: "ec2"}} assert.False(t, IsDefiniteAWSRefusal(recs, "ec2 m5.large: "+cut)) } + +// An account that fails just before the run deadline sends its own email, and a +// slow mailer must not keep the fan-out busy past the account slice: the email +// is cut at E+T/2 and the root still saves with its status inside its own slice. +func TestExecuteAndFinalize_AccountFailureEmailSharesTheAccountSlice(t *testing.T) { + f := newLateFixture(t, "2", nil) + f.blockCMail = true + start := time.Now() + failAt := start.Add(lateExecTimeout - 30*time.Millisecond) + f.client.On("PurchaseCommitment", mock.Anything, mock.MatchedBy(func(r common.Recommendation) bool { return r.ResourceType == "c.large" }), mock.Anything). + Run(func(mock.Arguments) { time.Sleep(time.Until(failAt)) }). + Return(common.PurchaseResult{}, errors.New("insufficient capacity")).Maybe() + + err := f.m.executeAndFinalize(context.Background(), f.exec) + elapsed := time.Since(start) + + var partial *multiAccountPartialError + require.ErrorAs(t, err, &partial) + root, ok := f.save("", false) + require.True(t, ok, "the root is saved") + assert.Equal(t, "partially_completed", root.exec.Status) + f.mu.Lock() + defer f.mu.Unlock() + require.False(t, f.cMailEnd.IsZero(), "C failed before the deadline and emailed on its own") + assert.WithinDuration(t, start.Add(lateExecTimeout+lateTerminalT/2), f.cMailEnd, 80*time.Millisecond, "C's email is cut at E+T/2") + assert.Less(t, elapsed, lateExecTimeout+lateTerminalT+lateMailDelay+150*time.Millisecond, "inside E+T+N") +}