fix(purchase): alert on failed fan-out roots and every terminal-failure writer

internal/purchase/claim_pin_test.go

@@ -143,14 +143,16 @@ 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)
 	require.Error(t, err)
 	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)
 }
 

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)

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)
+			}
+		})
+	}
+}