feat(config): complete a target-met ramp step as a no-op and clear a stale baseline

internal/analytics/collector_test.go

@@ -905,6 +905,10 @@ func (m *mockConfigStore) ListPlanCommittedTx(_ context.Context, _ pgx.Tx, _ str
 	return nil, nil
 }
 
+func (m *mockConfigStore) CompleteTargetMetStepTx(_ context.Context, _ pgx.Tx, _ string, _ int, _, _ string) error {
+	return nil
+}
+
 func (m *mockConfigStore) SetExecutionRecommendationsIfEmptyTx(_ context.Context, _ pgx.Tx, _ string, _ []config.RecommendationRecord, _, _ float64) (bool, error) {
 	return false, nil
 }

internal/config/interfaces.go

@@ -108,6 +108,11 @@ type StoreInterface interface {
 	// 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)
+	// CompleteTargetMetStepTx records a plan step whose cumulative target is
+	// already held as a completed no-op row (new, or rootID's empty root row
+	// completed in place) and advances the ramp over it in the same
+	// transaction. The caller holds LockPurchasePlanTx.
+	CompleteTargetMetStepTx(ctx context.Context, tx pgx.Tx, planID string, stepNumber int, rootID, note string) 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.go

@@ -798,7 +798,25 @@ func (s *PostgresStore) persistRampPosition(ctx context.Context, tx pgx.Tx, plan
 func (s *PostgresStore) UpdatePurchasePlan(ctx context.Context, plan *PurchasePlan) error {
 	pending := *plan
 	if err := s.WithTx(ctx, func(tx pgx.Tx) error {
-		return s.UpdatePurchasePlanTx(ctx, tx, &pending)
+		old, err := s.LockPurchasePlanTx(ctx, tx, pending.ID)
+		if err != nil {
+			return err
+		}
+		if updErr := s.UpdatePurchasePlanTx(ctx, tx, &pending); updErr != nil {
+			return updErr
+		}
+		if old == nil || !rampBaselineStale(old, &pending) {
+			return nil
+		}
+		// After the update, never before: clearing bumps updated_at, which
+		// would fail the update's own optimistic-lock check.
+		err = tx.QueryRow(ctx,
+			`UPDATE purchase_plans SET ramp_baseline = NULL WHERE id = $1 AND ramp_baseline IS NOT NULL RETURNING updated_at`,
+			pending.ID).Scan(&pending.UpdatedAt)
+		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+			return fmt.Errorf("failed to clear ramp baseline of plan %s: %w", pending.ID, err)
+		}
+		return nil
 	}); err != nil {
 		return err
 	}

internal/config/store_postgres_pgxmock_test.go

@@ -616,6 +616,7 @@ func TestPGXMock_UpdatePurchasePlan_NotFound(t *testing.T) {
 
 	plan := &PurchasePlan{ID: "missing-id", Services: map[string]ServiceConfig{}, RampSchedule: RampSchedule{}}
 	mock.ExpectBegin()
+	mock.ExpectQuery("FOR UPDATE").WithArgs("missing-id").WillReturnError(pgx.ErrNoRows)
 	mock.ExpectQuery("UPDATE").WithArgs(anyArgsCfg(12)...).
 		WillReturnRows(pgxmock.NewRows([]string{"updated_at"}))
 	mock.ExpectQuery("SELECT EXISTS").WithArgs("missing-id").

internal/config/store_postgres_plan_update_test.go

@@ -18,6 +18,7 @@ func TestUpdatePurchasePlanPreservesTokenOnCommitFailure(t *testing.T) {
 	plan := &PurchasePlan{ID: "plan", UpdatedAt: readVersion}
 	commitErr := errors.New("commit failed")
 	mock.ExpectBegin()
+	mock.ExpectQuery("FOR UPDATE").WithArgs("plan").WillReturnError(pgx.ErrNoRows)
 	mock.ExpectQuery(`UPDATE purchase_plans[\s\S]*WHERE id = \$1 AND updated_at = \$8[\s\S]*RETURNING updated_at`).
 		WithArgs(anyArgsCfg(12)...).
 		WillReturnRows(pgxmock.NewRows([]string{"updated_at"}).AddRow(readVersion.Add(time.Second)))
@@ -35,6 +36,7 @@ func TestUpdatePurchasePlanPropagatesConflictProbeFailure(t *testing.T) {
 	readVersion := plan.UpdatedAt
 	probeErr := errors.New("probe failed")
 	mock.ExpectBegin()
+	mock.ExpectQuery("FOR UPDATE").WithArgs(plan.ID).WillReturnError(pgx.ErrNoRows)
 	mock.ExpectQuery("UPDATE purchase_plans").WithArgs(anyArgsCfg(12)...).
 		WillReturnRows(pgxmock.NewRows([]string{"updated_at"}))
 	mock.ExpectQuery("SELECT EXISTS").WithArgs(plan.ID).WillReturnError(probeErr)

internal/config/store_postgres_ramp_baseline.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sort"
+	"strings"
 
 	"github.com/jackc/pgx/v5"
 )
@@ -180,3 +182,26 @@ func recHourlyCommitment(rec *RecommendationRecord) (float64, error) {
 	}
 	return *d.Hourly, nil
 }
+
+// rampBaselineStale reports whether an edit changes what the frozen baseline
+// describes: which services are bought (provider, service, term, payment of
+// the enabled ones) or the ramp's shape (type, percent, steps, interval).
+// Coverage and the current position are deliberately not part of it: the
+// baseline holds demand counts, and the target scales with coverage. A stale
+// baseline is cleared rather than reused, so the next step fails loud (every
+// identity is skipped) instead of sizing against the wrong demand.
+func rampBaselineStale(old, next *PurchasePlan) bool {
+	shape := func(p *PurchasePlan) string {
+		var svcs []string
+		for k := range p.Services {
+			svc := p.Services[k]
+			if svc.Enabled {
+				svcs = append(svcs, fmt.Sprintf("%s|%s|%d|%s", strings.ToLower(svc.Provider), svc.Service, svc.Term, svc.Payment))
+			}
+		}
+		sort.Strings(svcs)
+		r := p.RampSchedule
+		return fmt.Sprintf("%s;%s|%v|%d|%d", strings.Join(svcs, ","), r.Type, r.PercentPerStep, r.TotalSteps, r.StepIntervalDays)
+	}
+	return shape(old) != shape(next)
+}

internal/config/store_postgres_ramp_baseline_integration_test.go

@@ -140,3 +140,47 @@ func TestPlanRampBaseline_ImmutableAndOutOfFullRowWrites(t *testing.T) {
 		return nil
 	}))
 }
+
+// An edit that changes what the baseline describes clears it (the next step
+// then fails loud for want of a baseline); one that does not leaves it.
+func TestPlanRampBaseline_ClearedOnlyByServiceOrRampShapeEdit(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 1, 0, 4)
+	key := baselineKey(t, baselineRec(f.accounts[0], "m5.large", 100))
+	freeze := func() {
+		require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+			_, err := f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{key: {Count: 100}})
+			return err
+		}))
+	}
+	edit := func(mutate func(p *PurchasePlan)) {
+		p, err := f.store.GetPurchasePlan(ctx, f.planID)
+		require.NoError(t, err)
+		mutate(p)
+		require.NoError(t, f.store.UpdatePurchasePlan(ctx, p))
+	}
+	edit(func(p *PurchasePlan) {
+		p.Services = map[string]ServiceConfig{"ec2": {Provider: "aws", Service: "ec2", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 80}}
+	})
+	freeze()
+	edit(func(p *PurchasePlan) { p.Name = "renamed" })
+	assert.NotNil(t, f.baseline(ctx, t), "a rename keeps the baseline")
+	edit(func(p *PurchasePlan) {
+		svc := p.Services["ec2"]
+		svc.Coverage = 50
+		p.Services["ec2"] = svc
+	})
+	assert.NotNil(t, f.baseline(ctx, t), "a coverage change keeps it: the target scales with coverage")
+
+	edit(func(p *PurchasePlan) { p.RampSchedule.TotalSteps = 5 })
+	assert.Nil(t, f.baseline(ctx, t), "a ramp shape change clears it")
+
+	freeze()
+	edit(func(p *PurchasePlan) {
+		for k, svc := range p.Services {
+			svc.Term = 3
+			p.Services[k] = svc
+		}
+	})
+	assert.Nil(t, f.baseline(ctx, t), "a service identity change clears it")
+}

internal/config/store_postgres_ramp_target_met.go

@@ -0,0 +1,84 @@
+package config
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+)
+
+// PreApprovalSkipRampTargetMet marks the completed no-op row a ramp step gets
+// when its cumulative target is already held (platform#631). The advance gate
+// (requireRampStepBought) is unchanged and still counts only completed or
+// partially_completed rows as bought, so a failed zero-buy fan-out can never
+// advance the ramp; only CompleteTargetMetStepTx writes this row, in the claim
+// transaction, and the marker tells it apart from a purchase.
+const PreApprovalSkipRampTargetMet = "ramp-target-met"
+
+// CompleteTargetMetStepTx records a plan step whose cumulative target is
+// already held as a completed no-op row and advances the ramp over it, all in
+// the caller's transaction. Without a shared transaction a crash between the
+// row and the advance would leave a settled step the ramp never moves past.
+//
+// rootID names the pre-created empty root row of the step to complete in place
+// (compare-and-set: it must still be pending/notified and empty; otherwise
+// ErrExecutionNotInExpectedStatus), or is "" to insert a new row. The note is
+// stored as the row's error text; the number of failed rows of unknown outcome
+// that were counted as committed is appended to it so an advance taken on such
+// capacity is visible.
+//
+// The plan is re-read under the lock AFTER any baseline write in this
+// transaction: that write bumps updated_at, and advancing from a plan read
+// before it would lose the optimistic-lock check in UpdatePurchasePlanTx.
+func (s *PostgresStore) CompleteTargetMetStepTx(ctx context.Context, tx pgx.Tx, planID string, stepNumber int, rootID, note string) error {
+	if stepNumber <= 0 {
+		return fmt.Errorf("cannot complete ramp step %d of plan %s: step numbers are 1-based", stepNumber, planID)
+	}
+	var unknown int
+	if err := tx.QueryRow(ctx, `
+		SELECT count(*) FROM purchase_executions
+		 WHERE plan_id = $1 AND status = $2 AND NOT no_submission
+		   AND CASE WHEN jsonb_typeof(recommendations) = 'array' THEN jsonb_array_length(recommendations) ELSE 0 END > 0`,
+		planID, StatusFailed).Scan(&unknown); err != nil {
+		return fmt.Errorf("failed to count unknown-outcome executions of plan %s: %w", planID, err)
+	}
+	note = fmt.Sprintf("%s; %d failed row(s) of unknown outcome counted as committed", note, unknown)
+	now := time.Now()
+	if rootID == "" {
+		marker := PreApprovalSkipRampTargetMet
+		exec := &PurchaseExecution{
+			PlanID: planID, ExecutionID: uuid.New().String(), Status: StatusCompleted, StepNumber: stepNumber,
+			ScheduledDate: now, CompletedAt: &now, Error: note,
+			PreApprovalSkipReason: &marker,
+		}
+		if err := s.SavePurchaseExecutionTx(ctx, tx, exec); err != nil {
+			return fmt.Errorf("failed to record target-met plan step %d of plan %s: %w", stepNumber, planID, err)
+		}
+	} else {
+		tag, err := tx.Exec(ctx, `
+			UPDATE purchase_executions
+			   SET status = $2, completed_at = $3, error = $4, pre_approval_skip_reason = $5
+			 WHERE execution_id = $1 AND status IN ('pending', 'notified')
+			   AND CASE WHEN jsonb_typeof(recommendations) = 'array' THEN jsonb_array_length(recommendations) ELSE 0 END = 0`,
+			rootID, StatusCompleted, now, note, PreApprovalSkipRampTargetMet)
+		if err != nil {
+			return fmt.Errorf("failed to complete target-met root row %s: %w", rootID, err)
+		}
+		if tag.RowsAffected() != 1 {
+			return fmt.Errorf("%w: root row %s", ErrExecutionNotInExpectedStatus, rootID)
+		}
+	}
+	plan, err := s.LockPurchasePlanTx(ctx, tx, planID)
+	if err != nil {
+		return err
+	}
+	if plan == nil {
+		return fmt.Errorf("%w: purchase plan %s", ErrNotFound, planID)
+	}
+	if err := advanceRampStep(ctx, tx, plan, stepNumber); err != nil {
+		return err
+	}
+	return s.persistRampPosition(ctx, tx, plan)
+}

internal/config/store_postgres_ramp_target_met_integration_test.go

@@ -0,0 +1,130 @@
+//go:build integration
+// +build integration
+
+package config
+
+import (
+	"context"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The no-op row and the ramp advance commit together, after a baseline write
+// that bumped updated_at in the same transaction; a failed zero-buy fan-out
+// still cannot advance through the unchanged gate.
+func TestCompleteTargetMetStepTx_RealPlan(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 1, 1, 4)
+	r := baselineRec(f.accounts[0], "m5.large", 100)
+	f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusCompleted, false, r)
+	failed := f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusFailed, false, r)
+	failed.Recommendations = nil
+	key := baselineKey(t, r)
+
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		_, err := f.store.LockPurchasePlanTx(ctx, tx, f.planID)
+		require.NoError(t, err)
+		_, err = f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{key: {Count: 100}})
+		require.NoError(t, err)
+		return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 2, "", "target already met: target 40, committed 40")
+	}))
+	assert.Equal(t, 2, f.currentStep(ctx, t))
+	assert.NotNil(t, f.baseline(ctx, t))
+	var reason, note, status string
+	require.NoError(t, f.store.db.QueryRow(ctx,
+		`SELECT pre_approval_skip_reason, error, status FROM purchase_executions WHERE plan_id = $1 AND step_number = 2`, f.planID).Scan(&reason, &note, &status))
+	assert.Equal(t, PreApprovalSkipRampTargetMet, reason)
+	assert.Equal(t, StatusCompleted, status)
+	assert.Contains(t, note, "1 failed row(s) of unknown outcome counted as committed")
+
+	// A step whose only row failed without buying is still refused.
+	f.rowWithRecs(ctx, t, 3, nil, StatusFailed, true)
+	require.ErrorIs(t, f.store.CompletePlanStep(ctx, f.planID, 3), ErrRampStepIncomplete)
+	assert.Equal(t, 2, f.currentStep(ctx, t))
+
+	// A root row that is no longer pending/notified is not completed in place.
+	done := f.rowWithRecs(ctx, t, 3, nil, StatusCompleted, false)
+	err := f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 3, done.ExecutionID, "n")
+	})
+	require.ErrorIs(t, err, ErrExecutionNotInExpectedStatus)
+	assert.Equal(t, 2, f.currentStep(ctx, t))
+}
+
+// G3: a zero-buy failed fan-out on the step refuses the target-met advance too.
+func TestTargetMet_TargetMetRefusedOverFailedFanOut(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 1, 1, 4)
+	f.rowWithRecs(ctx, t, 2, &f.accounts[0], StatusFailed, true)
+	err := f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		if _, err := f.store.LockPurchasePlanTx(ctx, tx, f.planID); err != nil {
+			return err
+		}
+		return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 2, "", "target already met")
+	})
+	require.ErrorIs(t, err, ErrRampStepIncomplete)
+	assert.Equal(t, 1, f.currentStep(ctx, t))
+	var n int
+	require.NoError(t, f.store.db.QueryRow(ctx, `SELECT count(*) FROM purchase_executions WHERE plan_id=$1 AND pre_approval_skip_reason=$2`, f.planID, PreApprovalSkipRampTargetMet).Scan(&n))
+	assert.Equal(t, 0, n)
+}
+
+// A pending root that already carries attached recommendations is never completed as a no-op.
+func TestTargetMet_RootWithRecsNotCompleted(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 1, 1, 4)
+	root := f.rowWithRecs(ctx, t, 2, nil, "pending", false, baselineRec(f.accounts[0], "m5.large", 5))
+	err := f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 2, root.ExecutionID, "n")
+	})
+	require.ErrorIs(t, err, ErrExecutionNotInExpectedStatus)
+	assert.Equal(t, 1, f.currentStep(ctx, t))
+	var st string
+	require.NoError(t, f.store.db.QueryRow(ctx, `SELECT status FROM purchase_executions WHERE execution_id=$1`, root.ExecutionID).Scan(&st))
+	assert.Equal(t, "pending", st)
+}
+
+// The unknown-outcome count excludes failed rows that never submitted.
+func TestTargetMet_UnknownCountExcludesNoSubmission(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 1, 1, 4)
+	r := baselineRec(f.accounts[0], "m5.large", 3)
+	f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusCompleted, false, r)
+	f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusFailed, true, r)
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 2, "", "met")
+	}))
+	var note string
+	require.NoError(t, f.store.db.QueryRow(ctx, `SELECT error FROM purchase_executions WHERE plan_id=$1 AND pre_approval_skip_reason=$2`, f.planID, PreApprovalSkipRampTargetMet).Scan(&note))
+	assert.Contains(t, note, "; 0 failed row(s) of unknown outcome")
+}
+
+// A failed, submitted row whose recommendations are JSON null (scalar) must not break the count.
+func TestTargetMet_UnknownCountScalarRecs(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 1, 1, 4)
+	r := baselineRec(f.accounts[0], "m5.large", 3)
+	f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusCompleted, false, r)
+	f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusFailed, false)
+	var typ string
+	require.NoError(t, f.store.db.QueryRow(ctx, `SELECT coalesce(jsonb_typeof(recommendations),'sqlnull') FROM purchase_executions WHERE plan_id=$1 AND status=$2`, f.planID, StatusFailed).Scan(&typ))
+	t.Logf("recommendations jsonb type of the failed row: %s", typ)
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 2, "", "met")
+	}))
+}
+
+// A pre-created empty root (recommendations JSON null) completes in place.
+func TestTargetMet_NullRootCompletes(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 1, 1, 4)
+	f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusCompleted, false, baselineRec(f.accounts[0], "m5.large", 3))
+	root := f.rowWithRecs(ctx, t, 2, nil, "pending", false)
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 2, root.ExecutionID, "met")
+	}))
+	assert.Equal(t, 2, f.currentStep(ctx, t))
+}

internal/mocks/stores.go

@@ -321,6 +321,11 @@ func (m *MockConfigStore) ListPlanCommittedTx(ctx context.Context, tx pgx.Tx, pl
 	return v, args.Error(1)
 }
 
+func (m *MockConfigStore) CompleteTargetMetStepTx(ctx context.Context, tx pgx.Tx, planID string, stepNumber int, rootID, note string) error {
+	m.record("CompleteTargetMetStepTx", ctx, tx, planID, stepNumber, rootID, note)
+	return m.Called(ctx, tx, planID, stepNumber, rootID, note).Error(0)
+}
+
 func (m *MockConfigStore) SetExecutionRecommendationsIfEmptyTx(ctx context.Context, tx pgx.Tx, executionID string, recs []config.RecommendationRecord, upfront, savings float64) (bool, error) {
 	m.record("SetExecutionRecommendationsIfEmptyTx", ctx, tx, executionID, recs, upfront, savings)
 	args := m.Called(ctx, tx, executionID, recs, upfront, savings)