Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
8 changes: 8 additions & 0 deletions internal/config/store_postgres_ladder_lifecycle.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down
82 changes: 68 additions & 14 deletions internal/config/store_postgres_ladder_lifecycle_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand All @@ -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)
Expand All @@ -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()
Expand All @@ -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))
Expand Down Expand Up @@ -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))

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
17 changes: 15 additions & 2 deletions internal/config/store_postgres_ladder_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down
4 changes: 2 additions & 2 deletions internal/purchase/ladder_reconcile_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
19 changes: 15 additions & 4 deletions internal/server/handler_ladder.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
//
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
//
Expand Down Expand Up @@ -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 {
Expand Down
Loading
Loading