From 9db25ae109a987676ea5d1749f09f5c52bfcee0e Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 13:16:29 +0200 Subject: [PATCH 1/3] fix(ladder): bound amend raises by the config commitment ceiling under the ladder lock (#842) Co-Authored-By: Claude Sonnet 5.5 --- internal/api/handler_ladder_amend.go | 2 + internal/api/handler_ladder_amend_test.go | 5 + internal/config/interfaces.go | 2 +- internal/config/ladder_amend.go | 2 +- .../config/store_postgres_ladder_amend.go | 35 ++- .../store_postgres_ladder_amend_ceiling.go | 70 ++++++ ...s_ladder_amend_ceiling_integration_test.go | 210 ++++++++++++++++++ ..._postgres_ladder_amend_integration_test.go | 2 +- internal/config/store_postgres_ladder_lock.go | 9 + 9 files changed, 329 insertions(+), 8 deletions(-) create mode 100644 internal/config/store_postgres_ladder_amend_ceiling.go create mode 100644 internal/config/store_postgres_ladder_amend_ceiling_integration_test.go diff --git a/internal/api/handler_ladder_amend.go b/internal/api/handler_ladder_amend.go index db249816..3495a125 100644 --- a/internal/api/handler_ladder_amend.go +++ b/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 } diff --git a/internal/api/handler_ladder_amend_test.go b/internal/api/handler_ladder_amend_test.go index 26c9577e..dd839755 100644 --- a/internal/api/handler_ladder_amend_test.go +++ b/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,6 +70,7 @@ 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} @@ -76,5 +78,8 @@ func TestLadderAmendmentStatusErrors(t *testing.T) { 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"]) + } } } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 6ed9bfe6..13097083 100644 --- a/internal/config/interfaces.go +++ b/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. diff --git a/internal/config/ladder_amend.go b/internal/config/ladder_amend.go index f80a0614..ce89e184 100644 --- a/internal/config/ladder_amend.go +++ b/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})?$`) ) diff --git a/internal/config/store_postgres_ladder_amend.go b/internal/config/store_postgres_ladder_amend.go index 1995a4f4..578838c0 100644 --- a/internal/config/store_postgres_ladder_amend.go +++ b/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 diff --git a/internal/config/store_postgres_ladder_amend_ceiling.go b/internal/config/store_postgres_ladder_amend_ceiling.go new file mode 100644 index 00000000..5878949c --- /dev/null +++ b/internal/config/store_postgres_ladder_amend_ceiling.go @@ -0,0 +1,70 @@ +package config + +import ( + "context" + "errors" + "fmt" + "math" + "time" + + "github.com/jackc/pgx/v5" +) + +const ladderMicros = 1e6 + +// 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 (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 + } + var anchor time.Time + var ceiling, oldF, newF float64 + 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(&anchor, &hasSnapshot, &ceiling, &oldF, &newF) + if errors.Is(err, pgx.ErrNoRows) { + return fmt.Errorf("%w: config has no non-canceled run", ErrLadderAmendInvalid) + } + if err != nil { + return err + } + if !hasSnapshot { + return fmt.Errorf("%w: latest run has no target snapshot to bound a raise", ErrLadderAmendInvalid) + } + inFlight, err := inFlightLadderCommit(ctx, tx, configID, time.Now()) + if err != nil { + return err + } + settled, err := settledLadderCommit(ctx, tx, configID, anchor) + if err != nil { + return err + } + committed := toLadderMicros(inFlight) - toLadderMicros(oldF) + toLadderMicros(newF) + toLadderMicros(settled) + if committed > toLadderMicros(ceiling) { + return ErrLadderAmendInvalid + } + return nil +} diff --git a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go new file mode 100644 index 00000000..1fee69d2 --- /dev/null +++ b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go @@ -0,0 +1,210 @@ +//go:build integration + +package config + +// Real PostgreSQL tests for the config-wide amend ceiling (#842 PR-B). + +import ( + "context" + "fmt" + "strconv" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +type ceilingFixture struct { + ctx context.Context + t *testing.T + store *PostgresStore + cfgID string + accountID string + runs int +} + +func newCeilingFixture(t *testing.T) *ceilingFixture { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + t.Cleanup(cancel) + store := setupLadderStore(ctx, t) + f := &ceilingFixture{ctx: ctx, t: t, store: store, cfgID: seedLadderConfig(ctx, t, store)} + require.NoError(t, store.db.QueryRow(ctx, "SELECT cloud_account_id FROM ladder_configs WHERE id=$1", f.cfgID).Scan(&f.accountID)) + return f +} + +func fptr(v float64) *float64 { return &v } + +// plan persists a planned run through PlanAndSaveLadderRun with the given +// snapshot (nil leaves the column NULL) and one scheduled tranche per amount. +// Runs start one minute apart, oldest first, all in the past. +func (f *ceilingFixture) plan(target, existing, baseline *float64, amounts ...float64) (string, []string) { + f.t.Helper() + f.runs++ + asOf := time.Now().UTC().Add(-time.Duration(10-f.runs) * time.Minute).Truncate(time.Microsecond) + runID := uuid.NewString() + var ids []string + _, err := f.store.PlanAndSaveLadderRun(f.ctx, f.cfgID, asOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + trs := make([]LadderTrancheDB, len(amounts)) + for i, a := range amounts { + trs[i] = lifecycleTranche(f.cfgID, a, ladder.TrancheStatusScheduled, time.Now().UTC().Add(time.Hour)) + trs[i].RunID = &runID + ids = append(ids, trs[i].ID) + } + return &LadderRunDB{ID: runID, ConfigID: &f.cfgID, StartedAt: asOf, Status: ladder.RunStatusPlanned, TotalHourlyCommit: 100, // the original_total term is not under test here + + TargetUSDHr: target, ExistingUSDHr: existing, BaselineUSDHr: baseline}, trs, nil + }) + require.NoError(f.t, err) + return runID, ids +} + +// amend sets the tranche to amount, moving its date one minute later. +func (f *ceilingFixture) amend(id string, amount float64) error { + f.t.Helper() + var rev int64 + var date time.Time + require.NoError(f.t, f.store.db.QueryRow(f.ctx, "SELECT revision,scheduled_date FROM ladder_tranches WHERE id=$1", id).Scan(&rev, &date)) + _, err := f.store.AmendLadderTranche(f.ctx, id, f.accountID, "aws", "actor", LadderAmendment{ + ExpectedRevision: &rev, ScheduledDate: date.Add(time.Minute), AmountUSDHr: strconv.FormatFloat(amount, 'f', 6, 64)}) + return err +} + +func (f *ceilingFixture) inFlight() float64 { + f.t.Helper() + v, err := inFlightLadderCommit(f.ctx, f.store.db, f.cfgID, time.Now()) + require.NoError(f.t, err) + return v +} + +func (f *ceilingFixture) move(id string, from, to ladder.TrancheStatus) { + f.t.Helper() + if from == ladder.TrancheStatusScheduled && to == ladder.TrancheStatusFired { + // Claiming is the executor's job; the test only needs the resulting row state. + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_tranches SET status='fired' WHERE id=$1", id) + require.NoError(f.t, err) + return + } + moved, err := f.store.TransitionLadderTranche(f.ctx, id, from, to, "test") + require.NoError(f.t, err) + require.True(f.t, moved) +} + +func TestLadderAmendCeiling_ThreeStepReplay(t *testing.T) { + f := newCeilingFixture(t) + _, first := f.plan(fptr(10), fptr(0), fptr(100), 5, 5) + require.NoError(t, f.amend(first[0], 1), "decrease") + require.NoError(t, f.amend(first[0], 2), "raise within headroom (7 of 10)") + require.NoError(t, f.amend(first[0], 1)) + // The planner tops up the gap: 6 in flight, 4 missing. + f.plan(fptr(10), fptr(0), fptr(100), 4) + require.Equal(t, 10.0, f.inFlight()) + require.ErrorIs(t, f.amend(first[0], 5), ErrLadderAmendInvalid, "raising A back would put 14 in flight against a 10 ceiling") + require.Equal(t, 10.0, f.inFlight()) + require.NoError(t, f.amend(first[0], 1), "re-date only passes at the ceiling") +} + +func TestLadderAmendCeiling_CompletedAfterRunCounts(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 5, 5) + f.move(ids[0], ladder.TrancheStatusScheduled, ladder.TrancheStatusFired) + require.ErrorIs(t, f.amend(ids[1], 6), ErrLadderAmendInvalid, "fired A (5) + B raised (6) > 10") + f.move(ids[0], ladder.TrancheStatusFired, ladder.TrancheStatusCompleted) + require.ErrorIs(t, f.amend(ids[1], 6), ErrLadderAmendInvalid, "A completed after the run started still counts: 5 + 6 > 10") + require.NoError(t, f.amend(ids[1], 5), "re-date only") + require.NoError(t, f.amend(ids[1], 4), "decrease") + require.ErrorIs(t, f.amend(ids[0], 5), ErrLadderAmendConflict, "a completed tranche is not amendable") +} + +func TestLadderAmendCeiling_SnapshotAndCoverage(t *testing.T) { + t.Run("lowered target never blocks decrease or re-date", func(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 5, 5) + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_runs SET target_usd_hr=4 WHERE config_id=$1", f.cfgID) + require.NoError(t, err) + require.NoError(t, f.amend(ids[0], 5), "re-date only with in-flight 10 > ceiling 4") + require.ErrorIs(t, f.amend(ids[0], 6), ErrLadderAmendInvalid) + require.NoError(t, f.amend(ids[0], 4)) + }) + t.Run("coverage lowered after the run caps a raise", func(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 1, 1) + require.NoError(t, f.amend(ids[0], 3), "80% of 100 does not bind") + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_configs SET target_coverage=4 WHERE id=$1", f.cfgID) + require.NoError(t, err) + require.ErrorIs(t, f.amend(ids[0], 4), ErrLadderAmendInvalid, "ceiling is now 4 but 5 would be in flight") + require.NoError(t, f.amend(ids[0], 2)) + }) + t.Run("gross existing lowers the ceiling", func(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(6), fptr(100), 1, 1) + require.NoError(t, f.amend(ids[0], 3)) + require.ErrorIs(t, f.amend(ids[0], 4), ErrLadderAmendInvalid) + }) + for name, snap := range map[string][3]*float64{ + "NULL target": {nil, fptr(0), fptr(100)}, "NULL existing": {fptr(10), nil, fptr(100)}, "NULL baseline": {fptr(10), fptr(0), nil}, + } { + t.Run(name+" refuses a raise only", func(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(snap[0], snap[1], snap[2], 2, 2) + require.ErrorIs(t, f.amend(ids[0], 3), ErrLadderAmendInvalid) + require.NoError(t, f.amend(ids[0], 2), "re-date only") + require.NoError(t, f.amend(ids[0], 1), "decrease") + }) + } +} + +func TestLadderAmendCeiling_LatestRunDecides(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 1, 1) + f.plan(fptr(3), fptr(0), fptr(100), 1) + require.ErrorIs(t, f.amend(ids[0], 2), ErrLadderAmendInvalid, "latest run's ceiling 3 < 4 in flight, older run's 10 must not apply") + runID, _ := f.plan(fptr(50), fptr(0), fptr(100)) + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_runs SET status='cancelled' WHERE id=$1", runID) + require.NoError(t, err) + require.ErrorIs(t, f.amend(ids[0], 2), ErrLadderAmendInvalid, "a canceled latest run is ignored") +} + +func TestLadderAmendCeiling_WaitsForPlanner(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 5, 3) + entered, gate, planned := make(chan struct{}), make(chan struct{}), make(chan error, 1) + asOf := time.Now().UTC().Truncate(time.Microsecond) + go func() { + runID := uuid.NewString() + _, err := f.store.PlanAndSaveLadderRun(f.ctx, f.cfgID, asOf, func(inFlight float64) (*LadderRunDB, []LadderTrancheDB, error) { + close(entered) + <-gate + tr := lifecycleTranche(f.cfgID, 10-inFlight, ladder.TrancheStatusScheduled, time.Now().UTC().Add(time.Hour)) + tr.RunID = &runID + return &LadderRunDB{ID: runID, ConfigID: &f.cfgID, StartedAt: asOf, Status: ladder.RunStatusPlanned, + TargetUSDHr: fptr(10), ExistingUSDHr: fptr(0), BaselineUSDHr: fptr(100)}, []LadderTrancheDB{tr}, nil + }) + planned <- err + }() + var once sync.Once + release := func() { once.Do(func() { close(gate) }) } + t.Cleanup(release) // a failed wait below must not strand the planner + <-entered + amended := make(chan error, 1) + go func() { amended <- f.amend(ids[0], 7) }() + waitForLadderWaiters(f.ctx, t, f.store, f.cfgID, 1) + release() + require.NoError(t, <-planned) + require.ErrorIs(t, <-amended, ErrLadderAmendInvalid, "amend must see the planner's committed top-up (8+2=10) and refuse +2") + require.Equal(t, 10.0, f.inFlight()) +} + +func TestLadderAmendCeiling_LockTimeoutIsTyped(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 1) + old := ladderLockTimeout + ladderLockTimeout = 200 * time.Millisecond + t.Cleanup(func() { ladderLockTimeout = old }) + release := holdLadderLock(f.ctx, t, f.store, f.cfgID) + err := f.amend(ids[0], 2) + require.ErrorIs(t, err, ErrLadderLockTimeout, fmt.Sprint(err)) + require.NoError(t, release()) +} diff --git a/internal/config/store_postgres_ladder_amend_integration_test.go b/internal/config/store_postgres_ladder_amend_integration_test.go index 642a3b0b..6e84303b 100644 --- a/internal/config/store_postgres_ladder_amend_integration_test.go +++ b/internal/config/store_postgres_ladder_amend_integration_test.go @@ -20,7 +20,7 @@ func TestLadderAmendmentAuditAndSiblingConcurrency(t *testing.T) { cfgID := seedLadderConfig(ctx, t, store) var accountID string require.NoError(t, store.db.QueryRow(ctx, "SELECT cloud_account_id FROM ladder_configs WHERE id=$1", cfgID).Scan(&accountID)) - run, err := store.SaveLadderRun(ctx, &LadderRunDB{ConfigID: &cfgID, Status: ladder.RunStatusPlanned, StartedAt: time.Now(), TotalHourlyCommit: 10, Plan: []byte(`{"original":true}`)}) + run, err := store.SaveLadderRun(ctx, &LadderRunDB{ConfigID: &cfgID, Status: ladder.RunStatusPlanned, StartedAt: time.Now(), TotalHourlyCommit: 10, TargetUSDHr: fptr(10), ExistingUSDHr: fptr(0), BaselineUSDHr: fptr(100), Plan: []byte(`{"original":true}`)}) require.NoError(t, err) date := time.Now().UTC().Add(time.Hour) ids := []string{uuid.NewString(), uuid.NewString()} diff --git a/internal/config/store_postgres_ladder_lock.go b/internal/config/store_postgres_ladder_lock.go index 77d85f8f..f6b9cafc 100644 --- a/internal/config/store_postgres_ladder_lock.go +++ b/internal/config/store_postgres_ladder_lock.go @@ -9,6 +9,7 @@ import ( "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" ) @@ -28,6 +29,10 @@ var ladderLockTimeout = 30 * time.Second // persist transaction: it must be pure compute (no provider or network I/O). type LadderPlanBuilder func(inFlightUSDHr float64) (*LadderRunDB, []LadderTrancheDB, error) +// ErrLadderLockTimeout means the per-config ladder lock was not acquired within +// ladderLockTimeout (SQLSTATE 55P03). The operation did nothing; retry later. +var ErrLadderLockTimeout = errors.New("ladder config lock timed out") + func ladderConfigLockKey(configID string) string { return ladderConfigLockPrefix + configID } // withLadderConfigLock runs fn in a READ COMMITTED transaction that holds the @@ -77,6 +82,10 @@ func acquireLadderConfigLock(ctx context.Context, tx pgx.Tx, configID string) er return fmt.Errorf("ladder config lock %s: set lock_timeout: %w", configID, err) } if _, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", ladderConfigLockKey(configID)); err != nil { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "55P03" { + return fmt.Errorf("ladder config lock %s: %w: %w", configID, ErrLadderLockTimeout, err) + } return fmt.Errorf("ladder config lock %s: %w", configID, err) } // The timeout was for the advisory wait only; do not leak it into fn's statements. From 7d5c7bf1a24592e6683fe09865c00b54b56e43b3 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 13:28:59 +0200 Subject: [PATCH 2/3] fix(ladder): anchor the amend settled term at the skew-margin anchor (#842) Co-Authored-By: Claude Sonnet 5.5 --- .../store_postgres_ladder_amend_ceiling.go | 5 ++-- ...s_ladder_amend_ceiling_integration_test.go | 30 +++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/internal/config/store_postgres_ladder_amend_ceiling.go b/internal/config/store_postgres_ladder_amend_ceiling.go index 5878949c..087f40e6 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling.go +++ b/internal/config/store_postgres_ladder_amend_ceiling.go @@ -21,7 +21,8 @@ func toLadderMicros(v float64) int64 { return int64(math.Round(v * ladderMicros) // 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 (their +// 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 @@ -58,7 +59,7 @@ func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmoun if err != nil { return err } - settled, err := settledLadderCommit(ctx, tx, configID, anchor) + settled, err := settledLadderCommit(ctx, tx, configID, ladderSettleAnchor(anchor)) if err != nil { return err } diff --git a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go index 1fee69d2..56ac97a2 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go +++ b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go @@ -208,3 +208,33 @@ func TestLadderAmendCeiling_LockTimeoutIsTyped(t *testing.T) { require.ErrorIs(t, err, ErrLadderLockTimeout, fmt.Sprint(err)) require.NoError(t, release()) } + +// settled_at is the database clock and the run's started_at the app clock. The +// offsets are literal on purpose: a margin of 0 misses the 30s case and a margin +// of an hour counts the 2m case. +func TestLadderAmendCeiling_SettledSkewMargin(t *testing.T) { + for _, c := range []struct { + name string + settledAgo time.Duration + wantRaise bool + }{ + {"settled 30s before the run started still counts", 30 * time.Second, false}, + {"settled 2m before the run started does not count", 2 * time.Minute, true}, + } { + t.Run(c.name, func(t *testing.T) { + f := newCeilingFixture(t) + runID, ids := f.plan(fptr(10), fptr(0), fptr(100), 5) + var started time.Time + require.NoError(t, f.store.db.QueryRow(f.ctx, "SELECT started_at FROM ladder_runs WHERE id=$1", runID).Scan(&started)) + done := lifecycleTranche(f.cfgID, 4, ladder.TrancheStatusCompleted, started.Add(-c.settledAgo)) + done.RunID = &runID + require.NoError(t, f.store.SaveLadderTranches(f.ctx, []LadderTrancheDB{done})) + err := f.amend(ids[0], 7) + if c.wantRaise { + require.NoError(t, err) + } else { + require.ErrorIs(t, err, ErrLadderAmendInvalid) + } + }) + } +} From 2d7ed48f951e4313c15bd64a71990dc2d2243790 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 14:09:19 +0200 Subject: [PATCH 3/3] fix(ladder): sweep overdue tranches and guard huge amounts before the amend ceiling check (#842) Co-Authored-By: Claude Sonnet 5.5 --- .../store_postgres_ladder_amend_ceiling.go | 68 +++++++++++++------ ...s_ladder_amend_ceiling_integration_test.go | 41 ++++++++++- 2 files changed, 89 insertions(+), 20 deletions(-) diff --git a/internal/config/store_postgres_ladder_amend_ceiling.go b/internal/config/store_postgres_ladder_amend_ceiling.go index 087f40e6..4ed012d9 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling.go +++ b/internal/config/store_postgres_ladder_amend_ceiling.go @@ -12,6 +12,9 @@ import ( 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)) } @@ -36,36 +39,63 @@ func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmoun if !raise { return nil } - var anchor time.Time - var ceiling, oldF, newF float64 - 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(&anchor, &hasSnapshot, &ceiling, &oldF, &newF) - if errors.Is(err, pgx.ErrNoRows) { - return fmt.Errorf("%w: config has no non-canceled run", ErrLadderAmendInvalid) - } + snap, err := loadLadderCeiling(ctx, tx, configID, oldAmount, newAmount) if err != nil { return err } - if !hasSnapshot { - return fmt.Errorf("%w: latest run has no target snapshot to bound a raise", ErrLadderAmendInvalid) + // 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, time.Now()) + inFlight, err := inFlightLadderCommit(ctx, tx, configID, now) if err != nil { return err } - settled, err := settledLadderCommit(ctx, tx, configID, ladderSettleAnchor(anchor)) + settled, err := settledLadderCommit(ctx, tx, configID, ladderSettleAnchor(snap.anchor)) if err != nil { return err } - committed := toLadderMicros(inFlight) - toLadderMicros(oldF) + toLadderMicros(newF) + toLadderMicros(settled) - if committed > toLadderMicros(ceiling) { + 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 +} diff --git a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go index 56ac97a2..bf075766 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go +++ b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go @@ -14,6 +14,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/stretchr/testify/require" ) @@ -203,7 +204,11 @@ func TestLadderAmendCeiling_LockTimeoutIsTyped(t *testing.T) { old := ladderLockTimeout ladderLockTimeout = 200 * time.Millisecond t.Cleanup(func() { ladderLockTimeout = old }) - release := holdLadderLock(f.ctx, t, f.store, f.cfgID) + holder := holdLadderLock(f.ctx, t, f.store, f.cfgID) + var once sync.Once + var holderErr error + release := func() error { once.Do(func() { holderErr = holder() }); return holderErr } + t.Cleanup(func() { _ = release() }) // a failed assertion must not leave the holder blocked err := f.amend(ids[0], 2) require.ErrorIs(t, err, ErrLadderLockTimeout, fmt.Sprint(err)) require.NoError(t, release()) @@ -238,3 +243,37 @@ func TestLadderAmendCeiling_SettledSkewMargin(t *testing.T) { }) } } + +func TestLadderAmendCeiling_CancelsOverdueBeforeCounting(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 5, 5) + // X is past the 24h grace window: neither counted in flight nor claimable at "now". + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_tranches SET scheduled_date=now()-interval '25 hours' WHERE id=$1", ids[0]) + require.NoError(t, err) + require.NoError(t, f.amend(ids[1], 10), "X is not in flight, so 10 fits the ceiling") + var status ladder.TrancheStatus + require.NoError(t, f.store.db.QueryRow(f.ctx, "SELECT status FROM ladder_tranches WHERE id=$1", ids[0]).Scan(&status)) + require.Equal(t, ladder.TrancheStatusCancelled, status, "the overdue row must be canceled, not left for a claim with an older asOf") + claimed, err := f.store.ClaimDueLadderTranches(f.ctx, f.cfgID, time.Now().Add(-time.Minute), 10, + func(context.Context, pgx.Tx, LadderTrancheDB) (string, error) { return uuid.NewString(), nil }) + require.NoError(t, err) + require.Empty(t, claimed) +} + +func TestLadderAmendCeiling_CeilingTruncatesNotRounds(t *testing.T) { + f := newCeilingFixture(t) + // 1.000015 * 33.33% = 0.3333049995: truncated 0.333304, rounded 0.333305. + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_configs SET target_coverage=33.33 WHERE id=$1", f.cfgID) + require.NoError(t, err) + _, ids := f.plan(fptr(10), fptr(0), fptr(1.000015), 0.1) + require.ErrorIs(t, f.amend(ids[0], 0.333305), ErrLadderAmendInvalid) + require.NoError(t, f.amend(ids[0], 0.333304)) +} + +func TestLadderAmendCeiling_HugeAmountRefused(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(1e13), fptr(0), fptr(9e13), 1) + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_runs SET total_hourly_commit=99999999999999 WHERE config_id=$1", f.cfgID) + require.NoError(t, err) // let the original-total term pass so only the guard can refuse + require.ErrorIs(t, f.amend(ids[0], 9999999999999), ErrLadderAmendInvalid) +}