From f2d750cd29dc9c350ad181630e802db540228f25 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 22 Oct 2026 10:11:47 +0100 Subject: [PATCH 1/2] fix(purchase): alert on a failed fan-out root whose children never notified executeAndFinalize skipped the failure notification for a multi-account fan-out root on the assumption that every per-account row had notified for itself. A child the fan-out skips once the context is done, or one whose panic it recovers, never stamps or saves its row, so its own deferred notification is a no-op and the run ended failed or partially_completed with no purchase_execution_failed marker and no email. The root now always logs the marker (the alarm source) and still sends no email, so a child that did notify is not mailed twice. Refs #1195 Co-Authored-By: Claude Opus 5.5 --- internal/purchase/failure_notify.go | 56 +++-- .../purchase/failure_notify_writers_test.go | 215 ++++++++++++++++++ internal/purchase/manager.go | 8 +- 3 files changed, 260 insertions(+), 19 deletions(-) create mode 100644 internal/purchase/failure_notify_writers_test.go diff --git a/internal/purchase/failure_notify.go b/internal/purchase/failure_notify.go index ce0da513d..db18e6ed1 100644 --- a/internal/purchase/failure_notify.go +++ b/internal/purchase/failure_notify.go @@ -30,20 +30,23 @@ const failureNotifyTimeout = 30 * time.Second // purchase-failed email listing the selected recommendations that did not // purchase. A completed (or any non-terminal) row is a no-op. // -// It runs after the terminal status is decided, once per row: the root row in -// executeAndFinalize (single-account and pre-fan-out failures) and each -// per-account row in executeForAccount (multi-account fan-out, whose root never -// carries per-rec Purchased flags). Every caller holds the row's "running" -// claim, so concurrent workers cannot both notify for the same row. +// It runs after the terminal status is decided, once per row, from every +// writer that ends a row failed or partially_completed (issue #1195): +// - executeAndFinalize, for the root row (single-account and pre-fan-out +// failures; a fan-out root gets the marker only, see notifyRootFailure); +// - executeForAccount, for each per-account fan-out row; +// - failClaimedBeforeSubmission, for a claimed row refused before submission; +// - safeFail and the reaper, for stranded rows the sweeps fail; +// - failExistingPlanStep and getOrCreateExecution, for a plan step that is +// failed without ever reaching the executor. +// +// Each of them notifies only after winning the row: the "running" claim, or the +// compare-and-set (or plan lock) that moved it to failed, so concurrent workers +// cannot both notify for the same row. // // Best effort: a send failure is logged and never changes the purchase outcome. func (m *Manager) notifyExecutionFailure(ctx context.Context, exec *config.PurchaseExecution) { - if exec.Status != "failed" && exec.Status != "partially_completed" { - return - } - logging.Errorf("%s execution_id=%s plan_id=%s status=%s", - PurchaseFailedLogMarker, exec.ExecutionID, exec.PlanID, exec.Status) - if m.email == nil { + if !logExecutionFailure(exec) || m.email == nil { return } ctx, cancel := ctxutil.DetachedWithTimeout(ctx, failureNotifyTimeout) @@ -81,10 +84,35 @@ func (m *Manager) buildPurchaseFailedData(ctx context.Context, exec *config.Purc return data } +// logExecutionFailure logs PurchaseFailedLogMarker, the alarm source, when exec +// ended failed or partially_completed, and reports whether it did. +func logExecutionFailure(exec *config.PurchaseExecution) bool { + if exec.Status != "failed" && exec.Status != "partially_completed" { + return false + } + logging.Errorf("%s execution_id=%s plan_id=%s status=%s", + PurchaseFailedLogMarker, exec.ExecutionID, exec.PlanID, exec.Status) + return true +} + +// notifyRootFailure is executeAndFinalize's notification for the root row. A +// fan-out root always logs the marker but sends no email: the per-account rows +// email for themselves, and the root could not list the failed recs (it never +// carries per-rec Purchased flags). The marker cannot be left to the children, +// because a child can fail without reaching its own notification: the fan-out +// skips an account once the context is done, and a panicking child is +// recovered with its row still non-terminal (issue #1195). +func (m *Manager) notifyRootFailure(ctx context.Context, exec *config.PurchaseExecution, execErr error) { + if isFanOutOutcome(execErr) { + logExecutionFailure(exec) + return + } + m.notifyExecutionFailure(ctx, exec) +} + // isFanOutOutcome reports whether execErr is the aggregate of a multi-account -// fan-out. Each per-account row already notified for itself, so the root row -// must not notify again (and could not list the failed recs: the root never -// carries per-rec Purchased flags). +// fan-out, whose root row gets the failure marker but no email (see +// notifyRootFailure). func isFanOutOutcome(execErr error) bool { var partial *multiAccountPartialError return errors.Is(execErr, errAllAccountsFailed) || errors.As(execErr, &partial) diff --git a/internal/purchase/failure_notify_writers_test.go b/internal/purchase/failure_notify_writers_test.go new file mode 100644 index 000000000..e12379b16 --- /dev/null +++ b/internal/purchase/failure_notify_writers_test.go @@ -0,0 +1,215 @@ +package purchase + +import ( + "bytes" + "context" + "errors" + "fmt" + "strings" + "sync" + "testing" + + "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/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// Issue #1195: every writer that ends an execution row failed or +// partially_completed must raise an alert signal: the PurchaseFailedLogMarker +// line the monitoring modules alarm on. These tests pin the marker per writer, +// including a fan-out root whose children never reached their own notification. + +// markerLog captures the logging package's output; the fan-out logs from +// goroutines, so writes are serialized. +type markerLog struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (l *markerLog) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.buf.Write(p) +} + +// count returns how many purchase-failed markers were logged for executionID. +func (l *markerLog) count(executionID string) int { + l.mu.Lock() + defer l.mu.Unlock() + return strings.Count(l.buf.String(), PurchaseFailedLogMarker+" execution_id="+executionID+" ") +} + +func captureMarkers(t *testing.T) *markerLog { + t.Helper() + l := &markerLog{} + prev := logging.SetOutput(l) + t.Cleanup(func() { logging.SetOutput(prev) }) + return l +} + +func TestLogExecutionFailure_LogsMarkerOnlyForFailedOrPartialRows(t *testing.T) { + logs := captureMarkers(t) + for _, status := range []string{"failed", "partially_completed", "completed", "running", "approved", "pending", ""} { + id := "exec-" + status + logged := logExecutionFailure(&config.PurchaseExecution{ExecutionID: id, PlanID: "plan-1", Status: status}) + want := status == "failed" || status == "partially_completed" + assert.Equal(t, want, logged, "status %q", status) + assert.Equal(t, map[bool]int{true: 1, false: 0}[want], logs.count(id), "status %q", status) + } + assert.Contains(t, logs.buf.String(), PurchaseFailedLogMarker+" execution_id=exec-failed plan_id=plan-1 status=failed") +} + +// A fan-out root logs the marker but sends no email, whatever happened to the +// children: a skipped child (the fan-out records ctx.Err() without running it) +// and a recovered panic both leave no notified child row behind. +func TestNotifyRootFailure_FanOutRootLogsMarkerWithoutEmail(t *testing.T) { + cases := map[string]struct { + execErr error + status string + }{ + "skipped child": { + fmt.Errorf("%w: account a: %w", errAllAccountsFailed, context.DeadlineExceeded), "failed", + }, + "panicked child beside a committed one": { + &multiAccountPartialError{committed: 1, errors: []string{"account b: panic during fan-out (account=b): boom"}}, + "partially_completed", + }, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + logs := captureMarkers(t) + sender := new(MockEmailSender) + m := &Manager{config: new(MockConfigStore), email: sender} + root := &config.PurchaseExecution{ExecutionID: "root", Status: "running"} + m.finalizeExecution(root, tc.execErr) + require.Equal(t, tc.status, root.Status) + + m.notifyRootFailure(context.Background(), root, tc.execErr) + + assert.Equal(t, 1, logs.count("root")) + sender.AssertNotCalled(t, "SendPurchaseFailedNotification", mock.Anything, mock.Anything) + }) + } + + t.Run("non fan-out root emails too", func(t *testing.T) { + logs := captureMarkers(t) + sender := allowFailedNotification(new(MockEmailSender)) + m := &Manager{config: new(MockConfigStore), email: sender} + root := &config.PurchaseExecution{ExecutionID: "root", Status: "failed"} + m.notifyRootFailure(context.Background(), root, errors.New("provider down")) + assert.Equal(t, 1, logs.count("root")) + sender.AssertNumberOfCalls(t, "SendPurchaseFailedNotification", 1) + }) +} + +// fanOutRun drives a two-account plan through the SQS executor. Account +// "acct-ok" buys; the other account's behavior is set by the caller through the +// credential store and the save hook. It returns the root's saved status and +// every purchase-failed email. +func fanOutRun(t *testing.T, other config.CloudAccount, loadRaw func(string) ([]byte, error), saveErr func(*config.PurchaseExecution) error) (string, []string) { + t.Helper() + ctx := context.Background() + store := new(MockConfigStore) + sender := new(MockEmailSender) + rec := &failedEmailRecorder{} + rec.expect(sender) + factory, prov, svc := new(MockProviderFactory), new(MockProvider), new(MockServiceClient) + + accounts := []config.CloudAccount{ + {ID: "acct-ok", Name: "OK", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys"}, + other, + } + exec := &config.PurchaseExecution{ + ExecutionID: "root-1195", IdempotencyKey: "lineage-1195", Status: "notified", PlanID: "plan-x", + Recommendations: []config.RecommendationRecord{scopedTestRec("acct-ok"), scopedTestRec(other.ID)}, + } + plan := &config.PurchasePlan{ID: "plan-x", Name: "Plan X", AutoPurchase: true, MaxStepCommitment: capOf(1e9)} + + store.On("GetGlobalConfig", mock.Anything).Return(failureNotifyCfg(), nil) + store.On("GetExecutionByID", ctx, "root-1195").Return(exec, nil) + running := *exec + running.Status = "running" + store.On("TransitionExecutionStatus", ctx, "root-1195", + []string{"approved", "pending", "notified"}, "running", (*string)(nil)).Return(&running, nil) + store.On("GetPurchasePlan", mock.Anything, "plan-x").Return(plan, nil).Maybe() + store.GetPlanAccountsFn = func(context.Context, string) ([]config.CloudAccount, error) { return accounts, nil } + store.On("CompletePlanStep", ctx, "plan-x", mock.Anything).Return(nil).Maybe() + var mu sync.Mutex + var rootStatus string + store.SavePurchaseExecutionFn = func(_ context.Context, e *config.PurchaseExecution) error { + mu.Lock() + defer mu.Unlock() + if e.CloudAccountID == nil { + rootStatus = e.Status + return nil + } + return saveErr(e) + } + store.On("SavePurchaseHistory", mock.Anything, mock.Anything).Return(nil) + sender.On("SendPurchaseConfirmation", mock.Anything, mock.Anything).Return(nil) + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil) + prov.On("GetServiceClient", mock.Anything, common.ServiceEC2, mock.Anything).Return(svc, nil) + svc.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything). + Return(common.PurchaseResult{Success: true, CommitmentID: "ri-ok"}, nil).Once() + keys := awsAccessKeyCredStore() + creds := &MockCredentialStore{LoadRawFn: func(ctx context.Context, accountID, credType string) ([]byte, error) { + if accountID == other.ID { + return loadRaw(accountID) + } + return keys.LoadRawFn(ctx, accountID, credType) + }} + + m := &Manager{config: store, email: sender, providerFactory: factory, credStore: creds} + require.NoError(t, m.handleExecutePurchase(ctx, &AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: "root-1195"})) + + mu.Lock() + defer mu.Unlock() + sent := rec.all() + emailed := make([]string, 0, len(sent)) + for _, d := range sent { + emailed = append(emailed, d.ExecutionID) + } + return rootStatus, emailed +} + +// A child that panics is recovered by the fan-out before it stamps or saves +// its row, so its own deferred notification is a no-op. Before #1195 the +// partially_completed root was skipped as well and nothing alerted. +func TestFanOutRootAlertsWhenPanickedChildNeverNotifies(t *testing.T) { + logs := captureMarkers(t) + panicky := config.CloudAccount{ID: "acct-panic", Name: "PANIC", Provider: "aws", ExternalID: "222222222222", AWSAuthMode: "access_keys"} + rootStatus, emailed := fanOutRun(t, panicky, + func(string) ([]byte, error) { panic("credential store blew up") }, + func(*config.PurchaseExecution) error { return nil }) + + assert.Equal(t, "partially_completed", rootStatus) + assert.Equal(t, 1, logs.count("root-1195"), "the root's marker is the one alert signal for the panicked child") + assert.Empty(t, emailed, "the root sends no email; the panicked child had no row to email for") +} + +// A child whose failed row cannot be saved (AUDIT LOSS) still notifies for +// itself; the root adds its marker without a second email. +func TestFanOutAuditLossChildStillNotifiesAndRootAddsMarker(t *testing.T) { + logs := captureMarkers(t) + // role_arn with no STS client wired: credential resolution fails. + bad := config.CloudAccount{ID: "acct-bad", Name: "BAD", Provider: "aws", ExternalID: "222222222222", AWSAuthMode: "role_arn"} + var childID string + rootStatus, emailed := fanOutRun(t, bad, + func(string) ([]byte, error) { return nil, errors.New("unused") }, + func(e *config.PurchaseExecution) error { + if e.Status == "failed" { + childID = e.ExecutionID + return errors.New("db down") + } + return nil + }) + + assert.Equal(t, "partially_completed", rootStatus) + require.NotEmpty(t, childID) + assert.Equal(t, 1, logs.count(childID), "the child alerts for itself") + assert.Equal(t, []string{childID}, emailed, "one email, for the child, none for the root") + assert.Equal(t, 1, logs.count("root-1195")) +} diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index 9d7ca8c60..d66e782ef 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -264,11 +264,9 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE saveCtx, cancelSave := ctxutil.DetachedWithTimeout(ctx, terminalSaveTimeout) defer cancelSave() // Operators hear about every failed or partial row (issue #714), even - // when the terminal save below fails. A fan-out root is skipped: each - // per-account row notified for itself in executeForAccount. - if !isFanOutOutcome(execErr) { - defer m.notifyExecutionFailure(saveCtx, exec) - } + // 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 { logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err) // Wrap with ErrAuditLoss regardless of whether executePurchase itself From cb66e8494fe6d224618db3d75fa8568e306aa4fd Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 22 Oct 2026 10:11:47 +0100 Subject: [PATCH 2/2] fix(purchase): alert from every terminal-failure writer Only executeAndFinalize and the fan-out children called notifyExecutionFailure. The other writers that end an execution row failed raised no purchase_execution_failed marker and sent no email: - failClaimedBeforeSubmission (post-claim cap and authorization refusals) - safeFail (stranded approved rows the recovery sweep fails) - the reaper (stuck approved/running rows) - failExistingPlanStep and the new failed row a notification tick records for a plan step that will never be bought Each now notifies once it has won the row (the running claim, the CAS to failed, or the plan lock), so racing workers cannot both alert. The reaper stays quiet when the executor recovered the row before the reap annotation landed, since that purchase did not fail. Existing tests that end a row failed now expect the one purchase-failed email. Refs #1195 Co-Authored-By: Claude Opus 5.5 --- internal/purchase/claim_pin_test.go | 4 +- .../purchase/failure_notify_paths_test.go | 155 ++++++++++++++++++ internal/purchase/manager.go | 2 + internal/purchase/manager_test.go | 3 + internal/purchase/notifications.go | 10 +- .../plan_step_atomic_integration_test.go | 38 ++++- internal/purchase/plan_step_attach.go | 2 +- .../plan_step_attach_integration_test.go | 10 +- internal/purchase/plan_step_no_recs_test.go | 9 +- internal/purchase/reaper.go | 44 +++-- internal/purchase/step_cap.go | 2 + 11 files changed, 252 insertions(+), 27 deletions(-) create mode 100644 internal/purchase/failure_notify_paths_test.go diff --git a/internal/purchase/claim_pin_test.go b/internal/purchase/claim_pin_test.go index 79d2d808c..294820ecd 100644 --- a/internal/purchase/claim_pin_test.go +++ b/internal/purchase/claim_pin_test.go @@ -143,7 +143,8 @@ func TestClaimPin_PostClaimPlanReadErrorFailsClosed(t *testing.T) { var saved *config.PurchaseExecution store.On("SavePurchaseExecutionTx", ctx, mock.Anything, mock.Anything). Run(func(args mock.Arguments) { saved = args.Get(2).(*config.PurchaseExecution) }).Return(nil) - manager := &Manager{config: store, email: new(MockEmailSender), providerFactory: new(MockProviderFactory)} + sender := allowFailedNotification(new(MockEmailSender)) + manager := &Manager{config: store, email: sender, providerFactory: new(MockProviderFactory)} didClaim, err := manager.claimAndExecute(ctx, &exec) require.True(t, didClaim) @@ -151,6 +152,7 @@ func TestClaimPin_PostClaimPlanReadErrorFailsClosed(t *testing.T) { assert.ErrorContains(t, err, "post-claim cap check") require.NotNil(t, saved) assert.Equal(t, "failed", saved.Status) + sender.AssertNumberOfCalls(t, "SendPurchaseFailedNotification", 1) // issue #1195 store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", ctx, mock.Anything, exec.ExecutionID) } diff --git a/internal/purchase/failure_notify_paths_test.go b/internal/purchase/failure_notify_paths_test.go new file mode 100644 index 000000000..8433655d9 --- /dev/null +++ b/internal/purchase/failure_notify_paths_test.go @@ -0,0 +1,155 @@ +package purchase + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// Issue #1195 (maintainer scope): every terminal-failure writer outside +// executeAndFinalize alerts too, once per row and only when it won the row. +// Each case runs one writer and reports the row it acted on. + +type writerCase struct { + name string + // run wires store and calls the writer for execution "exec-w". + run func(t *testing.T, m *Manager, store *MockConfigStore) + // alerts is whether the writer must alert (it won the row). + alerts bool +} + +func writerRow(status string) *config.PurchaseExecution { + return &config.PurchaseExecution{ + ExecutionID: "exec-w", PlanID: "plan-w", Status: status, + Recommendations: []config.RecommendationRecord{ + {Provider: "azure", Service: "savingsplans", ResourceType: "Compute", Region: "eastus", Count: 1, Selected: true}, + }, + } +} + +func failedRowFromCAS(store *MockConfigStore, method string, args ...interface{}) { + store.On(method, args...).Return(writerRow("failed"), nil) +} + +func strandedWriterCases() []writerCase { + transition := []interface{}{mock.Anything, "exec-w", []string{"approved"}, "failed", (*string)(nil)} + return []writerCase{ + {name: "safeFail won CAS", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + failedRowFromCAS(store, "TransitionExecutionStatus", transition...) + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil) + counted, err := m.safeFail(context.Background(), writerRow("approved")) + require.NoError(t, err) + assert.True(t, counted) + }}, + {name: "safeFail won CAS, reason save failed", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + failedRowFromCAS(store, "TransitionExecutionStatus", transition...) + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(errors.New("db down")) + counted, err := m.safeFail(context.Background(), writerRow("approved")) + require.NoError(t, err) + assert.True(t, counted) + }}, + {name: "safeFail lost CAS", run: func(t *testing.T, m *Manager, store *MockConfigStore) { + store.On("TransitionExecutionStatus", transition...).Return(nil, config.ErrExecutionNotInExpectedStatus) + counted, err := m.safeFail(context.Background(), writerRow("approved")) + require.NoError(t, err) + assert.False(t, counted) + }}, + {name: "reaper won CAS", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + failedRowFromCAS(store, "TransitionExecutionStatus", mock.Anything, "exec-w", []string{"running"}, failedStatus, (*string)(nil)) + store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{failedStatus}).Return(nil) + result := &ReapResult{} + m.reapOne(context.Background(), writerRow("running"), DefaultReapAfter, time.Now(), result) + assert.Equal(t, 1, result.Reaped) + }}, + {name: "reaper won CAS, reason save failed", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + failedRowFromCAS(store, "TransitionExecutionStatus", mock.Anything, "exec-w", []string{"running"}, failedStatus, (*string)(nil)) + store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{failedStatus}).Return(errors.New("db down")) + result := &ReapResult{} + m.reapOne(context.Background(), writerRow("running"), DefaultReapAfter, time.Now(), result) + assert.Equal(t, ReapResult{Reaped: 1, Errored: 1}, *result) + }}, + // The executor finished late and recovered the row to completed before + // the reap annotation: the purchase did not fail, so nothing alerts. + {name: "reaper won CAS, executor recovered the row", run: func(t *testing.T, m *Manager, store *MockConfigStore) { + failedRowFromCAS(store, "TransitionExecutionStatus", mock.Anything, "exec-w", []string{"running"}, failedStatus, (*string)(nil)) + store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{failedStatus}).Return(config.ErrExecutionNotInExpectedStatus) + result := &ReapResult{} + m.reapOne(context.Background(), writerRow("running"), DefaultReapAfter, time.Now(), result) + assert.Equal(t, ReapResult{Reaped: 1}, *result) + }}, + {name: "reaper lost CAS", run: func(t *testing.T, m *Manager, store *MockConfigStore) { + store.On("TransitionExecutionStatus", mock.Anything, "exec-w", []string{"running"}, failedStatus, (*string)(nil)). + Return(nil, config.ErrExecutionNotInExpectedStatus) + result := &ReapResult{} + m.reapOne(context.Background(), writerRow("running"), DefaultReapAfter, time.Now(), result) + assert.Equal(t, 1, result.RaceLost) + }}, + } +} + +func planStepWriterCases() []writerCase { + pinned := []interface{}{mock.Anything, "exec-w", []string{"pending", "notified"}, "failed", (*string)(nil), false} + return []writerCase{ + {name: "plan step won CAS", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + failedRowFromCAS(store, "TransitionExecutionStatusPinned", pinned...) + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil) + err := m.failExistingPlanStep(context.Background(), writerRow("notified"), "refused") + assert.ErrorIs(t, err, errExecutionNotNotifiable) + }}, + {name: "plan step won CAS, reason save failed", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + failedRowFromCAS(store, "TransitionExecutionStatusPinned", pinned...) + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(errors.New("db down")) + err := m.failExistingPlanStep(context.Background(), writerRow("notified"), "refused") + assert.ErrorContains(t, err, "db down") + }}, + {name: "plan step lost CAS", run: func(t *testing.T, m *Manager, store *MockConfigStore) { + store.On("TransitionExecutionStatusPinned", pinned...).Return(nil, config.ErrExecutionNotInExpectedStatus) + err := m.failExistingPlanStep(context.Background(), writerRow("notified"), "refused") + assert.ErrorIs(t, err, errExecutionNotNotifiable) + }}, + {name: "claimed row refused before submission", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil) + err := m.failClaimedBeforeSubmission(context.Background(), writerRow("running"), errors.New("cap exceeded")) + assert.ErrorContains(t, err, "cap exceeded") + }}, + {name: "claimed row refused, terminal save failed", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(errors.New("db down")) + err := m.failClaimedBeforeSubmission(context.Background(), writerRow("running"), errors.New("cap exceeded")) + assert.ErrorIs(t, err, config.ErrAuditLoss) + }}, + } +} + +func TestTerminalFailureWritersAlertOncePerWonRow(t *testing.T) { + for _, tc := range append(strandedWriterCases(), planStepWriterCases()...) { + t.Run(tc.name, func(t *testing.T) { + logs := captureMarkers(t) + store := new(MockConfigStore) + store.On("GetGlobalConfig", mock.Anything).Return(failureNotifyCfg(), nil) + sender := new(MockEmailSender) + rec := &failedEmailRecorder{} + rec.expect(sender) + m := &Manager{config: store, email: sender} + + tc.run(t, m, store) + + want := 0 + if tc.alerts { + want = 1 + } + assert.Equal(t, want, logs.count("exec-w"), "purchase-failed log marker") + sent := rec.all() + require.Len(t, sent, want, "purchase-failed email") + if want == 1 { + assert.Equal(t, "exec-w", sent[0].ExecutionID) + assert.Equal(t, failureNotifyRecipient, sent[0].RecipientEmail) + } + }) + } +} diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index d66e782ef..928e28d5e 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -664,6 +664,8 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution) } return false, fmt.Errorf("failed to transition stranded execution %s to failed: %w", exec.ExecutionID, txErr) } + // This sweep won the CAS, so it is the one writer that notifies (#1195). + defer m.notifyExecutionFailure(ctx, updated) next := "Verify on the cloud provider that no commitment was created, then Retry." if RedriveRefusalReason(exec) != "" { diff --git a/internal/purchase/manager_test.go b/internal/purchase/manager_test.go index 395760a84..cef3465b8 100644 --- a/internal/purchase/manager_test.go +++ b/internal/purchase/manager_test.go @@ -525,6 +525,8 @@ func TestManager_RecoverStrandedApprovals_FailsStrandedRow(t *testing.T) { mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")). Run(func(args mock.Arguments) { saved = args.Get(1).(*config.PurchaseExecution) }). Return(nil) + // The failed row reaches operators (issue #1195). + mockEmail.On("SendPurchaseFailedNotification", mock.Anything, mock.Anything).Return(nil) manager := &Manager{ config: mockStore, @@ -536,6 +538,7 @@ func TestManager_RecoverStrandedApprovals_FailsStrandedRow(t *testing.T) { recovered, err := manager.RecoverStrandedApprovals(ctx) require.NoError(t, err) assert.Equal(t, 1, recovered) + mockEmail.AssertNumberOfCalls(t, "SendPurchaseFailedNotification", 1) require.NotNil(t, saved) assert.Equal(t, "failed", saved.Status, "stranded Azure Savings Plans row must become terminally failed, never stay approved") diff --git a/internal/purchase/notifications.go b/internal/purchase/notifications.go index 5b34502fc..b7fe82e01 100644 --- a/internal/purchase/notifications.go +++ b/internal/purchase/notifications.go @@ -236,7 +236,10 @@ func (m *Manager) getOrCreateExecution(ctx context.Context, plan *config.Purchas if txErr != nil { return nil, "", false, txErr } - if step.failedNew { + if step.failedNew != nil { + // Notified only once the claim's transaction committed the row; the + // plan lock lets one tick record it (issue #1195). + m.notifyExecutionFailure(ctx, step.failedNew) return nil, "", false, bareStepFailure(plan.ID, step.number, step.failureReason) } if step.targetMet { @@ -268,7 +271,7 @@ type stepClaim struct { existing *config.PurchaseExecution emptyRoot *config.PurchaseExecution blocked error - failedNew bool + failedNew *config.PurchaseExecution targetMet bool failureReason string } @@ -307,6 +310,9 @@ func (m *Manager) failExistingPlanStep(ctx context.Context, exec *config.Purchas return fmt.Errorf("failed to fail plan step %d of plan %s: %w", exec.StepNumber, exec.PlanID, err) } stampFailedPlanStep(failed, reason) + // The won CAS has already failed the row, so notify even if the reason + // write below fails (issue #1195). + defer m.notifyExecutionFailure(ctx, failed) // The row is failed with NoSubmission, so its suppressions are released as // for any proven not-submitted failure (saveTerminalExecution); a bare row has none. if err := m.config.WithTx(ctx, func(tx pgx.Tx) error { diff --git a/internal/purchase/plan_step_atomic_integration_test.go b/internal/purchase/plan_step_atomic_integration_test.go index 2102e1924..4d3e993e5 100644 --- a/internal/purchase/plan_step_atomic_integration_test.go +++ b/internal/purchase/plan_step_atomic_integration_test.go @@ -50,7 +50,8 @@ func TestGetOrCreateExecution_ConcurrentTicksMintOneRow(t *testing.T) { store, _ := rampStepStore(ctx, t) plan := atomicStepPlan(ctx, t, store) // No max_step_commitment on the plan: the cap refuses the step, giving it one failed row. - m := &Manager{config: store, resolver: &fakeResolver{recs: []config.RecommendationRecord{attachRec("rec-a", 1)}}} + sender := newCountingSender(ctx) + m := &Manager{config: store, email: sender, resolver: &fakeResolver{recs: []config.RecommendationRecord{attachRec("rec-a", 1)}}} var wg sync.WaitGroup for i := 0; i < 4; i++ { @@ -65,6 +66,41 @@ func TestGetOrCreateExecution_ConcurrentTicksMintOneRow(t *testing.T) { rows := stepRowsInDB(ctx, t, store, plan.ID) require.Len(t, rows, 1, "concurrent ticks must mint exactly one row for the step") assert.Equal(t, "failed", rows[0].Status) + assert.Equal(t, 1, sender.failedSends, "only the tick that recorded the failed row reports it (issue #1195)") +} + +// Issue #1195: ticks racing to fail the same pre-created step row report it +// once. The compare-and-set lets one of them fail the row; the others lose it +// and stay quiet. +func TestFailExistingPlanStep_ConcurrentTicksReportOnce(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + attachGlobal(ctx, t, store) + plan := atomicStepPlan(ctx, t, store) + root := &config.PurchaseExecution{PlanID: plan.ID, ExecutionID: uuid.New().String(), Status: "pending", + StepNumber: 1, ScheduledDate: *plan.NextExecutionDate} + require.NoError(t, store.SavePurchaseExecution(ctx, root)) + sender := newCountingSender(ctx) + m := &Manager{config: store, email: sender} + + start := make(chan struct{}) + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + stale := *root + <-start + _ = m.failExistingPlanStep(ctx, &stale, "no recommendations") + }() + } + close(start) + wg.Wait() + + rows := stepRowsInDB(ctx, t, store, plan.ID) + require.Len(t, rows, 1) + assert.Equal(t, "failed", rows[0].Status) + assert.Equal(t, 1, sender.failedSends, "one purchase-failed email for the one row") } // E1: a settled root (failed here; partially_completed is the same branch) diff --git a/internal/purchase/plan_step_attach.go b/internal/purchase/plan_step_attach.go index 5689b2b6e..0d0e082c3 100644 --- a/internal/purchase/plan_step_attach.go +++ b/internal/purchase/plan_step_attach.go @@ -306,7 +306,7 @@ func (m *Manager) failStep(ctx context.Context, tx pgx.Tx, locked *config.Purcha if err := m.config.SavePurchaseExecutionTx(ctx, tx, fresh); err != nil { return stepClaim{}, fmt.Errorf("failed to record plan step %d of plan %s as failed: %w", claim.number, locked.ID, err) } - claim.failedNew = true + claim.failedNew = fresh return claim, nil } diff --git a/internal/purchase/plan_step_attach_integration_test.go b/internal/purchase/plan_step_attach_integration_test.go index 0bc70967b..5fc65b30c 100644 --- a/internal/purchase/plan_step_attach_integration_test.go +++ b/internal/purchase/plan_step_attach_integration_test.go @@ -55,13 +55,15 @@ func attachRec(id string, count int) config.RecommendationRecord { type countingSender struct { *MockEmailSender - mu sync.Mutex - sends int + mu sync.Mutex + sends int + failedSends int // purchase-failed emails (issue #1195) } func newCountingSender(ctx context.Context) *countingSender { s := &countingSender{MockEmailSender: new(MockEmailSender)} s.On("SendScheduledPurchaseNotification", ctx, mock.Anything).Run(func(mock.Arguments) { s.mu.Lock(); s.sends++; s.mu.Unlock() }).Return(nil) + s.On("SendPurchaseFailedNotification", mock.Anything, mock.Anything).Run(func(mock.Arguments) { s.mu.Lock(); s.failedSends++; s.mu.Unlock() }).Return(nil) return s } @@ -135,9 +137,11 @@ func TestAttachIntegration_NoMatchAndCap(t *testing.T) { lowered := *plan lowered.MaxStepCommitment = f64(1) require.NoError(t, store.UpdatePurchasePlan(ctx, &lowered)) - m := &Manager{config: store, email: newCountingSender(ctx), resolver: &fakeResolver{recs: []config.RecommendationRecord{attachRec("rec-a", 5)}}} + sender := newCountingSender(ctx) + m := &Manager{config: store, email: sender, resolver: &fakeResolver{recs: []config.RecommendationRecord{attachRec("rec-a", 5)}}} require.False(t, m.sendPlanNotification(ctx, &snapshot)) + assert.Equal(t, 1, sender.failedSends, "the failed step is reported once (issue #1195)") rows := stepRowsInDB(ctx, t, store, plan.ID) require.Len(t, rows, 1) diff --git a/internal/purchase/plan_step_no_recs_test.go b/internal/purchase/plan_step_no_recs_test.go index bbd4ee91f..7d4d69467 100644 --- a/internal/purchase/plan_step_no_recs_test.go +++ b/internal/purchase/plan_step_no_recs_test.go @@ -129,8 +129,9 @@ func scheduledBarePlanStepFails(t *testing.T, notificationDaysBefore int) { assert.Contains(t, rec.saved[0].Error, "no recommendations") } -func TestNotificationTickOnBarePlanStepCreatesOneFailedRowAndSendsNoEmail(t *testing.T) { +func TestNotificationTickOnBarePlanStepCreatesOneFailedRowAndSendsNoApprovalEmail(t *testing.T) { ctx := context.Background() + logs := captureMarkers(t) mockStore := new(MockConfigStore) mockEmail := new(MockEmailSender) t.Cleanup(func() { @@ -145,7 +146,7 @@ func TestNotificationTickOnBarePlanStepCreatesOneFailedRowAndSendsNoEmail(t *tes notifyTo := "notify@example.com" mockStore.On("ListPurchasePlans", ctx, config.PurchasePlanFilter{}).Return([]config.PurchasePlan{plan}, nil) - mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: ¬ifyTo, AutoPurchaseEnabled: true}, nil) + mockStore.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{NotificationEmail: ¬ifyTo, AutoPurchaseEnabled: true}, nil) // Tick 1 finds no row for the date; tick 2 finds whatever tick 1 saved. var stored config.PurchaseExecution @@ -165,6 +166,8 @@ func TestNotificationTickOnBarePlanStepCreatesOneFailedRowAndSendsNoEmail(t *tes // assertion failure instead of an unexpected-call panic. mockEmail.On("SendScheduledPurchaseNotification", ctx, mock.Anything).Return(nil).Maybe() mockStore.On("StampPlanNotificationSent", ctx, plan.ID, mock.Anything).Return(nil).Maybe() + // The failed row is reported to operators, once (issue #1195). + mockEmail.On("SendPurchaseFailedNotification", mock.Anything, mock.Anything).Return(nil) manager := &Manager{config: mockStore, email: mockEmail, notifyDays: 7, resolver: &fakeResolver{err: planstep.ErrNoMatch}} @@ -175,7 +178,9 @@ func TestNotificationTickOnBarePlanStepCreatesOneFailedRowAndSendsNoEmail(t *tes } mockEmail.AssertNotCalled(t, "SendScheduledPurchaseNotification", ctx, mock.Anything) + mockEmail.AssertNumberOfCalls(t, "SendPurchaseFailedNotification", 1) require.Len(t, saves, 1, "two ticks must leave exactly one row") + assert.Equal(t, 1, logs.count(saves[0].ExecutionID), "one purchase-failed marker for the row") assert.Equal(t, "failed", saves[0].Status) assert.Equal(t, 1, saves[0].StepNumber) assert.Contains(t, saves[0].Error, "no current recommendation matched") diff --git a/internal/purchase/reaper.go b/internal/purchase/reaper.go index 883f3210c..86caf20ab 100644 --- a/internal/purchase/reaper.go +++ b/internal/purchase/reaper.go @@ -280,24 +280,34 @@ func (m *Manager) reapOne(ctx context.Context, exec *config.PurchaseExecution, r } transitioned.Error = fmt.Sprintf("reaped after %dm in %s state — executor did not complete%s", ageMinutes, prevStatus, safeMsg) - if saveErr := m.config.SaveExecutionIfStatus(ctx, transitioned, []string{failedStatus}); saveErr != nil { - if errors.Is(saveErr, config.ErrExecutionNotInExpectedStatus) { - // The executor finished late and recovered the row (completed), so - // the annotation must not overwrite it. The reap itself succeeded. - logging.Infof("purchase reaper: execution %s left failed before the reap annotation; leaving it as is: %v", - exec.ExecutionID, saveErr) - result.Reaped++ - return - } - logging.Errorf("purchase reaper: failed to persist canonical error for execution %s (already flipped to failed): %v", - exec.ExecutionID, saveErr) - result.Errored++ - // Still count as Reaped — the status flip succeeded, only the - // error-message annotation failed. Otherwise Reaped would - // undercount real recoveries. - result.Reaped++ - return + if m.annotateReapedRow(ctx, transitioned, result) { + // The won CAS makes this sweep the one writer that notifies (#1195). + m.notifyExecutionFailure(ctx, transitioned) } +} +// annotateReapedRow persists the reap reason on the row the CAS failed and +// counts the reap. It reports whether the row is still failed: false only when +// the executor finished late and recovered the row before the annotation, so +// the reap must not be reported as a failed purchase. +func (m *Manager) annotateReapedRow(ctx context.Context, transitioned *config.PurchaseExecution, result *ReapResult) bool { + // Counted as Reaped on every path: the status flip succeeded, only the + // annotation may not have. Otherwise Reaped would undercount real recoveries. result.Reaped++ + saveErr := m.config.SaveExecutionIfStatus(ctx, transitioned, []string{failedStatus}) + switch { + case saveErr == nil: + return true + case errors.Is(saveErr, config.ErrExecutionNotInExpectedStatus): + // The executor finished late and recovered the row (completed), so + // the annotation must not overwrite it. The reap itself succeeded. + logging.Infof("purchase reaper: execution %s left failed before the reap annotation; leaving it as is: %v", + transitioned.ExecutionID, saveErr) + return false + default: + logging.Errorf("purchase reaper: failed to persist canonical error for execution %s (already flipped to failed): %v", + transitioned.ExecutionID, saveErr) + result.Errored++ + return true + } } diff --git a/internal/purchase/step_cap.go b/internal/purchase/step_cap.go index 5f4757986..90c169387 100644 --- a/internal/purchase/step_cap.go +++ b/internal/purchase/step_cap.go @@ -60,9 +60,11 @@ func (m *Manager) recheckStepCap(ctx context.Context, exec *config.PurchaseExecu // before any provider call, with the beforeSubmission class so its suppressions // release. It returns the refusal, joined with ErrAuditLoss when the terminal // save fails (the row would then be stranded in "running" for the reaper). +// Operators are notified either way, as executeAndFinalize does (issue #1195). func (m *Manager) failClaimedBeforeSubmission(ctx context.Context, exec *config.PurchaseExecution, cause error) error { execErr := beforeSubmission(cause) m.finalizeExecution(exec, execErr) + defer m.notifyExecutionFailure(ctx, exec) logging.Errorf("Refused claimed purchase %s before submission: %v", exec.ExecutionID, cause) if err := m.saveTerminalExecution(ctx, exec, execErr); err != nil { return fmt.Errorf("%w: terminal save failed (%w); original refusal: %w", config.ErrAuditLoss, err, execErr)