fix(purchase): send an outcome-unknown email, not 'retry manually', for reaped and swept rows

internal/email/purchase_outcome_unknown.go

@@ -0,0 +1,44 @@
+package email
+
+// purchaseOutcomeUnknownSubject replaces the purchase-failed subject when
+// NotificationData.OutcomeUnknown is set (issue #1415).
+const purchaseOutcomeUnknownSubject = "CUDly - Purchase Outcome Unknown - Verify in the Cloud Console"
+
+// purchaseOutcomeUnknownTemplate is the purchase-failed email for a row that a
+// recovery sweep (the stuck-row reaper or the stranded-approval sweep) failed
+// because its purchase run was interrupted. The provider may already have
+// created the commitment, so unlike purchaseFailedTemplate it never asks the
+// operator to retry: buying again without checking could create a duplicate
+// (issue #1415).
+const purchaseOutcomeUnknownTemplate = `CUDly - Purchase Outcome Unknown
+================================
+
+The purchase run for execution {{.ExecutionID}}{{if .PlanID}} (plan {{.PlanID}}){{end}} was interrupted
+before it recorded a result, and a recovery sweep marked it failed. CUDly
+cannot tell whether the commitments below were bought: the cloud provider
+may already have created them.
+
+Verify each one in the cloud provider's console before buying it again.
+Buying it again without checking can create a duplicate commitment.
+
+Outcome unknown:
+{{range .Recommendations}}
+- {{.Count}}x {{.ResourceType}}{{if .Engine}} ({{.Engine}}){{end}} in {{.Region}}
+  Service: {{.Service}}
+{{end}}
+
+Execution details:
+{{.DashboardURL}}/history
+
+This is an automated message from CUDly.
+`
+
+// purchaseFailedSubject returns failedSubject, the sender's usual
+// purchase-failed subject, or the outcome-unknown subject when data is for a
+// row whose purchase may already exist.
+func purchaseFailedSubject(data NotificationData, failedSubject string) string {
+	if data.OutcomeUnknown {
+		return purchaseOutcomeUnknownSubject
+	}
+	return failedSubject
+}

internal/email/purchase_outcome_unknown_test.go

@@ -0,0 +1,88 @@
+package email
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func outcomeUnknownData() NotificationData {
+	return NotificationData{
+		DashboardURL:   "https://dashboard.example.com",
+		ExecutionID:    "exec-1",
+		PlanID:         "plan-1",
+		OutcomeUnknown: true,
+		Recommendations: []RecommendationSummary{
+			{Service: "rds", ResourceType: "db.m5.large", Engine: "postgres", Region: "us-west-2", Count: 2},
+		},
+	}
+}
+
+// Issue #1415: a row a recovery sweep failed after an interrupted run may
+// already have bought its commitment, so its email asks for a console check,
+// names the execution and never says "retry".
+func TestRenderPurchaseFailedEmail_OutcomeUnknown(t *testing.T) {
+	body, err := RenderPurchaseFailedEmail(outcomeUnknownData())
+	require.NoError(t, err)
+
+	assert.NotContains(t, strings.ToLower(body), "retry")
+	assert.NotContains(t, body, "Failed Purchases")
+	assert.Contains(t, body, "Purchase Outcome Unknown")
+	assert.Contains(t, body, "execution exec-1 (plan plan-1)")
+	assert.Contains(t, body, "cloud provider's console")
+	assert.Contains(t, body, "- 2x db.m5.large (postgres) in us-west-2\n  Service: rds")
+	assert.Contains(t, body, "https://dashboard.example.com/history")
+}
+
+func TestRenderPurchaseFailedEmail_OutcomeUnknownWithoutPlan(t *testing.T) {
+	data := outcomeUnknownData()
+	data.PlanID = ""
+	body, err := RenderPurchaseFailedEmail(data)
+	require.NoError(t, err)
+	assert.Contains(t, body, "execution exec-1 was interrupted")
+	assert.NotContains(t, body, "(plan")
+}
+
+func TestPurchaseFailedSubject(t *testing.T) {
+	assert.Equal(t, "CUDly Purchase Failed", purchaseFailedSubject(NotificationData{}, "CUDly Purchase Failed"))
+	got := purchaseFailedSubject(outcomeUnknownData(), "CUDly Purchase Failed")
+	assert.Equal(t, purchaseOutcomeUnknownSubject, got)
+	assert.NotContains(t, strings.ToLower(got), "retry")
+	assert.LessOrEqual(t, len(got), snsMaxSubjectLen)
+}
+
+func TestSMTPSender_SendPurchaseFailedNotification_OutcomeUnknown(t *testing.T) {
+	for _, tc := range []struct {
+		name           string
+		outcomeUnknown bool
+		wantSubject    string
+		wantBody       string
+	}{
+		{name: "genuine failure", wantSubject: "CUDly Purchase Failed", wantBody: "Please review and retry manually."},
+		{name: "outcome unknown", outcomeUnknown: true, wantSubject: purchaseOutcomeUnknownSubject, wantBody: "Outcome unknown:"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			server := newMockSMTPServer(t, false)
+			server.start(t)
+			sender := &SMTPSender{
+				host:        "127.0.0.1",
+				port:        server.port,
+				fromEmail:   "notifications@example.com",
+				fromName:    "CUDly",
+				notifyEmail: "static-notify@example.com",
+			}
+			data := outcomeUnknownData()
+			data.OutcomeUnknown = tc.outcomeUnknown
+			require.NoError(t, sender.SendPurchaseFailedNotification(context.Background(), data))
+			server.stop()
+			server.mu.Lock()
+			msg := server.receivedMsg
+			server.mu.Unlock()
+			assert.Contains(t, msg, "Subject: "+tc.wantSubject)
+			assert.Contains(t, msg, tc.wantBody)
+		})
+	}
+}

internal/email/sender.go

@@ -602,6 +602,11 @@ type NotificationData struct {
 	// the purchase). Used in the post-execution notification body.
 	// Empty omits the field.
 	ExecutedBy string
+	// OutcomeUnknown switches the purchase-failed email to the outcome-unknown
+	// variant, for a row a recovery sweep failed after its purchase run was
+	// interrupted: the commitment may already exist, so the email asks the
+	// operator to verify in the cloud console instead of retrying (issue #1415).
+	OutcomeUnknown bool
 }
 
 // RecommendationSummary is a simplified recommendation for email display.

internal/email/smtp_sender.go

@@ -468,7 +468,7 @@ func (s *SMTPSender) SendPurchaseConfirmation(ctx context.Context, data Notifica
 // to data.RecipientEmail (the global notification address) when set and to
 // the static notify address otherwise (issue #714).
 func (s *SMTPSender) SendPurchaseFailedNotification(ctx context.Context, data NotificationData) error {
-	subject := "CUDly Purchase Failed"
+	subject := purchaseFailedSubject(data, "CUDly Purchase Failed")
 	body, err := RenderPurchaseFailedEmail(data)
 	if err != nil {
 		return fmt.Errorf("failed to render purchase failed email: %w", err)

internal/email/template_renderers.go

@@ -126,8 +126,13 @@ func RenderPurchaseConfirmationEmail(data NotificationData) (string, error) {
 	return renderTextTemplate("confirmation", purchaseConfirmationTemplate, data)
 }
 
-// RenderPurchaseFailedEmail renders the plain-text purchase failed email template.
+// RenderPurchaseFailedEmail renders the plain-text purchase failed email
+// template, or its outcome-unknown variant when data.OutcomeUnknown is set
+// (issue #1415).
 func RenderPurchaseFailedEmail(data NotificationData) (string, error) {
+	if data.OutcomeUnknown {
+		return renderTextTemplate("outcome-unknown", purchaseOutcomeUnknownTemplate, data)
+	}
 	return renderTextTemplate("failed", purchaseFailedTemplate, data)
 }
 

internal/email/templates.go

@@ -477,7 +477,7 @@ func (s *Sender) SendPurchaseFailedNotification(ctx context.Context, data Notifi
 		return fmt.Errorf("failed to render purchase failed email: %w", err)
 	}
 
-	subject := "CUDly - Purchase Failed - Action Required"
+	subject := purchaseFailedSubject(data, "CUDly - Purchase Failed - Action Required")
 	if data.RecipientEmail != "" {
 		return s.SendToEmailWithCC(ctx, data.RecipientEmail, data.CCEmails, subject, body)
 	}

internal/purchase/failure_notify.go

@@ -36,29 +36,50 @@ const failureNotifyTimeout = 30 * time.Second
 //     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.
 //
+// The sweeps that fail an interrupted run (safeFail, failClaimedRedrive and the
+// reaper) use notifyOutcomeUnknown instead.
+//
 // 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) {
+	m.sendFailureNotice(ctx, exec, false)
+}
+
+// notifyOutcomeUnknown is notifyExecutionFailure for a row a recovery sweep
+// failed because its purchase run was interrupted: safeFail and
+// failClaimedRedrive (stranded approved rows) and the reaper (stuck approved
+// or running rows). It logs the same marker, so the alarm fires as before, but
+// sends the outcome-unknown email: the run may have bought before it stopped,
+// so the email asks the operator to verify in the cloud console and never to
+// retry, which could buy the commitment twice (issue #1415).
+func (m *Manager) notifyOutcomeUnknown(ctx context.Context, exec *config.PurchaseExecution) {
+	m.sendFailureNotice(ctx, exec, true)
+}
+
+// sendFailureNotice logs the marker and sends the purchase-failed email,
+// flagged outcome-unknown when outcomeUnknown is set.
+func (m *Manager) sendFailureNotice(ctx context.Context, exec *config.PurchaseExecution, outcomeUnknown bool) {
 	if !logExecutionFailure(exec) || m.email == nil {
 		return
 	}
 	ctx, cancel := ctxutil.DetachedWithTimeout(ctx, failureNotifyTimeout)
 	defer cancel()
 	data := m.buildPurchaseFailedData(ctx, exec)
+	data.OutcomeUnknown = outcomeUnknown
 	if err := m.email.SendPurchaseFailedNotification(ctx, data); err != nil {
 		logging.Errorf("Failed to send purchase-failed notification for execution %s: %v", exec.ExecutionID, err)
 	}
 }
 
 // buildPurchaseFailedData lists the selected recommendations that did not
-// purchase and addresses the email to the global notification address (the
+// purchase (or, for an outcome-unknown row, whose purchase is unconfirmed) and
+// addresses the email to the global notification address (the
 // same recipient the scheduled-purchase notice uses). An empty recipient lets
 // the sender fall back to its broadcast channel.
 func (m *Manager) buildPurchaseFailedData(ctx context.Context, exec *config.PurchaseExecution) email.NotificationData {

internal/purchase/failure_notify_paths_test.go

@@ -3,10 +3,12 @@ package purchase
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/email"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -22,6 +24,9 @@ type writerCase struct {
 	run func(t *testing.T, m *Manager, store *MockConfigStore)
 	// alerts is whether the writer must alert (it won the row).
 	alerts bool
+	// outcomeUnknown is whether the alert is for a run a sweep failed after it
+	// was interrupted, whose commitment may already exist (issue #1415).
+	outcomeUnknown bool
 }
 
 func writerRow(status string) *config.PurchaseExecution {
@@ -40,20 +45,29 @@ func failedRowFromCAS(store *MockConfigStore, method string, args ...interface{}
 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) {
+		{name: "safeFail won CAS", alerts: true, outcomeUnknown: 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) {
+		{name: "safeFail won CAS, reason save failed", alerts: true, outcomeUnknown: 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)
 		}},
+		// The row was recovered before the guarded stamp: the failed outcome
+		// is stale, so nothing alerts (#1417).
+		{name: "safeFail won CAS, row recovered before the stamp", run: func(t *testing.T, m *Manager, store *MockConfigStore) {
+			failedRowFromCAS(store, "TransitionExecutionStatus", transition...)
+			store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{failedStatus}).Return(config.ErrExecutionNotInExpectedStatus)
+			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"))
@@ -63,26 +77,26 @@ func strandedWriterCases() []writerCase {
 		// 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) {
+		{name: "claimed redrive row failed", alerts: true, outcomeUnknown: 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) {
+		{name: "claimed redrive row failed, terminal save failed", alerts: true, outcomeUnknown: 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) {
+		{name: "reaper won CAS", alerts: true, outcomeUnknown: 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) {
+		{name: "reaper won CAS, reason save failed", alerts: true, outcomeUnknown: 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{}
@@ -174,7 +188,47 @@ func TestTerminalFailureWritersAlertOncePerWonRow(t *testing.T) {
 			if want == 1 {
 				assert.Equal(t, "exec-w", sent[0].ExecutionID)
 				assert.Equal(t, failureNotifyRecipient, sent[0].RecipientEmail)
+				assertFailureEmailBody(t, sent[0], tc.outcomeUnknown)
 			}
 		})
 	}
 }
+
+// genuineFailureEmail is the purchase-failed email for writerRow, unchanged
+// since before issue #1415.
+const genuineFailureEmail = `CUDly - Purchase Failed
+=======================
+
+Some purchases could not be completed. Please review and retry manually.
+
+Failed Purchases:
+
+- 1x Compute in eastus
+  Service: savingsplans
+
+
+Review failed purchases:
+/history
+
+This is an automated message from CUDly.
+`
+
+// assertFailureEmailBody renders the email the writer sent. A row a sweep
+// failed after an interrupted run may already have bought its commitment, so
+// its email must name the execution, ask for a console check and never say
+// "retry" (issue #1415). A genuinely failed row keeps the old email.
+func assertFailureEmailBody(t *testing.T, data email.NotificationData, outcomeUnknown bool) {
+	t.Helper()
+	body, err := email.RenderPurchaseFailedEmail(data)
+	require.NoError(t, err)
+	if !outcomeUnknown {
+		assert.Equal(t, genuineFailureEmail, body)
+		return
+	}
+	assert.NotContains(t, strings.ToLower(body), "retry")
+	assert.NotContains(t, body, "Failed Purchases")
+	assert.Contains(t, body, "Purchase Outcome Unknown")
+	assert.Contains(t, body, "execution exec-w (plan plan-w)")
+	assert.Contains(t, body, "Verify each one in the cloud provider's console before buying it again.")
+	assert.Contains(t, body, "Outcome unknown:\n\n- 1x Compute in eastus\n  Service: savingsplans\n")
+}

internal/purchase/manager.go

@@ -711,10 +711,12 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution)
 	// This sweep won the CAS, so it is the one writer that notifies (#1195),
 	// unless the guarded stamp below finds the row already recovered: then the
 	// in-memory failed row is stale and a failure email would be false.
+	// The interrupted run may already have bought, so the email reports an
+	// unknown outcome rather than a failure to retry (#1415).
 	notify := true
 	defer func() {
 		if notify {
-			m.notifyExecutionFailure(ctx, updated)
+			m.notifyOutcomeUnknown(ctx, updated)
 		}
 	}()
 

internal/purchase/reaper.go

@@ -282,7 +282,9 @@ func (m *Manager) reapOne(ctx context.Context, exec *config.PurchaseExecution, r
 		ageMinutes, prevStatus, safeMsg)
 	if m.annotateReapedRow(ctx, transitioned, result) {
 		// The won CAS makes this sweep the one writer that notifies (#1195).
-		m.notifyExecutionFailure(ctx, transitioned)
+		// The stuck executor may already have bought, so the email reports
+		// an unknown outcome rather than a failure to retry (#1415).
+		m.notifyOutcomeUnknown(ctx, transitioned)
 	}
 }
 

internal/purchase/stranded_redrive.go

@@ -45,7 +45,7 @@ func (m *Manager) failClaimedRedrive(ctx context.Context, exec *config.PurchaseE
 	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)
+	defer m.notifyOutcomeUnknown(ctx, exec) // may already have bought (#1415)
 	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)