fix(purchase): attach explicit manual plan steps

.github/workflows/ci.yml

@@ -430,7 +430,7 @@ jobs:
               continue
             fi
             echo "==> integration tests in $mod"
-            if ! (cd "$mod" && go test -v -race -tags=integration \
+            if ! (cd "$mod" && go test -v -race -tags=integration -timeout=20m \
                     -coverprofile="$RUNNER_TEMP/coverage-integration-${tag}.out" \
                     ./...) 2>&1 | tee "$log"; then
               echo "::error::integration tests failed in $mod"

internal/purchase/manual_plan_step_integration_test.go

@@ -0,0 +1,108 @@
+//go:build integration
+
+package purchase
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Real PostgreSQL and resolver; recommendation reads and email delivery are fixtures.
+func TestManualPlanStep_AzureSizingPersistsWithoutAutoPurchase(t *testing.T) {
+	ctx := context.Background()
+	store, pool := rampStepStore(ctx, t)
+	global, err := store.GetGlobalConfig(ctx)
+	require.NoError(t, err)
+	recipient := "fixture@example.invalid"
+	global.NotificationEmail, global.AutoPurchaseEnabled = &recipient, false
+	require.NoError(t, store.SaveGlobalConfig(ctx, global))
+	account := &config.CloudAccount{Name: "manual-azure", Provider: "azure", ExternalID: uuid.NewString(), Enabled: true}
+	require.NoError(t, store.CreateCloudAccount(ctx, account))
+	next := time.Now().Add(84 * time.Hour).Truncate(time.Microsecond)
+	plan := &config.PurchasePlan{Name: "manual-azure", Enabled: true, NotificationDaysBefore: 3,
+		Services:          map[string]config.ServiceConfig{"azure:compute": {Provider: "azure", Service: "compute", Enabled: true, Term: 1, Payment: "monthly", Coverage: 80}},
+		RampSchedule:      config.RampSchedule{Type: "weekly", PercentPerStep: 25, StepIntervalDays: 7, TotalSteps: 4, StartDate: next},
+		NextExecutionDate: &next}
+	require.NoError(t, plan.Validate())
+	require.NoError(t, store.CreatePurchasePlan(ctx, plan))
+	require.NoError(t, store.SetPlanAccounts(ctx, plan.ID, []string{account.ID}))
+	monthly := 1000.0
+	rec := config.RecommendationRecord{ID: "manual-azure-rec", Provider: "azure", Service: "compute", Region: "westeurope", ResourceType: "Standard_D2s_v3",
+		Count: 100, Term: 1, Payment: "monthly", MonthlyCost: &monthly, Savings: 500, CloudAccountID: &account.ID}
+	sender := newCountingSender(ctx)
+	m := &Manager{config: store, email: sender, resolver: planstepResolver{
+		src: &cumSource{shown: func() []config.RecommendationRecord { return []config.RecommendationRecord{rec} }}, store: cumStore{rows: []config.RecommendationRecord{rec}}}}
+	result, err := m.SendUpcomingPurchaseNotifications(ctx)
+	require.NoError(t, err)
+	assert.Zero(t, result.Notified, "manual plans do not implicitly mint roots")
+	assert.Empty(t, stepRowsInDB(ctx, t, store, plan.ID))
+	root := &config.PurchaseExecution{ExecutionID: uuid.NewString(), PlanID: plan.ID, Status: "pending", StepNumber: 1, ScheduledDate: next}
+	require.NoError(t, store.SavePurchaseExecution(ctx, root))
+	var wg sync.WaitGroup
+	errors := make(chan error, 4)
+	for range 4 {
+		wg.Add(1)
+		go func() { defer wg.Done(); _, tickErr := m.SendUpcomingPurchaseNotifications(ctx); errors <- tickErr }()
+	}
+	wg.Wait()
+	close(errors)
+	for err := range errors {
+		require.NoError(t, err)
+	}
+	rows := stepRowsInDB(ctx, t, store, plan.ID)
+	require.Len(t, rows, 1)
+	assert.Equal(t, root.ExecutionID, rows[0].ExecutionID)
+	assert.Equal(t, "notified", rows[0].Status)
+	assert.True(t, rows[0].ScheduledDate.Equal(next))
+	require.Len(t, rows[0].Recommendations, 1)
+	assert.Equal(t, 20, rows[0].Recommendations[0].Count)
+	require.NotNil(t, rows[0].Recommendations[0].MonthlyCost)
+	assert.Equal(t, 200.0, *rows[0].Recommendations[0].MonthlyCost)
+	assert.Equal(t, 2400.0, rows[0].Recommendations[0].TotalCommitment())
+	assert.Equal(t, 100.0, rows[0].EstimatedSavings)
+	assert.Zero(t, rows[0].TotalUpfrontCost)
+	eligible, err := m.executableByScheduler(ctx, &rows[0])
+	require.NoError(t, err)
+	assert.False(t, eligible)
+	var suppressed int
+	require.NoError(t, pool.QueryRow(ctx, `SELECT COALESCE(SUM(suppressed_count),0) FROM purchase_suppressions WHERE execution_id=$1`, root.ExecutionID).Scan(&suppressed))
+	assert.Equal(t, 20, suppressed)
+}
+
+func TestManualPlanStep_OverdueEmptyRootFailsOnceWithoutEmail(t *testing.T) {
+	ctx := context.Background()
+	store, _ := rampStepStore(ctx, t)
+	next := time.Now().Add(-30 * 24 * time.Hour).Truncate(time.Microsecond)
+	plan := attachPlan(next, 100)
+	plan.ID, plan.Name, plan.AutoPurchase = "", "overdue-manual", false
+	plan.Services["azure:compute"] = config.ServiceConfig{Provider: "azure", Service: "compute", Enabled: true, Term: 1, Payment: "monthly", Coverage: 80}
+	plan.RampSchedule = config.RampSchedule{Type: "weekly", PercentPerStep: 25, StepIntervalDays: 7, TotalSteps: 4, StartDate: next}
+	require.NoError(t, store.CreatePurchasePlan(ctx, plan))
+	root := &config.PurchaseExecution{ExecutionID: uuid.NewString(), PlanID: plan.ID, Status: "pending", StepNumber: 1, ScheduledDate: next}
+	require.NoError(t, store.SavePurchaseExecution(ctx, root))
+	future := time.Now().Add(20 * 24 * time.Hour).Truncate(time.Microsecond)
+	plan.NextExecutionDate = &future
+	require.NoError(t, store.UpdatePurchasePlan(ctx, plan))
+	m := &Manager{config: store}
+	for range 2 {
+		result, err := m.SendUpcomingPurchaseNotifications(ctx)
+		require.NoError(t, err)
+		assert.Zero(t, result.Notified)
+	}
+	rows := stepRowsInDB(ctx, t, store, plan.ID)
+	require.Len(t, rows, 1)
+	assert.Equal(t, "failed", rows[0].Status)
+	assert.True(t, rows[0].NoSubmission)
+	assert.Contains(t, rows[0].Error, "without recommendations")
+	assert.True(t, rows[0].ScheduledDate.Equal(next))
+	stored, err := store.GetPurchasePlan(ctx, plan.ID)
+	require.NoError(t, err)
+	assert.Zero(t, stored.RampSchedule.CurrentStep)
+}

internal/purchase/notifications.go

@@ -37,6 +37,10 @@ func (m *Manager) SendUpcomingPurchaseNotifications(ctx context.Context) (*Notif
 	notified := 0
 	for _rvc := range plans {
 		plan := plans[_rvc]
+		if err := m.failDueManualStep(ctx, &plan); err != nil {
+			logging.Errorf("Failed to inspect due manual plan step for plan %s: %v", plan.ID, err)
+			continue
+		}
 		if m.shouldNotifyPlan(plan, autoPurchaseEnabled) || shouldRetryOverduePlan(plan, autoPurchaseEnabled, global, time.Now()) {
 			if m.sendPlanNotification(ctx, &plan) {
 				notified++
@@ -52,7 +56,7 @@ func (m *Manager) SendUpcomingPurchaseNotifications(ctx context.Context) (*Notif
 // shouldNotifyPlan checks if a plan should trigger a notification.
 // autoPurchaseEnabled is the global scheduling.auto_purchase kill switch.
 func (m *Manager) shouldNotifyPlan(plan config.PurchasePlan, autoPurchaseEnabled bool) bool {
-	if !autoPurchaseEnabled || !plan.Enabled || !plan.AutoPurchase {
+	if !plan.Enabled || (plan.AutoPurchase && !autoPurchaseEnabled) {
 		return false
 	}
 
@@ -90,7 +94,7 @@ func planGraceDays(plan *config.PurchasePlan, global *config.GlobalConfig) int {
 // one day past the grace window, so the first tick at or after the deadline can
 // write the failed row. A step that already has a row is skipped by the claim.
 func shouldRetryOverduePlan(plan config.PurchasePlan, autoPurchaseEnabled bool, global *config.GlobalConfig, now time.Time) bool {
-	if !autoPurchaseEnabled || !plan.Enabled || !plan.AutoPurchase || plan.NextExecutionDate == nil || !plan.NextExecutionDate.Before(now) {
+	if !plan.Enabled || (plan.AutoPurchase && !autoPurchaseEnabled) || plan.NextExecutionDate == nil || !plan.NextExecutionDate.Before(now) {
 		return false
 	}
 	if plan.LastNotificationSent != nil && now.Sub(*plan.LastNotificationSent).Hours() < config.MinHoursBetweenNotifications {
@@ -99,6 +103,47 @@ func shouldRetryOverduePlan(plan config.PurchasePlan, autoPurchaseEnabled bool,
 	return now.Before(plan.NextExecutionDate.AddDate(0, 0, planGraceDays(&plan, global)+1))
 }
 
+func (m *Manager) failDueManualStep(ctx context.Context, plan *config.PurchasePlan) error {
+	if !plan.Enabled || plan.AutoPurchase {
+		return nil
+	}
+	var root *config.PurchaseExecution
+	if err := m.config.WithTx(ctx, func(tx pgx.Tx) error {
+		var err error
+		root, err = m.dueManualRootTx(ctx, tx, plan.ID)
+		return err
+	}); err != nil {
+		return fmt.Errorf("failed to inspect due manual plan step: %w", err)
+	}
+	if root == nil {
+		return nil
+	}
+	err := m.failExistingPlanStep(ctx, root, "scheduled plan step reached its scheduled date without recommendations; no purchase was submitted")
+	if errors.Is(err, errExecutionNotNotifiable) {
+		return nil
+	}
+	return err
+}
+
+func (m *Manager) dueManualRootTx(ctx context.Context, tx pgx.Tx, planID string) (*config.PurchaseExecution, error) {
+	locked, err := m.config.LockPurchasePlanTx(ctx, tx, planID)
+	if err != nil {
+		return nil, err
+	}
+	if locked == nil || !locked.Enabled || locked.AutoPurchase {
+		return nil, nil
+	}
+	rows, err := m.config.ListExecutionsForPlanStepTx(ctx, tx, planID, locked.RampSchedule.CurrentStep+1)
+	if err != nil || len(rows) == 0 {
+		return nil, err
+	}
+	candidate, blocked := adoptableRootRow(rows)
+	if blocked == nil && len(candidate.Recommendations) == 0 && !candidate.ScheduledDate.After(time.Now()) {
+		return candidate, nil
+	}
+	return nil, nil
+}
+
 // errExecutionNotNotifiable marks an existing execution for the plan's date
 // that has already left pending/notified; every notification tick until the
 // plan's NextExecutionDate advances hits it, so it is skipped quietly.
@@ -250,7 +295,7 @@ func adoptableRootRow(rows []config.PurchaseExecution) (*config.PurchaseExecutio
 // from the stale read overwriting the approval. A won CAS returns the fresh
 // row, which is the one the error text is written onto.
 func (m *Manager) failExistingPlanStep(ctx context.Context, exec *config.PurchaseExecution, reason string) error {
-	failed, err := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"pending", "notified"}, "failed", nil)
+	failed, err := m.config.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, []string{"pending", "notified"}, "failed", nil, true)
 	if errors.Is(err, config.ErrNotFound) || errors.Is(err, config.ErrExecutionNotInExpectedStatus) {
 		return fmt.Errorf("%w: %s changed state before it could be failed", errExecutionNotNotifiable, exec.ExecutionID)
 	}

internal/purchase/notifications_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/email"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/planstep"
+	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -82,6 +83,7 @@ func TestManager_SendUpcomingPurchaseNotifications_NotAutoPurchase(t *testing.T)
 	}
 
 	mockStore.On("ListPurchasePlans", ctx, config.PurchasePlanFilter{}).Return(plans, nil)
+	expectStepRows(mockStore, &plans[0])
 
 	manager := &Manager{
 		config:       mockStore,
@@ -278,7 +280,7 @@ func TestManager_GetOrCreateExecution_ExistingBareRowIsFailedNotRotated(t *testi
 	expectStepRows(mockStore, plan, *bare)
 	failed := *bare
 	failed.Status = "failed"
-	mockStore.On("TransitionExecutionStatus", ctx, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil)).Return(&failed, nil)
+	mockStore.On("TransitionExecutionStatusPinned", ctx, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil), true).Return(&failed, nil)
 	mockStore.On("SavePurchaseExecution", ctx, &failed).Return(nil)
 
 	manager := &Manager{config: mockStore, resolver: &fakeResolver{err: planstep.ErrNoMatch}}
@@ -307,7 +309,7 @@ func TestManager_GetOrCreateExecution_ExistingBareRowApprovedMeanwhileIsNotOverw
 		ExecutionID: "bare-exec-id", PlanID: "plan-123", Status: "pending", StepNumber: 1, ScheduledDate: nextExec,
 	}
 	expectStepRows(mockStore, plan, *stale)
-	mockStore.On("TransitionExecutionStatus", ctx, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil)).
+	mockStore.On("TransitionExecutionStatusPinned", ctx, "bare-exec-id", []string{"pending", "notified"}, "failed", (*string)(nil), true).
 		Return(nil, config.ErrExecutionNotInExpectedStatus)
 	// No SavePurchaseExecution expectation: an upsert here would clobber the approval.
 
@@ -538,6 +540,42 @@ func TestManager_SendUpcomingPurchaseNotifications_WithNotification(t *testing.T
 	mockEmail.AssertExpectations(t)
 }
 
+func TestManager_SendUpcomingPurchaseNotifications_ManualCleanupErrorDoesNotBlockLaterPlan(t *testing.T) {
+	ctx := context.Background()
+	store := new(MockConfigStore)
+	sender := new(MockEmailSender)
+	nextDate := time.Now().Add(3 * 24 * time.Hour)
+	plans := []config.PurchasePlan{
+		{ID: "manual-failed", Enabled: true, NotificationDaysBefore: 7, NextExecutionDate: &nextDate},
+		{ID: "automatic-next", Enabled: true, AutoPurchase: true, NotificationDaysBefore: 7, NextExecutionDate: &nextDate},
+	}
+	address := "fixture@example.invalid"
+	store.On("ListPurchasePlans", ctx, config.PurchasePlanFilter{}).Return(plans, nil)
+	store.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &address, AutoPurchaseEnabled: true}, nil)
+	store.On("WithTx", ctx, mock.Anything).Return(errors.New("manual cleanup transaction failed")).Once()
+	store.On("WithTx", ctx, mock.Anything).Return(nil).Run(func(args mock.Arguments) {
+		require.NoError(t, args.Get(1).(func(pgx.Tx) error)(nil))
+	}).Once()
+	expectStepRows(store, &plans[1], config.PurchaseExecution{
+		ExecutionID: "automatic-root", PlanID: plans[1].ID, Status: "pending", StepNumber: 1,
+		ScheduledDate: nextDate, Recommendations: []config.RecommendationRecord{{Provider: "aws", Service: "ec2", Count: 1}},
+	})
+	store.On("RotatePendingApprovalToken", ctx, "automatic-root", mock.Anything, mock.Anything).Return(true, nil).Once()
+	store.On("StampPlanNotificationSent", ctx, plans[1].ID, mock.AnythingOfType("time.Time")).Return(nil).Once()
+	sender.On("SendScheduledPurchaseNotification", ctx, mock.MatchedBy(func(data email.NotificationData) bool {
+		return data.PlanID == plans[1].ID
+	})).Return(nil).Once()
+
+	manager := &Manager{config: store, email: sender, dashboardURL: "https://dashboard.example.invalid"}
+	result, err := manager.SendUpcomingPurchaseNotifications(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 1, result.Notified)
+	store.AssertNotCalled(t, "LockPurchasePlanTx", mock.Anything, mock.Anything, plans[0].ID)
+	store.AssertNotCalled(t, "StampPlanNotificationSent", mock.Anything, plans[0].ID, mock.Anything)
+	store.AssertExpectations(t)
+	sender.AssertExpectations(t)
+}
+
 func TestManager_SendUpcomingPurchaseNotifications_TooFarAway(t *testing.T) {
 	ctx := context.Background()
 	mockStore := new(MockConfigStore)

internal/purchase/plan_step_atomic_test.go

@@ -17,7 +17,7 @@ import (
 
 func stepTestPlan() *config.PurchasePlan {
 	next := time.Now().Add(24 * time.Hour)
-	return &config.PurchasePlan{ID: "plan-s", RampSchedule: config.RampSchedule{CurrentStep: 1}, NextExecutionDate: &next}
+	return &config.PurchasePlan{ID: "plan-s", Enabled: true, AutoPurchase: true, RampSchedule: config.RampSchedule{CurrentStep: 1}, NextExecutionDate: &next}
 }
 
 func stepRow(status string, mutate func(*config.PurchaseExecution)) config.PurchaseExecution {

internal/purchase/plan_step_attach.go

@@ -104,6 +104,9 @@ func (m *Manager) claimFromRows(ctx context.Context, tx pgx.Tx, locked *config.P
 		}
 		return m.attachStep(ctx, tx, locked, claim, root)
 	}
+	if !locked.AutoPurchase {
+		return stepClaim{}, fmt.Errorf("%w: manual plan has no scheduled root", errPlanStepSkipped)
+	}
 	return m.attachStep(ctx, tx, locked, claim, nil)
 }
 
@@ -127,7 +130,7 @@ func planProviders(plan *config.PurchasePlan) []string {
 // compared at the microsecond precision PostgreSQL stores, so an in-memory plan
 // with nanoseconds (Linux clocks) is not mistaken for a changed one.
 func planUnchanged(listed, locked *config.PurchasePlan, providers []string) bool {
-	return locked.Enabled && locked.AutoPurchase && locked.NextExecutionDate != nil && listed.NextExecutionDate != nil &&
+	return locked.Enabled && locked.AutoPurchase == listed.AutoPurchase && locked.NextExecutionDate != nil && listed.NextExecutionDate != nil &&
 		locked.NextExecutionDate.Truncate(time.Microsecond).Equal(listed.NextExecutionDate.Truncate(time.Microsecond)) &&
 		strings.Join(planProviders(locked), ",") == strings.Join(providers, ",")
 }
@@ -167,8 +170,14 @@ func (m *Manager) attachStep(ctx context.Context, tx pgx.Tx, locked *config.Purc
 	}
 	recs := res.Recs
 	sort.Slice(recs, func(i, j int) bool { return recs[i].ID < recs[j].ID })
-	if capErr := CheckStepCommitmentCap(locked, recs); capErr != nil {
-		return m.failStep(ctx, tx, locked, claim, root, fmt.Sprintf("%v (recommendations: %s)", capErr, recIDs(recs)))
+	return m.attachWithinCap(ctx, tx, locked, claim, root, recs)
+}
+
+func (m *Manager) attachWithinCap(ctx context.Context, tx pgx.Tx, locked *config.PurchasePlan, claim stepClaim, root *config.PurchaseExecution, recs []config.RecommendationRecord) (stepClaim, error) {
+	if locked.AutoPurchase || locked.MaxStepCommitment != nil {
+		if capErr := CheckStepCommitmentCap(locked, recs); capErr != nil {
+			return m.failStep(ctx, tx, locked, claim, root, fmt.Sprintf("%v (recommendations: %s)", capErr, recIDs(recs)))
+		}
 	}
 	return m.writeAttached(ctx, tx, locked, claim, root, recs)
 }