From 501dc4ebf5630fbc15be6e3a80f821de1dd0e1d6 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 15 Oct 2026 10:38:35 +0200 Subject: [PATCH 1/4] fix(purchase): refuse unverified ramp sizing and adoption Mint fresh ramp lineage at creation and freeze it atomically. Preserve authorized quantities while refusing new ramp sizing or notification adoption from unverifiable histories. --- internal/config/store_postgres_plan_writes.go | 4 +- .../config/store_postgres_ramp_baseline.go | 40 +++- ...postgres_ramp_baseline_integration_test.go | 17 +- ...ostgres_ramp_retention_integration_test.go | 95 ++++++++ ...stgres_ramp_target_met_integration_test.go | 7 +- ...ore_postgres_retention_integration_test.go | 2 +- internal/purchase/plan_step_attach.go | 21 +- .../plan_step_cumulative_integration_test.go | 7 +- .../plan_step_retention_integration_test.go | 204 ++++++++++++++++++ .../ramp_step_progress_integration_test.go | 2 + 10 files changed, 372 insertions(+), 27 deletions(-) create mode 100644 internal/config/store_postgres_ramp_retention_integration_test.go create mode 100644 internal/purchase/plan_step_retention_integration_test.go diff --git a/internal/config/store_postgres_plan_writes.go b/internal/config/store_postgres_plan_writes.go index 6fb99f1cc..e4d1a96f7 100644 --- a/internal/config/store_postgres_plan_writes.go +++ b/internal/config/store_postgres_plan_writes.go @@ -34,8 +34,8 @@ func (s *PostgresStore) createPurchasePlan(ctx context.Context, executor globalC id, name, enabled, auto_purchase, notification_days_before, services, ramp_schedule, created_at, updated_at, next_execution_date, last_execution_date, last_notification_sent, - max_step_commitment - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) + max_step_commitment, ramp_ledger_state + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, 'new') RETURNING updated_at ` diff --git a/internal/config/store_postgres_ramp_baseline.go b/internal/config/store_postgres_ramp_baseline.go index 6611773b6..81cf29d4a 100644 --- a/internal/config/store_postgres_ramp_baseline.go +++ b/internal/config/store_postgres_ramp_baseline.go @@ -21,11 +21,21 @@ type PlanRampQuantity struct { // 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. +type rampLedgerState string + +const ( + rampLedgerNew rampLedgerState = "new" + rampLedgerTrusted rampLedgerState = "trusted" +) + +func (s rampLedgerState) allowsSizing(raw []byte, current int) bool { + return s == rampLedgerTrusted && raw != nil || s == rampLedgerNew && raw == nil && current == 0 +} + +var ErrRampLedgerUntrusted = errors.New("ramp purchase history cannot be verified; reconcile prior commitments before attaching another step") + +// SetPlanRampBaselineTx freezes a baseline only while retention can prove its lineage. +// The caller holds the plan lock; full-row plan writes cannot promote trust. 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) @@ -35,24 +45,36 @@ func (s *PostgresStore) SetPlanRampBaselineTx(ctx context.Context, tx pgx.Tx, pl 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) + `UPDATE purchase_plans SET ramp_baseline = $2, + ramp_ledger_state = CASE WHEN ramp_ledger_protected(ramp_schedule) THEN 'trusted' ELSE ramp_ledger_state END + WHERE id = $1 AND ramp_baseline IS NULL + AND (COALESCE((ramp_schedule ->> 'total_steps')::int, 1) <= 1 + OR (ramp_ledger_state = 'new' AND COALESCE((ramp_schedule ->> 'current_step')::int, 0) = 0))`, 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. +// GetPlanRampBaselineTx refuses cumulative sizing when retained history is untrusted. +// Fresh protected plans return nil until their baseline freezes under the plan lock. 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) + var state rampLedgerState + var total, current int + err := tx.QueryRow(ctx, `SELECT ramp_baseline, ramp_ledger_state, + COALESCE((ramp_schedule ->> 'total_steps')::int, 1), + COALESCE((ramp_schedule ->> 'current_step')::int, 0) + FROM purchase_plans WHERE id = $1`, planID).Scan(&raw, &state, &total, ¤t) 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 total > 1 && !state.allowsSizing(raw, current) { + return nil, ErrRampLedgerUntrusted + } if raw == nil { return nil, nil } diff --git a/internal/config/store_postgres_ramp_baseline_integration_test.go b/internal/config/store_postgres_ramp_baseline_integration_test.go index 9421c69c6..f37fcd1f9 100644 --- a/internal/config/store_postgres_ramp_baseline_integration_test.go +++ b/internal/config/store_postgres_ramp_baseline_integration_test.go @@ -159,8 +159,16 @@ func TestPlanRampBaseline_ClearedOnlyByServiceOrRampShapeEdit(t *testing.T) { mutate(p) require.NoError(t, f.store.UpdatePurchasePlan(ctx, p)) } + assertStale := func() { + var raw []byte + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT ramp_baseline FROM purchase_plans WHERE id=$1`, f.planID).Scan(&raw)) + assert.Nil(t, raw) + readErr := f.store.WithTx(ctx, func(tx pgx.Tx) error { _, err := f.store.GetPlanRampBaselineTx(ctx, tx, f.planID); return err }) + assert.ErrorIs(t, readErr, ErrRampLedgerUntrusted) + } + services := map[string]ServiceConfig{"ec2": {Provider: "aws", Service: "ec2", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 80}} edit(func(p *PurchasePlan) { - p.Services = map[string]ServiceConfig{"ec2": {Provider: "aws", Service: "ec2", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 80}} + p.Services = services }) freeze() edit(func(p *PurchasePlan) { p.Name = "renamed" }) @@ -173,8 +181,11 @@ func TestPlanRampBaseline_ClearedOnlyByServiceOrRampShapeEdit(t *testing.T) { 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") + assertStale() + f = newRampFixture(ctx, t, 1, 0, 4) + key = baselineKey(t, baselineRec(f.accounts[0], "m5.large", 100)) + edit(func(p *PurchasePlan) { p.Services = services }) freeze() edit(func(p *PurchasePlan) { for k, svc := range p.Services { @@ -182,5 +193,5 @@ func TestPlanRampBaseline_ClearedOnlyByServiceOrRampShapeEdit(t *testing.T) { p.Services[k] = svc } }) - assert.Nil(t, f.baseline(ctx, t), "a service identity change clears it") + assertStale() } diff --git a/internal/config/store_postgres_ramp_retention_integration_test.go b/internal/config/store_postgres_ramp_retention_integration_test.go new file mode 100644 index 000000000..3c1f2cda5 --- /dev/null +++ b/internal/config/store_postgres_ramp_retention_integration_test.go @@ -0,0 +1,95 @@ +//go:build integration + +package config + +import ( + "context" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +func TestRampRetention_NullServicesRemainEditable(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + p := &PurchasePlan{Name: "disabled fresh", RampSchedule: RampSchedule{Type: "weekly", PercentPerStep: 25, TotalSteps: 4, StepIntervalDays: 7}} + require.NoError(t, store.CreatePurchasePlan(ctx, p)) + p.Name = "renamed" + require.NoError(t, store.UpdatePurchasePlan(ctx, p)) + p.RampSchedule.StepIntervalDays = 30 + require.NoError(t, store.UpdatePurchasePlan(ctx, p)) + var state string + require.NoError(t, store.db.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, p.ID).Scan(&state)) + require.Equal(t, "new", state) +} + +func TestRampRetention_FreshEditsRequireContinuousUnheldLineage(t *testing.T) { + cases := []struct { + name, status, lineage string + noSubmission bool + count int + want string + }{ + {name: "fresh", want: "new"}, + {name: "empty pending", status: "pending", want: "legacy"}, + {name: "zero completed", status: StatusCompleted, want: "legacy"}, + {name: "selected completed", status: StatusCompleted, count: 20, want: "legacy"}, + {name: "unknown failed", status: StatusFailed, want: "legacy"}, + {name: "unsubmitted", status: StatusFailed, noSubmission: true, want: "new"}, + {name: "canceled", status: StatusCanceled, want: "new"}, + {name: "expired", status: StatusExpired, want: "new"}, + {name: "trusted", lineage: "trusted", want: "legacy"}, + {name: "legacy", lineage: "legacy", want: "legacy"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + f := newRampFixture(ctx, t, 0, 0, 4) + if tc.status != "" { + var recs []RecommendationRecord + if tc.count > 0 { + recs = []RecommendationRecord{{Selected: true, Count: tc.count}} + } + f.rowWithRecs(ctx, t, 1, nil, tc.status, tc.noSubmission, recs...) + } + if tc.lineage != "" { + _, err := f.store.db.Exec(ctx, `UPDATE purchase_plans SET ramp_baseline='{"fixture":{"count":100}}', ramp_ledger_state=$2 WHERE id=$1`, f.planID, tc.lineage) + require.NoError(t, err) + } + p, err := f.store.GetPurchasePlan(ctx, f.planID) + require.NoError(t, err) + p.RampSchedule.Type, p.RampSchedule.StepIntervalDays = "custom", 30 + require.NoError(t, f.store.UpdatePurchasePlan(ctx, p)) + var state string + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, f.planID).Scan(&state)) + require.Equal(t, tc.want, state) + }) + } +} + +func TestRampRetention_LegacyNeverRearmsAndFreezeIsAtomic(t *testing.T) { + ctx := context.Background() + f := newRampFixture(ctx, t, 0, 0, 4) + require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error { + _, err := f.store.LockPurchasePlanTx(ctx, tx, f.planID) + if err != nil { + return err + } + wrote, err := f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{"fixture": {Count: 100}}) + require.True(t, wrote) + return err + })) + _, err := f.store.db.Exec(ctx, `UPDATE purchase_plans SET ramp_baseline=NULL WHERE id=$1`, f.planID) + require.NoError(t, err) + _, err = f.store.db.Exec(ctx, `UPDATE purchase_plans SET ramp_ledger_state='new', ramp_schedule=jsonb_set(ramp_schedule,'{current_step}','0') WHERE id=$1`, f.planID) + require.NoError(t, err) + var state string + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, f.planID).Scan(&state)) + require.Equal(t, "legacy", state) + err = f.store.WithTx(ctx, func(tx pgx.Tx) error { + _, readErr := f.store.GetPlanRampBaselineTx(ctx, tx, f.planID) + return readErr + }) + require.ErrorIs(t, err, ErrRampLedgerUntrusted) +} diff --git a/internal/config/store_postgres_ramp_target_met_integration_test.go b/internal/config/store_postgres_ramp_target_met_integration_test.go index d062ec896..fe6c8ceda 100644 --- a/internal/config/store_postgres_ramp_target_met_integration_test.go +++ b/internal/config/store_postgres_ramp_target_met_integration_test.go @@ -17,7 +17,7 @@ import ( // still cannot advance through the unchanged gate. func TestCompleteTargetMetStepTx_RealPlan(t *testing.T) { ctx := context.Background() - f := newRampFixture(ctx, t, 1, 1, 4) + f := newRampFixture(ctx, t, 1, 0, 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) @@ -27,7 +27,10 @@ func TestCompleteTargetMetStepTx_RealPlan(t *testing.T) { 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}}) + wrote, err := f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{key: {Count: 100}}) + require.NoError(t, err) + require.True(t, wrote) + _, err = tx.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule=jsonb_set(ramp_schedule,'{current_step}','1') WHERE id=$1`, f.planID) require.NoError(t, err) return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 2, "", "target already met: target 40, committed 40") })) diff --git a/internal/config/store_postgres_retention_integration_test.go b/internal/config/store_postgres_retention_integration_test.go index ad964db48..49b698a5a 100644 --- a/internal/config/store_postgres_retention_integration_test.go +++ b/internal/config/store_postgres_retention_integration_test.go @@ -190,7 +190,7 @@ func TestRampRetention_CreationLineageAndLegacyWriters(t *testing.T) { f := newRampFixture(ctx, t, 0, 0, 4) var state string require.NoError(t, f.store.db.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, f.planID).Scan(&state)) - require.Equal(t, "legacy", state) + require.Equal(t, "new", state) legacyID := uuid.NewString() _, err := f.store.db.Exec(ctx, `INSERT INTO purchase_plans(id,name,ramp_schedule) SELECT $1,'old writer',ramp_schedule FROM purchase_plans WHERE id=$2`, legacyID, f.planID) require.NoError(t, err) diff --git a/internal/purchase/plan_step_attach.go b/internal/purchase/plan_step_attach.go index 1008b9e80..eb1e4910a 100644 --- a/internal/purchase/plan_step_attach.go +++ b/internal/purchase/plan_step_attach.go @@ -96,6 +96,9 @@ func (m *Manager) claimFromRows(ctx context.Context, tx pgx.Tx, locked *config.P return claim, nil } if len(root.Recommendations) > 0 { + if _, err := m.config.GetPlanRampBaselineTx(ctx, tx, locked.ID); err != nil { + return stepClaim{}, err + } if locked.NextExecutionDate.Before(time.Now()) { return stepClaim{}, fmt.Errorf("%w: step %d of plan %s already has a row and the plan date has passed", errExecutionNotNotifiable, claim.number, locked.ID) } @@ -159,16 +162,14 @@ func (m *Manager) attachStep(ctx context.Context, tx pgx.Tx, locked *config.Purc ids = append(ids, accounts[i].ID) } res, err := m.resolveCumulative(ctx, tx, locked, claim.number, ids) - if errors.Is(err, planstep.ErrNoMatch) { + switch { + case errors.Is(err, planstep.ErrNoMatch): return m.noMatch(ctx, tx, locked, claim, root, res.Skipped) - } - if errors.Is(err, planstep.ErrTargetMet) { + case errors.Is(err, planstep.ErrTargetMet): return m.targetMet(ctx, tx, locked, claim, root, res) - } - if errors.Is(err, planstep.ErrUnsupportedPlan) { + case errors.Is(err, planstep.ErrUnsupportedPlan), errors.Is(err, config.ErrRampLedgerUntrusted): return m.failStep(ctx, tx, locked, claim, root, err.Error()) - } - if err != nil { + case err != nil: return stepClaim{}, fmt.Errorf("plan %s step %d: %w", locked.ID, claim.number, err) } if len(res.Skipped) > 0 { @@ -209,9 +210,13 @@ func (m *Manager) resolveCumulative(ctx context.Context, tx pgx.Tx, locked *conf if step != 1 || baseline != nil || len(res.Baseline) == 0 { return res, err } - if _, setErr := m.config.SetPlanRampBaselineTx(ctx, tx, locked.ID, res.Baseline); setErr != nil { + wrote, setErr := m.config.SetPlanRampBaselineTx(ctx, tx, locked.ID, res.Baseline) + if setErr != nil { return planstep.Result{}, setErr } + if !wrote { + return planstep.Result{}, config.ErrRampLedgerUntrusted + } if errors.Is(err, planstep.ErrNoMatch) { return m.resolver.Resolve(ctx, locked, ids, step, res.Baseline, committed) } diff --git a/internal/purchase/plan_step_cumulative_integration_test.go b/internal/purchase/plan_step_cumulative_integration_test.go index 7faa80a16..c80c9cbae 100644 --- a/internal/purchase/plan_step_cumulative_integration_test.go +++ b/internal/purchase/plan_step_cumulative_integration_test.go @@ -4,6 +4,7 @@ package purchase import ( "context" + "encoding/json" "fmt" "sync" "testing" @@ -170,8 +171,10 @@ func TestCumulativeAttach_WeeklyRampBuysTwentyEachStep(t *testing.T) { f.buy(step) } assert.Equal(t, 4, f.plan().RampSchedule.CurrentStep) - base, err := f.baseline() - require.NoError(t, err) + var raw []byte + require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT ramp_baseline FROM purchase_plans WHERE id=$1`, f.planID).Scan(&raw)) + var base config.PlanRampBaseline + require.NoError(t, json.Unmarshal(raw, &base)) require.Len(t, base, 1) for _, q := range base { assert.Equal(t, 100, q.Count, "frozen at step 1 from shown") diff --git a/internal/purchase/plan_step_retention_integration_test.go b/internal/purchase/plan_step_retention_integration_test.go new file mode 100644 index 000000000..3179dc556 --- /dev/null +++ b/internal/purchase/plan_step_retention_integration_test.go @@ -0,0 +1,204 @@ +//go:build integration + +package purchase + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +func TestCumulativeAttach_RetentionKeepsCommittedCapacity(t *testing.T) { + f := newCumFixture(t, 100) + p := f.plan() + p.RampSchedule.Type, p.RampSchedule.StepIntervalDays = "custom", 30 + require.NoError(t, f.store.UpdatePurchasePlan(f.ctx, p)) + f.src.shown = func() []config.RecommendationRecord { + return []config.RecommendationRecord{cumRec(f.account, "m5.t0", 100)} + } + first, err := f.tick() + require.NoError(t, err) + require.Equal(t, 20, totalCount(first)) + f.buy(1) + f.src.shown = func() []config.RecommendationRecord { + return []config.RecommendationRecord{cumRec(f.account, "m5.t0", 80)} + } + before, err := f.baseline() + require.NoError(t, err) + require.Len(t, before, 1) + baselineCount := 0 + for _, q := range before { + baselineCount = q.Count + require.Equal(t, 100, q.Count) + } + require.NoError(t, f.store.WithTx(f.ctx, func(tx pgx.Tx) error { + _, updateErr := tx.Exec(f.ctx, `UPDATE purchase_executions SET scheduled_date=NOW()-INTERVAL '31 days' WHERE plan_id=$1 AND step_number=1`, f.planID) + return updateErr + })) + deleted, err := f.store.CleanupOldExecutions(f.ctx, 30) + require.NoError(t, err) + require.Zero(t, deleted) + after, err := f.baseline() + require.NoError(t, err) + require.Equal(t, before, after) + var committed map[string]config.PlanRampQuantity + require.NoError(t, f.store.WithTx(f.ctx, func(tx pgx.Tx) error { + _, lockErr := f.store.LockPurchasePlanTx(f.ctx, tx, f.planID) + if lockErr != nil { + return lockErr + } + committed, err = f.store.ListPlanCommittedTx(f.ctx, tx, f.planID) + return err + })) + heldCount := 0 + for _, q := range committed { + heldCount = q.Count + require.Equal(t, 20, q.Count) + } + require.Len(t, committed, 1) + next, err := f.tick() + require.NoError(t, err) + require.Equal(t, 20, totalCount(next), "target 40 less the retained first 20") + t.Logf("synthetic quantities: baseline=%d first=%d retained=%d next_target=40 next_attached=%d", baselineCount, totalCount(first), heldCount, totalCount(next)) +} + +func TestCumulativeAttach_FreshEditOverlapsInvisibleHeldInsertion(t *testing.T) { + f := newCumFixture(t, 100) + tx, err := f.pool.Begin(f.ctx) + require.NoError(t, err) + defer tx.Rollback(f.ctx) + held := cumRec(f.account, "m5.t0", 20) + held.Selected = true + require.NoError(t, f.store.SavePurchaseExecutionTx(f.ctx, tx, &config.PurchaseExecution{ + PlanID: f.planID, ExecutionID: uuid.NewString(), Status: "pending", StepNumber: 2, + ScheduledDate: time.Now().Add(-31 * 24 * time.Hour), TTL: time.Now().Add(-31 * 24 * time.Hour).Unix(), + Recommendations: []config.RecommendationRecord{held}, + })) + bounded, cancel := context.WithTimeout(f.ctx, 2*time.Second) + defer cancel() + _, err = f.pool.Exec(bounded, `UPDATE purchase_plans SET ramp_schedule=jsonb_set(jsonb_set(ramp_schedule,'{type}','"custom"'),'{step_interval_days}','30') WHERE id=$1`, f.planID) + require.NoError(t, err, "old writer edits while the new held row is still invisible") + require.NoError(t, tx.Commit(f.ctx)) + deleted, err := f.store.CleanupOldExecutions(f.ctx, 30) + require.NoError(t, err) + require.Zero(t, deleted) + f.src.shown = func() []config.RecommendationRecord { + return []config.RecommendationRecord{cumRec(f.account, "m5.t0", 100)} + } + _, err = f.tick() + require.ErrorIs(t, err, errExecutionNotNotifiable) + rows := f.stepRows(1) + require.Len(t, rows, 1) + require.Empty(t, rows[0].Recommendations, "first target20 is already held, so the freeze must not attach another20") + require.NotNil(t, rows[0].PreApprovalSkipReason) + require.Equal(t, config.PreApprovalSkipRampTargetMet, *rows[0].PreApprovalSkipReason) + base, err := f.baseline() + require.NoError(t, err) + require.Len(t, base, 1) +} + +func TestCumulativeAttach_UntrustedHistoryCannotSizeOrNotify(t *testing.T) { + cases := []struct { + name, rootStatus string + progress, held int + clear bool + }{ + {name: "all prior history missing", progress: 1}, + {name: "partial prior history missing", progress: 1, held: 10}, + {name: "progressed baseline missing", progress: 1, clear: true}, + {name: "legacy step zero baseline missing", clear: true}, + {name: "nonempty pending root", progress: 1, rootStatus: "pending"}, + {name: "nonempty notified root", progress: 1, rootStatus: "notified"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newCumFixture(t, 100) + rec := cumRec(f.account, "m5.t0", 100) + key, err := config.RecIdentityKey(&rec) + require.NoError(t, err) + raw, err := json.Marshal(config.PlanRampBaseline{key: {Count: 100}}) + require.NoError(t, err) + _, err = f.pool.Exec(f.ctx, `UPDATE purchase_plans SET ramp_baseline=$2, ramp_schedule=jsonb_set(ramp_schedule,'{current_step}',to_jsonb($3::int)) WHERE id=$1`, f.planID, raw, tc.progress) + require.NoError(t, err, "old writer freezes without a trust transition") + if tc.clear { + _, err = f.pool.Exec(f.ctx, `UPDATE purchase_plans SET ramp_baseline=NULL WHERE id=$1`, f.planID) + require.NoError(t, err) + } + if tc.held > 0 { + rec.Count, rec.Selected = tc.held, true + require.NoError(t, f.store.SavePurchaseExecution(f.ctx, &config.PurchaseExecution{PlanID: f.planID, ExecutionID: uuid.NewString(), Status: config.StatusCompleted, StepNumber: 1, ScheduledDate: time.Now(), Recommendations: []config.RecommendationRecord{rec}})) + } + if tc.rootStatus != "" { + rec.Selected = true + require.NoError(t, f.store.SavePurchaseExecution(f.ctx, &config.PurchaseExecution{PlanID: f.planID, ExecutionID: uuid.NewString(), Status: tc.rootStatus, StepNumber: tc.progress + 1, ScheduledDate: time.Now().Add(40 * 24 * time.Hour), Recommendations: []config.RecommendationRecord{rec}})) + } + calls := 0 + f.src.shown = func() []config.RecommendationRecord { calls++; return []config.RecommendationRecord{rec} } + _, err = f.tick() + require.Error(t, err) + rows := f.stepRows(tc.progress + 1) + require.Len(t, rows, 1) + if tc.rootStatus == "" { + require.Equal(t, config.StatusFailed, rows[0].Status) + require.Contains(t, rows[0].Error, "reconcile prior commitments") + require.Empty(t, rows[0].Recommendations) + } else { + require.ErrorIs(t, err, config.ErrRampLedgerUntrusted) + require.Equal(t, tc.rootStatus, rows[0].Status) + } + require.False(t, f.m.sendPlanNotification(f.ctx, f.plan())) + require.Zero(t, calls, "no new sizing or resizing") + require.Zero(t, f.m.email.(*countingSender).sends) + require.Zero(t, suppressedTotal(f.ctx, t, f.pool)) + require.Empty(t, rows[0].ApprovalToken) + }) + } +} + +func TestCumulativeAttach_AuthorizedLegacyQuantityRemainsImmutable(t *testing.T) { + for _, status := range []string{"scheduled", "approved"} { + t.Run(status, func(t *testing.T) { + ctx := context.Background() + f := newRampStepFixture(ctx, t) + require.NoError(t, f.store.SetPlanAccounts(ctx, f.plan.ID, []string{f.accounts[0].ID})) + root := f.saveRootExecution(ctx, t, 3) + root.Status = status + root.CloudAccountID = &f.accounts[0].ID + root.Recommendations = root.Recommendations[:1] + root.Recommendations[0].Count = 7 + due := time.Now().Add(-time.Minute) + root.ScheduledExecutionAt = &due + require.NoError(t, f.store.SavePurchaseExecution(ctx, root)) + if status == "scheduled" { + result, fireErr := f.manager.FireScheduledDelayedPurchases(ctx) + require.NoError(t, fireErr) + require.Equal(t, 1, result.Fired) + require.Zero(t, result.Errored) + } else { + require.NoError(t, f.execute(ctx, root)) + } + f.mu.Lock() + counts := append([]int(nil), f.counts...) + f.mu.Unlock() + require.Equal(t, []int{7}, counts, "the recording provider receives the authorized frozen quantity") + resolver := &fakeResolver{recs: root.Recommendations} + f.manager.resolver = resolver + err := f.store.WithTx(ctx, func(tx pgx.Tx) error { + p, lockErr := f.store.LockPurchasePlanTx(ctx, tx, f.plan.ID) + if lockErr != nil { + return lockErr + } + _, resolveErr := f.manager.resolveCumulative(ctx, tx, p, p.RampSchedule.CurrentStep+1, []string{f.accounts[0].ID}) + return resolveErr + }) + require.ErrorIs(t, err, config.ErrRampLedgerUntrusted, "the next step may not size from unverifiable history") + require.Zero(t, resolver.calls.Load()) + }) + } +} diff --git a/internal/purchase/ramp_step_progress_integration_test.go b/internal/purchase/ramp_step_progress_integration_test.go index c8889e9c6..8553c6420 100644 --- a/internal/purchase/ramp_step_progress_integration_test.go +++ b/internal/purchase/ramp_step_progress_integration_test.go @@ -68,6 +68,7 @@ type rampStepFixture struct { mu sync.Mutex purchased []string // one entry per commitment that reached the fake cloud + counts []int // onPurchase, when set, runs inside each fake provider call (issue #829 tests). onPurchase func() @@ -156,6 +157,7 @@ func newRampStepFixture(ctx context.Context, t *testing.T) *rampStepFixture { opts := args.Get(2).(common.PurchaseOptions) f.mu.Lock() f.purchased = append(f.purchased, opts.IdempotencyToken) + f.counts = append(f.counts, args.Get(1).(common.Recommendation).Count) f.mu.Unlock() if f.onPurchase != nil { f.onPurchase() From c9c92ed8752a11437658e67f4f9c071f01ee94e5 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 16 Oct 2026 04:02:23 +0200 Subject: [PATCH 2/4] test(plans): adapt exact-count allocation proof to the retention ledger Delta of 686b9f02 over #907: the planstep and config code is identical on main. With the retention runtime (#905) a completed protected ramp has an untrusted baseline, so the exact-count test must assert the refusal instead of reading the baseline; without this TestEqualCountAttachThree fails on main plus #905. Co-Authored-By: Claude Sonnet 5.5 --- .../plan_step_allocation_integration_test.go | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/internal/purchase/plan_step_allocation_integration_test.go b/internal/purchase/plan_step_allocation_integration_test.go index 9e218f2b8..aaaed46c6 100644 --- a/internal/purchase/plan_step_allocation_integration_test.go +++ b/internal/purchase/plan_step_allocation_integration_test.go @@ -20,6 +20,9 @@ func testEqualCountAttach(t *testing.T, count int, want []int) { } _, err := f.pool.Exec(f.ctx, `UPDATE purchase_plans SET ramp_schedule = ramp_schedule || jsonb_build_object('allocation_mode','equal_count','total_steps',$2::int,'percent_per_step',$3::float8), services = jsonb_set(services,'{ec2,coverage}','100') WHERE id=$1`, f.planID, count, percent) require.NoError(t, err) + var ledger string + require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, f.planID).Scan(&ledger)) + require.Equal(t, "new", ledger) f.shownSuppressed(97) for i, n := range want { exec, tickErr := f.tick() @@ -27,8 +30,17 @@ func testEqualCountAttach(t *testing.T, count int, want []int) { assert.Equal(t, n, totalCount(exec), "exact equal-count allocation step %d", i+1) f.buy(i + 1) } - b, err := f.baseline() - require.NoError(t, err) + var b config.PlanRampBaseline + if f.plan().RampSchedule.IsComplete() { + require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT ramp_baseline FROM purchase_plans WHERE id=$1`, f.planID).Scan(&b)) + require.ErrorIs(t, f.store.WithTx(f.ctx, func(tx pgx.Tx) error { + _, readErr := f.store.GetPlanRampBaselineTx(f.ctx, tx, f.planID) + return readErr + }), config.ErrRampLedgerUntrusted) + } else { + b, err = f.baseline() + require.NoError(t, err) + } for _, q := range b { assert.Equal(t, 97, q.Count) } From 00eeaeb808f2fc793afb4af8d2088d11a5dc5f70 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Sat, 17 Oct 2026 15:41:36 +0200 Subject: [PATCH 3/4] fix(purchase): fail a refused ramp step durably and release its suppressions When the retention ledger refuses to size a step whose root execution row already carries recommendations, fail the row like the other refusals (Error and NoSubmission recorded, later ticks skip quietly) instead of returning the raw sentinel, which logged an error every tick. The pinned failure now expects the row's real recommendation state, and a won failure releases the row's suppressions in the same transaction as the stamp, since the step provably bought nothing. The CAS still commits before the stamp; that pre-existing gap is recorded in known_issues/36 and #1052. Refs #631 Co-Authored-By: Claude Opus 5.5 --- internal/purchase/notifications.go | 14 +++++-- internal/purchase/plan_step_attach.go | 3 ++ .../plan_step_retention_integration_test.go | 38 +++++++++++++++++-- .../36_plan_step_fail_cas_not_atomic.md | 30 +++++++++++++++ 4 files changed, 79 insertions(+), 6 deletions(-) create mode 100644 known_issues/36_plan_step_fail_cas_not_atomic.md diff --git a/internal/purchase/notifications.go b/internal/purchase/notifications.go index 0cf946bac..5b34502fc 100644 --- a/internal/purchase/notifications.go +++ b/internal/purchase/notifications.go @@ -292,13 +292,14 @@ func adoptableRootRow(rows []config.PurchaseExecution) (*config.PurchaseExecutio } // failExistingPlanStep fails a pending/notified plan-step row that will never -// carry recommendations, with reason. The status change is a compare-and-set, +// be bought, with reason. The CAS pins the recommendations to what the caller +// saw (empty for a bare row, non-empty for a refused attached one). The status change is a compare-and-set, // so a human approving the row between this tick's read and now wins: the row // is no longer awaiting approval and the tick skips it, instead of an upsert // 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.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, []string{"pending", "notified"}, "failed", nil, true) + failed, err := m.config.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, []string{"pending", "notified"}, "failed", nil, len(exec.Recommendations) == 0) 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) } @@ -306,7 +307,14 @@ func (m *Manager) failExistingPlanStep(ctx context.Context, exec *config.Purchas return fmt.Errorf("failed to fail plan step %d of plan %s: %w", exec.StepNumber, exec.PlanID, err) } stampFailedPlanStep(failed, reason) - if err := m.config.SavePurchaseExecution(ctx, failed); err != nil { + // The row is failed with NoSubmission, so its suppressions are released as + // for any proven not-submitted failure (saveTerminalExecution); a bare row has none. + if err := m.config.WithTx(ctx, func(tx pgx.Tx) error { + if err := m.config.SavePurchaseExecutionTx(ctx, tx, failed); err != nil { + return err + } + return m.config.DeleteSuppressionsByExecutionTx(ctx, tx, failed.ExecutionID) + }); err != nil { return fmt.Errorf("failed to record plan step %d of plan %s as failed: %w", failed.StepNumber, failed.PlanID, err) } return bareStepFailure(failed.PlanID, failed.StepNumber, reason) diff --git a/internal/purchase/plan_step_attach.go b/internal/purchase/plan_step_attach.go index eb1e4910a..31ee87b74 100644 --- a/internal/purchase/plan_step_attach.go +++ b/internal/purchase/plan_step_attach.go @@ -97,6 +97,9 @@ func (m *Manager) claimFromRows(ctx context.Context, tx pgx.Tx, locked *config.P } if len(root.Recommendations) > 0 { if _, err := m.config.GetPlanRampBaselineTx(ctx, tx, locked.ID); err != nil { + if errors.Is(err, config.ErrRampLedgerUntrusted) { + return m.failStep(ctx, tx, locked, claim, root, err.Error()) + } return stepClaim{}, err } if locked.NextExecutionDate.Before(time.Now()) { diff --git a/internal/purchase/plan_step_retention_integration_test.go b/internal/purchase/plan_step_retention_integration_test.go index 3179dc556..8f2504b0a 100644 --- a/internal/purchase/plan_step_retention_integration_test.go +++ b/internal/purchase/plan_step_retention_integration_test.go @@ -134,9 +134,15 @@ func TestCumulativeAttach_UntrustedHistoryCannotSizeOrNotify(t *testing.T) { rec.Count, rec.Selected = tc.held, true require.NoError(t, f.store.SavePurchaseExecution(f.ctx, &config.PurchaseExecution{PlanID: f.planID, ExecutionID: uuid.NewString(), Status: config.StatusCompleted, StepNumber: 1, ScheduledDate: time.Now(), Recommendations: []config.RecommendationRecord{rec}})) } + var rootID string if tc.rootStatus != "" { rec.Selected = true - require.NoError(t, f.store.SavePurchaseExecution(f.ctx, &config.PurchaseExecution{PlanID: f.planID, ExecutionID: uuid.NewString(), Status: tc.rootStatus, StepNumber: tc.progress + 1, ScheduledDate: time.Now().Add(40 * 24 * time.Hour), Recommendations: []config.RecommendationRecord{rec}})) + rootID = uuid.NewString() + require.NoError(t, f.store.SavePurchaseExecution(f.ctx, &config.PurchaseExecution{PlanID: f.planID, ExecutionID: rootID, Status: tc.rootStatus, StepNumber: tc.progress + 1, ScheduledDate: time.Now().Add(40 * 24 * time.Hour), Recommendations: []config.RecommendationRecord{rec}})) + for _, id := range []string{rootID} { + require.NoError(t, f.store.CreateSuppression(f.ctx, &config.PurchaseSuppression{ExecutionID: id, AccountID: "sup-acct", Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.t0", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)})) + } + require.NoError(t, f.store.CreateSuppression(f.ctx, &config.PurchaseSuppression{ExecutionID: uuid.NewString(), AccountID: "other-acct", Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.t0", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)})) } calls := 0 f.src.shown = func() []config.RecommendationRecord { calls++; return []config.RecommendationRecord{rec} } @@ -149,8 +155,17 @@ func TestCumulativeAttach_UntrustedHistoryCannotSizeOrNotify(t *testing.T) { require.Contains(t, rows[0].Error, "reconcile prior commitments") require.Empty(t, rows[0].Recommendations) } else { - require.ErrorIs(t, err, config.ErrRampLedgerUntrusted) - require.Equal(t, tc.rootStatus, rows[0].Status) + require.ErrorIs(t, err, errExecutionNotNotifiable, "a durable refusal is a quiet skip, not a raw error") + require.NotErrorIs(t, err, config.ErrRampLedgerUntrusted) + require.Equal(t, config.StatusFailed, rows[0].Status) + require.True(t, rows[0].NoSubmission) + require.Contains(t, rows[0].Error, config.ErrRampLedgerUntrusted.Error()) + require.NotEmpty(t, rows[0].Recommendations, "the attached recommendations stay on the failed row") + var released, others int + require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT COUNT(*) FROM purchase_suppressions WHERE execution_id = $1`, rootID).Scan(&released)) + require.Zero(t, released, "a proven not-submitted failure releases its suppressions") + require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT COUNT(*) FROM purchase_suppressions WHERE account_id = 'other-acct'`).Scan(&others)) + require.Equal(t, 1, others, "another execution's suppressions are untouched") } require.False(t, f.m.sendPlanNotification(f.ctx, f.plan())) require.Zero(t, calls, "no new sizing or resizing") @@ -202,3 +217,20 @@ func TestCumulativeAttach_AuthorizedLegacyQuantityRemainsImmutable(t *testing.T) }) } } + +func TestFailExistingPlanStep_LostRaceKeepsSuppressions(t *testing.T) { + f := newCumFixture(t, 100) + rec := cumRec(f.account, "m5.t0", 100) + rec.Selected = true + root := &config.PurchaseExecution{PlanID: f.planID, ExecutionID: uuid.NewString(), Status: "pending", StepNumber: 1, ScheduledDate: time.Now().Add(40 * 24 * time.Hour), Recommendations: []config.RecommendationRecord{rec}} + require.NoError(t, f.store.SavePurchaseExecution(f.ctx, root)) + require.NoError(t, f.store.CreateSuppression(f.ctx, &config.PurchaseSuppression{ExecutionID: root.ExecutionID, AccountID: "sup-acct", Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.t0", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)})) + _, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET status='approved' WHERE execution_id=$1`, root.ExecutionID) + require.NoError(t, err, "a human approves after the tick read the row") + + err = f.m.failExistingPlanStep(f.ctx, root, "reason") + require.ErrorIs(t, err, errExecutionNotNotifiable) + var n int + require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT COUNT(*) FROM purchase_suppressions WHERE execution_id = $1`, root.ExecutionID).Scan(&n)) + require.Equal(t, 1, n, "the approved row keeps its suppressions") +} diff --git a/known_issues/36_plan_step_fail_cas_not_atomic.md b/known_issues/36_plan_step_fail_cas_not_atomic.md new file mode 100644 index 000000000..9a654d04c --- /dev/null +++ b/known_issues/36_plan_step_fail_cas_not_atomic.md @@ -0,0 +1,30 @@ +# Known Issues: plan step failure is not atomic with its stamp and suppression release + +> **Audit status (2026-10-17):** `1 needs triage ยท 0 resolved` + +## LOW: crash between the failed CAS and the stamp tx strands the row + +**Files:** `internal/purchase/notifications.go` (`failExistingPlanStep`), +`internal/config/store_postgres.go` (`TransitionExecutionStatusPinned`), +`internal/purchase/plan_step_attach.go` (`claimFromRows`, `failStep`) + +**Description:** `failExistingPlanStep` first flips the row to `failed` with a +pinned compare-and-set, then saves the `Error` and `NoSubmission` stamp and +deletes the row's suppressions in a second transaction. The CAS cannot join a +caller transaction, so a crash or DB error between the two leaves the row +`failed` with no `Error`, no `NoSubmission`, and its suppressions still in +place until their grace period expires. The `Error`/`NoSubmission` half of the +gap predates the suppression release; the release makes it slightly wider. + +**Why it fails closed:** `adoptableRootRow` only adopts pending/notified roots, +so a failed row is skipped on every later tick and the step never gets a second +root. Nothing is bought twice. The cost is a step that stays failed without a +recorded reason and recommendations hidden until the suppressions expire. + +**Fix direction:** add a Tx variant of `TransitionExecutionStatusPinned` so the +CAS, the stamp and `DeleteSuppressionsByExecutionTx` commit together. + +**Why deferred:** out of scope for the retention-ledger runtime change (#905); +the window is a crash between two statements and the failure mode is safe. + +**Status:** Open. Tracked in #1052. From 8c0123d1271f38d10a39fa51dae39c7700b62f2c Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Sat, 17 Oct 2026 16:45:00 +0200 Subject: [PATCH 4/4] fix(plans): let never-started legacy ramps size again Migration 000127 marks a protected multi-step ramp 'new' again when its retention ledger is 'legacy' but it is still at step 0, has no frozen baseline and has no surviving purchase execution rows. Such plans then freeze a baseline at their first step like any new plan. The ledger guard trigger is disabled only around that update, inside the migration transaction. Mid-ramp legacy plans are left for a follow-up change. The untrusted-ledger refusal now says what is wrong instead of naming a reconcile action that does not exist. Refs #631 Co-Authored-By: Claude Opus 5.5 --- .../config/store_postgres_ramp_baseline.go | 2 +- ...never_started_legacy_ramps_to_new.down.sql | 4 + ...7_never_started_legacy_ramps_to_new.up.sql | 30 +++++ .../plan_step_cumulative_integration_test.go | 8 +- ..._step_ledger_migration_integration_test.go | 109 ++++++++++++++++++ .../plan_step_retention_integration_test.go | 2 +- 6 files changed, 152 insertions(+), 3 deletions(-) create mode 100644 internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.down.sql create mode 100644 internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.up.sql create mode 100644 internal/purchase/plan_step_ledger_migration_integration_test.go diff --git a/internal/config/store_postgres_ramp_baseline.go b/internal/config/store_postgres_ramp_baseline.go index 81cf29d4a..6c87cbcdd 100644 --- a/internal/config/store_postgres_ramp_baseline.go +++ b/internal/config/store_postgres_ramp_baseline.go @@ -32,7 +32,7 @@ func (s rampLedgerState) allowsSizing(raw []byte, current int) bool { return s == rampLedgerTrusted && raw != nil || s == rampLedgerNew && raw == nil && current == 0 } -var ErrRampLedgerUntrusted = errors.New("ramp purchase history cannot be verified; reconcile prior commitments before attaching another step") +var ErrRampLedgerUntrusted = errors.New("this ramp's purchase history from before the retention ledger cannot be verified, so it cannot size further steps") // SetPlanRampBaselineTx freezes a baseline only while retention can prove its lineage. // The caller holds the plan lock; full-row plan writes cannot promote trust. diff --git a/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.down.sql b/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.down.sql new file mode 100644 index 000000000..78c35476d --- /dev/null +++ b/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.down.sql @@ -0,0 +1,4 @@ +-- Intentionally a no-op: the up migration cannot tell the plans it promoted +-- from plans created as 'new', and demoting them would re-block sizing of +-- ramps that have never bought anything. +SELECT 1; diff --git a/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.up.sql b/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.up.sql new file mode 100644 index 000000000..265065132 --- /dev/null +++ b/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.up.sql @@ -0,0 +1,30 @@ +-- Promote never-started legacy ramps to the retention ledger (platform#631). +-- +-- Owner decision 2026-10-17: no backwards compatibility. A plan stamped +-- 'legacy' by 000124's default that is a protected multi-step ramp, has not +-- advanced (current_step = 0, no frozen baseline) and has no surviving +-- purchase_executions row of any status is treated as having no pre-ledger +-- history to distrust, so it becomes 'new' and sizes its first step normally. +-- This is not proof it never bought: retention cleanup could have deleted a +-- completed pre-000124 row of a plan whose step never advanced. That narrow +-- risk is accepted under the owner's decision. Mid-ramp legacy plans and any +-- plan with an execution row are deliberately left legacy. +-- +-- guard_ramp_ledger_state forces OLD.ramp_ledger_state = 'legacy' to stay +-- legacy on every UPDATE, so the guard is disabled for this one statement. +-- DISABLE TRIGGER names only that trigger on this table (unlike +-- session_replication_role, it needs no superuser and leaves +-- retain_ramp_ledger_execution and every other trigger live). The migration +-- runs in one transaction, so the guard is re-enabled before commit and a +-- failure rolls both back. +ALTER TABLE purchase_plans DISABLE TRIGGER guard_ramp_ledger_state; + +UPDATE purchase_plans p +SET ramp_ledger_state = 'new' +WHERE p.ramp_ledger_state = 'legacy' + AND ramp_ledger_protected(p.ramp_schedule) + AND COALESCE((p.ramp_schedule ->> 'current_step')::INT, 0) = 0 + AND p.ramp_baseline IS NULL + AND NOT EXISTS (SELECT 1 FROM purchase_executions e WHERE e.plan_id = p.id); + +ALTER TABLE purchase_plans ENABLE TRIGGER guard_ramp_ledger_state; diff --git a/internal/purchase/plan_step_cumulative_integration_test.go b/internal/purchase/plan_step_cumulative_integration_test.go index c80c9cbae..77ec5219a 100644 --- a/internal/purchase/plan_step_cumulative_integration_test.go +++ b/internal/purchase/plan_step_cumulative_integration_test.go @@ -65,9 +65,15 @@ func cumRec(account, typ string, count int) config.RecommendationRecord { const cumCoverage = 80 func newCumFixture(t *testing.T, stored ...int) *cumFixture { + t.Helper() + store, pool := rampStepStore(context.Background(), t) + return newCumFixtureOn(t, store, pool, stored...) +} + +// newCumFixtureOn builds the fixture on an already migrated store. +func newCumFixtureOn(t *testing.T, store *config.PostgresStore, pool *pgxpool.Pool, stored ...int) *cumFixture { t.Helper() ctx := context.Background() - store, pool := rampStepStore(ctx, t) attachGlobal(ctx, t, store) acct := &config.CloudAccount{Name: "cum-acct", Provider: "aws", ExternalID: "111111111110", Enabled: true, AWSAuthMode: "role_arn", AWSRoleARN: "arn:aws:iam::111111111110:role/cudly"} diff --git a/internal/purchase/plan_step_ledger_migration_integration_test.go b/internal/purchase/plan_step_ledger_migration_integration_test.go new file mode 100644 index 000000000..714971fb8 --- /dev/null +++ b/internal/purchase/plan_step_ledger_migration_integration_test.go @@ -0,0 +1,109 @@ +//go:build integration + +package purchase + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +const neverStartedMigration = "000127_never_started_legacy_ramps_to_new" + +// stagedMigrations returns a directory holding every migration but the one +// under test, and a func that adds it, so the test runs the real runner twice. +func stagedMigrations(t *testing.T) (dir string, addTarget func()) { + t.Helper() + src := rampMigrationsPath() + dir = t.TempDir() + entries, err := os.ReadDir(src) + require.NoError(t, err) + copyFile := func(name string) { + b, readErr := os.ReadFile(filepath.Join(src, name)) + require.NoError(t, readErr) + require.NoError(t, os.WriteFile(filepath.Join(dir, name), b, 0o600)) + } + for _, e := range entries { + if filepath.Ext(e.Name()) == ".sql" && !strings.HasPrefix(e.Name(), neverStartedMigration) { + copyFile(e.Name()) + } + } + return dir, func() { + copyFile(neverStartedMigration + ".up.sql") + copyFile(neverStartedMigration + ".down.sql") + } +} + +func TestMigration_NeverStartedLegacyRampsBecomeNew(t *testing.T) { + ctx := context.Background() + container, err := testhelpers.SetupPostgresContainer(ctx, t) + require.NoError(t, err) + t.Cleanup(func() { _ = container.Cleanup(ctx) }) + dir, addTarget := stagedMigrations(t) + require.NoError(t, migrations.RunMigrations(ctx, container.DB.Pool(), dir, "", "")) + store, pool := config.NewPostgresStore(container.DB), container.DB.Pool() + + f := newCumFixtureOn(t, store, pool, 100) // plan (a): never started, no executions + clone := func(name string, total, current int) string { + p := f.plan() + p.ID, p.Name = "", name + p.RampSchedule.TotalSteps = total + require.NoError(t, store.CreatePurchasePlan(ctx, p)) + _, execErr := pool.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule=jsonb_set(ramp_schedule,'{current_step}',to_jsonb($2::int)) WHERE id=$1`, p.ID, current) + require.NoError(t, execErr) + return p.ID + } + withExec := clone("b-has-execution", 4, 0) + require.NoError(t, store.SavePurchaseExecution(ctx, &config.PurchaseExecution{PlanID: withExec, ExecutionID: uuid.NewString(), Status: config.StatusFailed, NoSubmission: true, StepNumber: 1, ScheduledDate: time.Now()})) + midRamp := clone("c-mid-ramp", 4, 2) + singleStep := clone("d-single-step", 1, 0) + trusted := clone("e-trusted", 4, 0) + withBaseline := clone("f-legacy-baseline", 4, 0) + + // Force the states the old writer left behind; the guard would otherwise rewrite them. + _, err = pool.Exec(ctx, `ALTER TABLE purchase_plans DISABLE TRIGGER guard_ramp_ledger_state`) + require.NoError(t, err) + _, err = pool.Exec(ctx, `UPDATE purchase_plans SET ramp_ledger_state='legacy' WHERE id <> $1`, trusted) + require.NoError(t, err) + _, err = pool.Exec(ctx, `UPDATE purchase_plans SET ramp_baseline='{"k":{"count":1}}'::jsonb WHERE id = $1`, withBaseline) + require.NoError(t, err) + _, err = pool.Exec(ctx, `UPDATE purchase_plans SET ramp_ledger_state='trusted', ramp_baseline='{"k":{"count":1}}'::jsonb WHERE id = $1`, trusted) + require.NoError(t, err) + _, err = pool.Exec(ctx, `ALTER TABLE purchase_plans ENABLE TRIGGER guard_ramp_ledger_state`) + require.NoError(t, err) + + addTarget() + require.NoError(t, migrations.RunMigrations(ctx, container.DB.Pool(), dir, "", "")) + + state := func(id string) (s string) { + require.NoError(t, pool.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, id).Scan(&s)) + return s + } + require.Equal(t, "new", state(f.planID), "never started, no executions") + require.Equal(t, "legacy", state(withExec), "an execution row of any status keeps it legacy") + require.Equal(t, "legacy", state(midRamp), "mid-ramp plans are not touched") + require.Equal(t, "legacy", state(singleStep), "single-step plans are not protected") + require.Equal(t, "legacy", state(withBaseline), "a frozen baseline is never discarded") + require.Equal(t, "trusted", state(trusted)) + var enabled string + require.NoError(t, pool.QueryRow(ctx, `SELECT tgenabled::text FROM pg_trigger WHERE tgname='guard_ramp_ledger_state'`).Scan(&enabled)) + require.Equal(t, "O", enabled, "the guard is live again") + + rec := cumRec(f.account, "m5.t0", 100) + f.src.shown = func() []config.RecommendationRecord { return []config.RecommendationRecord{rec} } + exec, err := f.tick() + require.NoError(t, err, "the promoted ramp sizes step 1 normally") + require.NotEmpty(t, exec.Recommendations) + base, err := f.baseline() + require.NoError(t, err) + require.NotEmpty(t, base, "step 1 froze a baseline") +} diff --git a/internal/purchase/plan_step_retention_integration_test.go b/internal/purchase/plan_step_retention_integration_test.go index 8f2504b0a..f1ea65cd4 100644 --- a/internal/purchase/plan_step_retention_integration_test.go +++ b/internal/purchase/plan_step_retention_integration_test.go @@ -152,7 +152,7 @@ func TestCumulativeAttach_UntrustedHistoryCannotSizeOrNotify(t *testing.T) { require.Len(t, rows, 1) if tc.rootStatus == "" { require.Equal(t, config.StatusFailed, rows[0].Status) - require.Contains(t, rows[0].Error, "reconcile prior commitments") + require.Contains(t, rows[0].Error, "cannot be verified") require.Empty(t, rows[0].Recommendations) } else { require.ErrorIs(t, err, errExecutionNotNotifiable, "a durable refusal is a quiet skip, not a raw error")