diff --git a/internal/config/ladder_provenance.go b/internal/config/ladder_provenance.go index b0b44b76..48a59e65 100644 --- a/internal/config/ladder_provenance.go +++ b/internal/config/ladder_provenance.go @@ -2,6 +2,7 @@ package config import ( "fmt" + "math" "math/big" ) @@ -29,9 +30,9 @@ func trancheCommitmentRate(tr *LadderTrancheDB) (*string, error) { if tr.CommitmentRate == nil { return nil, fmt.Errorf("commitment amount requires its stored rate") } - rate, ok := new(big.Rat).SetString(*tr.CommitmentRate) - if !ok || rate.Sign() <= 0 || rate.Cmp(big.NewRat(1, 1)) > 0 { - return nil, fmt.Errorf("stored commitment rate must be a rational in (0, 1]") + rate, err := parseStoredCommitmentRate(*tr.CommitmentRate) + if err != nil { + return nil, err } canonical := rate.RatString() return &canonical, nil @@ -39,3 +40,48 @@ func trancheCommitmentRate(tr *LadderTrancheDB) (*string, error) { return nil, fmt.Errorf("unknown ladder amount basis %q", *tr.AmountBasis) } } + +func parseStoredCommitmentRate(text string) (*big.Rat, error) { + rate, ok := new(big.Rat).SetString(text) + if !ok || rate.Sign() <= 0 || rate.Cmp(big.NewRat(1, 1)) > 0 { + return nil, fmt.Errorf("stored commitment rate must be a rational in (0, 1]") + } + return rate, nil +} + +func nettingOnDemand(amount string, tr *LadderTrancheDB) (*big.Rat, error) { + canonical, err := trancheCommitmentRate(tr) + if err != nil { + return nil, err + } + if tr.AmountBasis == nil { + return nil, fmt.Errorf("cannot net an unproven amount basis") + } + total, ok := new(big.Rat).SetString(amount) + if !ok || total.Sign() < 0 { + return nil, fmt.Errorf("netting amount must be finite and non-negative") + } + if canonical != nil { + rate, err := parseStoredCommitmentRate(*canonical) + if err != nil { + return nil, err + } + total.Quo(total, rate) + } + return total, nil +} + +func nettingFloat(total *big.Rat) (float64, error) { + value, _ := total.Float64() + if math.IsInf(value, 0) { + return 0, fmt.Errorf("netting total exceeds float64") + } + // The old SDK accepts one float; never round already committed OD down. + if new(big.Rat).SetFloat64(value).Cmp(total) < 0 { + value = math.Nextafter(value, math.Inf(1)) + } + if math.IsInf(value, 0) { + return 0, fmt.Errorf("netting total exceeds float64") + } + return value, nil +} diff --git a/internal/config/ladder_provenance_test.go b/internal/config/ladder_provenance_test.go new file mode 100644 index 00000000..0c8fcdf4 --- /dev/null +++ b/internal/config/ladder_provenance_test.go @@ -0,0 +1,47 @@ +package config + +import ( + "math" + "math/big" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestLadderNetting_ExactRatesAndBoundary(t *testing.T) { + od, native := LadderAmountOnDemand, LadderAmountCommitment + third, half, bad, above := "1/3", "1/2", "bad", "2" + for _, tc := range []struct { + amount string + basis *LadderAmountBasis + rate *string + want string + }{ + {"55", &od, nil, "55"}, {"3.5", &od, nil, "7/2"}, + {"1", &native, &third, "3"}, {"1", &native, &half, "2"}, {"0", &native, &half, "0"}, + {"0", nil, nil, ""}, {"1", &native, nil, ""}, {"1", &od, &half, ""}, + {"1", &native, &bad, ""}, {"1", &native, &above, ""}, {"-1", &od, nil, ""}, {"NaN", &od, nil, ""}, + } { + value, err := nettingOnDemand(tc.amount, &LadderTrancheDB{AmountBasis: tc.basis, CommitmentRate: tc.rate}) + if tc.want == "" { + require.Error(t, err) + continue + } + require.NoError(t, err) + require.Equal(t, tc.want, value.RatString()) + } + for _, text := range []string{"55", "3", "7/2", "55/3", "1/10"} { + exact, ok := new(big.Rat).SetString(text) + require.True(t, ok) + value, err := nettingFloat(exact) + require.NoError(t, err) + require.GreaterOrEqual(t, new(big.Rat).SetFloat64(value).Cmp(exact), 0) + require.Less(t, new(big.Rat).SetFloat64(math.Nextafter(value, math.Inf(-1))).Cmp(exact), 0) + if text == "55" || text == "3" || text == "7/2" { + nearest, _ := exact.Float64() + require.Equal(t, nearest, value) + } + } + _, err := nettingFloat(new(big.Rat).SetInt(new(big.Int).Lsh(big.NewInt(1), 2048))) + require.Error(t, err) +} diff --git a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go index bf075766..37e816fa 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go +++ b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go @@ -51,6 +51,8 @@ func (f *ceilingFixture) plan(target, existing, baseline *float64, amounts ...fl trs := make([]LadderTrancheDB, len(amounts)) for i, a := range amounts { trs[i] = lifecycleTranche(f.cfgID, a, ladder.TrancheStatusScheduled, time.Now().UTC().Add(time.Hour)) + basis := LadderAmountOnDemand + trs[i].AmountBasis = &basis trs[i].RunID = &runID ids = append(ids, trs[i].ID) } @@ -179,6 +181,8 @@ func TestLadderAmendCeiling_WaitsForPlanner(t *testing.T) { close(entered) <-gate tr := lifecycleTranche(f.cfgID, 10-inFlight, ladder.TrancheStatusScheduled, time.Now().UTC().Add(time.Hour)) + basis := LadderAmountOnDemand + tr.AmountBasis = &basis tr.RunID = &runID return &LadderRunDB{ID: runID, ConfigID: &f.cfgID, StartedAt: asOf, Status: ladder.RunStatusPlanned, TargetUSDHr: fptr(10), ExistingUSDHr: fptr(0), BaselineUSDHr: fptr(100)}, []LadderTrancheDB{tr}, nil @@ -188,7 +192,13 @@ func TestLadderAmendCeiling_WaitsForPlanner(t *testing.T) { var once sync.Once release := func() { once.Do(func() { close(gate) }) } t.Cleanup(release) // a failed wait below must not strand the planner - <-entered + select { + case <-entered: + case err := <-planned: + t.Fatalf("planner exited before entering callback: %v", err) + case <-f.ctx.Done(): + t.Fatal(f.ctx.Err()) + } amended := make(chan error, 1) go func() { amended <- f.amend(ids[0], 7) }() waitForLadderWaiters(f.ctx, t, f.store, f.cfgID, 1) diff --git a/internal/config/store_postgres_ladder_lock.go b/internal/config/store_postgres_ladder_lock.go index beb3feea..6f5a2f2d 100644 --- a/internal/config/store_postgres_ladder_lock.go +++ b/internal/config/store_postgres_ladder_lock.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log" + "math/big" "time" "github.com/google/uuid" @@ -25,7 +26,7 @@ const ladderConfigLockPrefix = "cudly:ladder-config:" var ladderLockTimeout = 30 * time.Second // LadderPlanBuilder computes a run and its new tranches from the config's -// in-flight commitment. It runs while the per-config lock is held inside the +// measured on-demand-equivalent total. It runs while the per-config lock is held inside the // persist transaction: it must be pure compute (no provider or network I/O). type LadderPlanBuilder func(inFlightUSDHr float64) (*LadderRunDB, []LadderTrancheDB, error) @@ -169,16 +170,44 @@ func (s *PostgresStore) PlanAndSaveLadderRun(ctx context.Context, configID strin // a fresh READ COMMITTED snapshot, so a tranche fired at the first and // completed before the second is counted by both (safe); the reverse order // could miss one completing between the two. -func lockedLadderCommit(ctx context.Context, q queryRower, configID string, asOf time.Time) (float64, error) { - inFlight, err := inFlightLadderCommit(ctx, q, configID, asOf) +func lockedLadderCommit(ctx context.Context, q dbConn, configID string, asOf time.Time) (float64, error) { + inFlight, err := ladderNettingRows(ctx, q, + "status = $2 OR (status = $3 AND scheduled_date > $4)", + configID, string(ladder.TrancheStatusFired), string(ladder.TrancheStatusScheduled), ladderGraceCutoff(asOf)) if err != nil { return 0, err } - settled, err := settledLadderCommit(ctx, q, configID, ladderSettleAnchor(asOf)) + settled, err := ladderNettingRows(ctx, q, "status = 'completed' AND settled_at > $2", + configID, ladderSettleAnchor(asOf)) if err != nil { return 0, err } - return inFlight + settled, nil + return nettingFloat(inFlight.Add(inFlight, settled)) +} + +func ladderNettingRows(ctx context.Context, q dbConn, predicate string, args ...any) (*big.Rat, error) { + rows, err := q.Query(ctx, "SELECT id,layer_type,amount_usd_hr::text,amount_basis,commitment_rate FROM ladder_tranches WHERE config_id=$1 AND ("+predicate+")", args...) + if err != nil { + return nil, fmt.Errorf("read ladder netting rows: %w", err) + } + defer rows.Close() + total := new(big.Rat) + for rows.Next() { + var tr LadderTrancheDB + var amount string + if err := rows.Scan(&tr.ID, &tr.LayerType, &amount, &tr.AmountBasis, &tr.CommitmentRate); err != nil { + return nil, err + } + od, err := nettingOnDemand(amount, &tr) + if err != nil { + return nil, fmt.Errorf("net tranche %s (%s): %w", tr.ID, tr.LayerType, err) + } + total.Add(total, od) + } + if err := rows.Err(); err != nil { + return nil, err + } + return total, nil } // queryRower is satisfied by pgx.Tx and the pool connection. diff --git a/internal/config/store_postgres_ladder_lock_integration_test.go b/internal/config/store_postgres_ladder_lock_integration_test.go index 74bc7892..0cc9f832 100644 --- a/internal/config/store_postgres_ladder_lock_integration_test.go +++ b/internal/config/store_postgres_ladder_lock_integration_test.go @@ -81,7 +81,10 @@ func TestLadderLock_ClaimWaitsForPlanner(t *testing.T) { defer cancel() store := setupLadderStore(ctx, t) cfgID := seedLadderConfigWithExtID(ctx, t, store, "711111111111") - require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, time.Now().UTC().Add(-time.Hour))})) + tr := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, time.Now().UTC().Add(-time.Hour)) + basis := LadderAmountOnDemand + tr.AmountBasis = &basis + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) release := holdLadderLock(ctx, t, store, cfgID) var claimEntered atomic.Bool diff --git a/internal/config/store_postgres_ladder_netting_integration_test.go b/internal/config/store_postgres_ladder_netting_integration_test.go new file mode 100644 index 00000000..9657283a --- /dev/null +++ b/internal/config/store_postgres_ladder_netting_integration_test.go @@ -0,0 +1,165 @@ +//go:build integration + +package config + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/require" +) + +func TestLadderNetting_MixedStoredBasis(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + cfg := seedLadderConfig(ctx, t, store) + od, native, rate := LadderAmountOnDemand, LadderAmountCommitment, "1/2" + rows := []LadderTrancheDB{ + lifecycleTranche(cfg, 15, ladder.TrancheStatusScheduled, lockTestAsOf.Add(time.Hour)), + lifecycleTranche(cfg, 20, ladder.TrancheStatusFired, lockTestAsOf.Add(-time.Hour)), + } + rows[0].AmountBasis = &od + rows[1].AmountBasis, rows[1].CommitmentRate = &native, &rate + require.NoError(t, store.SaveLadderTranches(ctx, rows)) + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(odTotal float64) (*LadderRunDB, []LadderTrancheDB, error) { + require.Equal(t, 55.0, odTotal) + return lockTestRun(cfg), nil, nil + }) + require.NoError(t, err) + var amount, storedRate string + require.NoError(t, store.db.QueryRow(ctx, "SELECT amount_usd_hr::text,commitment_rate FROM ladder_tranches WHERE id=$1", rows[1].ID).Scan(&amount, &storedRate)) + require.Equal(t, "20.000000", amount) + require.Equal(t, rate, storedRate) +} + +func TestLadderNetting_UnknownRefusesBeforeBuilder(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + cfg := seedLadderConfig(ctx, t, store) + row := lifecycleTranche(cfg, 5, ladder.TrancheStatusFired, lockTestAsOf) + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{row})) + built := false + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + built = true + return lockTestRun(cfg), nil, nil + }) + require.Error(t, err) + require.False(t, built) + require.Zero(t, countRuns(ctx, t, store, cfg)) + _, err = store.db.Exec(ctx, "UPDATE ladder_tranches SET amount_basis='commitment',commitment_rate='bad' WHERE id=$1", row.ID) + require.NoError(t, err) + _, err = store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + built = true + return lockTestRun(cfg), nil, nil + }) + require.Error(t, err) + require.False(t, built) + require.Zero(t, countRuns(ctx, t, store, cfg)) +} + +func TestLadderNetting_SelectionBoundaries(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + cfg := seedLadderConfig(ctx, t, store) + od, native := LadderAmountOnDemand, LadderAmountCommitment + cutoff, anchor := ladderGraceCutoff(lockTestAsOf), ladderSettleAnchor(lockTestAsOf) + before, after := anchor.Add(-time.Microsecond), anchor.Add(time.Microsecond) + add := func(status ladder.TrancheStatus, amount float64, at time.Time, settled *time.Time, rate string) { + tr := lifecycleTranche(cfg, amount, status, at) + tr.AmountBasis, tr.SettledAt = &od, settled + if status == ladder.TrancheStatusCompleted { + tr.LayerType = ladder.LayerEC2InstanceSP + } + if rate != "" { + tr.AmountBasis, tr.CommitmentRate = &native, &rate + } + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) + } + add(ladder.TrancheStatusFired, 1, lockTestAsOf.Add(-24*time.Hour), nil, "1/2") + add(ladder.TrancheStatusScheduled, 2, cutoff.Add(time.Microsecond), nil, "") + add(ladder.TrancheStatusScheduled, 4, cutoff, nil, "") + add(ladder.TrancheStatusScheduled, 8, cutoff.Add(-time.Microsecond), nil, "") + add(ladder.TrancheStatusCompleted, 16, lockTestAsOf, &after, "1/4") + add(ladder.TrancheStatusCompleted, 32, lockTestAsOf, &anchor, "") + add(ladder.TrancheStatusCompleted, 64, lockTestAsOf, &before, "") + add(ladder.TrancheStatusFailed, 128, lockTestAsOf, nil, "") + add(ladder.TrancheStatusCancelled, 256, lockTestAsOf, nil, "") + other := seedLadderConfigWithExtID(ctx, t, store, "fixture-other") + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{lifecycleTranche(other, 512, ladder.TrancheStatusFired, lockTestAsOf)})) + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(total float64) (*LadderRunDB, []LadderTrancheDB, error) { + require.Equal(t, 68.0, total) + return lockTestRun(cfg), nil, nil + }) + require.NoError(t, err) + var overdue int + require.NoError(t, store.db.QueryRow(ctx, "SELECT count(*) FROM ladder_tranches WHERE config_id=$1 AND status='cancelled' AND amount_usd_hr IN (4,8)", cfg).Scan(&overdue)) + require.Equal(t, 2, overdue) +} + +type nettingHookDB struct { + dbConn + hook func() +} +type nettingHookTx struct { + pgx.Tx + hook func() + reads int +} + +func (db nettingHookDB) Begin(ctx context.Context) (pgx.Tx, error) { + tx, err := db.dbConn.Begin(ctx) + if err != nil { + return nil, err + } + return &nettingHookTx{Tx: tx, hook: db.hook}, nil +} +func (tx *nettingHookTx) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) { + tx.reads++ + if tx.reads == 2 { + tx.hook() + } + return tx.Tx.Query(ctx, sql, args...) +} + +func TestLadderNetting_CompletionBetweenReadsCountsTwice(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + cfg := seedLadderConfig(ctx, t, store) + native, rate := LadderAmountCommitment, "1/3" + tr := lifecycleTranche(cfg, 1, ladder.TrancheStatusFired, lockTestAsOf.Add(-time.Hour)) + tr.AmountBasis, tr.CommitmentRate = &native, &rate + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) + original := store.db + transitioned := false + store.db = nettingHookDB{dbConn: original, hook: func() { + _, err := original.Exec(ctx, "UPDATE ladder_tranches SET status='completed',settled_at=$2 WHERE id=$1", tr.ID, lockTestAsOf) + require.NoError(t, err) + transitioned = true + }} + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(total float64) (*LadderRunDB, []LadderTrancheDB, error) { + require.True(t, transitioned) + require.Equal(t, 6.0, total) + return lockTestRun(cfg), nil, nil + }) + require.NoError(t, err) +} + +func TestLadderNetting_ReadFailureDoesNotBecomeZero(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + store := setupLadderStore(ctx, t) + cfg := seedLadderConfig(ctx, t, store) + original := store.db + store.db = nettingHookDB{dbConn: original, hook: cancel} + built := false + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + built = true + return lockTestRun(cfg), nil, nil + }) + require.Error(t, err) + require.False(t, built) + require.Zero(t, countRuns(context.Background(), t, store, cfg)) +} diff --git a/internal/config/store_postgres_ladder_settled_plan_integration_test.go b/internal/config/store_postgres_ladder_settled_plan_integration_test.go index 01cdd267..bee8528f 100644 --- a/internal/config/store_postgres_ladder_settled_plan_integration_test.go +++ b/internal/config/store_postgres_ladder_settled_plan_integration_test.go @@ -28,6 +28,8 @@ func TestPostgresStoreDB_PlanAndSaveSettledTerm(t *testing.T) { add := func(status ladder.TrancheStatus, amount float64, settled *time.Time) { t.Helper() tr := lifecycleTranche(cfgID, amount, status, asOf.Add(time.Hour)) + basis := LadderAmountOnDemand + tr.AmountBasis = &basis tr.SettledAt = settled require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) } diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go index fb906bd6..eb8f35d5 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -340,7 +340,7 @@ func (app *Application) executeLadderRun( } // ladderRunPlanner holds the read-side data of one run and turns the locked -// in-flight figure into the run row and tranche rows to persist. plan is pure +// on-demand-equivalent netting total into the run row and tranche rows to persist. plan is pure // compute (it runs while the per-config lock is held). type ladderRunPlanner struct { dbCfg *config.LadderConfigDB diff --git a/internal/server/handler_ladder_lock_integration_test.go b/internal/server/handler_ladder_lock_integration_test.go index e3e4f4c2..71b7c22f 100644 --- a/internal/server/handler_ladder_lock_integration_test.go +++ b/internal/server/handler_ladder_lock_integration_test.go @@ -15,6 +15,7 @@ import ( pkgladder "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -106,3 +107,32 @@ func TestLadderLock_ConcurrentRunsPlanTheGapOnce(t *testing.T) { // Target is 10: the second run must have netted the first's rows (gap 0). assert.InDelta(t, 10.0, e.live(ctx, t), 1e-9, "two concurrent runs must not both plan the gap") } + +func TestLadderLock_MixedBasisReachesActualPlanner(t *testing.T) { + ctx := context.Background() + e := newLadderB1Env(ctx, t, ladderB1Conn(ctx, t), ladderB1TwoStep) + od, native, rate := config.LadderAmountOnDemand, config.LadderAmountCommitment, "1/2" + seed := []config.LadderTrancheDB{ + {ID: uuid.New().String(), ConfigID: &e.cfg.ID, LayerType: pkgladder.LayerConvertibleRI, AmountUSDHr: 1, AmountBasis: &od, Term: pkgladder.Term1Year, PaymentOption: pkgladder.PaymentNoUpfront, Status: pkgladder.TrancheStatusScheduled, ScheduledDate: e.start.Add(time.Hour)}, + {ID: uuid.New().String(), ConfigID: &e.cfg.ID, LayerType: pkgladder.LayerConvertibleRI, AmountUSDHr: 2, AmountBasis: &native, CommitmentRate: &rate, Term: pkgladder.Term1Year, PaymentOption: pkgladder.PaymentNoUpfront, Status: pkgladder.TrancheStatusFired, ScheduledDate: e.start.Add(-time.Hour)}, + } + require.NoError(t, e.app.Config.SaveLadderTranches(ctx, seed)) + id := e.run(ctx, t, e.start) + _, total := e.planAndTotal(ctx, t, id) + require.Equal(t, 5.0, total) + var basis string + require.NoError(t, e.conn.QueryRow(ctx, "SELECT amount_basis FROM ladder_tranches WHERE run_id=$1 LIMIT 1", id).Scan(&basis)) + require.Equal(t, string(od), basis) +} + +func TestLadderLock_UnknownBasisPersistsNoRun(t *testing.T) { + ctx := context.Background() + e := newLadderB1Env(ctx, t, ladderB1Conn(ctx, t), ladderB1TwoStep) + seed := config.LadderTrancheDB{ID: uuid.New().String(), ConfigID: &e.cfg.ID, LayerType: pkgladder.LayerConvertibleRI, AmountUSDHr: 2, Term: pkgladder.Term1Year, PaymentOption: pkgladder.PaymentNoUpfront, Status: pkgladder.TrancheStatusFired, ScheduledDate: e.start} + require.NoError(t, e.app.Config.SaveLadderTranches(ctx, []config.LadderTrancheDB{seed})) + err := e.app.executeLadderRun(ctx, e.cfg, e.cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, e.start) + require.Error(t, err) + var runs int + require.NoError(t, e.conn.QueryRow(ctx, "SELECT count(*) FROM ladder_runs WHERE config_id=$1", e.cfg.ID).Scan(&runs)) + require.Zero(t, runs) +}