From 6af64138035bec57a46d37dc915b68eebe72085f Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 22 Oct 2026 01:36:54 +0100 Subject: [PATCH] fix(ladder): convert existing coverage to on-demand-equivalent dollars and never claim legacy tranches The AWS provider reports LayerState.ExistingUSDPerHour and ExpiringUSDPerHour in commitment dollars (SavingsPlan.Commitment for the SP layers, amortized cost for the RI layer), but Allocate compares them with an on-demand-equivalent target. The planner passed them through unconverted, so existing coverage was understated by the commitment rate and the gap overstated: a 2.5 commitment $/hr EC2 Instance SP at rate 0.5 was netted as 2.5 OD instead of 5. executeLadderRun now converts each layer with od = commitment / rate, using the GetCommitmentRates rate the run already fetched (option 2 on the issue). Existing rounds up and expiring down, so net coverage is never understated. A layer with non-zero existing or expiring commitment and no rate fails the run instead of being netted 1:1. ladder_runs.existing_usd_hr and the plan JSON existing are now on-demand-equivalent. Per the maintainer's note on the issue, ClaimDueLadderTranches now claims only amount_basis = 'commitment' rows. Legacy NULL-basis and on_demand_equivalent rows stay netted until the overdue sweep cancels them, so an executor can never buy their OD amount as commitment dollars. Closes #946 Co-Authored-By: Claude Opus 5.5 --- ...ladder_reconcile_retry_integration_test.go | 3 +- ...s_ladder_amend_ceiling_integration_test.go | 3 + .../config/store_postgres_ladder_lifecycle.go | 8 + ...tgres_ladder_lifecycle_integration_test.go | 82 ++++++-- ...e_postgres_ladder_lock_integration_test.go | 6 +- ...ostgres_ladder_resolve_integration_test.go | 2 +- internal/config/store_postgres_ladder_test.go | 17 +- .../ladder_reconcile_integration_test.go | 4 +- internal/server/handler_ladder.go | 19 +- internal/server/ladder_existing_units.go | 131 +++++++++++++ internal/server/ladder_existing_units_test.go | 180 ++++++++++++++++++ known_issues/05_config_store_postgres.md | 6 +- 12 files changed, 431 insertions(+), 30 deletions(-) create mode 100644 internal/server/ladder_existing_units.go create mode 100644 internal/server/ladder_existing_units_test.go diff --git a/internal/api/handler_ladder_reconcile_retry_integration_test.go b/internal/api/handler_ladder_reconcile_retry_integration_test.go index 938950768..881c2e72a 100644 --- a/internal/api/handler_ladder_reconcile_retry_integration_test.go +++ b/internal/api/handler_ladder_reconcile_retry_integration_test.go @@ -70,9 +70,10 @@ func newLadderRetryRig(t *testing.T) *ladderRetryRig { // tranche id as its idempotency key, the link Retry uses to find the tranche. func (r *ladderRetryRig) addFiredTranche(t *testing.T) { ctx := context.Background() + basis, rate := config.LadderAmountCommitment, "1" // only commitment-dollar rows are claimable tr := config.LadderTrancheDB{ID: uuid.New().String(), ConfigID: &r.cfgID, LayerType: ladder.LayerComputeSP, Term: ladder.Term1Year, PaymentOption: ladder.PaymentNoUpfront, Status: ladder.TrancheStatusScheduled, - AmountUSDHr: 5, ScheduledDate: time.Now().Add(-time.Hour)} + AmountUSDHr: 5, ScheduledDate: time.Now().Add(-time.Hour), AmountBasis: &basis, CommitmentRate: &rate} require.NoError(t, r.store.SaveLadderTranches(ctx, []config.LadderTrancheDB{tr})) var failedID string _, err := r.store.ClaimDueLadderTranches(ctx, r.cfgID, time.Now(), 1, 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 b6134dde6..4b841e4eb 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go +++ b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go @@ -257,6 +257,9 @@ 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) + // Commitment basis at rate 1 (same netting) so only the cancel, not the + // claim's basis filter, keeps X from being claimed below. + f.setBasis(ids[0], "commitment", "1") // 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) diff --git a/internal/config/store_postgres_ladder_lifecycle.go b/internal/config/store_postgres_ladder_lifecycle.go index 3c4bf1d28..7fd703640 100644 --- a/internal/config/store_postgres_ladder_lifecycle.go +++ b/internal/config/store_postgres_ladder_lifecycle.go @@ -100,6 +100,13 @@ func (s *PostgresStore) TransitionLadderTrancheTx(ctx context.Context, tx pgx.Tx // than the planner's, a claim could otherwise fire a row the planner's netting // read had already excluded as expired, and the row would be bought AND // re-planned. A callback slower than ladderLockTimeout makes planners fail loudly. +// +// Only rows stored in commitment dollars (amount_basis = 'commitment', which +// the table constraint pairs with a commitment_rate) are claimable: a buyer +// spends amount_usd_hr as commitment dollars, and a legacy NULL-basis or +// on_demand_equivalent row would over-buy by 1/rate (#946). Those rows stay +// netted until the overdue sweep cancels them, and the next run re-plans +// their gap in commitment dollars. func (s *PostgresStore) ClaimDueLadderTranches( ctx context.Context, configID string, @@ -150,6 +157,7 @@ func selectDueLadderTranchesTx(ctx context.Context, tx pgx.Tx, configID string, WHERE config_id = $1 AND status = 'scheduled' AND execution_id IS NULL + AND amount_basis = 'commitment' AND scheduled_date <= $2 AND scheduled_date > $3 ORDER BY scheduled_date, id diff --git a/internal/config/store_postgres_ladder_lifecycle_integration_test.go b/internal/config/store_postgres_ladder_lifecycle_integration_test.go index 2d74ba025..ae237fff8 100644 --- a/internal/config/store_postgres_ladder_lifecycle_integration_test.go +++ b/internal/config/store_postgres_ladder_lifecycle_integration_test.go @@ -34,6 +34,16 @@ func lifecycleTranche(cfgID string, amount float64, status ladder.TrancheStatus, return tr } +// claimableTranche is a scheduled lifecycleTranche planned in commitment +// dollars, the only basis ClaimDueLadderTranches fires (#946). Rate 1 keeps +// its on-demand-equivalent netting equal to its amount. +func claimableTranche(cfgID string, amount float64, at time.Time) LadderTrancheDB { + tr := lifecycleTranche(cfgID, amount, ladder.TrancheStatusScheduled, at) + basis, rate := LadderAmountCommitment, "1" + tr.AmountBasis, tr.CommitmentRate = &basis, &rate + return tr +} + func trancheStatus(ctx context.Context, t *testing.T, store *PostgresStore, id string) ladder.TrancheStatus { t.Helper() var st string @@ -134,11 +144,11 @@ func TestClaimDueLadderTranches_ClaimsOnlyDueUnexpiredAndKeepsNetting(t *testing var calls atomic.Int64 create := newExecCreator(ctx, t, store, &calls) - due := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, asOf.Add(-time.Hour)) - dueNow := lifecycleTranche(cfgID, 2, ladder.TrancheStatusScheduled, asOf) - future := lifecycleTranche(cfgID, 4, ladder.TrancheStatusScheduled, asOf.Add(time.Microsecond)) - expired := lifecycleTranche(cfgID, 8, ladder.TrancheStatusScheduled, asOf.Add(-LadderFireGrace)) - otherCfg := lifecycleTranche(otherID, 16, ladder.TrancheStatusScheduled, asOf.Add(-time.Hour)) + due := claimableTranche(cfgID, 1, asOf.Add(-time.Hour)) + dueNow := claimableTranche(cfgID, 2, asOf) + future := claimableTranche(cfgID, 4, asOf.Add(time.Microsecond)) + expired := claimableTranche(cfgID, 8, asOf.Add(-LadderFireGrace)) + otherCfg := claimableTranche(otherID, 16, asOf.Add(-time.Hour)) require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{due, dueNow, future, expired, otherCfg})) before := inFlight(ctx, t, store, cfgID, asOf) assert.Equal(t, 1.0+2+4, before, "expired row is not netted") @@ -163,8 +173,8 @@ func TestClaimDueLadderTranches_ClaimsOnlyDueUnexpiredAndKeepsNetting(t *testing require.NoError(t, err) assert.Empty(t, again, "a second claim finds nothing: idempotent") - limited := lifecycleTranche(cfgID, 32, ladder.TrancheStatusScheduled, asOf.Add(-time.Minute)) - limited2 := lifecycleTranche(cfgID, 64, ladder.TrancheStatusScheduled, asOf.Add(-2*time.Minute)) + limited := claimableTranche(cfgID, 32, asOf.Add(-time.Minute)) + limited2 := claimableTranche(cfgID, 64, asOf.Add(-2*time.Minute)) require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{limited, limited2})) one, err := store.ClaimDueLadderTranches(ctx, cfgID, asOf, 1, create) require.NoError(t, err) @@ -175,6 +185,50 @@ func TestClaimDueLadderTranches_ClaimsOnlyDueUnexpiredAndKeepsNetting(t *testing require.Error(t, err) } +// Legacy rows (NULL basis from before migration 000121, or +// on_demand_equivalent) hold on-demand-equivalent dollars, but an executor buys +// amount_usd_hr as commitment dollars, so firing one would over-buy by 1/rate. +// They are never claimed, even ahead of a due commitment row under a limit; +// they stay netted while within grace, and the overdue sweep cancels them +// after it, so the next run re-plans their gap in commitment dollars (#946). +func TestClaimDueLadderTranches_NeverClaimsOnDemandOrUnknownBasisRows(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + cfgID := seedLadderConfigWithExtID(ctx, t, store, "688888888888") + asOf := time.Now().UTC().Truncate(time.Microsecond) + var calls atomic.Int64 + create := newExecCreator(ctx, t, store, &calls) + + unknown := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, asOf.Add(-3*time.Hour)) + od := lifecycleTranche(cfgID, 2, ladder.TrancheStatusScheduled, asOf.Add(-2*time.Hour)) + odBasis := LadderAmountOnDemand + od.AmountBasis = &odBasis + native := claimableTranche(cfgID, 4, asOf.Add(-time.Hour)) + half := "1/2" + native.CommitmentRate = &half + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{unknown, od, native})) + assert.Equal(t, 1.0+2+8, inFlight(ctx, t, store, cfgID, asOf), "4 commitment at rate 1/2 nets as 8 on-demand-equivalent") + + claimed, err := store.ClaimDueLadderTranches(ctx, cfgID, asOf, 1, create) + require.NoError(t, err) + require.Len(t, claimed, 1, "the older legacy rows must not use up the limit") + assert.Equal(t, native.ID, claimed[0].ID) + again, err := store.ClaimDueLadderTranches(ctx, cfgID, asOf, 10, create) + require.NoError(t, err) + assert.Empty(t, again, "legacy rows are never claimable") + assert.Equal(t, int64(1), calls.Load(), "no execution row for a legacy tranche") + assert.Equal(t, ladder.TrancheStatusScheduled, trancheStatus(ctx, t, store, unknown.ID)) + assert.Equal(t, ladder.TrancheStatusScheduled, trancheStatus(ctx, t, store, od.ID)) + assert.Equal(t, 1.0+2+8, inFlight(ctx, t, store, cfgID, asOf), "legacy rows stay netted within grace") + + later := asOf.Add(LadderFireGrace) + _, err = store.SaveLadderRunWithTranches(ctx, &LadderRunDB{ConfigID: &cfgID, StartedAt: later, Status: ladder.RunStatusPlanned}, nil) + require.NoError(t, err) + assert.Equal(t, ladder.TrancheStatusCancelled, trancheStatus(ctx, t, store, unknown.ID), "past grace the sweep cancels it") + assert.Equal(t, ladder.TrancheStatusCancelled, trancheStatus(ctx, t, store, od.ID), "past grace the sweep cancels it") +} + func TestClaimDueLadderTranches_CallbackFailureRollsBackClaimAndExecution(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) defer cancel() @@ -183,8 +237,8 @@ func TestClaimDueLadderTranches_CallbackFailureRollsBackClaimAndExecution(t *tes asOf := time.Now().UTC().Truncate(time.Microsecond) var calls atomic.Int64 create := newExecCreator(ctx, t, store, &calls) - first := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, asOf.Add(-2*time.Hour)) - second := lifecycleTranche(cfgID, 2, ladder.TrancheStatusScheduled, asOf.Add(-time.Hour)) + first := claimableTranche(cfgID, 1, asOf.Add(-2*time.Hour)) + second := claimableTranche(cfgID, 2, asOf.Add(-time.Hour)) require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{first, second})) var execsBefore int require.NoError(t, store.db.QueryRow(ctx, `SELECT count(*) FROM purchase_executions`).Scan(&execsBefore)) @@ -220,10 +274,10 @@ func TestTransitionLadderTranche_EveryLegalEdgeAndCompareAndSet(t *testing.T) { create := newExecCreator(ctx, t, store, &calls) // scheduled -> canceled, once. - canceled := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, asOf.Add(time.Hour)) + canceled := claimableTranche(cfgID, 1, asOf.Add(time.Hour)) // scheduled -> fired (claim) -> completed / failed. - toCompleted := lifecycleTranche(cfgID, 2, ladder.TrancheStatusScheduled, asOf.Add(-time.Hour)) - toFailed := lifecycleTranche(cfgID, 4, ladder.TrancheStatusScheduled, asOf.Add(-time.Hour)) + toCompleted := claimableTranche(cfgID, 2, asOf.Add(-time.Hour)) + toFailed := claimableTranche(cfgID, 4, asOf.Add(-time.Hour)) require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{canceled, toCompleted, toFailed})) require.Equal(t, 7.0, inFlight(ctx, t, store, cfgID, asOf)) @@ -273,7 +327,7 @@ func TestClaimDueLadderTranches_ConcurrentClaimersExactlyOneWinner(t *testing.T) asOf := time.Now().UTC().Truncate(time.Microsecond) var calls atomic.Int64 create := newExecCreator(ctx, t, store, &calls) - tr := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, asOf.Add(-time.Hour)) + tr := claimableTranche(cfgID, 1, asOf.Add(-time.Hour)) require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) const n = 8 @@ -317,7 +371,7 @@ func TestClaimVsCancelAndOverdueSweep_ExactlyOneOutcome(t *testing.T) { create := newExecCreator(ctx, t, store, &calls) for i := 0; i < 25; i++ { - tr := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, asOf.Add(-time.Hour)) + tr := claimableTranche(cfgID, 1, asOf.Add(-time.Hour)) require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) var claimWon, cancelWon bool var claimErr, cancelErr, sweepErr error diff --git a/internal/config/store_postgres_ladder_lock_integration_test.go b/internal/config/store_postgres_ladder_lock_integration_test.go index 92546b6be..36ef41cdb 100644 --- a/internal/config/store_postgres_ladder_lock_integration_test.go +++ b/internal/config/store_postgres_ladder_lock_integration_test.go @@ -81,9 +81,7 @@ func TestLadderLock_ClaimWaitsForPlanner(t *testing.T) { defer cancel() store := setupLadderStore(ctx, t) cfgID := seedLadderConfigWithExtID(ctx, t, store, "711111111111") - tr := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, time.Now().UTC().Add(-time.Hour)) - basis := LadderAmountOnDemand - tr.AmountBasis = &basis + tr := claimableTranche(cfgID, 1, time.Now().UTC().Add(-time.Hour)) require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) release := holdLadderLock(ctx, t, store, cfgID) @@ -226,7 +224,7 @@ func TestLadderLock_TimeoutBoundsOnlyTheWaitAndIsolationIsReadCommitted(t *testi pool.Reset() // new connections pick up the database default past := time.Now().UTC().Add(-time.Hour) - require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, past)})) + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{claimableTranche(cfgID, 1, past)})) inner := newExecCreator(ctx, t, store, new(atomic.Int64)) var lockTimeout, isolation string _, err = store.ClaimDueLadderTranches(ctx, cfgID, time.Now().UTC(), 1, func(c context.Context, tx pgx.Tx, tr LadderTrancheDB) (string, error) { diff --git a/internal/config/store_postgres_ladder_resolve_integration_test.go b/internal/config/store_postgres_ladder_resolve_integration_test.go index b16b9d966..c1cf301bf 100644 --- a/internal/config/store_postgres_ladder_resolve_integration_test.go +++ b/internal/config/store_postgres_ladder_resolve_integration_test.go @@ -37,7 +37,7 @@ func newResolveRig(ctx context.Context, t *testing.T, store *PostgresStore, node cfgID := seedLadderConfigWithExtID(ctx, t, store, ext) r := &resolveRig{store: store, external: ext} require.NoError(t, store.db.QueryRow(ctx, `SELECT cloud_account_id::text FROM ladder_configs WHERE id = $1`, cfgID).Scan(&r.accountID)) - tr := lifecycleTranche(cfgID, 5, ladder.TrancheStatusScheduled, time.Now().Add(-time.Hour)) + tr := claimableTranche(cfgID, 5, time.Now().Add(-time.Hour)) require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) r.trancheID = tr.ID var ids []string diff --git a/internal/config/store_postgres_ladder_test.go b/internal/config/store_postgres_ladder_test.go index 814747ebc..0fa027dad 100644 --- a/internal/config/store_postgres_ladder_test.go +++ b/internal/config/store_postgres_ladder_test.go @@ -84,8 +84,21 @@ func TestLadderTrancheProvenance_InsertClaimAndReconcile(t *testing.T) { return create(ctx, tx, tr) }) require.NoError(t, err) - require.Len(t, claimed, len(rows)) - require.Equal(t, int64(3), calls.Load()) + // Only the commitment-dollar row is claimable (#946); fire the legacy NULL + // and on-demand rows directly so the fired-row reader still sees every basis. + require.Len(t, claimed, 1) + require.Equal(t, rows[2].ID, claimed[0].ID) + require.Equal(t, int64(1), calls.Load()) + for _, legacy := range rows[:2] { + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + execID, createErr := create(ctx, tx, legacy) + if createErr != nil { + return createErr + } + _, execErr := tx.Exec(ctx, `UPDATE ladder_tranches SET status = 'fired', execution_id = $2 WHERE id = $1`, legacy.ID, execID) + return execErr + })) + } fired, err := store.ListFiredLadderTranches(ctx, 10) require.NoError(t, err) require.Len(t, fired, len(rows)) diff --git a/internal/purchase/ladder_reconcile_integration_test.go b/internal/purchase/ladder_reconcile_integration_test.go index 36cbe15a1..595cd2740 100644 --- a/internal/purchase/ladder_reconcile_integration_test.go +++ b/internal/purchase/ladder_reconcile_integration_test.go @@ -72,10 +72,10 @@ func (r *ladderRig) link(from, to string) { // firedTranche claims a real scheduled tranche with a real execution row. func (r *ladderRig) firedTranche(headStatus string, headProof bool, age time.Duration) (trancheID, headExec string) { - basis := config.LadderAmountOnDemand + basis, rate := config.LadderAmountCommitment, "1" // only commitment-dollar rows are claimable tr := config.LadderTrancheDB{ID: uuid.New().String(), ConfigID: &r.cfgID, LayerType: ladder.LayerComputeSP, Term: ladder.Term1Year, PaymentOption: ladder.PaymentNoUpfront, Status: ladder.TrancheStatusScheduled, - AmountUSDHr: 5, ScheduledDate: time.Now().Add(-time.Hour), AmountBasis: &basis} + AmountUSDHr: 5, ScheduledDate: time.Now().Add(-time.Hour), AmountBasis: &basis, CommitmentRate: &rate} require.NoError(r.t, r.store.SaveLadderTranches(r.ctx, []config.LadderTrancheDB{tr})) claimed, err := r.store.ClaimDueLadderTranches(r.ctx, r.cfgID, time.Now(), 10, func(ctx context.Context, tx pgx.Tx, _ config.LadderTrancheDB) (string, error) { diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go index 163381334..b365c5fbc 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -278,8 +278,10 @@ func (app *Application) processOneLadderConfig( // have not yet been bought. Tranches completed after `now` are counted too (see // below): a safe under-buy, and a deliberate deviation from go's InFlight doc. // -// Units: Allocate sizes the gap in on-demand-equivalent dollars and emits -// tranches in commitment dollars with the rate they were sized with. Rows are +// Units: the provider's existing and expiring coverage arrive in commitment +// dollars and are first converted to on-demand-equivalent dollars with the +// same rates (ladderLayerStatesOnDemand). Allocate sizes the gap in +// on-demand-equivalent dollars and emits tranches in commitment dollars with the rate they were sized with. Rows are // stored as basis=commitment plus that rate; the store nets each row back to // on-demand-equivalent dollars by its own rate. // @@ -329,6 +331,13 @@ func (app *Application) executeLadderRun( return fmt.Errorf("GetCommitmentRates: %w", err) } + // Providers report existing coverage in commitment dollars; Allocate and + // the plan snapshot compare it with an on-demand-equivalent target (#946). + layerStates, err = ladderLayerStatesOnDemand(layerStates, rates) + if err != nil { + return fmt.Errorf("existing coverage: %w", err) + } + // L5 netting runs inside the persist transaction (PlanAndSaveLadderRun): // the store reads the config's in-flight commitment (scheduled tranches // still within config.LadderFireGrace plus fired, unresolved ones) under @@ -553,7 +562,8 @@ func ladderConfigToEngine(dbCfg *config.LadderConfigDB, accountID string) (pkgla // The monetary snapshot is derived to stay consistent with the authoritative // allocation rather than recomputed independently: // - target = baseline low-water * targetCoveragePct/100 (the coverage goal), -// - existing = sum of per-layer ExistingUSDPerHour, +// - existing = sum of per-layer ExistingUSDPerHour, already converted to +// on-demand-equivalent dollars by ladderLayerStatesOnDemand, // - gap = sum of the planned allocations' on-demand-equivalent gaps (the unit // of target and existing), NOT target - existing. // @@ -597,7 +607,8 @@ func assembleLadderPlan( // Holds are informational; append last. actions = append(actions, allocResult.Holds...) - // existing = sum of per-layer existing commitment (always reportable). + // existing = sum of per-layer on-demand-equivalent existing coverage + // (always reportable). var existingTotal float64 for _, ls := range layerStates { if ls.ExistingUSDPerHour != nil { diff --git a/internal/server/ladder_existing_units.go b/internal/server/ladder_existing_units.go new file mode 100644 index 000000000..1588cade2 --- /dev/null +++ b/internal/server/ladder_existing_units.go @@ -0,0 +1,131 @@ +package server + +import ( + "errors" + "fmt" + "math" + "math/big" + + pkgladder "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" +) + +// Existing coverage units (#946). +// +// Providers report LayerState.ExistingUSDPerHour and ExpiringUSDPerHour in +// native commitment dollars per hour: the AWS provider sums +// SavingsPlan.Commitment for the Savings Plan layers and the amortized RI cost +// for the RI layer. Allocate compares them with an on-demand-equivalent (OD) +// target, and the pkg/ladder LayerState contract says callers must convert +// them first. With a commitment rate r in (0, 1], the commitment dollars paid +// per OD dollar covered: +// +// od = commitment / r +// +// so 5 commitment $/hr at r = 0.5 covers 10 OD $/hr. The rate is today's +// new-purchase rate for the config's term and payment option, from +// GetCommitmentRates (option 2 on #946): DescribeSavingsPlans does not expose +// each plan's own rate, so plans bought at a different discount are +// approximated. +// +// Rounding favors under-buying. When the exact quotient is not a float64, +// existing is rounded up and expiring down, so the net existing coverage +// (existing - expiring) Allocate subtracts from the target is never +// understated. +// +// A layer with non-zero existing or expiring commitment and no rate cannot be +// converted. 1:1 would understate its coverage and over-buy, and no value is a +// safe upper bound, so the run fails instead of guessing. + +// errNoRateForExistingCoverage marks a layer whose existing commitment cannot +// be converted to on-demand-equivalent dollars. +var errNoRateForExistingCoverage = errors.New("no commitment rate to convert existing commitment dollars to on-demand-equivalent dollars") + +// ladderLayerStatesOnDemand returns a copy of states with ExistingUSDPerHour +// and ExpiringUSDPerHour converted from commitment dollars to +// on-demand-equivalent dollars by each layer's rate. The input map and its +// values are not modified. Nil amounts stay nil for Allocate to reject. +func ladderLayerStatesOnDemand( + states map[pkgladder.LayerType]pkgladder.LayerState, + rates map[pkgladder.LayerType]float64, +) (map[pkgladder.LayerType]pkgladder.LayerState, error) { + out := make(map[pkgladder.LayerType]pkgladder.LayerState, len(states)) + for layer, st := range states { + rate, hasRate := rates[layer] + od, err := layerStateOnDemand(st, rate, hasRate) + if err != nil { + return nil, fmt.Errorf("layer %s: %w", layer, err) + } + out[layer] = od + } + return out, nil +} + +// layerStateOnDemand converts one layer's existing and expiring commitment. +// A layer with no commitment (both amounts nil or zero) needs no rate. +func layerStateOnDemand(st pkgladder.LayerState, rate float64, hasRate bool) (pkgladder.LayerState, error) { + if isZeroOrNil(st.ExistingUSDPerHour) && isZeroOrNil(st.ExpiringUSDPerHour) { + return st, nil + } + if !hasRate { + return st, fmt.Errorf("%w (existing %s, expiring %s commitment USD/hr); refusing to assume 1:1", + errNoRateForExistingCoverage, fmtFloatPtr(st.ExistingUSDPerHour), fmtFloatPtr(st.ExpiringUSDPerHour)) + } + r, err := commitmentRateRat(rate) + if err != nil { + return st, err + } + existing, err := commitmentToOnDemand("existing", st.ExistingUSDPerHour, r, true) + if err != nil { + return st, err + } + expiring, err := commitmentToOnDemand("expiring", st.ExpiringUSDPerHour, r, false) + if err != nil { + return st, err + } + st.ExistingUSDPerHour, st.ExpiringUSDPerHour = existing, expiring + return st, nil +} + +func isZeroOrNil(v *float64) bool { return v == nil || *v == 0 } + +func fmtFloatPtr(v *float64) string { + if v == nil { + return "nil" + } + return fmt.Sprintf("%g", *v) +} + +// commitmentRateRat validates a provider rate the same way Allocate does: +// finite and in (0, 1]. +func commitmentRateRat(rate float64) (*big.Rat, error) { + if math.IsNaN(rate) || math.IsInf(rate, 0) || rate <= 0 || rate > 1 { + return nil, fmt.Errorf("commitment rate %g must be in (0, 1]", rate) + } + return new(big.Rat).SetFloat64(rate), nil +} + +// commitmentToOnDemand returns commitment/rate as a float64, rounded up when +// roundUp is set and down otherwise. Nil stays nil. +func commitmentToOnDemand(name string, commitment *float64, rate *big.Rat, roundUp bool) (*float64, error) { + if commitment == nil { + return nil, nil + } + c := ratFromFloat(*commitment) + if c == nil || c.Sign() < 0 { + return nil, fmt.Errorf("%s commitment %g USD/hr must be finite and non-negative", name, *commitment) + } + exact := c.Quo(c, rate) + od, _ := exact.Float64() + if math.IsInf(od, 0) { + return nil, fmt.Errorf("%s on-demand-equivalent amount exceeds float64", name) + } + if cmp := new(big.Rat).SetFloat64(od).Cmp(exact); roundUp && cmp < 0 { + od = math.Nextafter(od, math.Inf(1)) + } else if !roundUp && cmp > 0 { + od = math.Nextafter(od, 0) + } + if math.IsInf(od, 0) { + return nil, fmt.Errorf("%s on-demand-equivalent amount exceeds float64", name) + } + return &od, nil +} diff --git a/internal/server/ladder_existing_units_test.go b/internal/server/ladder_existing_units_test.go new file mode 100644 index 000000000..203e07e67 --- /dev/null +++ b/internal/server/ladder_existing_units_test.go @@ -0,0 +1,180 @@ +package server + +import ( + "encoding/json" + "math" + "math/big" + "testing" + + pkgladder "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Units in these tests: "commitment" is commitment USD/hr as the provider +// reports it (SavingsPlan.Commitment); "OD" is on-demand-equivalent USD/hr. +// od = commitment / rate. + +func f64(v float64) *float64 { return &v } + +func spState(existing, expiring *float64) pkgladder.LayerState { + return pkgladder.LayerState{Layer: pkgladder.LayerEC2InstanceSP, ExistingUSDPerHour: existing, ExpiringUSDPerHour: expiring} +} + +func TestLadderLayerStatesOnDemand_PinsCommitmentToOnDemandConversion(t *testing.T) { + cases := []struct { + name string + existing, expiring float64 + rate float64 + wantExisting, wantExpg float64 + }{ + {"half rate doubles", 5, 2, 0.5, 10, 4}, + {"quarter rate quadruples", 3, 1, 0.25, 12, 4}, + {"rate one is identity", 7, 3, 1, 7, 3}, + {"zero expiring stays zero", 6, 0, 0.75, 8, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + states := map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerEC2InstanceSP: spState(f64(tc.existing), f64(tc.expiring)), + } + out, err := ladderLayerStatesOnDemand(states, map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: tc.rate}) + require.NoError(t, err) + got := out[pkgladder.LayerEC2InstanceSP] + assert.Equal(t, tc.wantExisting, *got.ExistingUSDPerHour, "existing OD") + assert.Equal(t, tc.wantExpg, *got.ExpiringUSDPerHour, "expiring OD") + assert.Equal(t, pkgladder.LayerEC2InstanceSP, got.Layer) + }) + } +} + +// 1/0.3 is not a float64. Existing must round up and expiring down, so the +// net existing coverage (existing - expiring) is never understated, which +// would over-buy. +func TestLadderLayerStatesOnDemand_RoundsTowardUnderBuying(t *testing.T) { + const rate = 0.3 + states := map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerEC2InstanceSP: spState(f64(1), f64(1)), + } + out, err := ladderLayerStatesOnDemand(states, map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: rate}) + require.NoError(t, err) + exact := new(big.Rat).Quo(big.NewRat(1, 1), new(big.Rat).SetFloat64(rate)) + + existing := *out[pkgladder.LayerEC2InstanceSP].ExistingUSDPerHour + require.GreaterOrEqual(t, new(big.Rat).SetFloat64(existing).Cmp(exact), 0, "existing must not be below the exact OD") + require.Less(t, new(big.Rat).SetFloat64(math.Nextafter(existing, 0)).Cmp(exact), 0, "existing is the nearest float at or above") + + expiring := *out[pkgladder.LayerEC2InstanceSP].ExpiringUSDPerHour + require.LessOrEqual(t, new(big.Rat).SetFloat64(expiring).Cmp(exact), 0, "expiring must not be above the exact OD") + require.Greater(t, new(big.Rat).SetFloat64(math.Nextafter(expiring, math.Inf(1))).Cmp(exact), 0, "expiring is the nearest float at or below") +} + +func TestLadderLayerStatesOnDemand_DoesNotMutateInput(t *testing.T) { + existing, expiring := f64(5), f64(2) + states := map[pkgladder.LayerType]pkgladder.LayerState{pkgladder.LayerEC2InstanceSP: spState(existing, expiring)} + _, err := ladderLayerStatesOnDemand(states, map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: 0.5}) + require.NoError(t, err) + assert.Equal(t, 5.0, *existing) + assert.Equal(t, 2.0, *expiring) + assert.Same(t, existing, states[pkgladder.LayerEC2InstanceSP].ExistingUSDPerHour) +} + +// A layer with no commitment needs no rate (the AWS provider has none for +// Compute SP or Convertible RI today); nil amounts stay nil for Allocate. +func TestLadderLayerStatesOnDemand_EmptyLayersNeedNoRate(t *testing.T) { + states := map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerComputeSP: {Layer: pkgladder.LayerComputeSP, ExistingUSDPerHour: f64(0), ExpiringUSDPerHour: f64(0)}, + pkgladder.LayerConvertibleRI: {Layer: pkgladder.LayerConvertibleRI}, + } + out, err := ladderLayerStatesOnDemand(states, map[pkgladder.LayerType]float64{}) + require.NoError(t, err) + assert.Equal(t, 0.0, *out[pkgladder.LayerComputeSP].ExistingUSDPerHour) + assert.Nil(t, out[pkgladder.LayerConvertibleRI].ExistingUSDPerHour) + assert.Nil(t, out[pkgladder.LayerConvertibleRI].ExpiringUSDPerHour) +} + +func TestLadderLayerStatesOnDemand_RejectsUnconvertibleCoverage(t *testing.T) { + cases := []struct { + name string + state pkgladder.LayerState + rates map[pkgladder.LayerType]float64 + wantErr string + }{ + {"existing without a rate", spState(f64(5), f64(0)), map[pkgladder.LayerType]float64{pkgladder.LayerComputeSP: 0.5}, "no commitment rate"}, + {"expiring without a rate", spState(f64(0), f64(1)), nil, "no commitment rate"}, + {"zero rate", spState(f64(5), f64(0)), map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: 0}, "must be in (0, 1]"}, + {"rate above one", spState(f64(5), f64(0)), map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: 1.5}, "must be in (0, 1]"}, + {"NaN rate", spState(f64(5), f64(0)), map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: math.NaN()}, "must be in (0, 1]"}, + {"negative existing", spState(f64(-1), f64(0)), map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: 0.5}, "non-negative"}, + {"NaN expiring", spState(f64(1), f64(math.NaN())), map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: 0.5}, "non-negative"}, + {"overflow", spState(f64(math.MaxFloat64), f64(0)), map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: 0.5}, "exceeds float64"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, err := ladderLayerStatesOnDemand(map[pkgladder.LayerType]pkgladder.LayerState{pkgladder.LayerEC2InstanceSP: tc.state}, tc.rates) + require.ErrorContains(t, err, tc.wantErr) + assert.ErrorContains(t, err, string(pkgladder.LayerEC2InstanceSP)) + }) + } +} + +// End to end through executeLadderRun: low-water 12.5 at 80% coverage is a +// target of 10 OD. An existing EC2 Instance SP of 2.5 commitment at rate 0.5 +// covers 5 OD, so the run places 5 OD = 2.5 commitment. Treating the 2.5 as +// OD (the bug) placed 7.5 OD = 3.75 commitment. +func TestExecuteLadderRun_ExistingSavingsPlanCoverageIsOnDemandEquivalent(t *testing.T) { + store := &ladderTestStore{} + cap := &fakeLadderCapability{t: t, baseline: testBaselineLowWater(12.5), + rates: map[pkgladder.LayerType]float64{pkgladder.LayerConvertibleRI: 0.5, pkgladder.LayerEC2InstanceSP: 0.5}, + layerStates: map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerConvertibleRI: {Layer: pkgladder.LayerConvertibleRI, ExistingUSDPerHour: f64(0), ExpiringUSDPerHour: f64(0)}, + pkgladder.LayerEC2InstanceSP: spState(f64(2.5), f64(0)), + }} + require.NoError(t, runWithRates(t, store, cap)) + + run := store.savedRun + require.NotNil(t, run) + require.NotNil(t, run.ExistingUSDHr) + assert.InDelta(t, 5.0, *run.ExistingUSDHr, 1e-9, "existing_usd_hr is OD, comparable with target") + require.NotNil(t, run.GapUSDHr) + assert.InDelta(t, 5.0, *run.GapUSDHr, 1e-9, "gap OD = target 10 - existing 5") + assert.InDelta(t, 2.5, run.TotalHourlyCommit, 1e-9, "5 OD at rate 1/2 is 2.5 commitment") + + var dto ladderPlanJSONDTO + require.NoError(t, json.Unmarshal(run.Plan, &dto)) + require.NotNil(t, dto.ExistingUSDHr) + assert.InDelta(t, 5.0, *dto.ExistingUSDHr, 1e-9, "plan JSON existing is OD") +} + +// Expiring coverage converts by the same rate: existing 4 and expiring 2 +// commitment at rate 0.5 leave a net 4 OD, so the gap is 10 - 4 = 6 OD. +func TestExecuteLadderRun_ExpiringCoverageIsOnDemandEquivalent(t *testing.T) { + store := &ladderTestStore{} + cap := &fakeLadderCapability{t: t, baseline: testBaselineLowWater(12.5), + rates: map[pkgladder.LayerType]float64{pkgladder.LayerConvertibleRI: 0.5, pkgladder.LayerEC2InstanceSP: 0.5}, + layerStates: map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerConvertibleRI: {Layer: pkgladder.LayerConvertibleRI, ExistingUSDPerHour: f64(0), ExpiringUSDPerHour: f64(0)}, + pkgladder.LayerEC2InstanceSP: spState(f64(4), f64(2)), + }} + require.NoError(t, runWithRates(t, store, cap)) + require.NotNil(t, store.savedRun.GapUSDHr) + assert.InDelta(t, 6.0, *store.savedRun.GapUSDHr, 1e-9) + assert.InDelta(t, 3.0, store.savedRun.TotalHourlyCommit, 1e-9) +} + +// Existing commitment on a layer with no rate cannot be converted; the run +// fails and persists nothing instead of netting it 1:1 and over-buying. +func TestExecuteLadderRun_ExistingCoverageWithoutRateIsErrored(t *testing.T) { + store := &ladderTestStore{} + cap := &fakeLadderCapability{t: t, baseline: testBaselineLowWater(12.5), + rates: map[pkgladder.LayerType]float64{pkgladder.LayerEC2InstanceSP: 0.5}, + layerStates: map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerConvertibleRI: {Layer: pkgladder.LayerConvertibleRI, ExistingUSDPerHour: f64(3), ExpiringUSDPerHour: f64(0)}, + pkgladder.LayerEC2InstanceSP: spState(f64(0), f64(0)), + }} + err := runWithRates(t, store, cap) + require.ErrorContains(t, err, "no commitment rate") + require.ErrorContains(t, err, string(pkgladder.LayerConvertibleRI)) + assert.Nil(t, store.savedRun) + assert.Empty(t, store.savedTranches) +} diff --git a/known_issues/05_config_store_postgres.md b/known_issues/05_config_store_postgres.md index a301acddc..9c1177320 100644 --- a/known_issues/05_config_store_postgres.md +++ b/known_issues/05_config_store_postgres.md @@ -129,9 +129,11 @@ **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 +## ~~MEDIUM: nothing stops a future ladder executor from claiming a legacy on-demand row~~ — RESOLVED **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 +**Status:** ✔️ Resolved + +**Resolved by:** the #946 fix. `selectDueLadderTranchesTx` (used by `ClaimDueLadderTranches`) now selects only `amount_basis = 'commitment'` rows, which the validated `ladder_tranche_amount_provenance` constraint pairs with a stored `commitment_rate`. NULL-basis and `on_demand_equivalent` rows are never claimed: they stay netted while within `LadderFireGrace`, then the overdue sweep cancels them and the next run re-plans the gap in commitment dollars. `TestClaimDueLadderTranches_NeverClaimsOnDemandOrUnknownBasisRows` covers this against real PostgreSQL, including a limit-1 claim where older legacy rows must not starve a due commitment row.