From f838030cddd8c7ce5e090d6255b29db12761cef7 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 16 Oct 2026 13:11:07 +0200 Subject: [PATCH 1/2] fix(ladder): net in-flight tranches per row basis in on-demand-equivalent dollars LadderPlanBuilder now receives per-row LadderInFlightRow values (layer, commitment, rate) instead of one float. Commitment rows carry their stored rate, legacy on-demand rows carry rate 1, and a NULL basis still fails loud in netting. The handler adapter sums the rows exactly and rounds up only when the float is below the exact total, so legacy-only data is unchanged. The amend ceiling reads the same rows (NULL basis assumed on-demand there, per #886). GetInFlightLadderCommitUSDHr, a raw mixed-basis SUM with no production caller, is removed and its tests assert the on-demand netting. Refs #799 Refs LeanerCloud/cloud-commitments-go#349 Co-Authored-By: Claude Opus 5.5 --- internal/analytics/collector_test.go | 2 +- internal/config/ladder_provenance.go | 88 +++++++++++++++++ internal/config/ladder_provenance_test.go | 35 +++++++ internal/config/store_postgres_ladder.go | 30 +----- .../store_postgres_ladder_amend_ceiling.go | 24 ++++- ...s_ladder_amend_ceiling_integration_test.go | 10 +- ..._postgres_ladder_basis_integration_test.go | 96 +++++++++++++++++++ ...tgres_ladder_lifecycle_integration_test.go | 5 +- internal/config/store_postgres_ladder_lock.go | 73 ++++++-------- ...e_postgres_ladder_lock_integration_test.go | 22 +++-- ...ostgres_ladder_netting_integration_test.go | 18 ++-- .../config/store_postgres_ladder_settled.go | 20 +--- ...ostgres_ladder_settled_integration_test.go | 8 +- ...es_ladder_settled_plan_integration_test.go | 4 +- internal/config/store_postgres_ladder_test.go | 35 ++++--- internal/mocks/stores.go | 2 +- .../ladder_reconcile_integration_test.go | 18 +++- internal/server/handler_ladder.go | 14 ++- .../handler_ladder_lock_integration_test.go | 2 +- internal/server/handler_ladder_test.go | 8 +- internal/server/test_helpers_test.go | 2 +- 21 files changed, 355 insertions(+), 161 deletions(-) create mode 100644 internal/config/store_postgres_ladder_basis_integration_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index eee0fad1..f4916c13 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -876,7 +876,7 @@ func (m *mockConfigStore) SaveLadderRun(_ context.Context, run *config.LadderRun return run, nil } func (m *mockConfigStore) PlanAndSaveLadderRun(_ context.Context, _ string, _ time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { - run, _, err := build(0) + run, _, err := build(nil) return run, err } func (m *mockConfigStore) GetLadderRun(_ context.Context, _ string) (*config.LadderRunDB, error) { diff --git a/internal/config/ladder_provenance.go b/internal/config/ladder_provenance.go index 48a59e65..a9aa624b 100644 --- a/internal/config/ladder_provenance.go +++ b/internal/config/ladder_provenance.go @@ -4,6 +4,8 @@ import ( "fmt" "math" "math/big" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" ) type LadderAmountBasis string @@ -85,3 +87,89 @@ func nettingFloat(total *big.Rat) (float64, error) { } return value, nil } + +// LadderInFlightRow is one netted tranche. Rate is rounded down and the +// commitment up, so CommitmentUSDPerHour/Rate never understates the exact +// on-demand-equivalent amount. Legacy on-demand rows carry Rate 1. +type LadderInFlightRow struct { + Layer ladder.LayerType + CommitmentUSDPerHour float64 + Rate float64 + exactOD *big.Rat // set by the store so legacy totals stay bit-identical to the exact sum +} + +// assumeOnDemandIfUnknown covers only the amend path: rows from before 000121 +// have no basis and #886 deliberately left them unknown, so the amend ceiling +// ASSUMES on-demand-equivalent (its behavior before the basis existed). +// Netting never calls this and keeps failing loud on an unknown basis. +func assumeOnDemandIfUnknown(tr *LadderTrancheDB) *LadderTrancheDB { + if tr.AmountBasis != nil || tr.CommitmentRate != nil { + return tr + } + assumed, basis := *tr, LadderAmountOnDemand + assumed.AmountBasis = &basis + return &assumed +} + +func newLadderInFlightRow(amount string, tr *LadderTrancheDB) (LadderInFlightRow, error) { + od, err := nettingOnDemand(amount, tr) + if err != nil { + return LadderInFlightRow{}, err + } + canonical, err := trancheCommitmentRate(tr) + if err != nil { + return LadderInFlightRow{}, err + } + rate := big.NewRat(1, 1) + if canonical != nil { + if rate, err = parseStoredCommitmentRate(*canonical); err != nil { + return LadderInFlightRow{}, err + } + } + commitment, _ := new(big.Rat).SetString(amount) + commitmentUp, err := nettingFloat(commitment) + if err != nil { + return LadderInFlightRow{}, err + } + rateDown, _ := rate.Float64() + if new(big.Rat).SetFloat64(rateDown).Cmp(rate) > 0 { + rateDown = math.Nextafter(rateDown, 0) + } + rateDown = max(rateDown, math.SmallestNonzeroFloat64) + return LadderInFlightRow{Layer: tr.LayerType, CommitmentUSDPerHour: commitmentUp, Rate: rateDown, exactOD: od}, nil +} + +func (r LadderInFlightRow) onDemand() (*big.Rat, error) { + if r.exactOD != nil { + return r.exactOD, nil + } + commitment := new(big.Rat).SetFloat64(r.CommitmentUSDPerHour) + if commitment == nil || commitment.Sign() < 0 || !(r.Rate > 0 && r.Rate <= 1) { + return nil, fmt.Errorf("invalid in-flight row: commitment %v at rate %v", r.CommitmentUSDPerHour, r.Rate) + } + return commitment.Quo(commitment, new(big.Rat).SetFloat64(r.Rate)), nil +} + +func sumLadderOnDemand(rows ...[]LadderInFlightRow) (*big.Rat, error) { + total := new(big.Rat) + for _, set := range rows { + for _, r := range set { + od, err := r.onDemand() + if err != nil { + return nil, err + } + total.Add(total, od) + } + } + return total, nil +} + +// LadderInFlightOnDemand is the rows' on-demand-equivalent total, summed +// exactly and rounded up to the next float only when the float is below it. +func LadderInFlightOnDemand(rows []LadderInFlightRow) (float64, error) { + total, err := sumLadderOnDemand(rows) + if err != nil { + return 0, err + } + return nettingFloat(total) +} diff --git a/internal/config/ladder_provenance_test.go b/internal/config/ladder_provenance_test.go index 0c8fcdf4..6652b325 100644 --- a/internal/config/ladder_provenance_test.go +++ b/internal/config/ladder_provenance_test.go @@ -45,3 +45,38 @@ func TestLadderNetting_ExactRatesAndBoundary(t *testing.T) { _, err := nettingFloat(new(big.Rat).SetInt(new(big.Int).Lsh(big.NewInt(1), 2048))) require.Error(t, err) } + +func TestLadderInFlightOnDemand_RoundsAwayFromUnderstating(t *testing.T) { + native, third := LadderAmountCommitment, "1/3" + tr := &LadderTrancheDB{AmountBasis: &native, CommitmentRate: &third} + row, err := newLadderInFlightRow("1.000001", tr) + require.NoError(t, err) + exact := big.NewRat(3000003, 1000000) + require.LessOrEqual(t, new(big.Rat).SetFloat64(row.Rate).Cmp(big.NewRat(1, 3)), 0, "rate rounds down") + require.GreaterOrEqual(t, new(big.Rat).SetFloat64(row.CommitmentUSDPerHour).Cmp(big.NewRat(1000001, 1000000)), 0, "commitment rounds up") + row.exactOD = nil // consumers that only see the floats must still not understate + total, err := LadderInFlightOnDemand([]LadderInFlightRow{row}) + require.NoError(t, err) + require.GreaterOrEqual(t, new(big.Rat).SetFloat64(total).Cmp(exact), 0) +} + +func TestLadderInFlightOnDemand_RejectsUnusableRows(t *testing.T) { + for _, row := range []LadderInFlightRow{ + {CommitmentUSDPerHour: 1, Rate: 0}, {CommitmentUSDPerHour: 1, Rate: 1.5}, {CommitmentUSDPerHour: 1, Rate: math.NaN()}, + {CommitmentUSDPerHour: -1, Rate: 1}, {CommitmentUSDPerHour: math.Inf(1), Rate: 1}, + } { + _, err := LadderInFlightOnDemand([]LadderInFlightRow{row}) + require.Error(t, err, "%+v", row) + } +} + +func TestAssumeOnDemandIfUnknown_OnlyForUnprovenBasis(t *testing.T) { + assumed := assumeOnDemandIfUnknown(&LadderTrancheDB{}) + require.NotNil(t, assumed.AmountBasis) + require.Equal(t, LadderAmountOnDemand, *assumed.AmountBasis) + rate := "1/2" + orphan := &LadderTrancheDB{CommitmentRate: &rate} + require.Same(t, orphan, assumeOnDemandIfUnknown(orphan), "a rate without a basis is corrupt, not legacy") + _, err := nettingOnDemand("1", &LadderTrancheDB{}) + require.Error(t, err, "netting still refuses an unknown basis") +} diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index 6515485c..71994fbb 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -327,7 +327,7 @@ func insertLadderTranchesTx(ctx context.Context, tx pgx.Tx, tranches []LadderTra // cancelOverdueLadderTranchesTx marks the config's scheduled tranches that have // outlived LadderFireGrace as cancelled: scheduled_date <= ladderGraceCutoff(asOf), -// the exact complement of the netting bound in GetInFlightLadderCommitUSDHr. +// the exact complement of the netting bound in inFlightLadderRows. // No executor transitions tranches yet (issue #118), so without this the // overdue amount grows with every run and a future due-row sweep would buy all // of it at once (#544). Rows already linked to an execution are left alone. @@ -410,34 +410,6 @@ func insertLadderRunWithTranchesTx(ctx context.Context, tx pgx.Tx, run *LadderRu return &scanned, nil } -// GetInFlightLadderCommitUSDHr returns the total hourly USD commitment already -// in flight for the given config: the SUM of amount_usd_hr for ladder_tranches -// rows where config_id=$1 and either -// - status = 'scheduled' and scheduled_date > ladderGraceCutoff(asOf): not yet -// bought and still fireable (future, or due but within LadderFireGrace), or -// - status = 'fired': claimed for purchase but not yet resolved (approval -// pending, in progress, or outcome unknown). Fired nets regardless of age; -// only a transition to completed or failed releases it. -// -// Excluded: completed (counted in the engine's ExistingUSDPerHour, but only if -// that figure was read after the tranche became visible: a caller that read -// ExistingUSDPerHour before this call must also add settledLadderCommit with -// ladderSettleAnchor(asOf), as PlanAndSaveLadderRun does, or a tranche that -// completes in between is netted nowhere), failed and cancelled (never bought; counting -// them would silently under-buy), and scheduled rows past the grace window -// (cancelled by cancelOverdueLadderTranchesTx with the complementary bound). -// -// Returns a non-nil *float64 (zero when nothing is in flight); never returns -// nil without a non-nil error, so callers can pass it directly to -// AllocationInput.InFlightUSDPerHour without an extra nil-guard. -func (s *PostgresStore) GetInFlightLadderCommitUSDHr(ctx context.Context, configID string, asOf time.Time) (*float64, error) { - total, err := inFlightLadderCommit(ctx, s.db, configID, asOf) - if err != nil { - return nil, err - } - return &total, nil -} - // LatestLadderRunStartedAt returns the maximum started_at for the given // config_id, or nil when no run has been recorded yet. Drives the per-cadence // self-gate in handleLadderRun (Q6). diff --git a/internal/config/store_postgres_ladder_amend_ceiling.go b/internal/config/store_postgres_ladder_amend_ceiling.go index 4ed012d9..a87df688 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling.go +++ b/internal/config/store_postgres_ladder_amend_ceiling.go @@ -51,11 +51,7 @@ func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmoun if err != nil { return err } - inFlight, err := inFlightLadderCommit(ctx, tx, configID, now) - if err != nil { - return err - } - settled, err := settledLadderCommit(ctx, tx, configID, ladderSettleAnchor(snap.anchor)) + inFlight, settled, err := ladderCommittedOnDemand(ctx, tx, configID, now, ladderSettleAnchor(snap.anchor)) if err != nil { return err } @@ -71,6 +67,24 @@ func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmoun return nil } +// ladderCommittedOnDemand nets the in-flight and settled-after-anchor rows to +// on-demand-equivalent dollars, assuming on-demand for rows with no stored basis. +func ladderCommittedOnDemand(ctx context.Context, tx pgx.Tx, configID string, now, anchor time.Time) (inFlight, settled float64, err error) { + inFlightRows, err := inFlightLadderRows(ctx, tx, true, configID, now) + if err != nil { + return 0, 0, err + } + settledRows, err := settledLadderRows(ctx, tx, true, configID, anchor) + if err != nil { + return 0, 0, err + } + if inFlight, err = LadderInFlightOnDemand(inFlightRows); err != nil { + return 0, 0, err + } + settled, err = LadderInFlightOnDemand(settledRows) + return inFlight, settled, err +} + type ladderCeilingSnapshot struct { anchor time.Time ceiling, oldAmount, newAmount float64 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 37e816fa..b6134dde 100644 --- a/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go +++ b/internal/config/store_postgres_ladder_amend_ceiling_integration_test.go @@ -47,7 +47,7 @@ func (f *ceilingFixture) plan(target, existing, baseline *float64, amounts ...fl asOf := time.Now().UTC().Add(-time.Duration(10-f.runs) * time.Minute).Truncate(time.Microsecond) runID := uuid.NewString() var ids []string - _, err := f.store.PlanAndSaveLadderRun(f.ctx, f.cfgID, asOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err := f.store.PlanAndSaveLadderRun(f.ctx, f.cfgID, asOf, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { trs := make([]LadderTrancheDB, len(amounts)) for i, a := range amounts { trs[i] = lifecycleTranche(f.cfgID, a, ladder.TrancheStatusScheduled, time.Now().UTC().Add(time.Hour)) @@ -77,9 +77,9 @@ func (f *ceilingFixture) amend(id string, amount float64) error { func (f *ceilingFixture) inFlight() float64 { f.t.Helper() - v, err := inFlightLadderCommit(f.ctx, f.store.db, f.cfgID, time.Now()) + rows, err := inFlightLadderRows(f.ctx, f.store.db, true, f.cfgID, time.Now()) require.NoError(f.t, err) - return v + return mustOnDemand(f.t, rows) } func (f *ceilingFixture) move(id string, from, to ladder.TrancheStatus) { @@ -177,10 +177,10 @@ func TestLadderAmendCeiling_WaitsForPlanner(t *testing.T) { asOf := time.Now().UTC().Truncate(time.Microsecond) go func() { runID := uuid.NewString() - _, err := f.store.PlanAndSaveLadderRun(f.ctx, f.cfgID, asOf, func(inFlight float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err := f.store.PlanAndSaveLadderRun(f.ctx, f.cfgID, asOf, func(rows []LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { close(entered) <-gate - tr := lifecycleTranche(f.cfgID, 10-inFlight, ladder.TrancheStatusScheduled, time.Now().UTC().Add(time.Hour)) + tr := lifecycleTranche(f.cfgID, 10-mustOnDemand(f.t, rows), ladder.TrancheStatusScheduled, time.Now().UTC().Add(time.Hour)) basis := LadderAmountOnDemand tr.AmountBasis = &basis tr.RunID = &runID diff --git a/internal/config/store_postgres_ladder_basis_integration_test.go b/internal/config/store_postgres_ladder_basis_integration_test.go new file mode 100644 index 00000000..6a861ac1 --- /dev/null +++ b/internal/config/store_postgres_ladder_basis_integration_test.go @@ -0,0 +1,96 @@ +//go:build integration + +package config + +// Amend-path behavior per stored amount basis (PR A). Commitment rows are +// seeded directly: no writer produces them yet. + +import ( + "math" + "math/big" + "strconv" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/stretchr/testify/require" +) + +func mustOnDemand(t *testing.T, rows []LadderInFlightRow) float64 { + t.Helper() + v, err := LadderInFlightOnDemand(rows) + require.NoError(t, err) + return v +} + +func (f *ceilingFixture) setBasis(id, basis, rate string) { + f.t.Helper() + _, err := f.store.db.Exec(f.ctx, "UPDATE ladder_tranches SET amount_basis=NULLIF($2,''),commitment_rate=NULLIF($3,'') WHERE id=$1", id, basis, rate) + require.NoError(f.t, err) +} + +func seedNettingRows(t *testing.T, f *ceilingFixture, basis LadderAmountBasis, rate string, amounts ...string) { + t.Helper() + for _, a := range amounts { + v, err := strconv.ParseFloat(a, 64) + require.NoError(t, err) + tr := lifecycleTranche(f.cfgID, v, ladder.TrancheStatusFired, lockTestAsOf.Add(-time.Hour)) + tr.AmountBasis = &basis + if rate != "" { + tr.CommitmentRate = &rate + } + require.NoError(t, f.store.SaveLadderTranches(f.ctx, []LadderTrancheDB{tr})) + } +} + +func plannedRows(t *testing.T, f *ceilingFixture) []LadderInFlightRow { + t.Helper() + var got []LadderInFlightRow + _, err := f.store.PlanAndSaveLadderRun(f.ctx, f.cfgID, lockTestAsOf, func(rows []LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { + got = rows + return lockTestRun(f.cfgID), nil, nil + }) + require.NoError(t, err) + return got +} + +// The reference is the pre-change algorithm: sum the decimals exactly, then +// round up to a float only when the nearest float is below the sum. +func TestLadderNetting_LegacyOnlyRowsKeepTheExactTotalBitForBit(t *testing.T) { + f := newCeilingFixture(t) + amounts := []string{"0.1", "0.2", "0.3", "1234.567891", "0.000001", "0.000001", "7.7", "19.999999"} + seedNettingRows(t, f, LadderAmountOnDemand, "", amounts...) + exact := new(big.Rat) + for _, a := range amounts { + v, ok := new(big.Rat).SetString(a) + require.True(t, ok) + exact.Add(exact, v) + } + want, err := nettingFloat(exact) + require.NoError(t, err) + got := mustOnDemand(t, plannedRows(t, f)) + require.Equal(t, math.Float64bits(want), math.Float64bits(got)) +} + +func TestLadderNetting_ThirdRateRowNeverNetsBelowTheExactRational(t *testing.T) { + f := newCeilingFixture(t) + seedNettingRows(t, f, LadderAmountCommitment, "1/3", "1.000001") + rows := plannedRows(t, f) + require.Len(t, rows, 1) + require.LessOrEqual(t, new(big.Rat).SetFloat64(rows[0].Rate).Cmp(big.NewRat(1, 3)), 0) + got := mustOnDemand(t, rows) + require.GreaterOrEqual(t, new(big.Rat).SetFloat64(got).Cmp(big.NewRat(3000003, 1000000)), 0) +} + +func TestLadderNetting_NullBasisStillFailsLoudBeforeBuilder(t *testing.T) { + f := newCeilingFixture(t) + _, ids := f.plan(fptr(10), fptr(0), fptr(100), 5) + f.setBasis(ids[0], "", "") + built := false + _, err := f.store.PlanAndSaveLadderRun(f.ctx, f.cfgID, time.Now().UTC().Truncate(time.Microsecond), func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { + built = true + return nil, nil, nil + }) + require.Error(t, err) + require.False(t, built) +} diff --git a/internal/config/store_postgres_ladder_lifecycle_integration_test.go b/internal/config/store_postgres_ladder_lifecycle_integration_test.go index b446deab..2d74ba02 100644 --- a/internal/config/store_postgres_ladder_lifecycle_integration_test.go +++ b/internal/config/store_postgres_ladder_lifecycle_integration_test.go @@ -57,10 +57,9 @@ func newExecCreator(ctx context.Context, t *testing.T, store *PostgresStore, cal func inFlight(ctx context.Context, t *testing.T, store *PostgresStore, cfgID string, asOf time.Time) float64 { t.Helper() - v, err := store.GetInFlightLadderCommitUSDHr(ctx, cfgID, asOf) + rows, err := inFlightLadderRows(ctx, store.db, true, cfgID, asOf) require.NoError(t, err) - require.NotNil(t, v) - return *v + return mustOnDemand(t, rows) } // Property: for every status x scheduled_date offset around the grace diff --git a/internal/config/store_postgres_ladder_lock.go b/internal/config/store_postgres_ladder_lock.go index 91e76873..f0bdeec3 100644 --- a/internal/config/store_postgres_ladder_lock.go +++ b/internal/config/store_postgres_ladder_lock.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "math/big" "time" "github.com/google/uuid" @@ -25,9 +24,9 @@ const ladderConfigLockPrefix = "cudly:ladder-config:" var ladderLockTimeout = 30 * time.Second // LadderPlanBuilder computes a run and its new tranches from the config's -// measured on-demand-equivalent total. It runs while the per-config lock is held inside the +// netted tranches. 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) +type LadderPlanBuilder func(inFlight []LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) // ErrLadderLockTimeout means the per-config ladder lock was not acquired within // ladderLockTimeout (SQLSTATE 55P03). The operation did nothing; retry later. @@ -123,11 +122,11 @@ func ladderSettleAnchor(asOf time.Time) time.Time { return asOf.Add(-ladderSettl func (s *PostgresStore) PlanAndSaveLadderRun(ctx context.Context, configID string, asOf time.Time, build LadderPlanBuilder) (*LadderRunDB, error) { var result LadderRunDB err := s.withLadderConfigLock(ctx, configID, func(tx pgx.Tx) error { - inFlight, err := lockedLadderCommit(ctx, tx, configID, asOf) + rows, err := lockedLadderCommit(ctx, tx, configID, asOf) if err != nil { return err } - run, tranches, err := build(inFlight) + run, tranches, err := build(rows) if err != nil { return err } @@ -158,70 +157,60 @@ func (s *PostgresStore) PlanAndSaveLadderRun(ctx context.Context, configID strin return &result, nil } -// lockedLadderCommit is the figure PlanAndSaveLadderRun hands to build: -// in-flight plus tranches settled after the skew-widened anchor. +// lockedLadderCommit is the set of rows PlanAndSaveLadderRun hands to build: +// in-flight plus tranches settled after the skew-widened anchor. Netting these +// is a deliberate deviation from go's InFlight doc (fired and settled-after-anchor +// rows are included), which errs toward under-buying. // // Order is load-bearing: in-flight FIRST, settled SECOND. Each statement takes // 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 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)) +func lockedLadderCommit(ctx context.Context, q dbConn, configID string, asOf time.Time) ([]LadderInFlightRow, error) { + inFlight, err := inFlightLadderRows(ctx, q, false, configID, asOf) if err != nil { - return 0, err + return nil, err } - settled, err := ladderNettingRows(ctx, q, "status = 'completed' AND settled_at > $2", - configID, ladderSettleAnchor(asOf)) + settled, err := settledLadderRows(ctx, q, false, configID, ladderSettleAnchor(asOf)) if err != nil { - return 0, err + return nil, err } - return nettingFloat(inFlight.Add(inFlight, settled)) + return append(inFlight, settled...), nil } -func ladderNettingRows(ctx context.Context, q dbConn, predicate string, args ...any) (*big.Rat, error) { +// inFlightLadderRows are the config's tranches that count as committed but +// unsettled: fired (resolved only by completed or failed) and scheduled rows +// still within the grace window. assumeOD is for the amend path only. +func inFlightLadderRows(ctx context.Context, q dbConn, assumeOD bool, configID string, asOf time.Time) ([]LadderInFlightRow, error) { + return ladderNettingRows(ctx, q, assumeOD, "status = $2 OR (status = $3 AND scheduled_date > $4)", + configID, string(ladder.TrancheStatusFired), string(ladder.TrancheStatusScheduled), ladderGraceCutoff(asOf)) +} + +func ladderNettingRows(ctx context.Context, q dbConn, assumeOD bool, predicate string, args ...any) ([]LadderInFlightRow, 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) + var out []LadderInFlightRow 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) + trp := &tr + if assumeOD { + trp = assumeOnDemandIfUnknown(trp) + } + row, err := newLadderInFlightRow(amount, trp) if err != nil { return nil, fmt.Errorf("net tranche %s (%s): %w", tr.ID, tr.LayerType, err) } - total.Add(total, od) + out = append(out, row) } if err := rows.Err(); err != nil { return nil, err } - return total, nil -} - -// queryRower is satisfied by pgx.Tx and the pool connection. -type queryRower interface { - QueryRow(ctx context.Context, sql string, args ...any) pgx.Row -} - -// inFlightLadderCommit sums the config's in-flight commitment; see -// GetInFlightLadderCommitUSDHr for the row classes counted and excluded. -func inFlightLadderCommit(ctx context.Context, q queryRower, configID string, asOf time.Time) (float64, error) { - var total float64 - err := q.QueryRow(ctx, ` - SELECT COALESCE(SUM(amount_usd_hr), 0) - FROM ladder_tranches - WHERE config_id = $1 - AND (status = $2 OR (status = $3 AND scheduled_date > $4)) - `, configID, string(ladder.TrancheStatusFired), string(ladder.TrancheStatusScheduled), ladderGraceCutoff(asOf)).Scan(&total) - if err != nil { - return 0, fmt.Errorf("in-flight ladder commit config_id=%s: %w", configID, err) - } - return total, nil + return out, nil } diff --git a/internal/config/store_postgres_ladder_lock_integration_test.go b/internal/config/store_postgres_ladder_lock_integration_test.go index 0cc9f832..fa7c8552 100644 --- a/internal/config/store_postgres_ladder_lock_integration_test.go +++ b/internal/config/store_postgres_ladder_lock_integration_test.go @@ -54,7 +54,7 @@ func holdLadderLock(ctx context.Context, t *testing.T, store *PostgresStore, cfg t.Helper() entered, gate, done := make(chan struct{}), make(chan struct{}), make(chan error, 1) go func() { - _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { close(entered) <-gate return lockTestRun(cfgID), nil, nil @@ -112,7 +112,7 @@ func TestLadderLock_IsPerConfig(t *testing.T) { a := seedLadderConfigWithExtID(ctx, t, store, "721111111111") b := seedLadderConfigWithExtID(ctx, t, store, "722222222222") release := holdLadderLock(ctx, t, store, a) - _, err := store.PlanAndSaveLadderRun(ctx, b, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err := store.PlanAndSaveLadderRun(ctx, b, lockTestAsOf, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { return lockTestRun(b), nil, nil }) require.NoError(t, err, "a held lock on config A must not block config B") @@ -124,14 +124,18 @@ func TestLadderLock_ReleasedOnBuildErrorPanicAndCancel(t *testing.T) { defer cancel() store := setupLadderStore(ctx, t) cfgID := seedLadderConfigWithExtID(ctx, t, store, "731111111111") - ok := func(float64) (*LadderRunDB, []LadderTrancheDB, error) { return lockTestRun(cfgID), nil, nil } //nolint:unparam // matches LadderPlanBuilder + ok := func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { + return lockTestRun(cfgID), nil, nil + } //nolint:unparam // matches LadderPlanBuilder plan := func(c context.Context, b LadderPlanBuilder) error { _, err := store.PlanAndSaveLadderRun(c, cfgID, lockTestAsOf, b) return err } t.Run("build error", func(t *testing.T) { - err := plan(ctx, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { return nil, nil, errors.New("boom") }) + err := plan(ctx, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { + return nil, nil, errors.New("boom") + }) require.ErrorContains(t, err, "boom") before := countRuns(ctx, t, store, cfgID) require.NoError(t, plan(ctx, ok)) @@ -139,7 +143,7 @@ func TestLadderLock_ReleasedOnBuildErrorPanicAndCancel(t *testing.T) { }) t.Run("build panic", func(t *testing.T) { assert.Panics(t, func() { - _ = plan(ctx, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { panic("boom") }) + _ = plan(ctx, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { panic("boom") }) }) before := countRuns(ctx, t, store, cfgID) require.NoError(t, plan(ctx, ok), "the lock must be free after a panic") @@ -166,7 +170,7 @@ func TestLadderLock_EmptyConfigIDRejected(t *testing.T) { defer cancel() store := setupLadderStore(ctx, t) built := false - _, err := store.PlanAndSaveLadderRun(ctx, "", time.Now().UTC(), func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err := store.PlanAndSaveLadderRun(ctx, "", time.Now().UTC(), func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { built = true return nil, nil, nil }) @@ -181,9 +185,9 @@ func TestLadderLock_RejectsNilRunAndStartedAtMismatch(t *testing.T) { defer cancel() store := setupLadderStore(ctx, t) cfgID := seedLadderConfigWithExtID(ctx, t, store, "751111111111") - _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { return nil, nil, nil }) + _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { return nil, nil, nil }) require.ErrorContains(t, err, "nil run") - _, err = store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf.Add(time.Second), func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err = store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf.Add(time.Second), func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { return lockTestRun(cfgID), nil, nil }) require.ErrorContains(t, err, "does not match asOf") @@ -203,7 +207,7 @@ func TestLadderLock_TimeoutBoundsOnlyTheWaitAndIsolationIsReadCommitted(t *testi defer func() { ladderLockTimeout = old }() release := holdLadderLock(ctx, t, store, cfgID) - _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { return lockTestRun(cfgID), nil, nil }) var pgErr *pgconn.PgError diff --git a/internal/config/store_postgres_ladder_netting_integration_test.go b/internal/config/store_postgres_ladder_netting_integration_test.go index 9657283a..9af689cd 100644 --- a/internal/config/store_postgres_ladder_netting_integration_test.go +++ b/internal/config/store_postgres_ladder_netting_integration_test.go @@ -24,8 +24,8 @@ func TestLadderNetting_MixedStoredBasis(t *testing.T) { 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) + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(rows []LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { + require.Equal(t, 55.0, mustOnDemand(t, rows)) return lockTestRun(cfg), nil, nil }) require.NoError(t, err) @@ -42,7 +42,7 @@ func TestLadderNetting_UnknownRefusesBeforeBuilder(t *testing.T) { 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) { + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { built = true return lockTestRun(cfg), nil, nil }) @@ -51,7 +51,7 @@ func TestLadderNetting_UnknownRefusesBeforeBuilder(t *testing.T) { 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) { + _, err = store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { built = true return lockTestRun(cfg), nil, nil }) @@ -89,8 +89,8 @@ func TestLadderNetting_SelectionBoundaries(t *testing.T) { 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) + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(rows []LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { + require.Equal(t, 68.0, mustOnDemand(t, rows)) return lockTestRun(cfg), nil, nil }) require.NoError(t, err) @@ -139,9 +139,9 @@ func TestLadderNetting_CompletionBetweenReadsCountsTwice(t *testing.T) { require.NoError(t, err) transitioned = true }} - _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(total float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(rows []LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { require.True(t, transitioned) - require.Equal(t, 6.0, total) + require.Equal(t, 6.0, mustOnDemand(t, rows)) return lockTestRun(cfg), nil, nil }) require.NoError(t, err) @@ -155,7 +155,7 @@ func TestLadderNetting_ReadFailureDoesNotBecomeZero(t *testing.T) { original := store.db store.db = nettingHookDB{dbConn: original, hook: cancel} built := false - _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + _, err := store.PlanAndSaveLadderRun(ctx, cfg, lockTestAsOf, func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { built = true return lockTestRun(cfg), nil, nil }) diff --git a/internal/config/store_postgres_ladder_settled.go b/internal/config/store_postgres_ladder_settled.go index 7b4bf8ee..f74bfaf4 100644 --- a/internal/config/store_postgres_ladder_settled.go +++ b/internal/config/store_postgres_ladder_settled.go @@ -2,27 +2,15 @@ package config import ( "context" - "fmt" "time" ) -// settledLadderCommit sums the config's completed tranches that settled after +// settledLadderRows are the config's completed tranches that settled after // anchor (strictly: settled_at == anchor is not counted). Rows completed before // the column existed were backfilled with the migration time, so they are // counted for any anchor older than it. failed and canceled rows are -// never counted. Consumers add it to the in-flight sum when the +// never counted. Consumers add them to the in-flight rows when the // existing commitment they compare against was read at or before anchor. -func settledLadderCommit(ctx context.Context, q queryRower, configID string, anchor time.Time) (float64, error) { - var total float64 - err := q.QueryRow(ctx, ` - SELECT COALESCE(SUM(amount_usd_hr), 0) - FROM ladder_tranches - WHERE config_id = $1 - AND status = 'completed' - AND settled_at > $2 - `, configID, anchor).Scan(&total) - if err != nil { - return 0, fmt.Errorf("settled ladder commit config_id=%s: %w", configID, err) - } - return total, nil +func settledLadderRows(ctx context.Context, q dbConn, assumeOD bool, configID string, anchor time.Time) ([]LadderInFlightRow, error) { + return ladderNettingRows(ctx, q, assumeOD, "status = 'completed' AND settled_at > $2", configID, anchor) } diff --git a/internal/config/store_postgres_ladder_settled_integration_test.go b/internal/config/store_postgres_ladder_settled_integration_test.go index 3c73cf14..26c1a79b 100644 --- a/internal/config/store_postgres_ladder_settled_integration_test.go +++ b/internal/config/store_postgres_ladder_settled_integration_test.go @@ -95,9 +95,9 @@ func TestPostgresStoreDB_LadderSettledAt(t *testing.T) { } before, equal, after := anchor.Add(-time.Microsecond), anchor, anchor.Add(time.Microsecond) sum := func() float64 { - v, err := settledLadderCommit(ctx, pool, otherCfg, anchor) + rows, err := settledLadderRows(ctx, pool, true, otherCfg, anchor) require.NoError(t, err) - return v + return mustOnDemand(t, rows) } assert.InDelta(t, 0, sum(), 1e-9, "no rows") add(ladder.TrancheStatusCompleted, 1, &before) @@ -138,9 +138,9 @@ func settledTestStore(ctx context.Context, t *testing.T) (*PostgresStore, *pgxpo func mustSettled(ctx context.Context, t *testing.T, pool *pgxpool.Pool, cfgID string, anchor time.Time) float64 { t.Helper() - v, err := settledLadderCommit(ctx, pool, cfgID, anchor) + rows, err := settledLadderRows(ctx, pool, true, cfgID, anchor) require.NoError(t, err) - return v + return mustOnDemand(t, rows) } func requireCheckViolation(t *testing.T, err error) { 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 bee8528f..4cc26279 100644 --- a/internal/config/store_postgres_ladder_settled_plan_integration_test.go +++ b/internal/config/store_postgres_ladder_settled_plan_integration_test.go @@ -36,8 +36,8 @@ func TestPostgresStoreDB_PlanAndSaveSettledTerm(t *testing.T) { planned := func() float64 { t.Helper() var got float64 - _, err := store.PlanAndSaveLadderRun(ctx, cfgID, asOf, func(inFlight float64) (*LadderRunDB, []LadderTrancheDB, error) { - got = inFlight + _, err := store.PlanAndSaveLadderRun(ctx, cfgID, asOf, func(rows []LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { + got = mustOnDemand(t, rows) return lockTestRunAt(cfgID, asOf), nil, nil }) require.NoError(t, err) diff --git a/internal/config/store_postgres_ladder_test.go b/internal/config/store_postgres_ladder_test.go index 22d779f4..20e39061 100644 --- a/internal/config/store_postgres_ladder_test.go +++ b/internal/config/store_postgres_ladder_test.go @@ -508,15 +508,16 @@ func seedLadderConfigWithExtID(ctx context.Context, t *testing.T, store *Postgre return saved.ID } -// TestPostgresStore_GetInFlightLadderCommitUSDHr tests the in-flight sum -// query (L5 netting). It verifies: -// - zero is returned (not nil) when no scheduled tranches exist. +// TestPostgresStore_InFlightLadderRows tests the in-flight netting rows +// (L5 netting), summed to on-demand-equivalent dollars. Seeds have no stored +// basis, so they use the amend path's assume-on-demand reading. It verifies: +// - zero is returned when no scheduled tranches exist. // - scheduled tranches still within LadderFireGrace and fired tranches are // summed; completed (already in ExistingUSDPerHour), failed and canceled // are excluded. // - scheduled tranches past the grace window are excluded (issue #118). // - tranches from a different config are not included. -func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { +func TestPostgresStore_InFlightLadderRows(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) defer cancel() store := setupLadderStore(ctx, t) @@ -545,6 +546,12 @@ func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { } require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) } + inFlightOD := func(t *testing.T, cID string) float64 { + t.Helper() + rows, err := inFlightLadderRows(ctx, store.db, true, cID, asOf) + require.NoError(t, err) + return mustOnDemand(t, rows) + } insertTranche := func(t *testing.T, cID *string, amount float64, status ladder.TrancheStatus) { t.Helper() insertTrancheAt(t, cID, amount, status, fire) @@ -553,10 +560,7 @@ func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { t.Run("zero when no tranches exist", func(t *testing.T) { // Use the other config which has no tranches yet. otherID := otherConfigID - result, err := store.GetInFlightLadderCommitUSDHr(ctx, otherID, asOf) - require.NoError(t, err) - require.NotNil(t, result, "must return non-nil *float64 even when sum is zero") - assert.InDelta(t, 0.0, *result, 1e-6) + assert.InDelta(t, 0.0, inFlightOD(t, otherID), 1e-6) }) // Insert tranches with various statuses for the main config. @@ -577,21 +581,15 @@ func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { insertTranche(t, &otherID, 99.0, ladder.TrancheStatusScheduled) t.Run("sums fireable scheduled and fired", func(t *testing.T) { - result, err := store.GetInFlightLadderCommitUSDHr(ctx, configID, asOf) - require.NoError(t, err) - require.NotNil(t, result) // 3.0 future + 6.0 due-within-grace scheduled + 2.0 fired. Completed is // already counted in ExistingUSDPerHour (netting it would double-subtract); // failed, canceled and past-grace scheduled rows were never bought. - assert.InDelta(t, 11.0, *result, 1e-6, "in-flight = fireable scheduled + fired") + assert.InDelta(t, 11.0, inFlightOD(t, configID), 1e-6, "in-flight = fireable scheduled + fired") }) t.Run("other config is not contaminated", func(t *testing.T) { - result, err := store.GetInFlightLadderCommitUSDHr(ctx, otherConfigID, asOf) - require.NoError(t, err) - require.NotNil(t, result) // Only the 99.0 (scheduled) tranche we just inserted for the other config. - assert.InDelta(t, 99.0, *result, 1e-6, "other config must see only its own tranches") + assert.InDelta(t, 99.0, inFlightOD(t, otherConfigID), 1e-6, "other config must see only its own tranches") }) } @@ -642,10 +640,9 @@ func TestPostgresStore_SaveLadderRunWithTranches_AppendOnly(t *testing.T) { // In-flight sum reflects BOTH generations (append-only accumulation until // tranches fire); netting will cap future runs, not the store. - inFlight, err := store.GetInFlightLadderCommitUSDHr(ctx, cfgID, time.Now().UTC()) + inFlight, err := inFlightLadderRows(ctx, store.db, true, cfgID, time.Now().UTC()) require.NoError(t, err) - require.NotNil(t, inFlight) - assert.InDelta(t, 5.5, *inFlight, 1e-6, "in-flight must sum both live scheduled generations (3.0 + 2.5)") + assert.InDelta(t, 5.5, mustOnDemand(t, inFlight), 1e-6, "in-flight must sum both live scheduled generations (3.0 + 2.5)") } // TestPostgresStore_SaveLadderRunWithTranches_CancelsOverdue proves #544: with diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index f578c310..62fc2e3e 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -2023,7 +2023,7 @@ func (m *MockConfigStore) SaveLadderRun(ctx context.Context, run *config.LadderR func (m *MockConfigStore) PlanAndSaveLadderRun(ctx context.Context, configID string, asOf time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { m.record("PlanAndSaveLadderRun", ctx, configID, asOf) if !isExpected(&m.Mock, "PlanAndSaveLadderRun") { - run, _, err := build(0) + run, _, err := build(nil) return run, err } args := m.Called(ctx, configID, asOf) diff --git a/internal/purchase/ladder_reconcile_integration_test.go b/internal/purchase/ladder_reconcile_integration_test.go index bc0bf52a..36cbe15a 100644 --- a/internal/purchase/ladder_reconcile_integration_test.go +++ b/internal/purchase/ladder_reconcile_integration_test.go @@ -8,6 +8,7 @@ package purchase import ( "context" + "errors" "testing" "time" @@ -71,9 +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 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} 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) { @@ -99,10 +101,16 @@ func (r *ladderRig) state(trancheID string) (status string, reason *string) { } func (r *ladderRig) inFlight() float64 { - v, err := r.store.GetInFlightLadderCommitUSDHr(r.ctx, r.cfgID, time.Now()) - require.NoError(r.t, err) - require.NotNil(r.t, v) - return *v + var v float64 + abort := errors.New("read only") + _, err := r.store.PlanAndSaveLadderRun(r.ctx, r.cfgID, time.Now(), func(rows []config.LadderInFlightRow) (*config.LadderRunDB, []config.LadderTrancheDB, error) { + var err error + v, err = config.LadderInFlightOnDemand(rows) + require.NoError(r.t, err) + return nil, nil, abort + }) + require.ErrorIs(r.t, err, abort) + return v } func TestReconcileLadderTranches_Mappings(t *testing.T) { diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go index eb8f35d5..6a0dbc40 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -275,13 +275,13 @@ func (app *Application) processOneLadderConfig( // commitment (scheduled tranches still fireable, plus fired tranches not yet // resolved). The Allocate engine subtracts this from the raw gap so each daily // run plans only what is still needed, preventing pile-up when prior tranches -// have not yet been bought. Completed tranches are deliberately NOT counted -// here: they are executed purchases already reflected in the engine's -// ExistingUSDPerHour, so counting them again would double-subtract. +// 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: tranche amounts are stored exactly as Allocate emits them, which is // ON-DEMAND-EQUIVALENT $/hr until Allocate converts to commitment dollars -// (go#349). Do not convert here. +// (go#349). Do not convert here. The store nets each stored row to +// on-demand-equivalent dollars by its own basis and rate. // // Persistence is APPEND-ONLY (see persistLadderRun): every run inserts its new // scheduled tranches and NEVER supersedes existing ones. Netting already @@ -355,7 +355,11 @@ type ladderRunPlanner struct { allocations, tranches, holds int // set by plan, for the log line } -func (p *ladderRunPlanner) plan(inFlight float64) (*config.LadderRunDB, []config.LadderTrancheDB, error) { +func (p *ladderRunPlanner) plan(rows []config.LadderInFlightRow) (*config.LadderRunDB, []config.LadderTrancheDB, error) { + inFlight, err := config.LadderInFlightOnDemand(rows) + if err != nil { + return nil, nil, fmt.Errorf("net in-flight tranches: %w", err) + } allocResult, err := pkgladder.Allocate(&pkgladder.AllocationInput{ Now: p.now, Baseline: p.baseline, diff --git a/internal/server/handler_ladder_lock_integration_test.go b/internal/server/handler_ladder_lock_integration_test.go index 71b7c22f..2478fe12 100644 --- a/internal/server/handler_ladder_lock_integration_test.go +++ b/internal/server/handler_ladder_lock_integration_test.go @@ -31,7 +31,7 @@ type hookedLadderStore struct { } func (h *hookedLadderStore) PlanAndSaveLadderRun(ctx context.Context, configID string, asOf time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { - return h.PostgresStore.PlanAndSaveLadderRun(ctx, configID, asOf, func(inFlight float64) (*config.LadderRunDB, []config.LadderTrancheDB, error) { + return h.PostgresStore.PlanAndSaveLadderRun(ctx, configID, asOf, func(inFlight []config.LadderInFlightRow) (*config.LadderRunDB, []config.LadderTrancheDB, error) { if h.calls.Add(1) == 1 { close(h.entered) <-h.gate diff --git a/internal/server/handler_ladder_test.go b/internal/server/handler_ladder_test.go index df820a1f..edbd7ca1 100644 --- a/internal/server/handler_ladder_test.go +++ b/internal/server/handler_ladder_test.go @@ -194,11 +194,11 @@ func (s *ladderTestStore) PlanAndSaveLadderRun(_ context.Context, _ string, _ ti if s.inFlightErr != nil { return nil, s.inFlightErr } - inFlight := 0.0 + var rows []config.LadderInFlightRow if s.inFlightUSDHr != nil { - inFlight = *s.inFlightUSDHr + rows = []config.LadderInFlightRow{{Layer: pkgladder.LayerComputeSP, CommitmentUSDPerHour: *s.inFlightUSDHr, Rate: 1}} } - run, tranches, err := build(inFlight) + run, tranches, err := build(rows) if err != nil { return nil, err } @@ -1226,7 +1226,7 @@ func (s *ladderStatefulStore) PlanAndSaveLadderRun(_ context.Context, configID s inFlight += tr.AmountUSDHr } } - run, tranches, err := build(inFlight) + run, tranches, err := build([]config.LadderInFlightRow{{Layer: pkgladder.LayerComputeSP, CommitmentUSDPerHour: inFlight, Rate: 1}}) if err != nil { return nil, err } diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 80ad2ec6..2938c89f 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -431,7 +431,7 @@ func (m *mockConfigStoreForHealth) SaveLadderRun(_ context.Context, run *config. return run, nil } func (m *mockConfigStoreForHealth) PlanAndSaveLadderRun(_ context.Context, _ string, _ time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { - run, _, err := build(0) + run, _, err := build(nil) return run, err } func (m *mockConfigStoreForHealth) GetLadderRun(_ context.Context, _ string) (*config.LadderRunDB, error) { From f4a286af7a82a2f1f6e7c78293f441a85384ea0b Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Fri, 16 Oct 2026 14:02:13 +0200 Subject: [PATCH 2/2] style(ladder): keep the unparam suppression on the builder func line Co-Authored-By: Claude Opus 5.5 --- .../config/store_postgres_ladder_lock_integration_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/config/store_postgres_ladder_lock_integration_test.go b/internal/config/store_postgres_ladder_lock_integration_test.go index fa7c8552..92546b6b 100644 --- a/internal/config/store_postgres_ladder_lock_integration_test.go +++ b/internal/config/store_postgres_ladder_lock_integration_test.go @@ -124,9 +124,9 @@ func TestLadderLock_ReleasedOnBuildErrorPanicAndCancel(t *testing.T) { defer cancel() store := setupLadderStore(ctx, t) cfgID := seedLadderConfigWithExtID(ctx, t, store, "731111111111") - ok := func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { + ok := func([]LadderInFlightRow) (*LadderRunDB, []LadderTrancheDB, error) { //nolint:unparam // matches LadderPlanBuilder return lockTestRun(cfgID), nil, nil - } //nolint:unparam // matches LadderPlanBuilder + } plan := func(c context.Context, b LadderPlanBuilder) error { _, err := store.PlanAndSaveLadderRun(c, cfgID, lockTestAsOf, b) return err