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.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_paths_test.go b/internal/purchase/failure_notify_paths_test.go new file mode 100644 index 000000000..fa56f5d2e --- /dev/null +++ b/internal/purchase/failure_notify_paths_test.go @@ -0,0 +1,180 @@ +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) + }}, + // claimAndRedrive won the approved -> running claim but the claimed row + // no longer qualifies for a re-drive (#1037): the claim holder alerts, + // even when its terminal save fails. + {name: "claimed redrive row failed", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{"running"}).Return(nil) + counted, err := m.failClaimedRedrive(context.Background(), writerRow("running")) + require.NoError(t, err) + assert.True(t, counted) + }}, + {name: "claimed redrive row failed, terminal save failed", alerts: true, run: func(t *testing.T, m *Manager, store *MockConfigStore) { + store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{"running"}).Return(errors.New("db down")) + counted, err := m.failClaimedRedrive(context.Background(), writerRow("running")) + require.ErrorIs(t, err, config.ErrAuditLoss) + 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) + }}, + // The CAS, reason write and suppression release share one transaction + // (#1052), so a later error rolls the CAS back: the row stays pending + // for the next tick and nothing alerts. + {name: "plan step won CAS, reason save rolled back", 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 won CAS, suppression release rolled back", run: func(t *testing.T, m *Manager, store *MockConfigStore) { + failedRowFromCAS(store, "TransitionExecutionStatusPinned", pinned...) + store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil) + store.On("DeleteSuppressionsByExecutionTx", mock.Anything, mock.Anything, "exec-w").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/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 a299a190a..ceb40ade6 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -291,11 +291,9 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE saveCtx, cancelSave := ctxutil.DetachedWithTimeout(ctx, 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 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 @@ -708,6 +706,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) updated.Error = strandedRecoveryError(exec) if saveErr := m.config.SavePurchaseExecution(ctx, updated); saveErr != nil { 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 45b1b264c..1dae5f509 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 } @@ -318,6 +321,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) } logFailedPlanStep(failed) + // Notified only once the transaction committed the failed row: a lost CAS + // or an error before commit leaves it pending and quiet (issue #1195). + m.notifyExecutionFailure(ctx, failed) return bareStepFailure(failed.PlanID, failed.StepNumber, reason) } 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/redrive_claimed_row_integration_test.go b/internal/purchase/redrive_claimed_row_integration_test.go index 071d30d3f..d449df75e 100644 --- a/internal/purchase/redrive_claimed_row_integration_test.go +++ b/internal/purchase/redrive_claimed_row_integration_test.go @@ -137,6 +137,10 @@ func TestRecoverStrandedApprovals_ClaimedRowTurnedUnsafeIsFailedNotBought(t *tes }) }) f.manager.config = store + sender, ok := f.manager.email.(*MockEmailSender) + require.True(t, ok) + failedMail := &failedEmailRecorder{} + failedMail.expect(sender) recovered, err := f.manager.RecoverStrandedApprovals(ctx) require.NoError(t, store.editErr) @@ -150,4 +154,7 @@ func TestRecoverStrandedApprovals_ClaimedRowTurnedUnsafeIsFailedNotBought(t *tes assert.Contains(t, got.Error, "start a new purchase") require.Len(t, got.Recommendations, 1) assert.Equal(t, "ec2", got.Recommendations[0].Service, "the failed row keeps the claimed recommendations") + sent := failedMail.all() + require.Len(t, sent, 1, "the claim holder reports the failed row once (issue #1195)") + assert.Equal(t, row.ExecutionID, sent[0].ExecutionID) } 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) diff --git a/internal/purchase/stranded_redrive.go b/internal/purchase/stranded_redrive.go index 0f999dd84..f5fd055ff 100644 --- a/internal/purchase/stranded_redrive.go +++ b/internal/purchase/stranded_redrive.go @@ -38,11 +38,14 @@ func redrivable(exec *config.PurchaseExecution) bool { // bought, so no_submission is not stamped and plan suppressions are kept. // Returns (true, nil) when the row was failed, matching safeFail's count, and // an ErrAuditLoss error when the terminal save fails (the row would then sit -// in "running" for the reaper). +// in "running" for the reaper). This worker holds the claim, so it is the one +// writer that alerts, and it alerts either way, as failClaimedBeforeSubmission +// does (issue #1195). func (m *Manager) failClaimedRedrive(ctx context.Context, exec *config.PurchaseExecution) (bool, error) { cause := errors.New(strandedRecoveryError(exec)) logging.Errorf("Recovering stranded approved execution %s: the claimed row no longer qualifies for an in-place re-drive; failing it for visibility", exec.ExecutionID) m.finalizeExecution(exec, cause) + defer m.notifyExecutionFailure(ctx, exec) if err := m.saveTerminalExecution(ctx, exec, cause); err != nil { return false, fmt.Errorf("%w: failed to save stranded execution %s as failed after claiming it (row stranded in running): %w", config.ErrAuditLoss, exec.ExecutionID, err)