feat(config): ramp baseline column and plan committed ledger (631 C2)

internal/analytics/collector_test.go

@@ -893,6 +893,18 @@ func (m *mockConfigStore) ListExecutionsForPlanStepTx(_ context.Context, _ pgx.T
 	return nil, nil
 }
 
+func (m *mockConfigStore) SetPlanRampBaselineTx(_ context.Context, _ pgx.Tx, _ string, _ config.PlanRampBaseline) (bool, error) {
+	return false, nil
+}
+
+func (m *mockConfigStore) GetPlanRampBaselineTx(_ context.Context, _ pgx.Tx, _ string) (config.PlanRampBaseline, error) {
+	return nil, nil
+}
+
+func (m *mockConfigStore) ListPlanCommittedTx(_ context.Context, _ pgx.Tx, _ string) (map[string]config.PlanRampQuantity, error) {
+	return nil, nil
+}
+
 func (m *mockConfigStore) SetExecutionRecommendationsIfEmptyTx(_ context.Context, _ pgx.Tx, _ string, _ []config.RecommendationRecord, _, _ float64) (bool, error) {
 	return false, nil
 }

internal/config/interfaces.go

@@ -97,6 +97,17 @@ type StoreInterface interface {
 	// get-or-create of a plan step reads it under LockPurchasePlanTx so any
 	// existing row, settled or not, blocks minting a second root (platform#631).
 	ListExecutionsForPlanStepTx(ctx context.Context, tx pgx.Tx, planID string, stepNumber int) ([]PurchaseExecution, error)
+	// SetPlanRampBaselineTx freezes the plan's ramp baseline if none is set and
+	// reports whether it wrote. Immutable once set; not part of
+	// UpdatePurchasePlanTx, so a stale plan save cannot touch it. The caller
+	// holds LockPurchasePlanTx.
+	SetPlanRampBaselineTx(ctx context.Context, tx pgx.Tx, planID string, baseline PlanRampBaseline) (bool, error)
+	// GetPlanRampBaselineTx returns the frozen baseline, nil when none is set.
+	GetPlanRampBaselineTx(ctx context.Context, tx pgx.Tx, planID string) (PlanRampBaseline, error)
+	// ListPlanCommittedTx returns what the plan's executions already hold per
+	// recommendation identity (see its implementation for the status rules).
+	// The caller holds LockPurchasePlanTx.
+	ListPlanCommittedTx(ctx context.Context, tx pgx.Tx, planID string) (map[string]PlanRampQuantity, error)
 	// SetExecutionRecommendationsIfEmptyTx attaches resolved recommendations
 	// and their totals to a pending/notified execution that has none yet.
 	// Returns false when another resolver already attached them (or the row

internal/config/store_postgres_ramp_baseline.go

@@ -0,0 +1,182 @@
+package config
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// PlanRampQuantity is a quantity of one recommendation identity: whole units
+// for count-based recs, the summed hourly_commitment for Savings Plans.
+type PlanRampQuantity struct {
+	Count  int     `json:"count"`
+	Hourly float64 `json:"hourly,omitempty"`
+}
+
+// PlanRampBaseline maps RecIdentityKey to the base a ramp plan sizes against.
+type PlanRampBaseline map[string]PlanRampQuantity
+
+// SetPlanRampBaselineTx freezes a plan's ramp baseline. It writes only while
+// the column is NULL and reports whether it did: the baseline is immutable
+// once set. It has its own UPDATE and the plan's full-row UPDATE
+// (UpdatePurchasePlanTx) never names the column, so a plan saved from a stale
+// in-memory struct cannot clear or replace it.
+func (s *PostgresStore) SetPlanRampBaselineTx(ctx context.Context, tx pgx.Tx, planID string, baseline PlanRampBaseline) (bool, error) {
+	if len(baseline) == 0 {
+		return false, fmt.Errorf("refusing to freeze an empty ramp baseline for plan %s", planID)
+	}
+	raw, err := json.Marshal(baseline)
+	if err != nil {
+		return false, fmt.Errorf("failed to marshal ramp baseline of plan %s: %w", planID, err)
+	}
+	tag, err := tx.Exec(ctx,
+		`UPDATE purchase_plans SET ramp_baseline = $2 WHERE id = $1 AND ramp_baseline IS NULL`, planID, raw)
+	if err != nil {
+		return false, fmt.Errorf("failed to write ramp baseline of plan %s: %w", planID, err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+// GetPlanRampBaselineTx returns the frozen baseline, or nil when none was
+// written. It errors when the plan does not exist.
+func (s *PostgresStore) GetPlanRampBaselineTx(ctx context.Context, tx pgx.Tx, planID string) (PlanRampBaseline, error) {
+	var raw []byte
+	err := tx.QueryRow(ctx, `SELECT ramp_baseline FROM purchase_plans WHERE id = $1`, planID).Scan(&raw)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, fmt.Errorf("%w: purchase plan %s", ErrNotFound, planID)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to read ramp baseline of plan %s: %w", planID, err)
+	}
+	if raw == nil {
+		return nil, nil
+	}
+	var baseline PlanRampBaseline
+	if err := json.Unmarshal(raw, &baseline); err != nil {
+		return nil, fmt.Errorf("failed to decode ramp baseline of plan %s: %w", planID, err)
+	}
+	return baseline, nil
+}
+
+// ListPlanCommittedTx returns, per recommendation identity, what the plan's
+// executions already hold, committed or still in flight. It is the ledger the
+// plan's next step subtracts from its cumulative target. It reads the
+// executions themselves, never the suppressions, because a suppression expires
+// while the capacity it covered is still held.
+//
+// A row counts unless it is provably unbought: canceled (both spellings),
+// expired, or failed with NoSubmission set. A failed row without NoSubmission
+// is an unknown outcome and counts, as does any status this code does not know
+// (the fail-safe direction: over-counting only under-buys). Every selected rec
+// of a counted row counts, including a failed rec of a partially_completed
+// row, because a rec error can be a timeout after the provider accepted it.
+//
+// Within one step the root row of a fan-out, its per-account rows and a
+// retry chain all carry the same recs, so each identity counts once per step
+// as the largest quantity any counted row holds; steps then add up.
+//
+// The caller must hold the plan's lock (LockPurchasePlanTx) in tx or the
+// answer is advisory.
+func (s *PostgresStore) ListPlanCommittedTx(ctx context.Context, tx pgx.Tx, planID string) (map[string]PlanRampQuantity, error) {
+	rows, err := tx.Query(ctx, `
+		SELECT 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,
+		       cloud_account_id, source, approved_by, canceled_by, capacity_percent,
+		       created_by_user_id, retry_execution_id, retry_attempt_n,
+		       approval_token_expires_at,
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key, scheduled_execution_at, no_submission
+		FROM purchase_executions
+		WHERE plan_id = $1`, planID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query executions of plan %s: %w", planID, err)
+	}
+	defer rows.Close()
+	execs, err := scanExecutionRows(rows)
+	if err != nil {
+		return nil, err
+	}
+	return sumPlanCommitted(execs)
+}
+
+// holdsCapacity reports whether an execution row may hold bought or in-flight
+// capacity. See ListPlanCommittedTx for the rule.
+func holdsCapacity(e *PurchaseExecution) bool {
+	switch e.Status {
+	case StatusCanceled, LegacyStatusCanceled, StatusExpired:
+		return false
+	case StatusFailed:
+		return !e.NoSubmission
+	}
+	return true
+}
+
+func sumPlanCommitted(execs []PurchaseExecution) (map[string]PlanRampQuantity, error) {
+	type stepKey struct {
+		identity string
+		step     int
+	}
+	perStep := map[stepKey]PlanRampQuantity{}
+	for i := range execs {
+		e := &execs[i]
+		if !holdsCapacity(e) {
+			continue
+		}
+		held := map[string]PlanRampQuantity{}
+		for j := range e.Recommendations {
+			rec := &e.Recommendations[j]
+			if !rec.Selected {
+				continue
+			}
+			identity, err := RecIdentityKey(rec)
+			if err != nil {
+				return nil, fmt.Errorf("execution %s: %w", e.ExecutionID, err)
+			}
+			hourly, err := recHourlyCommitment(rec)
+			if err != nil {
+				return nil, fmt.Errorf("execution %s: %w", e.ExecutionID, err)
+			}
+			q := held[identity]
+			q.Count += rec.Count
+			q.Hourly += hourly
+			held[identity] = q
+		}
+		for identity, q := range held {
+			k := stepKey{identity, e.StepNumber}
+			cur := perStep[k]
+			cur.Count = max(cur.Count, q.Count)
+			cur.Hourly = max(cur.Hourly, q.Hourly)
+			perStep[k] = cur
+		}
+	}
+	out := map[string]PlanRampQuantity{}
+	for k, q := range perStep {
+		sum := out[k.identity]
+		sum.Count += q.Count
+		sum.Hourly += q.Hourly
+		out[k.identity] = sum
+	}
+	return out, nil
+}
+
+// recHourlyCommitment reads Details.hourly_commitment: 0 when the rec carries
+// none, an error when it carries one that is not a number.
+func recHourlyCommitment(rec *RecommendationRecord) (float64, error) {
+	if len(rec.Details) == 0 {
+		return 0, nil
+	}
+	var d struct {
+		Hourly *float64 `json:"hourly_commitment"`
+	}
+	if err := json.Unmarshal(rec.Details, &d); err != nil {
+		return 0, fmt.Errorf("recommendation details unreadable: %w", err)
+	}
+	if d.Hourly == nil {
+		return 0, nil
+	}
+	return *d.Hourly, nil
+}

internal/config/store_postgres_ramp_baseline_integration_test.go

@@ -0,0 +1,142 @@
+//go:build integration
+// +build integration
+
+package config
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func (f *rampFixture) rowWithRecs(ctx context.Context, t *testing.T, step int, account *string, status string, noSubmission bool, recs ...RecommendationRecord) *PurchaseExecution {
+	t.Helper()
+	exec := &PurchaseExecution{
+		ExecutionID: uuid.New().String(), IdempotencyKey: uuid.New().String(), PlanID: f.planID,
+		CloudAccountID: account, Status: status, StepNumber: step, ScheduledDate: time.Now(),
+		Recommendations: recs, NoSubmission: noSubmission,
+	}
+	require.NoError(t, f.store.SavePurchaseExecution(ctx, exec))
+	return exec
+}
+
+func (f *rampFixture) committed(ctx context.Context, t *testing.T) map[string]PlanRampQuantity {
+	t.Helper()
+	var got map[string]PlanRampQuantity
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		var err error
+		got, err = f.store.ListPlanCommittedTx(ctx, tx, f.planID)
+		return err
+	}))
+	return got
+}
+
+func (f *rampFixture) baseline(ctx context.Context, t *testing.T) PlanRampBaseline {
+	t.Helper()
+	var got PlanRampBaseline
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		var err error
+		got, err = f.store.GetPlanRampBaselineTx(ctx, tx, f.planID)
+		return err
+	}))
+	return got
+}
+
+// A 4-step weekly plan over two accounts, with every row shape the ledger has
+// to classify, read back from a real Postgres.
+func TestListPlanCommittedTx_RealRowShapes(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 2, 0, 4)
+	a1, a2 := &f.accounts[0], &f.accounts[1]
+	r1 := baselineRec(*a1, "m5.large", 20)
+	r2 := baselineRec(*a2, "m5.large", 20)
+	k1, k2 := baselineKey(t, r1), baselineKey(t, r2)
+
+	// Step 1: a fan-out. The root row carries both accounts' recs, each account
+	// row its own; both accounts bought.
+	f.rowWithRecs(ctx, t, 1, nil, StatusCompleted, false, r1, r2)
+	f.rowWithRecs(ctx, t, 1, a1, StatusCompleted, false, r1)
+	f.rowWithRecs(ctx, t, 1, a2, StatusCompleted, false, r2)
+
+	// Step 2: account 1 failed with an unknown outcome and was retried; the
+	// retry completed. Account 2 completed.
+	failed := f.rowWithRecs(ctx, t, 2, a1, StatusFailed, false, r1)
+	retry := f.retryInOneTx(ctx, t, failed, uuid.New().String(), StatusCompleted)
+	retry.Recommendations = []RecommendationRecord{r1}
+	require.NoError(t, f.store.SavePurchaseExecution(ctx, retry))
+	f.rowWithRecs(ctx, t, 2, a2, StatusCompleted, false, r2)
+
+	// Step 3: canceled, legacy-canceled and a failed row proven unsubmitted hold nothing.
+	f.rowWithRecs(ctx, t, 3, a1, StatusCanceled, false, r1)
+	f.rowWithRecs(ctx, t, 3, a1, LegacyStatusCanceled, false, r1)
+	f.rowWithRecs(ctx, t, 3, a2, StatusFailed, true, r2)
+
+	got := f.committed(ctx, t)
+	assert.Equal(t, PlanRampQuantity{Count: 40}, got[k1], "steps 1 and 2 for account 1, once each")
+	assert.Equal(t, PlanRampQuantity{Count: 40}, got[k2], "steps 1 and 2 for account 2, once each")
+
+	// Step 3 retried by an operator: rows that DO hold capacity join the ledger.
+	f.rowWithRecs(ctx, t, 3, a1, "paused", false, r1)
+	f.rowWithRecs(ctx, t, 3, a2, StatusFailed, false, r2)
+	got = f.committed(ctx, t)
+	assert.Equal(t, 60, got[k1].Count, "a paused row holds its step")
+	assert.Equal(t, 60, got[k2].Count, "a failed row with an unknown outcome is assumed bought")
+
+	partial := baselineRec(*a1, "c5.large", 6)
+	partial.Error = "context deadline exceeded"
+	f.rowWithRecs(ctx, t, 4, a1, StatusPartiallyCompleted, false, partial)
+	got = f.committed(ctx, t)
+	assert.Equal(t, 6, got[baselineKey(t, partial)].Count, "a partially_completed row counts its errored rec")
+}
+
+// The baseline is written once, by its own method, and nothing else touches it.
+func TestPlanRampBaseline_ImmutableAndOutOfFullRowWrites(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 1, 0, 4)
+	r := baselineRec(f.accounts[0], "m5.large", 100)
+	key := baselineKey(t, r)
+
+	// A plan struct read BEFORE the baseline exists, as a handler's GET would hold.
+	stale, err := f.store.GetPurchasePlan(ctx, f.planID)
+	require.NoError(t, err)
+	assert.Nil(t, f.baseline(ctx, t), "no baseline until step 1 writes one")
+
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		wrote, setErr := f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{key: {Count: 100}})
+		require.NoError(t, setErr)
+		assert.True(t, wrote)
+		wrote, setErr = f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{key: {Count: 999}})
+		require.NoError(t, setErr)
+		assert.False(t, wrote, "a second write is refused")
+		return nil
+	}))
+	assert.Equal(t, PlanRampBaseline{key: {Count: 100}}, f.baseline(ctx, t))
+
+	// The ramp advances through CompletePlanStep (a full-row plan UPDATE).
+	f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusCompleted, false, r)
+	require.NoError(t, f.store.CompletePlanStep(ctx, f.planID, 1))
+	assert.Equal(t, PlanRampBaseline{key: {Count: 100}}, f.baseline(ctx, t), "CompletePlanStep must not touch the baseline")
+
+	// A plan edit from a struct read before the baseline existed.
+	fresh, err := f.store.GetPurchasePlan(ctx, f.planID)
+	require.NoError(t, err)
+	fresh.Name = "renamed"
+	require.NoError(t, f.store.UpdatePurchasePlan(ctx, fresh))
+	assert.Equal(t, PlanRampBaseline{key: {Count: 100}}, f.baseline(ctx, t), "a plan edit must not touch the baseline")
+	stale.Name = "stale rename"
+	_ = f.store.UpdatePurchasePlan(ctx, stale) // rejected as stale or applied: either way the baseline survives
+	assert.Equal(t, PlanRampBaseline{key: {Count: 100}}, f.baseline(ctx, t), "a stale plan save must not clear the baseline")
+
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		_, err := f.store.SetPlanRampBaselineTx(ctx, tx, uuid.New().String(), PlanRampBaseline{})
+		require.Error(t, err, "an empty baseline is refused")
+		_, err = f.store.GetPlanRampBaselineTx(ctx, tx, uuid.New().String())
+		require.ErrorIs(t, err, ErrNotFound)
+		return nil
+	}))
+}