fix(ladder): net in-flight tranches per row basis in on-demand-equivalent dollars

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) {

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)
+}

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")
+}

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).

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

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

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)
+}

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