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.