From 2952deb0b3fa1b0ace1fc0d8732d3cf16d959f4d Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 16 Oct 2026 13:14:32 +0200 Subject: [PATCH] fix(ladder): convert amend ceiling amounts by the amended tranche's own basis The config-wide amend ceiling is in on-demand-equivalent dollars, but the amended tranche's old and new amounts were subtracted and added as stored numbers. A raise on a commitment-basis row (rate 0.5) could therefore pass while its on-demand equivalent exceeded the ceiling. Old and new amounts now convert by that row's basis and rate, and the comparison is exact in big.Rat. A NULL basis (rows from before 000121, left unknown by #886) is assumed on-demand-equivalent on this path, as before; netting still fails loud. Records the open risk that a future executor could claim a legacy on-demand row as commitment dollars. Refs #799 Refs LeanerCloud/cloud-commitments-go#349 Co-Authored-By: Claude Opus 5.5 --- .../config/store_postgres_ladder_amend.go | 8 ++- .../store_postgres_ladder_amend_ceiling.go | 69 +++++++++++-------- ..._postgres_ladder_basis_integration_test.go | 31 +++++++++ known_issues/05_config_store_postgres.md | 7 ++ 4 files changed, 85 insertions(+), 30 deletions(-) diff --git a/internal/config/store_postgres_ladder_amend.go b/internal/config/store_postgres_ladder_amend.go index 578838c0b..1a4981be4 100644 --- a/internal/config/store_postgres_ladder_amend.go +++ b/internal/config/store_postgres_ladder_amend.go @@ -56,6 +56,8 @@ type lockedLadderTranche struct { executionID *string revision int64 amount string + basis *LadderAmountBasis + rate *string date time.Time future bool } @@ -96,7 +98,7 @@ func amendLadderTrancheTx(ctx context.Context, tx pgx.Tx, configID, id, accountI if err != nil { return err } - err = checkLadderConfigCeiling(ctx, tx, run.configID, tranche.amount, amendment.AmountUSDHr) + err = checkLadderConfigCeiling(ctx, tx, run.configID, tranche, amendment.AmountUSDHr) if err != nil { return err } @@ -140,10 +142,10 @@ func lockLadderRun(ctx context.Context, tx pgx.Tx, id, accountID, provider strin func lockLadderTranche(ctx context.Context, tx pgx.Tx, id string, run lockedLadderRun, proposed time.Time) (lockedLadderTranche, error) { var t lockedLadderTranche - err := tx.QueryRow(ctx, `SELECT status,execution_id,revision,amount_usd_hr::text,scheduled_date, + err := tx.QueryRow(ctx, `SELECT status,execution_id,revision,amount_usd_hr::text,amount_basis,commitment_rate,scheduled_date, scheduled_date>clock_timestamp() AND $4::timestamptz>clock_timestamp() FROM ladder_tranches WHERE id=$1 AND config_id=$2 AND run_id=$3 FOR UPDATE`, id, run.configID, run.runID, proposed). - Scan(&t.status, &t.executionID, &t.revision, &t.amount, &t.date, &t.future) + Scan(&t.status, &t.executionID, &t.revision, &t.amount, &t.basis, &t.rate, &t.date, &t.future) if errors.Is(err, pgx.ErrNoRows) { return t, ErrLadderAmendNotFound } diff --git a/internal/config/store_postgres_ladder_amend_ceiling.go b/internal/config/store_postgres_ladder_amend_ceiling.go index a87df6880..125b3357f 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling.go +++ b/internal/config/store_postgres_ladder_amend_ceiling.go @@ -5,12 +5,13 @@ import ( "errors" "fmt" "math" + "math/big" "time" "github.com/jackc/pgx/v5" ) -const ladderMicros = 1e6 +const ladderMicros = 1_000_000 // ladderMaxUSDHr keeps micro-dollar sums far below int64 overflow (about 9.2e12 USD/hr). const ladderMaxUSDHr = 1e12 @@ -26,20 +27,22 @@ func toLadderMicros(v float64) int64 { return int64(math.Round(v * ladderMicros) // 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). +// commitment may not be in the run's existing_usd_hr snapshot yet). Every term +// is netted per row in on-demand-equivalent dollars, the unit of the ceiling; +// the amended row's old and new amounts convert by its own basis and rate. // 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 { +func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID string, amended lockedLadderTranche, newAmount string) error { var raise bool - if err := tx.QueryRow(ctx, `SELECT $1::numeric>$2::numeric`, newAmount, oldAmount).Scan(&raise); err != nil { + if err := tx.QueryRow(ctx, `SELECT $1::numeric>$2::numeric`, newAmount, amended.amount).Scan(&raise); err != nil { return err } if !raise { return nil } - snap, err := loadLadderCeiling(ctx, tx, configID, oldAmount, newAmount) + snap, err := loadLadderCeiling(ctx, tx, configID) if err != nil { return err } @@ -51,57 +54,69 @@ func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmoun if err != nil { return err } - inFlight, settled, err := ladderCommittedOnDemand(ctx, tx, configID, now, ladderSettleAnchor(snap.anchor)) + committed, err := ladderCommittedWithAmend(ctx, tx, configID, now, ladderSettleAnchor(snap.anchor), amended, newAmount) 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) - } + if snap.ceiling > 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) { + if committed.Cmp(big.NewRat(toLadderMicros(snap.ceiling), ladderMicros)) > 0 { return ErrLadderAmendInvalid } return nil } -// ladderCommittedOnDemand nets the in-flight and settled-after-anchor rows to -// on-demand-equivalent dollars, assuming on-demand for rows with no stored basis. -func ladderCommittedOnDemand(ctx context.Context, tx pgx.Tx, configID string, now, anchor time.Time) (inFlight, settled float64, err error) { +// ladderCommittedWithAmend is the on-demand-equivalent committed total with the +// amended row swapped from its old to its new amount, each converted by that +// row's own basis and rate. The rows still hold the old amount. +func ladderCommittedWithAmend(ctx context.Context, tx pgx.Tx, configID string, now, anchor time.Time, amended lockedLadderTranche, newAmount string) (*big.Rat, error) { inFlightRows, err := inFlightLadderRows(ctx, tx, true, configID, now) if err != nil { - return 0, 0, err + return nil, err } settledRows, err := settledLadderRows(ctx, tx, true, configID, anchor) if err != nil { - return 0, 0, err + return nil, err + } + committed, err := sumLadderOnDemand(inFlightRows, settledRows) + if err != nil { + return nil, err + } + tr := assumeOnDemandIfUnknown(&LadderTrancheDB{AmountBasis: amended.basis, CommitmentRate: amended.rate}) + oldOD, err := nettingOnDemand(amended.amount, tr) + if err != nil { + return nil, err + } + newOD, err := nettingOnDemand(newAmount, tr) + if err != nil { + return nil, err } - if inFlight, err = LadderInFlightOnDemand(inFlightRows); err != nil { - return 0, 0, err + maxOD := new(big.Rat).SetInt64(ladderMaxUSDHr) + for _, v := range []*big.Rat{committed, oldOD, newOD} { + if v.Cmp(maxOD) > 0 { + return nil, fmt.Errorf("%w: amount beyond %.0f USD/hr", ErrLadderAmendInvalid, ladderMaxUSDHr) + } } - settled, err = LadderInFlightOnDemand(settledRows) - return inFlight, settled, err + return committed.Sub(committed, oldOD).Add(committed, newOD), nil } type ladderCeilingSnapshot struct { - anchor time.Time - ceiling, oldAmount, newAmount float64 + anchor time.Time + ceiling 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) { +func loadLadderCeiling(ctx context.Context, tx pgx.Tx, configID 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 + COALESCE(trunc(LEAST(r.target_usd_hr,r.baseline_usd_hr*c.target_coverage/100)-r.existing_usd_hr,6),0)::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) + configID).Scan(&snap.anchor, &hasSnapshot, &snap.ceiling) if errors.Is(err, pgx.ErrNoRows) { return snap, fmt.Errorf("%w: config has no non-canceled run", ErrLadderAmendInvalid) } diff --git a/internal/config/store_postgres_ladder_basis_integration_test.go b/internal/config/store_postgres_ladder_basis_integration_test.go index 6a861ac15..ec8a9f472 100644 --- a/internal/config/store_postgres_ladder_basis_integration_test.go +++ b/internal/config/store_postgres_ladder_basis_integration_test.go @@ -29,6 +29,37 @@ func (f *ceilingFixture) setBasis(id, basis, rate string) { require.NoError(f.t, err) } +func TestLadderAmendCeiling_RaiseOnCommitmentRowNetsInOnDemand(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 4) + f.setBasis(ids[0], "commitment", "1/2") + require.ErrorIs(t, f.amend(ids[0], 6), ErrLadderAmendInvalid, "6 commitment at rate 1/2 is 12 on-demand against a ceiling of 10") + require.NoError(t, f.amend(ids[0], 5), "5 commitment at rate 1/2 is exactly the ceiling of 10") +} + +func TestLadderAmendCeiling_NullBasisIsAssumedOnDemand(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 5, 5) + f.setBasis(ids[0], "", "") + f.setBasis(ids[1], "", "") + require.NoError(t, f.amend(ids[0], 4)) + require.ErrorIs(t, f.amend(ids[0], 6), ErrLadderAmendInvalid, "assumed on-demand: 6+5 > 10") + require.NoError(t, f.amend(ids[0], 5)) +} + +func TestLadderAmendRunTotal_SingleCommitmentBasisRun(t *testing.T) { + f := newCeilingFixture(t) + runID, ids := f.plan(fptr(100), fptr(0), fptr(100), 4, 4) + for _, id := range ids { + f.setBasis(id, "commitment", "1/2") + } + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_runs SET total_hourly_commit=8 WHERE id=$1", runID) + require.NoError(t, err) + require.ErrorIs(t, f.amend(ids[0], 5), ErrLadderAmendInvalid, "run total 9 > original 8") + require.NoError(t, f.amend(ids[0], 3)) + require.NoError(t, f.amend(ids[1], 5), "run total back to 8") +} + func seedNettingRows(t *testing.T, f *ceilingFixture, basis LadderAmountBasis, rate string, amounts ...string) { t.Helper() for _, a := range amounts { diff --git a/known_issues/05_config_store_postgres.md b/known_issues/05_config_store_postgres.md index 004323dbe..a301acddc 100644 --- a/known_issues/05_config_store_postgres.md +++ b/known_issues/05_config_store_postgres.md @@ -128,3 +128,10 @@ **Resolved by:** `internal/api/db_rate_limiter.go::Allow` no longer issues a `SELECT ... FOR UPDATE` followed by a separate INSERT/UPDATE. Replaced with a single atomic `INSERT ... ON CONFLICT (id) DO UPDATE` that uses `CASE WHEN rate_limits.reset_time < $now THEN 1 ELSE rate_limits.count + 1 END` to handle both the fresh-insert / window-expired-reset / increment branches in one statement, with `RETURNING count, reset_time` driving the allow/deny decision. The dead `rateLimitOutcome` and `execAndCommit` helpers were removed. Three new integration tests in `internal/api/db_rate_limiter_integration_test.go` (build tag `integration`): `TestDBRateLimiter_ConcurrentFirstRequest` (20 goroutines hit the same id, all allowed, no SQLSTATE 23505), `TestDBRateLimiter_WindowExpiry_AtomicReset` (pre-seeded past reset_time → next call resets count to 1), `TestDBRateLimiter_ExceedsLimitDenies` (sanity: 3-allow + 2-deny burst with MaxAttempts=3). Documented trade-off: under sustained over-limit traffic, `count` may temporarily drift past MaxAttempts because the upsert always runs; the limiter still denies correctly and `cleanup()` evicts expired rows on its 24h cycle. **Implementation plan:** replace the SELECT+INSERT with a single atomic `INSERT ... ON CONFLICT (id) DO UPDATE SET count = CASE WHEN reset_time < $now THEN 1 ELSE count + 1 END, reset_time = CASE WHEN reset_time < $now THEN $newResetTime ELSE reset_time END, updated_at = $now RETURNING count, reset_time`. Drop the `queryErr/count/existingResetTime` arguments from `rateLimitOutcome`. Tests: `TestDBRateLimiter_ConcurrentFirstRequest` (20 goroutines hit same id, assert no SQLSTATE 23505 + final count = 20) and `TestDBRateLimiter_WindowExpiry_AtomicReset` (pre-seed past reset_time, single call resets count to 1). + +## MEDIUM: nothing stops a future ladder executor from claiming a legacy on-demand row + +**File**: `internal/config/store_postgres_ladder_lifecycle.go`, `ClaimDueLadderTranches` +**Description**: Legacy `on_demand_equivalent` tranches net at rate 1 and the amend ceiling assumes NULL-basis rows are on-demand (#886 left them unknown). The claim query does not filter by basis, so an executor that buys `amount_usd_hr` as commitment dollars would overbuy a legacy row. No executor exists yet. +**Impact**: None today; blocks executor enablement. Track on #946 and the executor issue. +**Status:** Open