fix(ladder): bound amend raises by the config commitment ceiling (#842)

internal/api/handler_ladder_amend.go

@@ -32,6 +32,8 @@ func (h *Handler) amendLadderTranche(ctx context.Context, req *events.LambdaFunc
 		return nil, NewClientError(409, err.Error())
 	case errors.Is(err, config.ErrLadderAmendInvalid):
 		return nil, NewClientError(400, err.Error())
+	case errors.Is(err, config.ErrLadderLockTimeout):
+		return nil, NewClientErrorWithHeaders(503, "Ladder planning is busy; retry shortly", nil, map[string]string{"Retry-After": "5"})
 	default:
 		return result, err
 	}

internal/api/handler_ladder_amend_test.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
@@ -69,12 +70,16 @@ func TestLadderAmendmentStatusErrors(t *testing.T) {
 		code int
 	}{
 		{config.ErrLadderAmendConflict, 409}, {config.ErrLadderAmendInvalid, 400},
+		{fmt.Errorf("amend ladder tranche: %w", config.ErrLadderLockTimeout), 503},
 	} {
 		h, original := ladderScopedHandler(t)
 		h.config = &amendmentTestStore{MockConfigStore: original, accountID: scopedInAccount, err: example.err}
 		_, err := h.amendLadderTranche(context.Background(), ladderScopedReq(`{"expected_revision":0,"scheduled_date":"2030-01-01T00:00:00Z","amount_usd_hr":"1"}`), scopedInAccount)
 		clientErr, ok := IsClientError(err)
 		require.True(t, ok)
 		require.Equal(t, example.code, clientErr.code)
+		if example.code == 503 {
+			require.Equal(t, "5", clientErr.headers["Retry-After"])
+		}
 	}
 }

internal/config/interfaces.go

@@ -76,7 +76,7 @@ type StoreInterface interface {
 	// named provider, in sorted order, so two plan steps that resolve the same
 	// provider's recommendations cannot both attach the same capacity. Lock
 	// order is advisory lock, then LockPurchasePlanTx. The ladder config lock
-	// (PlanAndSaveLadderRun, ClaimDueLadderTranches) is always taken BEFORE both:
+	// (PlanAndSaveLadderRun, ClaimDueLadderTranches, AmendLadderTranche) is always taken BEFORE both:
 	// ClaimDueLadderTranches's create callback may take these locks, so they
 	// come after the ladder lock; code holding an execution or tranche row
 	// (retry linker, reconciler) must never take the ladder lock. No cycle.

internal/config/ladder_amend.go

@@ -11,7 +11,7 @@ import (
 var (
 	ErrLadderAmendNotFound = errors.New("ladder tranche not found")
 	ErrLadderAmendConflict = errors.New("ladder tranche changed or is no longer editable")
-	ErrLadderAmendInvalid  = errors.New("ladder amendment exceeds the original run total or current per-run cap")
+	ErrLadderAmendInvalid  = errors.New("ladder amendment exceeds the original run total, the current per-run cap or the config commitment ceiling")
 	ladderAmountPattern    = regexp.MustCompile(`^(0|[1-9]\d{0,13})(\.\d{1,6})?$`)
 )
 

internal/config/store_postgres_ladder_amend.go

@@ -29,9 +29,15 @@ func (s *PostgresStore) AmendLadderTranche(ctx context.Context, id, accountID, p
 	if actor == "" {
 		return nil, errors.New("ladder amendment actor is required")
 	}
+	configID, err := s.ladderTrancheConfigID(ctx, id, accountID, provider)
+	if err != nil {
+		return nil, fmt.Errorf("amend ladder tranche: %w", err)
+	}
 	result := &LadderAmendmentResult{ID: id}
-	err := s.WithTx(ctx, func(tx pgx.Tx) error {
-		return amendLadderTrancheTx(ctx, tx, id, accountID, provider, actor, amendment, result)
+	// The ladder lock is taken first, row locks after (see withLadderConfigLock),
+	// so the ceiling sum below is read against the planner's committed state.
+	err = s.withLadderConfigLock(ctx, configID, func(tx pgx.Tx) error {
+		return amendLadderTrancheTx(ctx, tx, configID, id, accountID, provider, actor, amendment, result)
 	})
 	if err != nil {
 		return nil, fmt.Errorf("amend ladder tranche: %w", err)
@@ -59,11 +65,26 @@ func (t lockedLadderTranche) editable(runStatus ladder.RunStatus, expectedRevisi
 		t.executionID == nil && t.revision == expectedRevision && t.future
 }
 
-func amendLadderTrancheTx(ctx context.Context, tx pgx.Tx, id, accountID, provider, actor string, amendment LadderAmendment, result *LadderAmendmentResult) error {
+// ladderTrancheConfigID resolves the tranche's config without any row lock, so
+// the caller can take the config's ladder lock before locking rows.
+func (s *PostgresStore) ladderTrancheConfigID(ctx context.Context, id, accountID, provider string) (string, error) {
+	var configID string
+	err := s.db.QueryRow(ctx, `SELECT t.config_id FROM ladder_tranches t
+ JOIN ladder_configs c ON c.id=t.config_id WHERE t.id=$1 AND c.cloud_account_id=$2 AND c.provider=$3`, id, accountID, provider).Scan(&configID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return "", ErrLadderAmendNotFound
+	}
+	return configID, err
+}
+
+func amendLadderTrancheTx(ctx context.Context, tx pgx.Tx, configID, id, accountID, provider, actor string, amendment LadderAmendment, result *LadderAmendmentResult) error {
 	run, err := lockLadderRun(ctx, tx, id, accountID, provider)
 	if err != nil {
 		return err
 	}
+	if run.configID != configID {
+		return ErrLadderAmendConflict
+	}
 	tranche, err := lockLadderTranche(ctx, tx, id, run, amendment.ScheduledDate)
 	if err != nil {
 		return err
@@ -75,6 +96,10 @@ func amendLadderTrancheTx(ctx context.Context, tx pgx.Tx, id, accountID, provide
 	if err != nil {
 		return err
 	}
+	err = checkLadderConfigCeiling(ctx, tx, run.configID, tranche.amount, amendment.AmountUSDHr)
+	if err != nil {
+		return err
+	}
 	err = tx.QueryRow(ctx, `UPDATE ladder_tranches SET amount_usd_hr=$2::numeric,scheduled_date=$3,revision=revision+1
  WHERE id=$1 AND revision=$4 AND scheduled_date>clock_timestamp() AND $3::timestamptz>clock_timestamp()
  RETURNING revision,amount_usd_hr::text,scheduled_date`, id, amendment.AmountUSDHr, amendment.ScheduledDate, amendment.ExpectedRevision).
@@ -96,7 +121,7 @@ func lockLadderRun(ctx context.Context, tx pgx.Tx, id, accountID, provider strin
 	err := tx.QueryRow(ctx, `SELECT c.id,c.max_hourly_commit_per_run::text
  FROM ladder_configs c JOIN cloud_accounts a ON a.id=c.cloud_account_id AND a.provider=c.provider
  WHERE c.cloud_account_id=$2 AND c.provider=$3
- AND c.id=(SELECT config_id FROM ladder_tranches WHERE id=$1) FOR UPDATE OF c`, id, accountID, provider).
+ AND c.id=(SELECT config_id FROM ladder_tranches WHERE id=$1) FOR NO KEY UPDATE OF c`, id, accountID, provider).
 		Scan(&run.configID, &run.cap)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return run, ErrLadderAmendNotFound
@@ -105,7 +130,7 @@ func lockLadderRun(ctx context.Context, tx pgx.Tx, id, accountID, provider strin
 		return run, err
 	}
 	err = tx.QueryRow(ctx, `SELECT id,status,total_hourly_commit::text FROM ladder_runs
- WHERE config_id=$2 AND id=(SELECT run_id FROM ladder_tranches WHERE id=$1) FOR UPDATE`, id, run.configID).
+ WHERE config_id=$2 AND id=(SELECT run_id FROM ladder_tranches WHERE id=$1) FOR NO KEY UPDATE`, id, run.configID).
 		Scan(&run.runID, &run.status, &run.originalTotal)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return run, ErrLadderAmendNotFound

internal/config/store_postgres_ladder_amend_ceiling.go

@@ -0,0 +1,101 @@
+package config
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+const ladderMicros = 1e6
+
+// ladderMaxUSDHr keeps micro-dollar sums far below int64 overflow (about 9.2e12 USD/hr).
+const ladderMaxUSDHr = 1e12
+
+// toLadderMicros converts a USD/hr amount (at most six fractional digits) to
+// integer micro-dollars, so the ceiling comparison is exact.
+func toLadderMicros(v float64) int64 { return int64(math.Round(v * ladderMicros)) }
+
+// checkLadderConfigCeiling refuses a RAISE that would push the config's
+// commitment past what its latest non-canceled run planned to leave room for.
+// Decreases, re-date-only edits and cancels always pass.
+//
+// Committed = in-flight (fired + scheduled, all runs, with the amended row at
+// its new amount) + tranches completed after that run started, less the
+// ladderSettleSkewMargin for clock skew (their
+// commitment may not be in the run's existing_usd_hr snapshot yet).
+// Ceiling C = min(target, baseline * current coverage%) - existing, from the
+// run's snapshot (existing is gross, so C errs low) and the config's current
+// coverage. A run without a full snapshot refuses a raise. Must run inside
+// withLadderConfigLock so the sums cannot move under it.
+func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmount, newAmount string) error {
+	var raise bool
+	if err := tx.QueryRow(ctx, `SELECT $1::numeric>$2::numeric`, newAmount, oldAmount).Scan(&raise); err != nil {
+		return err
+	}
+	if !raise {
+		return nil
+	}
+	snap, err := loadLadderCeiling(ctx, tx, configID, oldAmount, newAmount)
+	if err != nil {
+		return err
+	}
+	// Same sweep as the planner: a scheduled row past the grace window is neither
+	// counted nor claimable, so cancel it here or a later claim with an older asOf
+	// could still fire it on top of the raise.
+	now := time.Now()
+	err = cancelOverdueLadderTranchesTx(ctx, tx, &configID, now)
+	if err != nil {
+		return err
+	}
+	inFlight, err := inFlightLadderCommit(ctx, tx, configID, now)
+	if err != nil {
+		return err
+	}
+	settled, err := settledLadderCommit(ctx, tx, configID, ladderSettleAnchor(snap.anchor))
+	if err != nil {
+		return err
+	}
+	for _, v := range []float64{inFlight, settled, snap.ceiling, snap.oldAmount, snap.newAmount} {
+		if v > ladderMaxUSDHr {
+			return fmt.Errorf("%w: amount beyond %.0f USD/hr", ErrLadderAmendInvalid, ladderMaxUSDHr)
+		}
+	}
+	committed := toLadderMicros(inFlight) - toLadderMicros(snap.oldAmount) + toLadderMicros(snap.newAmount) + toLadderMicros(settled)
+	if committed > toLadderMicros(snap.ceiling) {
+		return ErrLadderAmendInvalid
+	}
+	return nil
+}
+
+type ladderCeilingSnapshot struct {
+	anchor                        time.Time
+	ceiling, oldAmount, newAmount float64
+}
+
+// loadLadderCeiling reads the latest non-canceled run's snapshot joined with the
+// config's current coverage. It refuses a run without a full snapshot.
+func loadLadderCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmount, newAmount string) (ladderCeilingSnapshot, error) {
+	var snap ladderCeilingSnapshot
+	var hasSnapshot bool
+	err := tx.QueryRow(ctx, `SELECT r.started_at,
+ r.target_usd_hr IS NOT NULL AND r.existing_usd_hr IS NOT NULL AND r.baseline_usd_hr IS NOT NULL,
+ COALESCE(trunc(LEAST(r.target_usd_hr,r.baseline_usd_hr*c.target_coverage/100)-r.existing_usd_hr,6),0)::float8,
+ $2::numeric::float8,$3::numeric::float8
+ FROM ladder_runs r JOIN ladder_configs c ON c.id=r.config_id
+ WHERE r.config_id=$1 AND r.status<>'cancelled' ORDER BY r.started_at DESC,r.created_at DESC,r.id DESC LIMIT 1`,
+		configID, oldAmount, newAmount).Scan(&snap.anchor, &hasSnapshot, &snap.ceiling, &snap.oldAmount, &snap.newAmount)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return snap, fmt.Errorf("%w: config has no non-canceled run", ErrLadderAmendInvalid)
+	}
+	if err != nil {
+		return snap, err
+	}
+	if !hasSnapshot {
+		return snap, fmt.Errorf("%w: latest run has no target snapshot to bound a raise", ErrLadderAmendInvalid)
+	}
+	return snap, nil
+}