fix(plans): count a ramp step only when every account has bought (#1880)

`CompletePlanStep` advanced the ramp when **any** execution for that step finished successfully. A multi-account plan fans one ramp step out into one execution per cloud account, each succeeding or failing independently, so an operator who repaired and retried a single failed account moved the plan to "step N done" while the other accounts had bought nothing for step N. The plan then proceeded to step N+1 having silently bought less commitment than the customer intended, and nothing durable recorded that it had. This predates #1669 and was unchanged by it: it is visible in that fix's own regression test, where a 3-account plan whose step-3 fan-out committed only account A reaches `CurrentStep = 3` as soon as account B's retry succeeds while account C is still failed.

## Design: sibling-completeness gate, not derived progress

The issue nominates deriving `CurrentStep` from the executions table as the preferred option. That was rejected because `CleanupOldExecutions` deletes `status = 'completed'` rows past the retention horizon, so a derived position falls back toward zero as rows age out and the plan re-buys its entire ramp. That trades a stored-convention problem for a data-lifetime problem whose failure direction is spending real money twice. The chosen gate has no equivalent hole: cleanup sweeps only `completed` and `canceled` rows, so the units that age out are exactly the ones that did buy, and the gate degrades permissively rather than toward a re-buy.

Deriving also does not avoid the freeze it is credited with avoiding. "Highest fully-bought step" either requires contiguity, which freezes identically on the incomplete step, or lets a later clean step jump over it, which is the overstatement the skipped-predecessor refusal exists to prevent.

"Target accounts at the time the step ran" is pinned to the execution rows the fan-out wrote, which are never rewritten. The root row is an aggregate of its children, so an all-accounts-failed step is not blocked by its own container; within each account only the latest attempt counts, ordered by `(retry_execution_id IS NULL) DESC, updated_at DESC`. Both keys are load-bearing: a retry successor shares its predecessor's transaction timestamp, while a root re-drive supersedes nothing.

## The gate needed a real exit

A gate that refuses to advance while any account is outstanding can freeze a ramp permanently. The first implementation's comment claimed an operator could "retry the account until it buys, or cancel its row". That exit did not exist: `IsCancelable` admits only `pending`, `notified` and `scheduled`, and `CancelExecutionAtomic` guards `status IN ('pending','notified')`, so a `failed` row is uncancelable by either path, while retry is separately refused whenever `RedriveRefusalReason` fires (Azure savings plans, unrecognised providers). An Azure-SP row that failed could be neither retried nor canceled.

Widening the cancel policy was rejected, since a row past `approved` may already have moved money. Instead the gate counts only units whose account the plan still targets: rows decide which units exist, current attachments decide whether a unit still matters, and detaching via `SetPlanAccounts` is the exit. Disabling an account is not an exit, because `GetPlanAccounts` has no `enabled` filter, and no claim is made that it is.

## Frozen ramps must not double-buy, and the guard must be atomic

While a ramp is held at step N-1, the create path stamps `StepNumber: CurrentStep + i + 1`, so a fresh create mints a root row for step N that re-fans-out across accounts that already bought. The per-account idempotency token derives from `idempotencyLineageKey(baseExec) + ":" + account.ID`, and a new root mints a fresh UUID key, so it is a different token, provider dedupe never engages, and the commitment is genuinely bought twice.

The first attempt at that guard ran the probe before `WithTx` opened, which made it advisory: two concurrent creates could both pass it, and a pending account execution could succeed between the check and the insert. The probe now runs inside the transaction under the per-plan ramp lock, against the state that lock protects, so a create cannot interleave with a concurrent create or a concurrent completion. The pre-transaction plan read is retained only to answer the 404 and is explicitly discarded as an unlocked snapshot.

Also fixed: "bought" is `EXISTS any succeeded row` rather than the latest attempt, since a purchase is irreversible and a later failed attempt would otherwise freeze the step and invite a retry that buys twice; the already-counted path no longer stamps an error note on a cleanly-completed row during a benign sibling race; the stuck-step report no longer describes non-ramp plans as blocked ramps; and the interface contract now names both duplicate-completion sentinels, since the #1669 scenario lands on the equal-step case that returns `ErrRampStepCountedBySibling` and callers branch on the distinction.

## How it was verified

The issue's exact scenario was reproduced first as a failing test (`expected: 2, actual: 3`) and re-confirmed against the final test code by disabling only the gate call. Tests drive the real executor against real Postgres. The retry fixture was corrected in the process: it previously modelled a retry no production path can produce, because it never stamped `retry_execution_id`.

Atomicity is tested as a genuine race, not a sequence: a holder transaction takes the ramp lock and stays **uncommitted** while a competing create runs, asserting it blocks and then observes the winner. Concurrent goroutines alone would prove nothing, because the loser would read the winner's committed row regardless.

A mutation harness covers seven mutants, all killed. Two survived the first attempt and were rewritten: making `ever_bought` independent of the representative row meant ordering tests written against a unit that had already bought could not fail. They now run against units that never bought, with execution IDs chosen so the dead attempt wins the last-resort tie-break, and the supersession case moved to a single-transaction retry so both rows share a timestamp as production produces.

Gates re-run on the final commit after rebasing onto `ac00d9d5e`: build, vet and the three touched packages all exit 0 with zero failures, and `gocyclo -over 10` is clean (confirmed to be scanning: 339 functions report at threshold 5).

## Known gap, deliberately not fixed here

A fan-out that fails to write one account's row advances anyway, because the gate derives its target set only from rows that exist and has no cross-check against fan-out width. Both migration-free remedies introduce a worse freeze: requiring all attached accounts freezes any plan that gains one mid-ramp, and a "row at an earlier step" heuristic freezes a detach and re-attach. `plan_accounts` carries no timestamp, so there is no sound "attached when the step ran" signal. Documented in-code, and not a regression: before this change that account never bought either and the ramp advanced regardless.

Deferred: `handler_history.go` describes any completed row carrying a non-empty error as an audit gap, which mis-describes a ramp note. Pre-existing since #1669, contained here by not stamping the newly-frequent transient case; a proper fix wants a dedicated column plus roughly eleven lockstep SELECT projections.

Closes #1861

Author: cristim

internal/analytics/collector_test.go

@@ -164,6 +164,18 @@ func (m *mockConfigStore) CompletePlanStep(ctx context.Context, planID string, s
 	return nil
 }
 
+func (m *mockConfigStore) GetStuckRampSteps(_ context.Context) (map[string]config.RampStepBlock, error) {
+	return nil, nil
+}
+
+func (m *mockConfigStore) LockPurchasePlanTx(_ context.Context, _ pgx.Tx, _ string) (*config.PurchasePlan, error) {
+	return nil, nil
+}
+
+func (m *mockConfigStore) OccupiedRampStepsInRangeTx(_ context.Context, _ pgx.Tx, _ string, _, _ int) ([]int, error) {
+	return nil, nil
+}
+
 func (m *mockConfigStore) UpdatePurchasePlanTx(ctx context.Context, _ pgx.Tx, plan *config.PurchasePlan) error {
 	return nil
 }

internal/api/handler_plans.go

@@ -79,8 +79,18 @@ func (h *Handler) attachPlanHealth(ctx context.Context, plans []config.PurchaseP
 		return result
 	}
 
+	// A blocked ramp is worth up to 25 points, so a score computed without it
+	// is not a partially-informed score, it is a wrong one -- and it errs
+	// healthy, on plans that are stopped. Withhold every score rather than
+	// publish that, exactly as the counts fetch above does.
+	stuckByPlan, err := h.config.GetStuckRampSteps(ctx)
+	if err != nil {
+		logging.Warnf("listPlans: GetStuckRampSteps failed, plan health reported as unknown: %v", err)
+		return result
+	}
+
 	for i := range plans {
-		score, factors := computePlanHealth(plans[i], now, countsByPlan[plans[i].ID])
+		score, factors := computePlanHealth(plans[i], now, countsByPlan[plans[i].ID], stuckByPlan[plans[i].ID])
 		result[i].HealthScore = &score
 		result[i].HealthFactors = factors
 	}
@@ -339,8 +349,12 @@ func (h *Handler) createPlannedPurchases(ctx context.Context, httpReq *events.La
 		return nil, err
 	}
 
-	plan, err := h.getPlanForPurchaseCreation(ctx, planID)
-	if err != nil {
+	// Validation only: this answers the 404 (and maps storage errors to a clean
+	// message) before a transaction is opened. The plan it returns is
+	// deliberately discarded -- it is an unlocked snapshot whose ramp position
+	// can be stale by the time the inserts run, and the authoritative read
+	// happens under the ramp lock inside the transaction below (issue #1861).
+	if _, err := h.getPlanForPurchaseCreation(ctx, planID); err != nil {
 		return nil, err
 	}
 
@@ -366,22 +380,95 @@ func (h *Handler) createPlannedPurchases(ctx context.Context, httpReq *events.La
 	creator := resolveCreatorUserID(session)
 	created := 0
 	if err := h.config.WithTx(ctx, func(tx pgx.Tx) error {
-		n, txErr := h.createPurchaseExecutionsTx(ctx, tx, plan, planID, req.Count, startDate, creator)
-		if txErr != nil {
-			return txErr
-		}
-		if planErr := h.updatePlanNextExecutionDateTx(ctx, tx, plan, startDate); planErr != nil {
-			return planErr
-		}
+		n, txErr := h.createPlannedPurchasesTx(ctx, tx, planID, req.Count, startDate, creator)
 		created = n
-		return nil
+		return txErr
 	}); err != nil {
 		return nil, err
 	}
 
 	return &CreatePlannedPurchasesResponse{Created: created}, nil
 }
 
+// createPlannedPurchasesTx is the transactional body of createPlannedPurchases:
+// take the per-plan ramp lock, decide against the state that lock protects, and
+// write, all before anyone else can read it.
+//
+// The lock is the point. The step range derives from the plan's CurrentStep and
+// the "is this step already covered" test reads the executions of those steps,
+// so both inputs are exactly the state a concurrent completion or a concurrent
+// create mutates. Read either outside the lock and the check becomes advisory:
+// two creates both see the step free and each mints a root row for it, and
+// approving both re-fans-out over accounts that already bought. CompletePlanStep
+// takes this same lock, so a completion cannot interleave either.
+func (h *Handler) createPlannedPurchasesTx(ctx context.Context, tx pgx.Tx, planID string, count int, startDate time.Time, creator *string) (int, error) {
+	plan, err := h.config.LockPurchasePlanTx(ctx, tx, planID)
+	if err != nil {
+		logging.Errorf("createPlannedPurchases: LockPurchasePlanTx failed (plan=%s): %v", planID, err)
+		return 0, NewClientError(503, "could not lock the plan for scheduling; try again")
+	}
+	if plan == nil {
+		return 0, NewClientError(404, "plan not found")
+	}
+
+	if refuseErr := h.refuseOccupiedRampSteps(ctx, tx, plan, planID, count); refuseErr != nil {
+		return 0, refuseErr
+	}
+
+	created, createErr := h.createPurchaseExecutionsTx(ctx, tx, plan, planID, count, startDate, creator)
+	if createErr != nil {
+		return 0, createErr
+	}
+	if planErr := h.updatePlanNextExecutionDateTx(ctx, tx, plan, startDate); planErr != nil {
+		return 0, planErr
+	}
+	return created, nil
+}
+
+// refuseOccupiedRampSteps blocks a create whose steps overlap one that is
+// already covered (issue #1861). Must be called with the plan's ramp lock held
+// for the same transaction, which createPlannedPurchasesTx does.
+//
+// createPurchaseExecutionsTx stamps CurrentStep+1 .. CurrentStep+count. The
+// completeness gate holds CurrentStep still while any account of a step is
+// outstanding, so on a plan an operator is trying to unstick, CurrentStep+1 is
+// the very step that is partly bought. The row minted for it is a ROOT row:
+// approving it re-fans-out across every account on the plan, including the ones
+// that already bought, under a fresh idempotency lineage (the new root's key is
+// a new UUID, and each account's token derives from it), so the provider-side
+// dedupe never engages and the commitment is genuinely bought twice. Per-account
+// retry of the outstanding rows is the operation that actually finishes the step.
+//
+// Fails closed: an unreadable check refuses the create rather than minting rows
+// that might double-buy.
+func (h *Handler) refuseOccupiedRampSteps(ctx context.Context, tx pgx.Tx, plan *config.PurchasePlan, planID string, count int) error {
+	from := plan.RampSchedule.CurrentStep + 1
+	occupied, err := h.config.OccupiedRampStepsInRangeTx(ctx, tx, planID, from, from+count-1)
+	if err != nil {
+		logging.Errorf("createPlannedPurchases: OccupiedRampStepsInRangeTx failed (plan=%s): %v", planID, err)
+		return NewClientError(503, "could not verify which ramp steps are already covered; try again")
+	}
+	if len(occupied) == 0 {
+		return nil
+	}
+	return NewClientError(409, fmt.Sprintf(
+		"ramp step(s) %s of this plan already have purchase executions that bought or are still in flight; "+
+			"finish or cancel those instead of scheduling the same step again",
+		formatRampSteps(occupied)))
+}
+
+// formatRampSteps renders step numbers for an operator-facing message.
+func formatRampSteps(steps []int) string {
+	out := ""
+	for i, s := range steps {
+		if i > 0 {
+			out += ", "
+		}
+		out += fmt.Sprintf("%d", s)
+	}
+	return out
+}
+
 // parseCreatePurchasesRequest parses and validates the create purchases request.
 func (h *Handler) parseCreatePurchasesRequest(body string) (*CreatePlannedPurchasesRequest, time.Time, error) {
 	var req CreatePlannedPurchasesRequest