feat(purchase): make notified real and pin the scheduler claim to the gated snapshot

internal/analytics/collector_test.go

@@ -303,6 +303,10 @@ func (m *mockConfigStore) TransitionExecutionStatus(ctx context.Context, executi
 	return nil, nil
 }
 
+func (m *mockConfigStore) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*config.PurchaseExecution, error) {
+	return nil, nil
+}
+
 func (m *mockConfigStore) SetCancelledBy(_ context.Context, _ string, _ string) error {
 	return nil
 }

internal/config/interfaces.go

@@ -177,6 +177,12 @@ type StoreInterface interface {
 	// When non-nil the actor is stamped onto transitioned_by + transitioned_at; when nil,
 	// transitioned_by is set to NULL and transitioned_at is still set to NOW() for ordering.
 	TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string) (*PurchaseExecution, error)
+	// TransitionExecutionStatusPinned is TransitionExecutionStatus that also
+	// requires the row's recommendations to be empty (sawEmptyRecommendations)
+	// or non-empty exactly as the caller saw them, so a claim cannot land on a
+	// row whose recommendations were attached after the caller's gate check. A
+	// lost pin returns ErrExecutionNotInExpectedStatus like any lost CAS.
+	TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error)
 	// SetCancelledBy stamps canceled_by and the legacy cancelled_by column on
 	// an execution without overwriting any other columns. Used after
 	// TransitionExecutionStatus to fold the actor attribution into the same

internal/config/store_postgres.go

@@ -1168,12 +1168,15 @@ func (s *PostgresStore) LinkRetryExecutionAtomic(ctx context.Context, tx pgx.Tx,
 	return tag.RowsAffected() == 1, nil
 }
 
-// RotatePendingApprovalToken implements StoreInterface.
+// RotatePendingApprovalToken implements StoreInterface. The same statement
+// that stores the emailed token promotes the row to 'notified' (a no-op for a
+// row already there), so the status never claims a notice that was not sent.
 func (s *PostgresStore) RotatePendingApprovalToken(ctx context.Context, executionID, tokenHash string, expiresAt time.Time) (bool, error) {
 	q := `
 		UPDATE purchase_executions
 		   SET approval_token_hash = $2,
 		       approval_token_expires_at = $3,
+		       status = 'notified',
 		       updated_at = NOW()
 		 WHERE execution_id = $1
 		   AND status IN ('pending', 'notified')
@@ -1194,6 +1197,19 @@ func (s *PostgresStore) RotatePendingApprovalToken(ctx context.Context, executio
 // and the legacy cancelled_by (issue #119), matching what the other cancel paths
 // record and what History reads.
 func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string) (*PurchaseExecution, error) {
+	return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, nil)
+}
+
+// TransitionExecutionStatusPinned is TransitionExecutionStatus that also
+// requires the row's recommendations to still be empty (or not empty), as the
+// caller saw them. A recommendation attach landing between a gate check and the
+// claim makes the pin fail with ErrExecutionNotInExpectedStatus, so a row is
+// never run on a snapshot different from the one that was gated.
+func (s *PostgresStore) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) {
+	return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, &sawEmptyRecommendations)
+}
+
+func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmpty *bool) (*PurchaseExecution, error) {
 	var cancelActor *string
 	if toStatus == StatusCanceled {
 		cancelActor = actor
@@ -1205,6 +1221,7 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution
 		    canceled_by = COALESCE((SELECT email FROM users WHERE id = $5), canceled_by),
 		    cancelled_by = COALESCE((SELECT email FROM users WHERE id = $5), cancelled_by)
 		WHERE execution_id = $1 AND status = ANY($3)
+		  AND ($6::boolean IS NULL OR (recommendations IN ('[]'::jsonb, 'null'::jsonb)) = $6)
 		RETURNING plan_id, execution_id, status, step_number, scheduled_date,
 		          notification_sent, approval_token_hash, recommendations,
 		          total_upfront_cost, estimated_savings, completed_at, error, expires_at,
@@ -1215,7 +1232,7 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution
 		          idempotency_key, scheduled_execution_at
 	`
 
-	records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses, actor, cancelActor)
+	records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty)
 	if err != nil {
 		return nil, err
 	}

internal/config/store_postgres_db_test.go

@@ -1661,7 +1661,8 @@ func TestPostgresStoreDB_LinkRetryExecutionAtomic(t *testing.T) {
 
 // TestPostgresStoreDB_RotatePendingApprovalToken (issue #103): the
 // notification path's token rotation writes approval_token_hash and its
-// expiry only while the row is pending/notified, and touches nothing else
+// expiry, and promotes the row to notified, only while it is pending/notified,
+// and touches nothing else
 // (a full-row upsert of a stale copy could revert a concurrent approve).
 // The hash round-trips through the approval_token_hash column.
 func TestPostgresStoreDB_RotatePendingApprovalToken(t *testing.T) {
@@ -1707,7 +1708,7 @@ func TestPostgresStoreDB_RotatePendingApprovalToken(t *testing.T) {
 			assert.Equal(t, HashApprovalToken("fresh-token"), reread.ApprovalToken)
 			require.NotNil(t, reread.ApprovalTokenExpiresAt)
 			assert.True(t, expiresAt.Equal(reread.ApprovalTokenExpiresAt.UTC().Truncate(time.Second)))
-			assert.Equal(t, status, reread.Status, "rotation must not touch status")
+			assert.Equal(t, "notified", reread.Status, "the token rotation is also the pending->notified CAS")
 		})
 	}
 
@@ -1720,6 +1721,65 @@ func TestPostgresStoreDB_RotatePendingApprovalToken(t *testing.T) {
 		reread, err := store.GetExecutionByID(ctx, exec.ExecutionID)
 		require.NoError(t, err)
 		assert.Equal(t, HashApprovalToken("original-token"), reread.ApprovalToken)
+		assert.Equal(t, "completed", reread.Status, "a row that left pending must not be revived to notified")
+	})
+}
+
+// TestPostgresStoreDB_TransitionExecutionStatusPinned: the claim only wins
+// while the row's recommendations are still as the caller saw them, so a
+// recommendation attach between a gate check and the claim loses the pin and
+// leaves the row untouched.
+func TestPostgresStoreDB_TransitionExecutionStatusPinned(t *testing.T) {
+	conn := setupTestContainerDB(t)
+	if conn == nil {
+		return
+	}
+	store := NewPostgresStore(conn)
+	ctx := context.Background()
+	plan := &PurchasePlan{Name: "pinned claim", Enabled: true, Services: map[string]ServiceConfig{}, RampSchedule: PresetRampSchedules["immediate"]}
+	require.NoError(t, store.CreatePurchasePlan(ctx, plan))
+	rec := []RecommendationRecord{{ID: "r1", Provider: "aws", Service: "ec2", Count: 1}}
+	newRow := func(t *testing.T, recs []RecommendationRecord) *PurchaseExecution {
+		t.Helper()
+		exec := &PurchaseExecution{PlanID: plan.ID, Status: "notified", StepNumber: 1, ScheduledDate: time.Now(), Recommendations: recs}
+		require.NoError(t, store.SavePurchaseExecution(ctx, exec))
+		return exec
+	}
+	from := []string{"approved", "pending", "notified"}
+
+	for _, empty := range []([]RecommendationRecord){nil, {}} {
+		exec := newRow(t, empty)
+		_, err := store.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, from, "running", nil, false)
+		require.ErrorIs(t, err, ErrExecutionNotInExpectedStatus, "gate saw recommendations but the row is empty")
+		reread, err := store.GetExecutionByID(ctx, exec.ExecutionID)
+		require.NoError(t, err)
+		assert.Equal(t, "notified", reread.Status)
+		got, err := store.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, from, "running", nil, true)
+		require.NoError(t, err)
+		assert.Equal(t, "running", got.Status)
+	}
+
+	t.Run("attach after the gate saw an empty row loses the claim", func(t *testing.T) {
+		exec := newRow(t, nil)
+		require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error {
+			ok, err := store.SetExecutionRecommendationsIfEmptyTx(ctx, tx, exec.ExecutionID, rec, 1, 1)
+			require.True(t, ok)
+			return err
+		}))
+		_, err := store.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, from, "running", nil, true)
+		require.ErrorIs(t, err, ErrExecutionNotInExpectedStatus)
+		reread, err := store.GetExecutionByID(ctx, exec.ExecutionID)
+		require.NoError(t, err)
+		assert.Equal(t, "notified", reread.Status)
+		got, err := store.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, from, "running", nil, false)
+		require.NoError(t, err)
+		assert.Equal(t, "running", got.Status, "the next tick gates the attached row and claims it")
+	})
+
+	t.Run("unpinned transition is unchanged", func(t *testing.T) {
+		got, err := store.TransitionExecutionStatus(ctx, newRow(t, rec).ExecutionID, from, "running", nil)
+		require.NoError(t, err)
+		assert.Equal(t, "running", got.Status)
 	})
 }
 

internal/config/store_postgres_pgxmock_test.go

@@ -2773,7 +2773,7 @@ func TestPGXMock_TransitionExecutionStatus_ProbeHardErrorNotMappedToNotFound(t *
 
 	// CAS UPDATE matches zero rows (status already transitioned or row gone).
 	mock.ExpectQuery(`UPDATE purchase_executions`).
-		WithArgs(anyArgsCfg(5)...).
+		WithArgs(anyArgsCfg(6)...).
 		WillReturnRows(pgxmock.NewRows(execCols))
 
 	// Probe fails with a hard DB error, not an empty result.

internal/mocks/stores.go

@@ -357,6 +357,26 @@ func (m *MockConfigStore) TransitionExecutionStatus(ctx context.Context, executi
 	return v, args.Error(1)
 }
 
+// TransitionExecutionStatusPinned mocks the pinned claim. With no expectation
+// registered it delegates to TransitionExecutionStatus, so tests that mock the
+// plain transition keep working; a test exercising a lost pin registers an
+// expectation returning config.ErrExecutionNotInExpectedStatus.
+func (m *MockConfigStore) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*config.PurchaseExecution, error) {
+	if !isExpected(&m.Mock, "TransitionExecutionStatusPinned") {
+		return m.TransitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor)
+	}
+	m.record("TransitionExecutionStatusPinned", ctx, executionID, fromStatuses, toStatus, actor, sawEmptyRecommendations)
+	args := m.Called(ctx, executionID, fromStatuses, toStatus, actor, sawEmptyRecommendations)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	v, ok := args.Get(0).(*config.PurchaseExecution)
+	if !ok {
+		panic(fmt.Sprintf("mock: expected *config.PurchaseExecution, got %T", args.Get(0)))
+	}
+	return v, args.Error(1)
+}
+
 // SetCancelledBy mocks the SetCancelledBy targeted-update operation.
 func (m *MockConfigStore) SetCancelledBy(ctx context.Context, executionID, cancelledBy string) error {
 	m.record("SetCancelledBy", ctx, executionID, cancelledBy)

internal/planstep/precheck.go

@@ -0,0 +1,71 @@
+package planstep
+
+import (
+	"errors"
+	"fmt"
+	"sort"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+)
+
+// ErrUnsupportedPlan marks a plan whose settings Resolve cannot honor. Every
+// such plan is refused before any recommendation is read, rather than resolved
+// with the setting silently ignored.
+var ErrUnsupportedPlan = errors.New("plan settings are not supported by recommendation resolution")
+
+// Precheck refuses, with a plan-specific error, the plan settings Resolve does
+// not apply: any Include*/Exclude*/MinCount filter on a service config (the
+// plan-step request never sets them, so honoring them would buy something other
+// than what the plan says), and an immediate ramp whose PercentPerStep is not
+// 100 (an immediate ramp ignores that field when validated but Resolve sizes
+// with it).
+func Precheck(plan *config.PurchasePlan) error {
+	keys := make([]string, 0, len(plan.Services))
+	for k := range plan.Services {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		svc := plan.Services[k]
+		if field := firstFilterSet(&svc); field != "" {
+			return fmt.Errorf("%w: plan %s service %q sets %s", ErrUnsupportedPlan, plan.ID, k, field)
+		}
+	}
+	if plan.RampSchedule.Type == "immediate" && plan.RampSchedule.PercentPerStep != 100 {
+		return fmt.Errorf("%w: plan %s has an immediate ramp with percent_per_step %v (must be 100)",
+			ErrUnsupportedPlan, plan.ID, plan.RampSchedule.PercentPerStep)
+	}
+	return nil
+}
+
+// firstFilterSet names the first filter field set on svc, or "".
+func firstFilterSet(svc *config.ServiceConfig) string {
+	switch {
+	case len(svc.IncludeEngines) > 0:
+		return "include_engines"
+	case len(svc.ExcludeEngines) > 0:
+		return "exclude_engines"
+	case len(svc.IncludeRegions) > 0:
+		return "include_regions"
+	case len(svc.ExcludeRegions) > 0:
+		return "exclude_regions"
+	case len(svc.IncludeTypes) > 0:
+		return "include_types"
+	case len(svc.ExcludeTypes) > 0:
+		return "exclude_types"
+	case svc.MinCount != 0:
+		return "min_count"
+	}
+	return ""
+}
+
+// checkStoredSPCount refuses a Savings Plans stored row whose Count is not 1:
+// an SP is sized by its hourly commitment, so any other count means the row is
+// not the one-commitment shape the sizing assumes.
+func checkStoredSPCount(stored *config.RecommendationRecord) error {
+	if common.IsSavingsPlan(common.ServiceType(stored.Service)) && stored.Count != 1 {
+		return fmt.Errorf("%w: savings plan stored row %s has count %d (must be 1)", ErrUnsupportedPlan, stored.ID, stored.Count)
+	}
+	return nil
+}

internal/planstep/precheck_test.go

@@ -0,0 +1,86 @@
+package planstep
+
+import (
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Fixtures are hand-built plans, not real configuration.
+
+func planWith(svc config.ServiceConfig, ramp config.RampSchedule) *config.PurchasePlan {
+	return &config.PurchasePlan{ID: "plan-pc", Services: map[string]config.ServiceConfig{"aws:ec2": svc}, RampSchedule: ramp}
+}
+
+func TestPrecheck(t *testing.T) {
+	ok := config.ServiceConfig{Provider: "aws", Service: "ec2", Enabled: true}
+	weekly := config.RampSchedule{Type: "weekly", PercentPerStep: 25}
+	cases := []struct {
+		name string
+		plan *config.PurchasePlan
+		want string // substring of the error; "" = accepted
+	}{
+		{"clean plan", planWith(ok, weekly), ""},
+		{"immediate at 100", planWith(ok, config.RampSchedule{Type: "immediate", PercentPerStep: 100}), ""},
+		{"immediate at 50", planWith(ok, config.RampSchedule{Type: "immediate", PercentPerStep: 50}), "immediate ramp"},
+		{"immediate at 0", planWith(ok, config.RampSchedule{Type: "immediate"}), "immediate ramp"},
+		{"weekly at 25 is not an immediate ramp", planWith(ok, weekly), ""},
+		{"include engines", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.IncludeEngines = []string{"mysql"} }), weekly), "include_engines"},
+		{"exclude engines", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.ExcludeEngines = []string{"mysql"} }), weekly), "exclude_engines"},
+		{"include regions", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.IncludeRegions = []string{"us-east-1"} }), weekly), "include_regions"},
+		{"exclude regions", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.ExcludeRegions = []string{"us-east-1"} }), weekly), "exclude_regions"},
+		{"include types", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.IncludeTypes = []string{"m5.large"} }), weekly), "include_types"},
+		{"exclude types", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.ExcludeTypes = []string{"m5.large"} }), weekly), "exclude_types"},
+		{"min count", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.MinCount = 1 }), weekly), "min_count"},
+		{"negative min count", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.MinCount = -1 }), weekly), "min_count"},
+		{"filter on a disabled service still refuses", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.Enabled = false; s.MinCount = 2 }), weekly), "min_count"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := Precheck(tc.plan)
+			if tc.want == "" {
+				assert.NoError(t, err)
+				return
+			}
+			require.ErrorIs(t, err, ErrUnsupportedPlan)
+			assert.ErrorContains(t, err, tc.want)
+			assert.ErrorContains(t, err, "plan-pc")
+		})
+	}
+}
+
+func withSvc(s config.ServiceConfig, f func(*config.ServiceConfig)) config.ServiceConfig {
+	f(&s)
+	return s
+}
+
+func TestCheckStoredSPCount(t *testing.T) {
+	sp := func(count int) *config.RecommendationRecord {
+		return &config.RecommendationRecord{ID: "r", Service: "savings-plans-compute", Count: count}
+	}
+	assert.NoError(t, checkStoredSPCount(sp(1)))
+	for _, n := range []int{0, 2, -1} {
+		require.ErrorIs(t, checkStoredSPCount(sp(n)), ErrUnsupportedPlan, "count %d", n)
+	}
+	// A count-denominated service is not constrained.
+	assert.NoError(t, checkStoredSPCount(&config.RecommendationRecord{Service: "ec2", Count: 7}))
+}
+
+func TestResolveRefusesUnsupportedPlanBeforeReading(t *testing.T) {
+	plan := azurePlan(100, 100)
+	svc := plan.Services["azure:compute"]
+	svc.MinCount = 3
+	plan.Services["azure:compute"] = svc
+	// A nil Source/Store would panic if Resolve read anything.
+	_, err := Resolve(t.Context(), nil, nil, plan, accts)
+	require.ErrorIs(t, err, ErrUnsupportedPlan)
+}
+
+func TestResolveRefusesStoredSPWithCountNotOne(t *testing.T) {
+	stored := spRec(`{"hourly_commitment":1.234}`)
+	stored.Count = 3
+	_, err := Resolve(t.Context(), fakeSource{shown: []config.RecommendationRecord{stored}}, fakeStore{rows: []config.RecommendationRecord{stored}}, spPlan(80, 25), accts)
+	require.ErrorIs(t, err, ErrUnsupportedPlan)
+}

internal/planstep/resolve.go

@@ -46,6 +46,9 @@ type Store interface {
 // excluded. Every failure is returned; there is no unsized or over-show
 // fallback. An empty result is ErrNoMatch.
 func Resolve(ctx context.Context, src Source, store Store, plan *config.PurchasePlan, accountIDs []string) ([]config.RecommendationRecord, error) {
+	if err := Precheck(plan); err != nil {
+		return nil, err
+	}
 	if len(accountIDs) == 0 {
 		return nil, fmt.Errorf("plan %s has no target accounts", plan.ID)
 	}
@@ -213,6 +216,9 @@ func size(shown, stored *config.RecommendationRecord, product float64) (config.R
 	if stored.Count <= 0 {
 		return config.RecommendationRecord{}, false, fmt.Errorf("stored count %d cannot price a share", stored.Count)
 	}
+	if err := checkStoredSPCount(stored); err != nil {
+		return config.RecommendationRecord{}, false, err
+	}
 	out := *shown
 	out.SuppressedCount, out.SuppressionExpiresAt, out.PrimarySuppressionExecutionID = 0, nil, nil
 	out.Selected = true

internal/planstep/resolve_test.go

@@ -216,9 +216,8 @@ func TestResolveSavingsPlanRoundsDownAndKeepsRawKeys(t *testing.T) {
 	stored := spRec(`{"hourly_commitment":1.234,"platform":"Linux/UNIX","tenancy":"default","plan_type":"Compute","extra":{"a":[1,2]}}`)
 	before, err := stored.ComputeIdentity()
 	require.NoError(t, err)
-	stored.Count = 3
+	stored.Count = 1
 	shown := stored
-	shown.Count = 2 // partially suppressed placeholder count
 	got, err := Resolve(t.Context(), fakeSource{shown: []config.RecommendationRecord{shown}}, fakeStore{rows: []config.RecommendationRecord{stored}}, spPlan(80, 25), accts)
 	require.NoError(t, err)
 	require.Len(t, got, 1)