fix(ladder): net locked tranches in on-demand units

internal/config/ladder_provenance.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"fmt"
+	"math"
 	"math/big"
 )
 
@@ -29,13 +30,58 @@ 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
 	default:
 		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
+}

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

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)

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.

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

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