From 7a97c676949d357bd753a2b69c5ab26ff81e9903 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 16 Jul 2026 21:14:22 +0300 Subject: [PATCH 1/4] feat(ladder): L5 tranche-netting convergence fix Fixes the G2+G4 convergence gap: without netting, each daily run ignores its own not-yet-fired scheduled tranches, re-measures the full gap, piles up tranches, and overshoots the low-water floor. - AllocationInput.InFlightUSDPerHour (*float64, required non-nil): gap is now min(target-E, lowWater-E) - inFlight so repeated runs on the same config converge to ~zero rather than accumulate. - StoreInterface gains GetInFlightLadderCommitUSDHr (SUM of amount_usd_hr for status IN ('scheduled','fired') per config) and SaveLadderRunWithTranchesAndSupersede (atomic cancel-and- replace: cancels prior scheduled tranches then inserts run+tranches in one tx, at most one generation per config). - executeLadderRun fetches in-flight before Allocate; persistLadderRun calls SaveLadderRunWithTranchesAndSupersede. - All mocks and test helpers updated (analytics, server, shared mock). - Convergence regression: 3 sequential same-usage runs -> only run 1 allocates, runs 2-3 hold; canary confirms zero-in-flight diverges. - Integration tests: sum+cancel round-trip, per-config isolation, atomicity rollback on duplicate tranche ID. --- internal/analytics/collector_test.go | 7 + internal/config/interfaces.go | 16 + internal/config/store_postgres_ladder.go | 68 +++ internal/config/store_postgres_ladder_test.go | 213 +++++++ internal/mocks/stores.go | 37 ++ internal/server/handler_ladder.go | 60 +- internal/server/handler_ladder_test.go | 45 +- internal/server/test_helpers_test.go | 7 + pkg/ladder/allocate.go | 61 +- pkg/ladder/allocate_test.go | 561 +++++++++++------- 10 files changed, 831 insertions(+), 244 deletions(-) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index adcd046eb..907970251 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -759,6 +759,13 @@ func (m *mockConfigStore) SaveLadderRun(_ context.Context, run *config.LadderRun func (m *mockConfigStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { return run, nil } +func (m *mockConfigStore) SaveLadderRunWithTranchesAndSupersede(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { + return run, nil +} +func (m *mockConfigStore) GetInFlightLadderCommitUSDHr(_ context.Context, _ string) (*float64, error) { + zero := 0.0 + return &zero, nil +} func (m *mockConfigStore) GetLadderRun(_ context.Context, _ string) (*config.LadderRunDB, error) { return nil, nil } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 175b9fd9e..2cc45073a 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -378,8 +378,24 @@ type StoreInterface interface { // transaction: a tranche-insert failure rolls back the run row too, so a // status=planned run never persists without its tranches (which would let // the cadence gate suppress the retry for a full window). + // + // SaveLadderRunWithTranchesAndSupersede is the atomic cancel-and-replace + // variant (L5 spec). Within a single transaction it: + // 1. Cancels all status=scheduled tranches for the run's config_id, + // 2. Inserts the new ladder_runs row, and + // 3. Inserts the new ladder_tranches rows. + // This guarantees at most one generation of scheduled tranches per config + // exists at any point; a partial failure rolls back all three steps. + // + // GetInFlightLadderCommitUSDHr returns the total hourly USD commitment in + // flight for the given config: the sum of amount_usd_hr for all tranches + // with status IN ('scheduled', 'fired'). Returns a non-nil pointer (zero + // when no in-flight tranches exist) so callers can pass it directly to + // AllocationInput.InFlightUSDPerHour. Never returns nil without an error. SaveLadderRun(ctx context.Context, run *LadderRunDB) (*LadderRunDB, error) SaveLadderRunWithTranches(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) + SaveLadderRunWithTranchesAndSupersede(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) + GetInFlightLadderCommitUSDHr(ctx context.Context, configID string) (*float64, error) GetLadderRun(ctx context.Context, id string) (*LadderRunDB, error) SaveLadderTranches(ctx context.Context, tranches []LadderTrancheDB) error LatestLadderRunStartedAt(ctx context.Context, configID string) (*time.Time, error) diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index 5a3097361..16535bb97 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -361,6 +361,74 @@ func (s *PostgresStore) SaveLadderRunWithTranches(ctx context.Context, run *Ladd return &result, nil } +// GetInFlightLadderCommitUSDHr returns the total hourly USD commitment already +// in flight for the given config: the SUM of amount_usd_hr for all +// ladder_tranches rows where config_id=$1 and status IN ('scheduled','fired'). +// Returns a non-nil *float64 (zero when no in-flight tranches exist); 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) (*float64, error) { + var total float64 + err := s.db.QueryRow(ctx, ` + SELECT COALESCE(SUM(amount_usd_hr), 0) + FROM ladder_tranches + WHERE config_id = $1 + AND status IN ('scheduled', 'fired') + `, configID).Scan(&total) + if err != nil { + return nil, fmt.Errorf("GetInFlightLadderCommitUSDHr config_id=%s: %w", configID, err) + } + return &total, nil +} + +// SaveLadderRunWithTranchesAndSupersede is the atomic cancel-and-replace +// variant of SaveLadderRunWithTranches (L5 netting spec). Within a single +// transaction it: +// 1. Cancels all status=scheduled tranches for the run's config_id (sets +// status='cancelled'), ensuring exactly one generation of scheduled +// tranches exists per config after the call. +// 2. Inserts the new ladder_runs row. +// 3. Inserts the new ladder_tranches rows. +// +// If any step fails the whole transaction is rolled back. If run.ConfigID is +// nil the cancel step is skipped (the run has no prior config context). If +// run.ID is empty a fresh UUID is generated before the insert. +func (s *PostgresStore) SaveLadderRunWithTranchesAndSupersede(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) { + if run.ID == "" { + run.ID = uuid.New().String() + } + var result LadderRunDB + err := s.WithTx(ctx, func(tx pgx.Tx) error { + // Step 1: cancel prior scheduled tranches for this config. + if run.ConfigID != nil { + _, err := tx.Exec(ctx, ` + UPDATE ladder_tranches + SET status = 'cancelled' + WHERE config_id = $1 AND status = 'scheduled' + `, *run.ConfigID) + if err != nil { + return fmt.Errorf("cancel prior scheduled tranches for config_id=%s: %w", *run.ConfigID, err) + } + } + // Step 2: insert the new run row. + row := tx.QueryRow(ctx, ladderRunInsertQuery, ladderRunInsertArgs(run, ladderRunPlanJSON(run))...) + scanned, err := scanLadderRun(row) + if err != nil { + return fmt.Errorf("failed to insert ladder_run id=%s: %w", run.ID, err) + } + // Step 3: insert the new tranche rows. + if err := insertLadderTranchesTx(ctx, tx, tranches); err != nil { + return err + } + result = scanned + return nil + }) + if err != nil { + return nil, err + } + return &result, 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_test.go b/internal/config/store_postgres_ladder_test.go index 6bf329ad9..02401d95b 100644 --- a/internal/config/store_postgres_ladder_test.go +++ b/internal/config/store_postgres_ladder_test.go @@ -359,3 +359,216 @@ func TestPostgresStore_SaveLadderRunWithTranches_Atomicity(t *testing.T) { assert.Equal(t, 0, count, "no tranche may survive a rolled-back transaction") }) } + +// seedLadderConfigWithExtID seeds a cloud_account (with the given externalID +// to avoid the UNIQUE(provider, external_id) constraint) and a ladder_config, +// returning the ladder_config id. +func seedLadderConfigWithExtID(ctx context.Context, t *testing.T, store *PostgresStore, externalID string) string { + t.Helper() + acctID := uuid.New().String() + _, err := store.db.Exec(ctx, ` + INSERT INTO cloud_accounts + (id, name, enabled, provider, external_id, aws_is_org_root, created_at, updated_at) + VALUES ($1, 'Ladder Test Account', true, 'aws', $2, false, now(), now()) + `, acctID, externalID) + require.NoError(t, err, "seed cloud_accounts (ext=%s)", externalID) + + cfg := &LadderConfigDB{ + CloudAccountID: acctID, + Provider: "aws", + Enabled: true, + Mode: "email_approval", + Cadence: "daily", + TargetCoverage: 80.0, + BufferFraction: 0.1, + BaselinePercentile: 5.0, + LookbackDays: 30, + BufferUtilizationThreshold: 50.0, + MaxActionsPerRun: 5, + RampSchedule: []byte(`{"steps":[{"after_days":0,"fraction":1.0}]}`), + } + saved, err := store.UpsertLadderConfig(ctx, cfg) + require.NoError(t, err, "seed ladder_config (ext=%s)", externalID) + require.NotEmpty(t, saved.ID) + return saved.ID +} + +// TestPostgresStore_GetInFlightLadderCommitUSDHr tests the in-flight sum +// query (L5 netting). It verifies: +// - zero is returned (not nil) when no tranches exist. +// - scheduled + fired tranches are summed; completed/cancelled/failed are excluded. +// - tranches from a different config are not included. +func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + // Use distinct external_ids to satisfy UNIQUE(provider, external_id). + configID := seedLadderConfigWithExtID(ctx, t, store, "111111111111") + otherConfigID := seedLadderConfigWithExtID(ctx, t, store, "222222222222") + cfgID := configID + fire := time.Now().UTC().Add(7 * 24 * time.Hour).Truncate(time.Microsecond) + + insertTranche := func(t *testing.T, cID *string, amount float64, status ladder.TrancheStatus) { + t.Helper() + tr := LadderTrancheDB{ + ID: uuid.New().String(), + ConfigID: cID, + LayerType: ladder.LayerConvertibleRI, + Term: ladder.Term1Year, + PaymentOption: ladder.PaymentNoUpfront, + Status: status, + AmountUSDHr: amount, + ScheduledDate: fire, + } + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) + } + + 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) + 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) + }) + + // Insert tranches with various statuses for the main config. + insertTranche(t, &cfgID, 3.0, ladder.TrancheStatusScheduled) // included + insertTranche(t, &cfgID, 2.0, ladder.TrancheStatusFired) // included + insertTranche(t, &cfgID, 5.0, ladder.TrancheStatusCompleted) // excluded (terminal) + insertTranche(t, &cfgID, 1.0, ladder.TrancheStatusCancelled) // excluded (terminal) + insertTranche(t, &cfgID, 4.0, ladder.TrancheStatusFailed) // excluded (terminal) + + // A tranche belonging to a different config must not be summed in. + otherID := otherConfigID + insertTranche(t, &otherID, 99.0, ladder.TrancheStatusScheduled) + + t.Run("sums scheduled and fired only", func(t *testing.T) { + result, err := store.GetInFlightLadderCommitUSDHr(ctx, configID) + require.NoError(t, err) + require.NotNil(t, result) + // Only 3.0 (scheduled) + 2.0 (fired) = 5.0 must be returned. + assert.InDelta(t, 5.0, *result, 1e-6, "in-flight sum must include scheduled+fired only") + }) + + t.Run("other config is not contaminated", func(t *testing.T) { + result, err := store.GetInFlightLadderCommitUSDHr(ctx, otherConfigID) + 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") + }) +} + +// TestPostgresStore_SaveLadderRunWithTranchesAndSupersede tests the L5 atomic +// cancel-and-replace: +// - prior scheduled tranches for the config are marked cancelled within the +// same transaction that inserts the new run+tranches. +// - fired/completed tranches are NOT touched. +// - a tranche-insert failure (duplicate ID) rolls back ALL three steps so no +// orphaned cancellations or run rows persist. +// - two configs remain isolated: supersede for config A does not cancel +// config B's tranches. +func TestPostgresStore_SaveLadderRunWithTranchesAndSupersede(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + // Use distinct external_ids to satisfy UNIQUE(provider, external_id). + configID := seedLadderConfigWithExtID(ctx, t, store, "333333333333") + otherConfigID := seedLadderConfigWithExtID(ctx, t, store, "444444444444") + cfgID := configID + otherID := otherConfigID + fire := time.Now().UTC().Add(7 * 24 * time.Hour).Truncate(time.Microsecond) + + insertTranche := func(t *testing.T, cID *string, status ladder.TrancheStatus) string { + t.Helper() + id := uuid.New().String() + tr := LadderTrancheDB{ + ID: id, + ConfigID: cID, + LayerType: ladder.LayerConvertibleRI, + Term: ladder.Term1Year, + PaymentOption: ladder.PaymentNoUpfront, + Status: status, + AmountUSDHr: 1.0, + ScheduledDate: fire, + } + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) + return id + } + + statusOf := func(t *testing.T, id string) string { + t.Helper() + var s string + require.NoError(t, store.db.QueryRow(ctx, + `SELECT status FROM ladder_tranches WHERE id = $1`, id).Scan(&s)) + return s + } + + // Seed prior generation for the main config. + priorScheduledID := insertTranche(t, &cfgID, ladder.TrancheStatusScheduled) + priorFiredID := insertTranche(t, &cfgID, ladder.TrancheStatusFired) // must survive supersede + // Seed a tranche for another config; it must not be touched. + otherScheduledID := insertTranche(t, &otherID, ladder.TrancheStatusScheduled) + + t.Run("cancel-and-replace succeeds atomically", func(t *testing.T) { + runID := uuid.New().String() + run := &LadderRunDB{ID: runID, ConfigID: &cfgID, StartedAt: time.Now().UTC(), Status: ladder.RunStatusPlanned} + newTr := LadderTrancheDB{ + ID: uuid.New().String(), + ConfigID: &cfgID, + RunID: &runID, + LayerType: ladder.LayerConvertibleRI, + Term: ladder.Term1Year, + PaymentOption: ladder.PaymentNoUpfront, + Status: ladder.TrancheStatusScheduled, + AmountUSDHr: 2.5, + ScheduledDate: fire, + } + + saved, err := store.SaveLadderRunWithTranchesAndSupersede(ctx, run, []LadderTrancheDB{newTr}) + require.NoError(t, err) + require.NotNil(t, saved) + assert.Equal(t, runID, saved.ID) + + // Prior scheduled tranche must now be cancelled. + assert.Equal(t, "cancelled", statusOf(t, priorScheduledID), "prior scheduled tranche must be cancelled") + // Fired tranche must be untouched (not in the scheduled->cancelled CAS). + assert.Equal(t, "fired", statusOf(t, priorFiredID), "fired tranche must not be touched") + // Other config's scheduled tranche must be untouched. + assert.Equal(t, "scheduled", statusOf(t, otherScheduledID), "other config's tranche must not be cancelled") + + // The new tranche must be persisted with the correct status. + var count int + require.NoError(t, store.db.QueryRow(ctx, + `SELECT count(*) FROM ladder_tranches WHERE run_id = $1 AND status = 'scheduled'`, runID).Scan(&count)) + assert.Equal(t, 1, count, "new scheduled tranche must be persisted") + }) + + t.Run("rollback on duplicate tranche ID: no orphan cancellations", func(t *testing.T) { + // Seed a fresh scheduled tranche that should NOT be cancelled if tx rolls back. + freshScheduledID := insertTranche(t, &cfgID, ladder.TrancheStatusScheduled) + + runID := uuid.New().String() + run := &LadderRunDB{ID: runID, ConfigID: &cfgID, StartedAt: time.Now().UTC(), Status: ladder.RunStatusPlanned} + // Two identical IDs will cause a PK violation on the second insert. + dupID := uuid.New().String() + bad1 := LadderTrancheDB{ID: dupID, ConfigID: &cfgID, RunID: &runID, + LayerType: ladder.LayerConvertibleRI, Term: ladder.Term1Year, + PaymentOption: ladder.PaymentNoUpfront, Status: ladder.TrancheStatusScheduled, + AmountUSDHr: 1.0, ScheduledDate: fire} + bad2 := bad1 // same ID + + _, err := store.SaveLadderRunWithTranchesAndSupersede(ctx, run, []LadderTrancheDB{bad1, bad2}) + require.Error(t, err, "duplicate tranche ID must fail the transaction") + + // The fresh scheduled tranche must still be scheduled (cancellations rolled back). + assert.Equal(t, "scheduled", statusOf(t, freshScheduledID), + "rollback must undo the cancellation of prior scheduled tranches") + + // The run row must not exist. + got, err := store.GetLadderRun(ctx, runID) + require.NoError(t, err) + assert.Nil(t, got, "run row must be rolled back on tranche insert failure") + }) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index cc7deb90c..eb1c6fac3 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1444,6 +1444,43 @@ func (m *MockConfigStore) LatestLadderRunStartedAt(ctx context.Context, configID return v, args.Error(1) } +// GetInFlightLadderCommitUSDHr mocks the GetInFlightLadderCommitUSDHr +// operation. Returns a pointer to zero (no in-flight commitment) when no +// expectation is registered, matching the happy-path default. +func (m *MockConfigStore) GetInFlightLadderCommitUSDHr(ctx context.Context, configID string) (*float64, error) { + if !isExpected(&m.Mock, "GetInFlightLadderCommitUSDHr") { + zero := 0.0 + return &zero, nil + } + args := m.Called(ctx, configID) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).(*float64) + if !ok { + panic(fmt.Sprintf("mock: expected *float64, got %T", args.Get(0))) + } + return v, args.Error(1) +} + +// SaveLadderRunWithTranchesAndSupersede mocks the atomic cancel-and-replace +// operation (L5 spec). Returns the run unchanged when no expectation is +// registered, mirroring the SaveLadderRunWithTranches default. +func (m *MockConfigStore) SaveLadderRunWithTranchesAndSupersede(ctx context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { + if !isExpected(&m.Mock, "SaveLadderRunWithTranchesAndSupersede") { + return run, nil + } + args := m.Called(ctx, run, tranches) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).(*config.LadderRunDB) + if !ok { + panic(fmt.Sprintf("mock: expected *config.LadderRunDB, got %T", args.Get(0))) + } + return v, args.Error(1) +} + // TransitionLadderRunStatus mocks the TransitionLadderRunStatus operation. // Returns (nil, nil) when no expectation is registered (CAS race-lost path). func (m *MockConfigStore) TransitionLadderRunStatus(ctx context.Context, id string, fromStatuses []ladder.RunStatus, toStatus ladder.RunStatus) (*config.LadderRunDB, error) { diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go index f3cd38007..e34dfee03 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -272,6 +272,18 @@ func (app *Application) processOneLadderConfig( // // It is split from handleLadderRun so tests can inject a hermetic // LadderCapability (fake) and a mock store without touching AWS or PostgreSQL. +// +// L5 netting: before calling Allocate, the handler fetches the total in-flight +// commitment (scheduled + fired non-terminal tranches plus any in-progress +// purchase executions linked to this config). 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 fired. +// +// The cancel-and-replace is performed atomically inside +// SaveLadderRunWithTranchesAndSupersede: the same transaction that inserts the +// new run and its tranches also cancels the config's prior scheduled (not yet +// fired) tranches, ensuring at most one generation of scheduled tranches exists +// for each config at any point in time. func (app *Application) executeLadderRun( ctx context.Context, dbCfg *config.LadderConfigDB, @@ -300,16 +312,35 @@ func (app *Application) executeLadderRun( return fmt.Errorf("GetLayerStates: %w", err) } + // L5: Fetch in-flight commitment BEFORE Allocate so the engine can subtract + // it from the gap. In-flight = scheduled (not-yet-fired) tranches + + // fired tranches whose execution has not reached a terminal state + + // any in-progress purchase executions linked to this config's runs. + // This query includes prior scheduled tranches (from earlier runs) so the + // engine correctly accounts for en-route commitment and produces a ~zero + // gap (Hold) when prior tranches already cover the target. + inFlight, err := app.Config.GetInFlightLadderCommitUSDHr(ctx, dbCfg.ID) + if err != nil { + return fmt.Errorf("GetInFlightLadderCommitUSDHr: %w", err) + } + if inFlight == nil { + // The store must return a non-nil value; nil signals an impossible + // query state (e.g. driver bug). Treat as fail-loud. + return fmt.Errorf("GetInFlightLadderCommitUSDHr returned nil for config %s", dbCfg.ID) + } + supportedLayers := capability.SupportedLayers() - // Run Allocate. + // Run Allocate with the in-flight figure so the gap is net of prior + // scheduled and fired tranches. allocResult, err := pkgladder.Allocate(&pkgladder.AllocationInput{ - Now: now, - Baseline: baseline, - LayerStates: layerStates, - Layers: supportedLayers, - Config: engineCfg, - DataSources: []string{dataSourceAWSCostExplorer}, + Now: now, + Baseline: baseline, + LayerStates: layerStates, + Layers: supportedLayers, + Config: engineCfg, + DataSources: []string{dataSourceAWSCostExplorer}, + InFlightUSDPerHour: inFlight, }) if err != nil { return fmt.Errorf("allocate: %w", err) @@ -334,6 +365,9 @@ func (app *Application) executeLadderRun( plan := assembleLadderPlan(scope, now, allocResult, layerStates, baseline, engineCfg.TargetCoveragePct, term, paymentOpt) // Validate and persist the run row + tranche audit rows. + // SaveLadderRunWithTranchesAndSupersede cancels any existing scheduled + // tranches for this config within the same transaction, then inserts the + // new run + tranches atomically (cancel-and-replace, L5 spec). return app.persistLadderRun(ctx, dbCfg, runID, now, plan, baseline, allocResult, trancheResult) } @@ -390,12 +424,14 @@ func (app *Application) persistLadderRun( return fmt.Errorf("buildTrancheDBRows: %w", err) } - // Persist the run row AND its tranches in a single transaction so a - // tranche-insert failure can never leave a status=planned run with no - // tranches (which the 20h cadence gate would then suppress a retry of). - savedRun, err := app.Config.SaveLadderRunWithTranches(ctx, runRow, trancheRows) + // L5: persist run + tranches atomically AND cancel any prior scheduled + // tranches for this config in the same transaction. This guarantees at + // most one generation of scheduled tranches per config exists at any + // point, preventing pile-up when repeated daily runs each see the full + // gap before prior tranches fire. + savedRun, err := app.Config.SaveLadderRunWithTranchesAndSupersede(ctx, runRow, trancheRows) if err != nil { - return fmt.Errorf("SaveLadderRunWithTranches: %w", err) + return fmt.Errorf("SaveLadderRunWithTranchesAndSupersede: %w", err) } log.Printf("ladder_run: config %s: persisted run %s (status=%s, allocations=%d, tranches=%d, holds=%d)", diff --git a/internal/server/handler_ladder_test.go b/internal/server/handler_ladder_test.go index da04c670b..1bf2849db 100644 --- a/internal/server/handler_ladder_test.go +++ b/internal/server/handler_ladder_test.go @@ -123,6 +123,11 @@ type ladderTestStore struct { cloudAcctByID map[string]*config.CloudAccount cloudAcctErrByID map[string]error + // L5: configurable in-flight hourly commitment returned by + // GetInFlightLadderCommitUSDHr. Defaults to zero (nil -> 0.0 pointer). + inFlightUSDHr *float64 + inFlightErr error + // Capture fields: non-nil if the call was made. savedRuns accumulates every // SaveLadderRun call so multi-config tests can assert on each persisted run; // savedRun mirrors the most recent one for single-config convenience. @@ -182,9 +187,9 @@ func (s *ladderTestStore) SaveLadderTranches(_ context.Context, tranches []confi return nil } -// SaveLadderRunWithTranches models the single-transaction persist the handler -// now uses. It mirrors atomicity: if either the run or the tranche insert is -// configured to fail, NOTHING is captured (the transaction rolls back). +// SaveLadderRunWithTranches models the single-transaction persist (without the +// cancel-and-replace step). Kept for tests that exercise the non-superseding +// path directly. func (s *ladderTestStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { if s.saveLadderRunErr != nil { return nil, s.saveLadderRunErr @@ -198,6 +203,36 @@ func (s *ladderTestStore) SaveLadderRunWithTranches(_ context.Context, run *conf return run, nil } +// SaveLadderRunWithTranchesAndSupersede models the L5 cancel-and-replace. It +// captures the same fields as SaveLadderRunWithTranches; a real DB would also +// mark prior scheduled tranches cancelled, but the unit test only needs to +// verify the handler wires the correct method. +func (s *ladderTestStore) SaveLadderRunWithTranchesAndSupersede(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { + if s.saveLadderRunErr != nil { + return nil, s.saveLadderRunErr + } + if s.saveLadderTranchesErr != nil { + return nil, s.saveLadderTranchesErr + } + s.savedRun = run + s.savedRuns = append(s.savedRuns, run) + s.savedTranches = tranches + return run, nil +} + +// GetInFlightLadderCommitUSDHr returns the configurable in-flight value. +// When inFlightUSDHr is nil, returns a pointer to 0.0 (no in-flight default). +func (s *ladderTestStore) GetInFlightLadderCommitUSDHr(_ context.Context, _ string) (*float64, error) { + if s.inFlightErr != nil { + return nil, s.inFlightErr + } + if s.inFlightUSDHr != nil { + return s.inFlightUSDHr, nil + } + zero := 0.0 + return &zero, nil +} + // ============================================================ // Shared test helpers // ============================================================ @@ -1055,8 +1090,10 @@ func TestRatToFloat64Ptr_PositiveValue(t *testing.T) { // Build a *big.Rat from an Allocation's GapUSDPerHour (same boundary path // used in the handler). ratToFloat64Ptr lives in this package so no import // of math/big is needed in the test. + zeroInflight := 0.0 allocResult, err := pkgladder.Allocate(&pkgladder.AllocationInput{ - Now: time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC), + Now: time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC), + InFlightUSDPerHour: &zeroInflight, Baseline: pkgladder.UsageBaseline{ LowWaterUSDPerHour: nonZeroFloat64(5.0), StableUSDPerHour: nonZeroFloat64(4.5), diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 6a628bde7..57787c7f2 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -341,6 +341,13 @@ func (m *mockConfigStoreForHealth) SaveLadderRun(_ context.Context, run *config. func (m *mockConfigStoreForHealth) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { return run, nil } +func (m *mockConfigStoreForHealth) SaveLadderRunWithTranchesAndSupersede(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { + return run, nil +} +func (m *mockConfigStoreForHealth) GetInFlightLadderCommitUSDHr(_ context.Context, _ string) (*float64, error) { + zero := 0.0 + return &zero, nil +} func (m *mockConfigStoreForHealth) GetLadderRun(_ context.Context, _ string) (*config.LadderRunDB, error) { return nil, nil } diff --git a/pkg/ladder/allocate.go b/pkg/ladder/allocate.go index 021b65e01..00c7a576a 100644 --- a/pkg/ladder/allocate.go +++ b/pkg/ladder/allocate.go @@ -35,6 +35,16 @@ func minAllocatableGap() *big.Rat { // and causes Allocate to return an error (fail loud: incomplete provider data // aborts the run). // +// InFlightUSDPerHour is the total hourly commitment already in flight for +// this config's scope: the sum of all scheduled (not-yet-fired) and fired +// (not-yet-terminal) ladder tranches, plus any in-progress purchase +// executions linked to this config's runs. It is required non-nil (fail +// loud) so callers must always explicitly account for in-flight commitment. +// A genuinely zero in-flight must be passed as a pointer to 0.0, not nil. +// Subtracting in-flight from the gap prevents new runs from re-planning +// commitment that is already en route, eliminating the pile-up that occurs +// when multiple daily runs each see the full gap before prior tranches fire. +// // DataSources lists the data feeds (e.g. "cost-explorer", "cloudwatch") used // to derive the inputs. The slice is propagated verbatim to every Allocation // and PlannedAction produced by Allocate so that approval emails and audit @@ -42,12 +52,13 @@ func minAllocatableGap() *big.Rat { // // Now must be set by the caller; Allocate never calls time.Now internally. type AllocationInput struct { - Now time.Time - LayerStates map[LayerType]LayerState - Layers []LayerSpec - DataSources []string - Baseline UsageBaseline - Config LadderConfig + Now time.Time + LayerStates map[LayerType]LayerState + Layers []LayerSpec + DataSources []string + Baseline UsageBaseline + Config LadderConfig + InFlightUSDPerHour *float64 } // Allocation is a per-layer sizing decision produced by Allocate. The @@ -125,12 +136,20 @@ func Allocate(in *AllocationInput) (*AllocateResult, error) { if err != nil { return nil, err } - // The second min arm is defensive: it guards against future target - // derivations exceeding the observed floor. Today target <= lowWater - // always holds because TargetCoveragePct is validated to be <= 100. - gap := minRat(new(big.Rat).Sub(target, existingTotal), new(big.Rat).Sub(lowWater, existingTotal)) + inFlight, err := ratFromFloat("in_flight_usd_hr", *in.InFlightUSDPerHour) + if err != nil { + return nil, fmt.Errorf("in_flight_usd_hr: %w", err) + } + // gap = min(target-E, lowWater-E) - inFlight. + // The two-arm min is defensive: it guards against future target derivations + // exceeding the observed floor (today target<=lowWater always holds because + // TargetCoveragePct is validated to be <=100). Subtracting inFlight nets + // out scheduled/fired tranches that are already en route so repeated runs + // on the same day do not re-plan commitment that is already in flight. + preInflightGap := minRat(new(big.Rat).Sub(target, existingTotal), new(big.Rat).Sub(lowWater, existingTotal)) + gap := new(big.Rat).Sub(preInflightGap, inFlight) if gap.Cmp(minAllocatableGap()) <= 0 { - return withMaintenance(gapBelowMinHold(gap, lowWater, target, existingTotal, in), reshapes, maintHolds), nil + return withMaintenance(gapBelowMinHold(gap, lowWater, target, existingTotal, inFlight, in), reshapes, maintHolds), nil } return buildResult(in, lowWater, target, existingTotal, gap, nets, reshapes, maintHolds) } @@ -176,9 +195,12 @@ func fmtRatUSD(r *big.Rat) string { return fmt.Sprintf("$%.4f/hr", f) } -// validateAllocateInput validates the config, layer list, and layer states in -// one pass. Returns the first error encountered. +// validateAllocateInput validates the config, layer list, layer states, and +// the required in-flight commitment figure. Returns the first error encountered. func validateAllocateInput(in *AllocationInput) error { + if in.InFlightUSDPerHour == nil { + return fmt.Errorf("InFlightUSDPerHour must not be nil: callers must always supply the in-flight commitment; pass a pointer to 0.0 when there is no in-flight commitment") + } if err := in.Config.Validate(); err != nil { return fmt.Errorf("config: %w", err) } @@ -415,14 +437,15 @@ func baselineUnavailableHold(in *AllocationInput) *AllocateResult { } // gapBelowMinHold returns a Hold result explaining that the target is already -// met. Concrete numbers (low-water, target %, existing, gap) are embedded in -// the rationale. -func gapBelowMinHold(gap, lowWater, target, existingTotal *big.Rat, in *AllocationInput) *AllocateResult { +// met (accounting for in-flight commitment). Concrete numbers (low-water, +// target %, existing, in-flight, gap) are embedded in the rationale so +// operators can distinguish "gap closed by prior commitment" from "truly met". +func gapBelowMinHold(gap, lowWater, target, existingTotal, inFlight *big.Rat, in *AllocationInput) *AllocateResult { flexLayer, _ := findLayerForRole(in.Layers, RoleFlex) rationale := fmt.Sprintf( - "target already met: low_water=%s, target=%.2f%%, existing=%s, target_commitment=%s, gap=%s <= min_allocatable=%s", - fmtRatUSD(lowWater), in.Config.TargetCoveragePct, fmtRatUSD(existingTotal), fmtRatUSD(target), - fmtRatUSD(gap), fmtRatUSD(minAllocatableGap()), + "target already met: low_water=%s, target=%.2f%%, existing=%s, in_flight=%s, target_commitment=%s, gap=%s <= min_allocatable=%s", + fmtRatUSD(lowWater), in.Config.TargetCoveragePct, fmtRatUSD(existingTotal), fmtRatUSD(inFlight), + fmtRatUSD(target), fmtRatUSD(gap), fmtRatUSD(minAllocatableGap()), ) return &AllocateResult{ Holds: []PlannedAction{ diff --git a/pkg/ladder/allocate_test.go b/pkg/ladder/allocate_test.go index b3d7f836b..8802aa807 100644 --- a/pkg/ladder/allocate_test.go +++ b/pkg/ladder/allocate_test.go @@ -131,12 +131,13 @@ func TestAllocate_NilBaseline(t *testing.T) { t.Parallel() layers := awsLayers() in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{}, // LowWaterUSDPerHour is nil - LayerStates: zeroStates(layers), - Now: nowFixed(), - DataSources: []string{"cost-explorer"}, + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{}, // LowWaterUSDPerHour is nil + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), + DataSources: []string{"cost-explorer"}, } result, err := Allocate(in) if err != nil { @@ -167,11 +168,12 @@ func TestAllocate_GapBelowMin(t *testing.T) { // B = $10/hr, Ctgt = $10/hr (100% coverage), gap = 0 -> below min states = withExisting(states, LayerComputeSP, 10.0) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(8.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(8.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -204,11 +206,12 @@ func TestAllocate_ExistingAboveTarget_Hold(t *testing.T) { // existing = $15/hr on flex vs B = Ctgt = $10/hr -> gap = -5 states = withExisting(states, LayerComputeSP, 15.0) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(8.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(8.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -255,12 +258,13 @@ func TestAllocate_AWS3LayerSplit(t *testing.T) { states = withExisting(states, LayerEC2InstanceSP, 1.0) // $1/hr on base in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), - DataSources: []string{"cost-explorer"}, + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), + DataSources: []string{"cost-explorer"}, } result, err := Allocate(in) if err != nil { @@ -306,11 +310,12 @@ func TestAllocate_BufferFractionZero(t *testing.T) { cfg.BufferFraction = 0 // no buffer allocation layers := awsLayers() in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -334,11 +339,12 @@ func TestAllocate_StableUnknown_BaseZeroFlexGetsAll(t *testing.T) { cfg.BufferFraction = 0.5 // exactly representable layers := awsLayers() in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, // StableUSDPerHour nil - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, // StableUSDPerHour nil + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -378,11 +384,12 @@ func TestAllocate_NoBaseLayer_NoteNamesActualReason(t *testing.T) { cfg := validConfigAWS() cfg.BufferFraction = 0.5 in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -418,11 +425,12 @@ func TestAllocate_ExpiringExceedsExisting_Error(t *testing.T) { states = withExpiring(states, LayerConvertibleRI, 5.0) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -446,11 +454,12 @@ func TestAllocate_UtilizationClamp(t *testing.T) { states = withExisting(states, LayerComputeSP, 3.0) // E = $3/hr in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } // B=$10, Ctgt=$10, E=$3, gap=min(7,7)=$7 result, err := Allocate(in) @@ -479,11 +488,12 @@ func TestAllocate_AzureMergedBaseBuffer(t *testing.T) { cfg := validConfigAzure() cfg.BufferFraction = 0.5 // exactly representable in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -519,11 +529,12 @@ func TestAllocate_CapScaling(t *testing.T) { layers := awsLayers() in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -573,12 +584,13 @@ func TestAllocate_MaxActionsExceeded_Truncates(t *testing.T) { cfg.MaxActionsPerRun = 2 // 3 allocations would normally be produced layers := awsLayers() in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), - DataSources: []string{"cost-explorer"}, + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), + DataSources: []string{"cost-explorer"}, } result, err := Allocate(in) // Regression: pre-fix this returned an error; post-fix must succeed. @@ -662,12 +674,13 @@ func TestAllocate_MaxActions_ReshapeRetained(t *testing.T) { states = withExisting(states, LayerConvertibleRI, 5.0) states = withBufferUtilization(states, 50.0) // below 90% threshold -> reshape in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), - DataSources: []string{"cost-explorer"}, + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), + DataSources: []string{"cost-explorer"}, } result, err := Allocate(in) if err != nil { @@ -717,12 +730,13 @@ func TestAllocate_MaxActions_ReshapesFillCap(t *testing.T) { states = withExisting(states, LayerConvertibleRI, 5.0) states = withBufferUtilization(states, 50.0) // below 90% threshold -> reshape in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), - DataSources: []string{"cost-explorer"}, + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), + DataSources: []string{"cost-explorer"}, } result, err := Allocate(in) if err != nil { @@ -780,12 +794,13 @@ func TestAllocate_ExactCap_Untouched(t *testing.T) { cfg.MaxActionsPerRun = 3 // exactly the number of allocations produced layers := awsLayers() in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), - DataSources: []string{"cost-explorer"}, + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), + DataSources: []string{"cost-explorer"}, } result, err := Allocate(in) if err != nil { @@ -810,11 +825,12 @@ func TestAllocate_MissingLayerState(t *testing.T) { delete(states, LayerConvertibleRI) // remove buffer layer state in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -834,11 +850,12 @@ func TestAllocate_NilExistingUSDPerHour(t *testing.T) { states[LayerConvertibleRI] = s in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -858,11 +875,12 @@ func TestAllocate_NilExpiringUSDPerHour(t *testing.T) { states[LayerConvertibleRI] = s in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -880,11 +898,12 @@ func TestAllocate_NaNInput(t *testing.T) { states = withExisting(states, LayerConvertibleRI, math.NaN()) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -902,11 +921,12 @@ func TestAllocate_NegativeInput(t *testing.T) { states = withExisting(states, LayerComputeSP, -1.0) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -921,11 +941,12 @@ func TestAllocate_NegativeBaseline(t *testing.T) { t.Parallel() layers := awsLayers() in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(-5.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(-5.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -946,11 +967,12 @@ func TestAllocate_ReshapeEmittedUnderThreshold(t *testing.T) { states = withExisting(states, LayerConvertibleRI, 2.0) states = withBufferUtilization(states, 70.0) in := &AllocationInput{ - Config: validConfigAWS(), // threshold=90% - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), // threshold=90% + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -982,11 +1004,12 @@ func TestAllocate_NoReshapeAtThreshold(t *testing.T) { states = withExisting(states, LayerConvertibleRI, 2.0) states = withBufferUtilization(states, DefaultBufferUtilizationThresholdPct) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1005,11 +1028,12 @@ func TestAllocate_NoReshapeAboveThreshold(t *testing.T) { states = withExisting(states, LayerConvertibleRI, 2.0) states = withBufferUtilization(states, 95.0) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1028,11 +1052,12 @@ func TestAllocate_UtilizationNil_InformationalHold(t *testing.T) { // informational hold. (Empty buffer layers are skipped and emit nothing.) states = withExisting(states, LayerConvertibleRI, 2.0) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1061,11 +1086,12 @@ func TestAllocate_EmptyBufferLayer_NoNoise(t *testing.T) { states := zeroStates(layers) // buffer existing = 0 states = withBufferUtilization(states, 0.0) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1125,11 +1151,12 @@ func TestAllocate_InvalidConfig(t *testing.T) { cfg := validConfigAWS() cfg.TargetCoveragePct = 0 // invalid in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -1140,11 +1167,12 @@ func TestAllocate_InvalidConfig(t *testing.T) { func TestAllocate_EmptyLayers(t *testing.T) { t.Parallel() in := &AllocationInput{ - Config: validConfigAWS(), - Layers: nil, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: map[LayerType]LayerState{}, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: nil, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: map[LayerType]LayerState{}, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -1161,11 +1189,12 @@ func TestAllocate_NoFlexLayer(t *testing.T) { {Type: LayerConvertibleRI, Roles: []LayerRole{RoleBuffer}}, } in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -1186,11 +1215,12 @@ func TestAllocate_TwoBaseLayers(t *testing.T) { {Type: LayerComputeSP, Roles: []LayerRole{RoleFlex}}, } in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -1210,11 +1240,12 @@ func TestAllocate_TwoFlexLayers(t *testing.T) { {Type: LayerEC2InstanceSP, Roles: []LayerRole{RoleFlex}}, } in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -1298,11 +1329,12 @@ func TestAllocate_LayerTopologyValidation(t *testing.T) { for _, c := range cases { t.Run(c.name, func(t *testing.T) { in := &AllocationInput{ - Config: validConfigAzure(), - Layers: c.layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(c.layers), - Now: nowFixed(), + Config: validConfigAzure(), + Layers: c.layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(c.layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if c.errContains == "" { @@ -1348,11 +1380,12 @@ func TestAllocate_UtilizationPctValidation(t *testing.T) { states = withExisting(states, LayerConvertibleRI, 2.0) states = withBufferUtilization(states, c.pct) in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if c.wantErr { @@ -1384,11 +1417,12 @@ func TestAllocate_NoBufferLayerWithBufferFraction_Error(t *testing.T) { } cfg := validConfigAWS() // BufferFraction = 0.10 > 0 in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -1414,11 +1448,12 @@ func TestAllocate_NoBufferLayerZeroFraction_OK(t *testing.T) { cfg := validConfigAWS() cfg.BufferFraction = 0 in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1452,11 +1487,12 @@ func TestAllocate_TargetMet_ReshapeStillEmitted(t *testing.T) { states = withExisting(states, LayerConvertibleRI, 2.0) states = withBufferUtilization(states, 70.0) // below 90% threshold in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(8.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(8.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1486,11 +1522,12 @@ func TestAllocate_NilBaseline_ReshapeStillEmitted(t *testing.T) { states = withExisting(states, LayerConvertibleRI, 2.0) states = withBufferUtilization(states, 70.0) // below 90% threshold in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{}, // LowWaterUSDPerHour nil - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{}, // LowWaterUSDPerHour nil + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1527,11 +1564,12 @@ func TestAllocate_UnknownLayerStateKey_Error(t *testing.T) { ExpiringUSDPerHour: ptr(0.0), } in := &AllocationInput{ - Config: validConfigAWS(), - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, - LayerStates: states, - Now: nowFixed(), + Config: validConfigAWS(), + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, + LayerStates: states, + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } _, err := Allocate(in) if err == nil { @@ -1556,11 +1594,12 @@ func TestAllocate_SubMinimumBufferSkipped(t *testing.T) { cfg.BufferFraction = 0.0005 layers := awsLayers() in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0), StableUSDPerHour: ptr(4.0)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1600,9 +1639,10 @@ func TestAllocate_AllSplitsSubMinimum(t *testing.T) { Layers: layers, // gap = $0.011 (just above the min-gap threshold); stable unknown so // base gets nothing and flex receives the whole core gap of $0.0099. - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(0.011)}, - LayerStates: zeroStates(layers), - Now: nowFixed(), + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(0.011)}, + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } result, err := Allocate(in) if err != nil { @@ -1635,11 +1675,12 @@ func TestAllocate_AzureMerged_StableUnknown_NoteOnFlexOnly(t *testing.T) { cfg := validConfigAzure() cfg.BufferFraction = 0.5 in := &AllocationInput{ - Config: cfg, - Layers: layers, - Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, // stable unknown - LayerStates: zeroStates(layers), - Now: nowFixed(), + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(10.0)}, // stable unknown + LayerStates: zeroStates(layers), + Now: nowFixed(), + InFlightUSDPerHour: ptr(0.0), } // bufferGap=5, coreGap=5, baseGap=0 (stable unknown), flexGap=5, merged=5 result, err := Allocate(in) @@ -1664,6 +1705,108 @@ func TestAllocate_AzureMerged_StableUnknown_NoteOnFlexOnly(t *testing.T) { } } +// TestAllocate_ConvergenceNetting is the L5 regression guard: three sequential +// daily runs with identical usage must not accumulate planned commitment beyond +// the initial gap. Without in-flight netting each run would re-plan the full +// gap and the system would diverge; with netting only the first run allocates +// and subsequent runs hold. +// +// To confirm the test guards the real behaviour, the second sub-test ("no +// netting diverges") shows that feeding zero in-flight to every run produces +// three allocations totalling 3x the gap -- this sub-test MUST FAIL if the L5 +// netting logic is removed from Allocate. +func TestAllocate_ConvergenceNetting(t *testing.T) { + t.Parallel() + + // Baseline: low-water $10/hr, target coverage 80% => target $8/hr. + // Existing = $0, so initial gap = $8. + const targetPct = 80.0 + const lowWater = 10.0 + const initialGap = 8.0 // min($8-$0, $10-$0) = $8 + + makeInput := func(inFlight float64) *AllocationInput { + layers := awsLayers() + states := zeroStates(layers) + cfg := validConfigAWS() + cfg.TargetCoveragePct = targetPct + return &AllocationInput{ + Config: cfg, + Layers: layers, + Baseline: UsageBaseline{LowWaterUSDPerHour: ptr(lowWater), StableUSDPerHour: ptr(lowWater * 0.9)}, + LayerStates: states, + Now: nowFixed(), + DataSources: []string{"cost-explorer"}, + InFlightUSDPerHour: &inFlight, + } + } + + t.Run("with netting converges", func(t *testing.T) { + t.Parallel() + var totalPlanned float64 + + // Run 1: no in-flight -> allocates the full gap. + r1, err := Allocate(makeInput(0.0)) + if err != nil { + t.Fatalf("run1 Allocate: %v", err) + } + for _, a := range r1.Allocations { + f, _ := a.GapUSDPerHour.Float64() + totalPlanned += f + } + if len(r1.Allocations) == 0 { + t.Fatalf("run1: expected at least one allocation with zero in-flight, got hold: %+v", r1.Holds) + } + + // Run 2: in-flight = totalPlanned from run 1 -> should hold (gap <= min). + r2, err := Allocate(makeInput(totalPlanned)) + if err != nil { + t.Fatalf("run2 Allocate: %v", err) + } + if len(r2.Allocations) != 0 { + t.Errorf("run2: expected Hold (gap covered by in-flight), got %d allocations", len(r2.Allocations)) + } + if len(r2.Holds) == 0 { + t.Errorf("run2: expected at least one Hold action, got none") + } + + // Run 3: same in-flight as run 2 -> still Hold. + r3, err := Allocate(makeInput(totalPlanned)) + if err != nil { + t.Fatalf("run3 Allocate: %v", err) + } + if len(r3.Allocations) != 0 { + t.Errorf("run3: expected Hold (gap covered by in-flight), got %d allocations", len(r3.Allocations)) + } + + // Total planned must be approx the initial gap (not 2x or 3x). + if totalPlanned < initialGap*0.5 || totalPlanned > initialGap*1.5 { + t.Errorf("total planned = %.4f, want approx %.4f (1x gap, not accumulated)", totalPlanned, initialGap) + } + }) + + t.Run("no netting diverges", func(t *testing.T) { + // This sub-test verifies that IGNORING in-flight (always passing 0) + // produces 3x the initial gap. It is a canary: if the L5 gap subtraction + // is removed from Allocate this sub-test changes meaning but the + // "with netting" sub-test above will fail first, which is the real guard. + t.Parallel() + var totalWithoutNetting float64 + for i := 0; i < 3; i++ { + r, err := Allocate(makeInput(0.0)) // zero in-flight every time + if err != nil { + t.Fatalf("run%d Allocate: %v", i+1, err) + } + for _, a := range r.Allocations { + f, _ := a.GapUSDPerHour.Float64() + totalWithoutNetting += f + } + } + if totalWithoutNetting < initialGap*2.0 { + t.Errorf("without netting: expected total >= 2x gap (%.4f), got %.4f; the divergence canary is broken", initialGap*2.0, totalWithoutNetting) + } + }) +} + // TestAllocateResult_IsNoOp verifies IsNoOp across the result shapes. func TestAllocateResult_IsNoOp(t *testing.T) { t.Parallel() From d656a7ddddace76890ca93de1f321742b6aff65e Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 16 Jul 2026 21:36:47 +0300 Subject: [PATCH 2/4] fix(ladder): net scheduled tranches only, not fired (L5 finding 1) GetInFlightLadderCommitUSDHr summed status IN ('scheduled','fired'), but fired/completed tranches are executed purchases already folded into the engine's ExistingUSDPerHour by the provider adapters. Counting them again double-subtracted from the gap and under-purchased. - Sum status = 'scheduled' ONLY; fired/completed belong solely to E. - Use typed pkg/ladder TrancheStatus constants in the sum and the supersede UPDATE instead of raw string literals (finding 5); the status set now lives in one place. - Update the InFlightUSDPerHour engine doc and the StoreInterface docs to say scheduled-tranches-only (finding 3, engine layer). - Update the integration test to assert scheduled-only (3.0, was 5.0). --- internal/config/interfaces.go | 16 +++++++--- internal/config/store_postgres_ladder.go | 32 +++++++++++++------ internal/config/store_postgres_ladder_test.go | 17 ++++++---- pkg/ladder/allocate.go | 20 ++++++------ 4 files changed, 56 insertions(+), 29 deletions(-) diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 2cc45073a..fb50cacbc 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -386,12 +386,20 @@ type StoreInterface interface { // 3. Inserts the new ladder_tranches rows. // This guarantees at most one generation of scheduled tranches per config // exists at any point; a partial failure rolls back all three steps. + // Callers invoke this ONLY when the run produced a materially new + // generation of scheduled tranches (>=1 new scheduled purchase tranche); + // a Hold run persists via the plain SaveLadderRunWithTranches path so the + // live scheduled ramp is preserved rather than cancelled every cadence. // // GetInFlightLadderCommitUSDHr returns the total hourly USD commitment in - // flight for the given config: the sum of amount_usd_hr for all tranches - // with status IN ('scheduled', 'fired'). Returns a non-nil pointer (zero - // when no in-flight tranches exist) so callers can pass it directly to - // AllocationInput.InFlightUSDPerHour. Never returns nil without an error. + // flight for the given config: the sum of amount_usd_hr for tranches with + // status = 'scheduled' ONLY. Fired/completed tranches are executed + // purchases already reflected in the engine's ExistingUSDPerHour (the + // provider adapters fold payment-pending and active commitments into E), + // so summing them here too would double-count and under-purchase. Returns + // a non-nil pointer (zero when no scheduled tranches exist) so callers can + // pass it directly to AllocationInput.InFlightUSDPerHour. Never returns + // nil without an error. SaveLadderRun(ctx context.Context, run *LadderRunDB) (*LadderRunDB, error) SaveLadderRunWithTranches(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) SaveLadderRunWithTranchesAndSupersede(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index 16535bb97..4e3cb4c43 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -362,9 +362,17 @@ func (s *PostgresStore) SaveLadderRunWithTranches(ctx context.Context, run *Ladd } // GetInFlightLadderCommitUSDHr returns the total hourly USD commitment already -// in flight for the given config: the SUM of amount_usd_hr for all -// ladder_tranches rows where config_id=$1 and status IN ('scheduled','fired'). -// Returns a non-nil *float64 (zero when no in-flight tranches exist); never +// in flight for the given config: the SUM of amount_usd_hr for ladder_tranches +// rows where config_id=$1 and status = 'scheduled'. +// +// SCHEDULED ONLY: fired/completed tranches are executed purchases already +// counted in the engine's ExistingUSDPerHour (the provider adapters fold +// payment-pending and active commitments into E). Summing them here as well +// would double-subtract them from the gap and cause under-purchasing. Only +// not-yet-fired (scheduled) tranches are genuinely "in flight" and absent +// from E, so they alone must be netted out of the gap. +// +// Returns a non-nil *float64 (zero when no scheduled tranches exist); 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) (*float64, error) { @@ -373,8 +381,8 @@ func (s *PostgresStore) GetInFlightLadderCommitUSDHr(ctx context.Context, config SELECT COALESCE(SUM(amount_usd_hr), 0) FROM ladder_tranches WHERE config_id = $1 - AND status IN ('scheduled', 'fired') - `, configID).Scan(&total) + AND status = $2 + `, configID, string(ladder.TrancheStatusScheduled)).Scan(&total) if err != nil { return nil, fmt.Errorf("GetInFlightLadderCommitUSDHr config_id=%s: %w", configID, err) } @@ -393,19 +401,25 @@ func (s *PostgresStore) GetInFlightLadderCommitUSDHr(ctx context.Context, config // If any step fails the whole transaction is rolled back. If run.ConfigID is // nil the cancel step is skipped (the run has no prior config context). If // run.ID is empty a fresh UUID is generated before the insert. +// +// Callers invoke this ONLY for a run that produced a materially new generation +// of scheduled tranches; a Hold run must use the plain SaveLadderRunWithTranches +// path so the existing scheduled ramp keeps its original fire dates. func (s *PostgresStore) SaveLadderRunWithTranchesAndSupersede(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) { if run.ID == "" { run.ID = uuid.New().String() } var result LadderRunDB err := s.WithTx(ctx, func(tx pgx.Tx) error { - // Step 1: cancel prior scheduled tranches for this config. + // Step 1: cancel prior scheduled tranches for this config. Typed + // TrancheStatus constants (not raw string literals) keep the status + // set consistent with the engine and GetInFlightLadderCommitUSDHr. if run.ConfigID != nil { _, err := tx.Exec(ctx, ` UPDATE ladder_tranches - SET status = 'cancelled' - WHERE config_id = $1 AND status = 'scheduled' - `, *run.ConfigID) + SET status = $2 + WHERE config_id = $1 AND status = $3 + `, *run.ConfigID, string(ladder.TrancheStatusCancelled), string(ladder.TrancheStatusScheduled)) if err != nil { return fmt.Errorf("cancel prior scheduled tranches for config_id=%s: %w", *run.ConfigID, err) } diff --git a/internal/config/store_postgres_ladder_test.go b/internal/config/store_postgres_ladder_test.go index 02401d95b..669dacf21 100644 --- a/internal/config/store_postgres_ladder_test.go +++ b/internal/config/store_postgres_ladder_test.go @@ -395,8 +395,9 @@ func seedLadderConfigWithExtID(ctx context.Context, t *testing.T, store *Postgre // TestPostgresStore_GetInFlightLadderCommitUSDHr tests the in-flight sum // query (L5 netting). It verifies: -// - zero is returned (not nil) when no tranches exist. -// - scheduled + fired tranches are summed; completed/cancelled/failed are excluded. +// - zero is returned (not nil) when no scheduled tranches exist. +// - SCHEDULED tranches ONLY are summed; fired/completed/cancelled/failed are +// all excluded (fired/completed are already in ExistingUSDPerHour). // - tranches from a different config are not included. func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) @@ -434,8 +435,8 @@ func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { // Insert tranches with various statuses for the main config. insertTranche(t, &cfgID, 3.0, ladder.TrancheStatusScheduled) // included - insertTranche(t, &cfgID, 2.0, ladder.TrancheStatusFired) // included - insertTranche(t, &cfgID, 5.0, ladder.TrancheStatusCompleted) // excluded (terminal) + insertTranche(t, &cfgID, 2.0, ladder.TrancheStatusFired) // excluded (already in E) + insertTranche(t, &cfgID, 5.0, ladder.TrancheStatusCompleted) // excluded (already in E) insertTranche(t, &cfgID, 1.0, ladder.TrancheStatusCancelled) // excluded (terminal) insertTranche(t, &cfgID, 4.0, ladder.TrancheStatusFailed) // excluded (terminal) @@ -443,12 +444,14 @@ func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { otherID := otherConfigID insertTranche(t, &otherID, 99.0, ladder.TrancheStatusScheduled) - t.Run("sums scheduled and fired only", func(t *testing.T) { + t.Run("sums scheduled only", func(t *testing.T) { result, err := store.GetInFlightLadderCommitUSDHr(ctx, configID) require.NoError(t, err) require.NotNil(t, result) - // Only 3.0 (scheduled) + 2.0 (fired) = 5.0 must be returned. - assert.InDelta(t, 5.0, *result, 1e-6, "in-flight sum must include scheduled+fired only") + // Only the 3.0 scheduled tranche must be returned. Fired/completed are + // executed purchases already counted in ExistingUSDPerHour, so netting + // them again would double-subtract. + assert.InDelta(t, 3.0, *result, 1e-6, "in-flight sum must include scheduled tranches ONLY") }) t.Run("other config is not contaminated", func(t *testing.T) { diff --git a/pkg/ladder/allocate.go b/pkg/ladder/allocate.go index 00c7a576a..c09ad7481 100644 --- a/pkg/ladder/allocate.go +++ b/pkg/ladder/allocate.go @@ -35,15 +35,17 @@ func minAllocatableGap() *big.Rat { // and causes Allocate to return an error (fail loud: incomplete provider data // aborts the run). // -// InFlightUSDPerHour is the total hourly commitment already in flight for -// this config's scope: the sum of all scheduled (not-yet-fired) and fired -// (not-yet-terminal) ladder tranches, plus any in-progress purchase -// executions linked to this config's runs. It is required non-nil (fail -// loud) so callers must always explicitly account for in-flight commitment. -// A genuinely zero in-flight must be passed as a pointer to 0.0, not nil. -// Subtracting in-flight from the gap prevents new runs from re-planning -// commitment that is already en route, eliminating the pile-up that occurs -// when multiple daily runs each see the full gap before prior tranches fire. +// InFlightUSDPerHour is the hourly commitment already in flight for this +// config's scope: the sum of its scheduled (not-yet-fired) ladder tranches +// ONLY. Fired/completed tranches are deliberately excluded because they are +// executed purchases already reflected in each layer's ExistingUSDPerHour; +// counting them here as well would double-subtract and under-purchase. It is +// required non-nil (fail loud) so callers must always explicitly account for +// in-flight commitment. A genuinely zero in-flight must be passed as a pointer +// to 0.0, not nil. Subtracting in-flight from the gap prevents new runs from +// re-planning commitment that is already scheduled, eliminating the pile-up +// that occurs when multiple daily runs each see the full gap before prior +// scheduled tranches fire. // // DataSources lists the data feeds (e.g. "cost-explorer", "cloudwatch") used // to derive the inputs. The slice is propagated verbatim to every Allocation From ca15efc1a3f908e57c4ad3348a95350b6c602ca6 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 16 Jul 2026 21:37:32 +0300 Subject: [PATCH 3/4] fix(ladder): make tranche supersede conditional to preserve the ramp (L5 finding 2) Netting scheduled tranches into the gap AND cancelling them every run were mutually destructive: run2 nets to a Hold, but unconditional SaveLadderRunWithTranchesAndSupersede cancelled the scheduled ramp anyway, so run3 re-planned from zero. Oscillation; tranches with fire dates beyond one cadence window never fired. Corrected design: - persistPlannedLadderRun chooses the persist path by whether THIS run produced >=1 new scheduled tranche (len(trancheRows) > 0): * >0 -> SaveLadderRunWithTranchesAndSupersede (cancel-and-replace the config's prior scheduled generation). * ==0 -> plain SaveLadderRunWithTranches: persist the run row and PRESERVE the existing scheduled ramp with its original fire dates (the live ramp the netting just relied on). - Net effect: run1 builds the ramp; steady-state runs Hold and preserve it; a drift-UP run generates a new generation and cancel-and-replaces. - Drift-DOWN over-commit (a Hold preserving a now-too-large ramp) is accepted for now, bounded by the low-water clamp; documented for a follow-up shrink pass. - Fix the executeLadderRun in-flight comment (finding 3, handler layer): scheduled-tranches-only, no "in-progress purchase executions". Tests (finding 4): handler-level ladderStatefulStore drives the real executeLadderRun -> Allocate -> BuildTranches -> persist pipeline. - Convergence+preservation: 3 constant-usage runs build the ramp once then Hold, preserving the exact scheduled tranche IDs; no accumulation. - Drift-up: higher usage re-opens the netted gap, cancels the old generation, and leaves exactly one live generation. Both fail on the pre-fix unconditional-supersede code and pass after. --- internal/server/handler_ladder.go | 77 ++++++---- internal/server/handler_ladder_test.go | 195 +++++++++++++++++++++++++ 2 files changed, 247 insertions(+), 25 deletions(-) diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go index e34dfee03..7c6e4ab1b 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -273,17 +273,21 @@ func (app *Application) processOneLadderConfig( // It is split from handleLadderRun so tests can inject a hermetic // LadderCapability (fake) and a mock store without touching AWS or PostgreSQL. // -// L5 netting: before calling Allocate, the handler fetches the total in-flight -// commitment (scheduled + fired non-terminal tranches plus any in-progress -// purchase executions linked to this config). 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 fired. +// L5 netting: before calling Allocate, the handler fetches the config's in-flight +// commitment (the sum of its status=scheduled, not-yet-fired ladder tranches +// ONLY). The Allocate engine subtracts this from the raw gap so each daily run +// plans only what is still needed, preventing pile-up when prior scheduled +// tranches have not yet fired. Fired/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. // -// The cancel-and-replace is performed atomically inside -// SaveLadderRunWithTranchesAndSupersede: the same transaction that inserts the -// new run and its tranches also cancels the config's prior scheduled (not yet -// fired) tranches, ensuring at most one generation of scheduled tranches exists -// for each config at any point in time. +// Cancel-and-replace is CONDITIONAL (see persistLadderRun): it runs only when +// this run produces a materially new generation of scheduled tranches. A Hold +// run (the netting says we are covered) persists the run row WITHOUT cancelling +// the existing scheduled ramp, so those tranches keep their original fire dates +// and remain the live ramp. Unconditional cancellation would fight the netting: +// run2 nets to a Hold, but cancelling its own scheduled ramp would let run3 +// re-plan from zero (oscillation; late tranches would never fire). func (app *Application) executeLadderRun( ctx context.Context, dbCfg *config.LadderConfigDB, @@ -313,12 +317,12 @@ func (app *Application) executeLadderRun( } // L5: Fetch in-flight commitment BEFORE Allocate so the engine can subtract - // it from the gap. In-flight = scheduled (not-yet-fired) tranches + - // fired tranches whose execution has not reached a terminal state + - // any in-progress purchase executions linked to this config's runs. - // This query includes prior scheduled tranches (from earlier runs) so the - // engine correctly accounts for en-route commitment and produces a ~zero - // gap (Hold) when prior tranches already cover the target. + // it from the gap. In-flight = the config's status=scheduled (not-yet-fired) + // ladder tranches ONLY. Fired/completed tranches are excluded because they + // are executed purchases already counted in ExistingUSDPerHour; netting them + // again would double-subtract. Including prior-run scheduled tranches lets + // the engine account for en-route commitment and produce a ~zero gap (Hold) + // when the scheduled ramp already covers the target. inFlight, err := app.Config.GetInFlightLadderCommitUSDHr(ctx, dbCfg.ID) if err != nil { return fmt.Errorf("GetInFlightLadderCommitUSDHr: %w", err) @@ -331,8 +335,8 @@ func (app *Application) executeLadderRun( supportedLayers := capability.SupportedLayers() - // Run Allocate with the in-flight figure so the gap is net of prior - // scheduled and fired tranches. + // Run Allocate with the in-flight figure so the gap is net of the config's + // prior scheduled (not-yet-fired) tranches. allocResult, err := pkgladder.Allocate(&pkgladder.AllocationInput{ Now: now, Baseline: baseline, @@ -424,14 +428,11 @@ func (app *Application) persistLadderRun( return fmt.Errorf("buildTrancheDBRows: %w", err) } - // L5: persist run + tranches atomically AND cancel any prior scheduled - // tranches for this config in the same transaction. This guarantees at - // most one generation of scheduled tranches per config exists at any - // point, preventing pile-up when repeated daily runs each see the full - // gap before prior tranches fire. - savedRun, err := app.Config.SaveLadderRunWithTranchesAndSupersede(ctx, runRow, trancheRows) + // L5: persist the run, using cancel-and-replace supersede ONLY when this + // run produced a materially new generation of scheduled tranches. + savedRun, err := app.persistPlannedLadderRun(ctx, runRow, trancheRows) if err != nil { - return fmt.Errorf("SaveLadderRunWithTranchesAndSupersede: %w", err) + return fmt.Errorf("persist ladder run: %w", err) } log.Printf("ladder_run: config %s: persisted run %s (status=%s, allocations=%d, tranches=%d, holds=%d)", @@ -440,6 +441,32 @@ func (app *Application) persistLadderRun( return nil } +// persistPlannedLadderRun writes the ladder_runs row, choosing between the +// cancel-and-replace supersede path and the plain (preserve) path based on +// whether this run produced a materially new generation of scheduled tranches: +// +// - len(trancheRows) > 0: this run planned a new scheduled ramp, so +// cancel-and-replace the config's prior scheduled tranches atomically +// (SaveLadderRunWithTranchesAndSupersede). Exactly one generation of +// scheduled tranches survives per config. +// - len(trancheRows) == 0: a Hold (the L5 netting says the config is already +// covered by its live scheduled ramp) or a buy-now-only run. Persist the +// run row via the plain path WITHOUT cancelling the existing scheduled +// tranches, so they keep their original fire dates and remain the live +// ramp. Superseding here would cancel the ramp the netting just relied on, +// causing the next run to re-plan from zero (oscillation). +// +// Drift-DOWN note: a Hold that preserves a now-too-large scheduled ramp (usage +// dropped after the ramp was planned) is intentionally accepted for now. It is +// bounded by the low-water clamp in Allocate; shrinking an over-large ramp on +// drift-down is a follow-up (tracked separately), not handled here. +func (app *Application) persistPlannedLadderRun(ctx context.Context, runRow *config.LadderRunDB, trancheRows []config.LadderTrancheDB) (*config.LadderRunDB, error) { + if len(trancheRows) > 0 { + return app.Config.SaveLadderRunWithTranchesAndSupersede(ctx, runRow, trancheRows) + } + return app.Config.SaveLadderRunWithTranches(ctx, runRow, trancheRows) +} + // ladderWithinCadenceWindow reports whether a new run should be skipped because // a recent run already covers the cadence window. It queries // LatestLadderRunStartedAt for the config and returns: diff --git a/internal/server/handler_ladder_test.go b/internal/server/handler_ladder_test.go index 1bf2849db..31b7eba3c 100644 --- a/internal/server/handler_ladder_test.go +++ b/internal/server/handler_ladder_test.go @@ -1139,3 +1139,198 @@ func TestRatToFloat64Ptr_PositiveValue(t *testing.T) { require.NotNil(t, ptr, "non-nil big.Rat must produce non-nil *float64") assert.Greater(t, *ptr, 0.0, "gap must be positive") } + +// ============================================================ +// L5 convergence + conditional-supersede (handler-level, real pipeline) +// ============================================================ + +// ladderStatefulStore is a faithful in-memory stand-in for the ladder tranche +// persistence that executeLadderRun drives. Unlike ladderTestStore (which +// returns a FIXED in-flight value), it tracks each persisted tranche's live +// status so the L5 netting + conditional-supersede interplay can be exercised +// end-to-end across multiple sequential runs. It mirrors PostgresStore +// semantics exactly: +// - GetInFlightLadderCommitUSDHr sums SCHEDULED tranches only, per config. +// - SaveLadderRunWithTranchesAndSupersede cancels the config's live scheduled +// tranches, then appends the run + the new generation (cancel-and-replace). +// - SaveLadderRunWithTranches appends the run + tranches WITHOUT cancelling +// (the Hold/preserve path). +// +// The tests below drive the REAL executeLadderRun -> Allocate -> BuildTranches +// -> persistPlannedLadderRun pipeline; only the store and capability are fakes. +type ladderStatefulStore struct { + mockConfigStoreForHealth + tranches []config.LadderTrancheDB // every persisted tranche, live status + runs []*config.LadderRunDB +} + +// scheduledSum totals the config's currently-scheduled tranche amounts. It is +// the single source of truth reused by GetInFlightLadderCommitUSDHr so the fake +// cannot drift from its own netting view. +func (s *ladderStatefulStore) scheduledSum(configID string) float64 { + total := 0.0 + for i := range s.tranches { + tr := &s.tranches[i] + if tr.ConfigID != nil && *tr.ConfigID == configID && tr.Status == pkgladder.TrancheStatusScheduled { + total += tr.AmountUSDHr + } + } + return total +} + +// scheduledTrancheIDs returns the set of currently-scheduled tranche IDs for +// the config, so preservation across runs can be asserted by identity. +func (s *ladderStatefulStore) scheduledTrancheIDs(configID string) map[string]bool { + ids := map[string]bool{} + for i := range s.tranches { + tr := &s.tranches[i] + if tr.ConfigID != nil && *tr.ConfigID == configID && tr.Status == pkgladder.TrancheStatusScheduled { + ids[tr.ID] = true + } + } + return ids +} + +func (s *ladderStatefulStore) GetInFlightLadderCommitUSDHr(_ context.Context, configID string) (*float64, error) { + total := s.scheduledSum(configID) + return &total, nil +} + +func (s *ladderStatefulStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { + s.runs = append(s.runs, run) + s.tranches = append(s.tranches, tranches...) + return run, nil +} + +func (s *ladderStatefulStore) SaveLadderRunWithTranchesAndSupersede(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { + // Cancel-and-replace: flip the config's live scheduled tranches to + // cancelled, then append the new generation. + if run.ConfigID != nil { + for i := range s.tranches { + tr := &s.tranches[i] + if tr.ConfigID != nil && *tr.ConfigID == *run.ConfigID && tr.Status == pkgladder.TrancheStatusScheduled { + tr.Status = pkgladder.TrancheStatusCancelled + } + } + } + s.runs = append(s.runs, run) + s.tranches = append(s.tranches, tranches...) + return run, nil +} + +// fullyDelayedRamp is a 2-step ramp with NO AfterDays==0 step, so ALL planned +// commitment lands as SCHEDULED tranches (no buy-now). With a static E=0 fake +// capability, that scheduled ramp is the only in-flight commitment, so the L5 +// netting alone decides whether a follow-up run Holds. +func fullyDelayedRamp(t *testing.T) json.RawMessage { + t.Helper() + raw, err := json.Marshal(pkgladder.RampSchedule{ + Steps: []pkgladder.RampStep{ + {AfterDays: 1, Fraction: 0.5}, + {AfterDays: 7, Fraction: 0.5}, + }, + }) + require.NoError(t, err) + return raw +} + +// TestExecuteLadderRun_L5Convergence_PreservesRampAcrossHolds is the core L5 +// regression: three sequential runs with CONSTANT usage must build the ramp +// once (run1) and then PRESERVE it (run2, run3 Hold), never accumulating and +// never cancelling the live scheduled ramp. +// +// FAILS on the pre-L5 code (unconditional supersede): run2 nets to a Hold but +// the old code still calls SaveLadderRunWithTranchesAndSupersede with an empty +// tranche set, cancelling the run1 ramp; the scheduled sum drops to 0, so this +// test's preservation assertions fail. PASSES after: run2/run3 take the plain +// preserve path. +func TestExecuteLadderRun_L5Convergence_PreservesRampAcrossHolds(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + const cfgID = "cfg-l5-converge" + dbCfg := validTestDBConfig(cfgID) + dbCfg.RampSchedule = fullyDelayedRamp(t) + + store := &ladderStatefulStore{} + app := &Application{Config: store} + // Constant usage across all three runs: low-water 10 -> target 8 (80%). + cap := &fakeLadderCapability{t: t, baseline: testBaseline(10.0)} + run := func(at time.Time) { + require.NoError(t, app.executeLadderRun(ctx, &dbCfg, cap, "123456789012", + pkgladder.Term1Year, pkgladder.PaymentNoUpfront, at)) + } + + // Run 1: builds the scheduled ramp. gap = 8, all delayed -> scheduled. + run(now) + gen1 := store.scheduledTrancheIDs(cfgID) + require.NotEmpty(t, gen1, "run1 must build a scheduled ramp") + assert.InDelta(t, 8.0, store.scheduledSum(cfgID), 1e-9, "run1 ramp must sum to the full gap (8)") + require.Len(t, store.runs, 1) + + // Run 2: SAME usage. Netting sees the 8 already scheduled -> Hold. The run + // row persists, but the run1 tranches must REMAIN scheduled (same IDs). + run(now.Add(24 * time.Hour)) + gen2 := store.scheduledTrancheIDs(cfgID) + assert.Equal(t, gen1, gen2, "run2 Hold must PRESERVE the exact run1 scheduled tranches (same IDs)") + assert.InDelta(t, 8.0, store.scheduledSum(cfgID), 1e-9, "run2 Hold must not change the scheduled ramp total") + require.Len(t, store.runs, 2, "run2 must still persist a run row") + assert.Equal(t, 0.0, store.runs[1].TotalHourlyCommit, "run2 must be a Hold (zero new commitment)") + + // Run 3: SAME usage again -> still Hold, still preserved. No accumulation. + run(now.Add(48 * time.Hour)) + gen3 := store.scheduledTrancheIDs(cfgID) + assert.Equal(t, gen1, gen3, "run3 Hold must still preserve the original ramp (same IDs)") + assert.InDelta(t, 8.0, store.scheduledSum(cfgID), 1e-9, "three constant-usage runs must not accumulate commitment") + assert.Equal(t, 0.0, store.runs[2].TotalHourlyCommit, "run3 must be a Hold (zero new commitment)") +} + +// TestExecuteLadderRun_L5DriftUp_CancelsAndReplaces verifies the drift-UP path: +// when usage rises so the netted gap re-opens, the run produces a NEW scheduled +// generation and cancel-and-replaces the old one, leaving exactly one live +// generation. +// +// FAILS on the pre-netting code (which never nets and never cancels +// conditionally): the old generation would survive alongside the new, so the +// "old IDs cancelled" and "single live generation" assertions fail. PASSES +// after: the drift-up run supersedes. +func TestExecuteLadderRun_L5DriftUp_CancelsAndReplaces(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + const cfgID = "cfg-l5-driftup" + dbCfg := validTestDBConfig(cfgID) + dbCfg.RampSchedule = fullyDelayedRamp(t) + + store := &ladderStatefulStore{} + app := &Application{Config: store} + + // Run 1: low-water 10 -> target 8. Build a scheduled ramp summing 8. + lowCap := &fakeLadderCapability{t: t, baseline: testBaseline(10.0)} + require.NoError(t, app.executeLadderRun(ctx, &dbCfg, lowCap, "123456789012", + pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now)) + gen1 := store.scheduledTrancheIDs(cfgID) + require.NotEmpty(t, gen1, "run1 must build a scheduled ramp") + assert.InDelta(t, 8.0, store.scheduledSum(cfgID), 1e-9) + + // Run 2: usage DRIFTS UP (low-water 20 -> target 16). Netting subtracts the + // 8 already scheduled, leaving a fresh gap of 8 that re-opens allocation. + // This produces a NEW scheduled generation and cancel-and-replaces the old. + highCap := &fakeLadderCapability{t: t, baseline: testBaseline(20.0)} + require.NoError(t, app.executeLadderRun(ctx, &dbCfg, highCap, "123456789012", + pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now.Add(24*time.Hour))) + + // The old generation must now be fully cancelled. + for id := range gen1 { + assert.False(t, store.scheduledTrancheIDs(cfgID)[id], + "drift-up must cancel the old scheduled generation (id %s must no longer be scheduled)", id) + } + // Exactly one new live generation remains, and it does not include any run1 ID. + gen2 := store.scheduledTrancheIDs(cfgID) + require.NotEmpty(t, gen2, "drift-up must produce a new scheduled generation") + for id := range gen2 { + assert.False(t, gen1[id], "the new generation must not reuse a cancelled run1 tranche ID") + } + // The new live scheduled ramp reflects the re-opened gap (8), not 8+8. + assert.InDelta(t, 8.0, store.scheduledSum(cfgID), 1e-9, + "only the new generation may be live; the cancelled old one must not add to the scheduled sum") + assert.Greater(t, store.runs[1].TotalHourlyCommit, 0.0, "drift-up run must plan positive new commitment") +} From cfd4364091a3532f47c5e661944b80e0dfa13777 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 16 Jul 2026 22:06:06 +0300 Subject: [PATCH 4/4] fix(ladder): use append-only tranche persistence, drop cancel-and-replace (L5) Conditional cancel-and-replace fought the netting on drift-up and silently capped coverage at the old ramp's size. Trace: run1 target8 -> ramp8; run2 target16, inFlight=8 -> gap=8, but supersede CANCELLED the old 8 and inserted a new 8 -> total scheduled 8 vs target 16; run3 stuck at 8 forever. The coverage-decay L5 must prevent, moved to drift-up. Root fix - netting + APPEND-ONLY: - persistLadderRun ALWAYS calls SaveLadderRunWithTranches: it inserts the run's new scheduled tranches and never cancels existing ones. - Remove SaveLadderRunWithTranchesAndSupersede entirely: the StoreInterface entry, the PostgresStore impl + its cancel UPDATE, and every mock/test stub (analytics, server, shared mock). - Why append-only converges: netting already subtracts existing scheduled tranches from the gap, so each run appends exactly the delta to reach target-E. run1 appends 8; drift-up run2 appends 8 more (gap=16-8) -> total 16 = target; run3 nets to gap 0 -> Hold, preserve. No double-commit (netting caps it), no pile-up past target, no re-dating (append keeps prior fire dates), no oscillation. Tests: - Drift-up rewritten to a 3-run APPEND convergence: run1 target8 -> scheduled 8; run2 target16 -> appends to total 16 with run1's IDs still present; run3 -> Hold, preserve 16. (Old assertion codified the bug by treating scheduled==8-while-target==16 as correct.) - Keep the constant-usage preservation test (build once, Hold+preserve). - Add an over-commit guard: constant usage across 3 runs -> total scheduled never exceeds target-E and Hold runs append no tranches. - ladderStatefulStore is now append-only (no supersede path). - Store integration test asserts the append-only contract (both generations stay scheduled; in-flight sums both). - Drift-DOWN (usage drop -> Hold -> too-large ramp preserved) stays a documented, separately-tracked follow-up. All new lint issues (misspell on British spellings) resolved. --- internal/analytics/collector_test.go | 3 - internal/config/interfaces.go | 19 +- internal/config/store_postgres_ladder.go | 54 ----- internal/config/store_postgres_ladder_test.go | 132 ++++--------- internal/mocks/stores.go | 18 -- internal/server/handler_ladder.go | 68 +++---- internal/server/handler_ladder_test.go | 185 ++++++++++-------- internal/server/test_helpers_test.go | 3 - pkg/ladder/allocate.go | 5 +- 9 files changed, 178 insertions(+), 309 deletions(-) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 907970251..292b6138e 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -759,9 +759,6 @@ func (m *mockConfigStore) SaveLadderRun(_ context.Context, run *config.LadderRun func (m *mockConfigStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { return run, nil } -func (m *mockConfigStore) SaveLadderRunWithTranchesAndSupersede(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { - return run, nil -} func (m *mockConfigStore) GetInFlightLadderCommitUSDHr(_ context.Context, _ string) (*float64, error) { zero := 0.0 return &zero, nil diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index fb50cacbc..21605e430 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -379,17 +379,13 @@ type StoreInterface interface { // status=planned run never persists without its tranches (which would let // the cadence gate suppress the retry for a full window). // - // SaveLadderRunWithTranchesAndSupersede is the atomic cancel-and-replace - // variant (L5 spec). Within a single transaction it: - // 1. Cancels all status=scheduled tranches for the run's config_id, - // 2. Inserts the new ladder_runs row, and - // 3. Inserts the new ladder_tranches rows. - // This guarantees at most one generation of scheduled tranches per config - // exists at any point; a partial failure rolls back all three steps. - // Callers invoke this ONLY when the run produced a materially new - // generation of scheduled tranches (>=1 new scheduled purchase tranche); - // a Hold run persists via the plain SaveLadderRunWithTranches path so the - // live scheduled ramp is preserved rather than cancelled every cadence. + // L5 append-only: every ladder run persists its new scheduled tranches via + // this method and leaves any existing scheduled tranches untouched. In-flight + // netting (GetInFlightLadderCommitUSDHr) already subtracts existing + // scheduled tranches from the gap, so each run appends exactly the delta + // needed to reach target-E. Appending (never superseding) keeps prior + // tranches' original fire dates, converges to target on drift-up (run N + // tops up the gap the netting leaves), and cannot oscillate. // // GetInFlightLadderCommitUSDHr returns the total hourly USD commitment in // flight for the given config: the sum of amount_usd_hr for tranches with @@ -402,7 +398,6 @@ type StoreInterface interface { // nil without an error. SaveLadderRun(ctx context.Context, run *LadderRunDB) (*LadderRunDB, error) SaveLadderRunWithTranches(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) - SaveLadderRunWithTranchesAndSupersede(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) GetInFlightLadderCommitUSDHr(ctx context.Context, configID string) (*float64, error) GetLadderRun(ctx context.Context, id string) (*LadderRunDB, error) SaveLadderTranches(ctx context.Context, tranches []LadderTrancheDB) error diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index 4e3cb4c43..929721bf8 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -389,60 +389,6 @@ func (s *PostgresStore) GetInFlightLadderCommitUSDHr(ctx context.Context, config return &total, nil } -// SaveLadderRunWithTranchesAndSupersede is the atomic cancel-and-replace -// variant of SaveLadderRunWithTranches (L5 netting spec). Within a single -// transaction it: -// 1. Cancels all status=scheduled tranches for the run's config_id (sets -// status='cancelled'), ensuring exactly one generation of scheduled -// tranches exists per config after the call. -// 2. Inserts the new ladder_runs row. -// 3. Inserts the new ladder_tranches rows. -// -// If any step fails the whole transaction is rolled back. If run.ConfigID is -// nil the cancel step is skipped (the run has no prior config context). If -// run.ID is empty a fresh UUID is generated before the insert. -// -// Callers invoke this ONLY for a run that produced a materially new generation -// of scheduled tranches; a Hold run must use the plain SaveLadderRunWithTranches -// path so the existing scheduled ramp keeps its original fire dates. -func (s *PostgresStore) SaveLadderRunWithTranchesAndSupersede(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) { - if run.ID == "" { - run.ID = uuid.New().String() - } - var result LadderRunDB - err := s.WithTx(ctx, func(tx pgx.Tx) error { - // Step 1: cancel prior scheduled tranches for this config. Typed - // TrancheStatus constants (not raw string literals) keep the status - // set consistent with the engine and GetInFlightLadderCommitUSDHr. - if run.ConfigID != nil { - _, err := tx.Exec(ctx, ` - UPDATE ladder_tranches - SET status = $2 - WHERE config_id = $1 AND status = $3 - `, *run.ConfigID, string(ladder.TrancheStatusCancelled), string(ladder.TrancheStatusScheduled)) - if err != nil { - return fmt.Errorf("cancel prior scheduled tranches for config_id=%s: %w", *run.ConfigID, err) - } - } - // Step 2: insert the new run row. - row := tx.QueryRow(ctx, ladderRunInsertQuery, ladderRunInsertArgs(run, ladderRunPlanJSON(run))...) - scanned, err := scanLadderRun(row) - if err != nil { - return fmt.Errorf("failed to insert ladder_run id=%s: %w", run.ID, err) - } - // Step 3: insert the new tranche rows. - if err := insertLadderTranchesTx(ctx, tx, tranches); err != nil { - return err - } - result = scanned - return nil - }) - if err != nil { - return nil, err - } - return &result, 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_test.go b/internal/config/store_postgres_ladder_test.go index 669dacf21..2a993727f 100644 --- a/internal/config/store_postgres_ladder_test.go +++ b/internal/config/store_postgres_ladder_test.go @@ -396,8 +396,8 @@ func seedLadderConfigWithExtID(ctx context.Context, t *testing.T, store *Postgre // TestPostgresStore_GetInFlightLadderCommitUSDHr tests the in-flight sum // query (L5 netting). It verifies: // - zero is returned (not nil) when no scheduled tranches exist. -// - SCHEDULED tranches ONLY are summed; fired/completed/cancelled/failed are -// all excluded (fired/completed are already in ExistingUSDPerHour). +// - SCHEDULED tranches ONLY are summed; all non-scheduled statuses are +// excluded (fired/completed are already in ExistingUSDPerHour). // - tranches from a different config are not included. func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) @@ -463,43 +463,18 @@ func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { }) } -// TestPostgresStore_SaveLadderRunWithTranchesAndSupersede tests the L5 atomic -// cancel-and-replace: -// - prior scheduled tranches for the config are marked cancelled within the -// same transaction that inserts the new run+tranches. -// - fired/completed tranches are NOT touched. -// - a tranche-insert failure (duplicate ID) rolls back ALL three steps so no -// orphaned cancellations or run rows persist. -// - two configs remain isolated: supersede for config A does not cancel -// config B's tranches. -func TestPostgresStore_SaveLadderRunWithTranchesAndSupersede(t *testing.T) { +// TestPostgresStore_SaveLadderRunWithTranches_AppendOnly proves the L5 +// append-only contract at the store layer: persisting a new run's tranches for +// a config MUST NOT touch that config's existing scheduled tranches. (The +// cancel-and-replace supersede variant was removed; netting caps the appends.) +func TestPostgresStore_SaveLadderRunWithTranches_AppendOnly(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) defer cancel() store := setupLadderStore(ctx, t) - // Use distinct external_ids to satisfy UNIQUE(provider, external_id). configID := seedLadderConfigWithExtID(ctx, t, store, "333333333333") - otherConfigID := seedLadderConfigWithExtID(ctx, t, store, "444444444444") cfgID := configID - otherID := otherConfigID fire := time.Now().UTC().Add(7 * 24 * time.Hour).Truncate(time.Microsecond) - insertTranche := func(t *testing.T, cID *string, status ladder.TrancheStatus) string { - t.Helper() - id := uuid.New().String() - tr := LadderTrancheDB{ - ID: id, - ConfigID: cID, - LayerType: ladder.LayerConvertibleRI, - Term: ladder.Term1Year, - PaymentOption: ladder.PaymentNoUpfront, - Status: status, - AmountUSDHr: 1.0, - ScheduledDate: fire, - } - require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) - return id - } - statusOf := func(t *testing.T, id string) string { t.Helper() var s string @@ -508,70 +483,35 @@ func TestPostgresStore_SaveLadderRunWithTranchesAndSupersede(t *testing.T) { return s } - // Seed prior generation for the main config. - priorScheduledID := insertTranche(t, &cfgID, ladder.TrancheStatusScheduled) - priorFiredID := insertTranche(t, &cfgID, ladder.TrancheStatusFired) // must survive supersede - // Seed a tranche for another config; it must not be touched. - otherScheduledID := insertTranche(t, &otherID, ladder.TrancheStatusScheduled) - - t.Run("cancel-and-replace succeeds atomically", func(t *testing.T) { - runID := uuid.New().String() - run := &LadderRunDB{ID: runID, ConfigID: &cfgID, StartedAt: time.Now().UTC(), Status: ladder.RunStatusPlanned} - newTr := LadderTrancheDB{ - ID: uuid.New().String(), - ConfigID: &cfgID, - RunID: &runID, - LayerType: ladder.LayerConvertibleRI, - Term: ladder.Term1Year, - PaymentOption: ladder.PaymentNoUpfront, - Status: ladder.TrancheStatusScheduled, - AmountUSDHr: 2.5, - ScheduledDate: fire, - } - - saved, err := store.SaveLadderRunWithTranchesAndSupersede(ctx, run, []LadderTrancheDB{newTr}) - require.NoError(t, err) - require.NotNil(t, saved) - assert.Equal(t, runID, saved.ID) - - // Prior scheduled tranche must now be cancelled. - assert.Equal(t, "cancelled", statusOf(t, priorScheduledID), "prior scheduled tranche must be cancelled") - // Fired tranche must be untouched (not in the scheduled->cancelled CAS). - assert.Equal(t, "fired", statusOf(t, priorFiredID), "fired tranche must not be touched") - // Other config's scheduled tranche must be untouched. - assert.Equal(t, "scheduled", statusOf(t, otherScheduledID), "other config's tranche must not be cancelled") - - // The new tranche must be persisted with the correct status. - var count int - require.NoError(t, store.db.QueryRow(ctx, - `SELECT count(*) FROM ladder_tranches WHERE run_id = $1 AND status = 'scheduled'`, runID).Scan(&count)) - assert.Equal(t, 1, count, "new scheduled tranche must be persisted") - }) - - t.Run("rollback on duplicate tranche ID: no orphan cancellations", func(t *testing.T) { - // Seed a fresh scheduled tranche that should NOT be cancelled if tx rolls back. - freshScheduledID := insertTranche(t, &cfgID, ladder.TrancheStatusScheduled) - - runID := uuid.New().String() - run := &LadderRunDB{ID: runID, ConfigID: &cfgID, StartedAt: time.Now().UTC(), Status: ladder.RunStatusPlanned} - // Two identical IDs will cause a PK violation on the second insert. - dupID := uuid.New().String() - bad1 := LadderTrancheDB{ID: dupID, ConfigID: &cfgID, RunID: &runID, - LayerType: ladder.LayerConvertibleRI, Term: ladder.Term1Year, - PaymentOption: ladder.PaymentNoUpfront, Status: ladder.TrancheStatusScheduled, - AmountUSDHr: 1.0, ScheduledDate: fire} - bad2 := bad1 // same ID - - _, err := store.SaveLadderRunWithTranchesAndSupersede(ctx, run, []LadderTrancheDB{bad1, bad2}) - require.Error(t, err, "duplicate tranche ID must fail the transaction") + // Generation 1: an existing scheduled tranche (as if from a prior run). + gen1ID := uuid.New().String() + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{{ + ID: gen1ID, ConfigID: &cfgID, LayerType: ladder.LayerConvertibleRI, + Term: ladder.Term1Year, PaymentOption: ladder.PaymentNoUpfront, + Status: ladder.TrancheStatusScheduled, AmountUSDHr: 3.0, ScheduledDate: fire, + }})) + + // Generation 2: a new run appends its tranche. Gen1 must be untouched. + runID := uuid.New().String() + run := &LadderRunDB{ID: runID, ConfigID: &cfgID, StartedAt: time.Now().UTC(), Status: ladder.RunStatusPlanned} + gen2ID := uuid.New().String() + newTr := LadderTrancheDB{ + ID: gen2ID, ConfigID: &cfgID, RunID: &runID, LayerType: ladder.LayerConvertibleRI, + Term: ladder.Term1Year, PaymentOption: ladder.PaymentNoUpfront, + Status: ladder.TrancheStatusScheduled, AmountUSDHr: 2.5, ScheduledDate: fire, + } + saved, err := store.SaveLadderRunWithTranches(ctx, run, []LadderTrancheDB{newTr}) + require.NoError(t, err) + require.NotNil(t, saved) - // The fresh scheduled tranche must still be scheduled (cancellations rolled back). - assert.Equal(t, "scheduled", statusOf(t, freshScheduledID), - "rollback must undo the cancellation of prior scheduled tranches") + // Both generations must be live (scheduled): append-only never supersedes. + assert.Equal(t, string(ladder.TrancheStatusScheduled), statusOf(t, gen1ID), "existing scheduled tranche must be preserved, never superseded") + assert.Equal(t, string(ladder.TrancheStatusScheduled), statusOf(t, gen2ID), "new tranche must be persisted as scheduled") - // The run row must not exist. - got, err := store.GetLadderRun(ctx, runID) - require.NoError(t, err) - assert.Nil(t, got, "run row must be rolled back on tranche insert failure") - }) + // 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) + 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)") } diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index eb1c6fac3..c31e4b7cf 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1463,24 +1463,6 @@ func (m *MockConfigStore) GetInFlightLadderCommitUSDHr(ctx context.Context, conf return v, args.Error(1) } -// SaveLadderRunWithTranchesAndSupersede mocks the atomic cancel-and-replace -// operation (L5 spec). Returns the run unchanged when no expectation is -// registered, mirroring the SaveLadderRunWithTranches default. -func (m *MockConfigStore) SaveLadderRunWithTranchesAndSupersede(ctx context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { - if !isExpected(&m.Mock, "SaveLadderRunWithTranchesAndSupersede") { - return run, nil - } - args := m.Called(ctx, run, tranches) - if args.Get(0) == nil { - return nil, args.Error(1) - } - v, ok := args.Get(0).(*config.LadderRunDB) - if !ok { - panic(fmt.Sprintf("mock: expected *config.LadderRunDB, got %T", args.Get(0))) - } - return v, args.Error(1) -} - // TransitionLadderRunStatus mocks the TransitionLadderRunStatus operation. // Returns (nil, nil) when no expectation is registered (CAS race-lost path). func (m *MockConfigStore) TransitionLadderRunStatus(ctx context.Context, id string, fromStatuses []ladder.RunStatus, toStatus ladder.RunStatus) (*config.LadderRunDB, error) { diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go index 7c6e4ab1b..4de3b406a 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -281,13 +281,12 @@ func (app *Application) processOneLadderConfig( // counted here: they are executed purchases already reflected in the engine's // ExistingUSDPerHour, so counting them again would double-subtract. // -// Cancel-and-replace is CONDITIONAL (see persistLadderRun): it runs only when -// this run produces a materially new generation of scheduled tranches. A Hold -// run (the netting says we are covered) persists the run row WITHOUT cancelling -// the existing scheduled ramp, so those tranches keep their original fire dates -// and remain the live ramp. Unconditional cancellation would fight the netting: -// run2 nets to a Hold, but cancelling its own scheduled ramp would let run3 -// re-plan from zero (oscillation; late tranches would never fire). +// Persistence is APPEND-ONLY (see persistLadderRun): every run inserts its new +// scheduled tranches and NEVER supersedes existing ones. Netting already +// subtracted the config's existing scheduled tranches from the gap, so each run +// appends exactly the delta needed to reach target-E: a Hold appends nothing +// and preserves the live ramp with its original fire dates; a drift-up run +// appends the top-up so total scheduled commitment converges to target. func (app *Application) executeLadderRun( ctx context.Context, dbCfg *config.LadderConfigDB, @@ -368,10 +367,9 @@ func (app *Application) executeLadderRun( // Assemble LadderPlan (Q5: interleaves purchases AND reshapes in one JSON). plan := assembleLadderPlan(scope, now, allocResult, layerStates, baseline, engineCfg.TargetCoveragePct, term, paymentOpt) - // Validate and persist the run row + tranche audit rows. - // SaveLadderRunWithTranchesAndSupersede cancels any existing scheduled - // tranches for this config within the same transaction, then inserts the - // new run + tranches atomically (cancel-and-replace, L5 spec). + // Validate and persist the run row + tranche audit rows. L5 append-only: + // SaveLadderRunWithTranches inserts the run and its NEW scheduled tranches + // and leaves any existing scheduled tranches untouched (see persistLadderRun). return app.persistLadderRun(ctx, dbCfg, runID, now, plan, baseline, allocResult, trancheResult) } @@ -428,9 +426,25 @@ func (app *Application) persistLadderRun( return fmt.Errorf("buildTrancheDBRows: %w", err) } - // L5: persist the run, using cancel-and-replace supersede ONLY when this - // run produced a materially new generation of scheduled tranches. - savedRun, err := app.persistPlannedLadderRun(ctx, runRow, trancheRows) + // L5 APPEND-ONLY: always persist via the plain path, which INSERTS the new + // scheduled tranches and NEVER cancels existing ones. In-flight netting + // (GetInFlightLadderCommitUSDHr, subtracted from the gap in Allocate) has + // already removed the config's existing scheduled tranches from the gap, so + // trancheRows carries exactly the delta needed to reach target-E: + // - Hold (gap ~0): trancheRows is empty; only the run row is written and + // the existing scheduled ramp is preserved with its original fire dates. + // - Drift-up (gap re-opens): trancheRows is the top-up; appending it brings + // the total scheduled commitment to target-E while leaving (not re-dating + // or discarding) the prior generation intact. + // Cancel-and-replace was removed because it fought the netting: on drift-up + // it discarded the amount the netting had subtracted, capping coverage below + // target and never converging. + // + // Drift-DOWN note: if usage drops after a ramp is planned, the too-large + // scheduled ramp is preserved uncancelled (a Hold). That over-commitment is + // an accepted, separately-tracked follow-up (drift-down shrink); it is + // bounded by the low-water clamp in Allocate and not handled here. + savedRun, err := app.Config.SaveLadderRunWithTranches(ctx, runRow, trancheRows) if err != nil { return fmt.Errorf("persist ladder run: %w", err) } @@ -441,32 +455,6 @@ func (app *Application) persistLadderRun( return nil } -// persistPlannedLadderRun writes the ladder_runs row, choosing between the -// cancel-and-replace supersede path and the plain (preserve) path based on -// whether this run produced a materially new generation of scheduled tranches: -// -// - len(trancheRows) > 0: this run planned a new scheduled ramp, so -// cancel-and-replace the config's prior scheduled tranches atomically -// (SaveLadderRunWithTranchesAndSupersede). Exactly one generation of -// scheduled tranches survives per config. -// - len(trancheRows) == 0: a Hold (the L5 netting says the config is already -// covered by its live scheduled ramp) or a buy-now-only run. Persist the -// run row via the plain path WITHOUT cancelling the existing scheduled -// tranches, so they keep their original fire dates and remain the live -// ramp. Superseding here would cancel the ramp the netting just relied on, -// causing the next run to re-plan from zero (oscillation). -// -// Drift-DOWN note: a Hold that preserves a now-too-large scheduled ramp (usage -// dropped after the ramp was planned) is intentionally accepted for now. It is -// bounded by the low-water clamp in Allocate; shrinking an over-large ramp on -// drift-down is a follow-up (tracked separately), not handled here. -func (app *Application) persistPlannedLadderRun(ctx context.Context, runRow *config.LadderRunDB, trancheRows []config.LadderTrancheDB) (*config.LadderRunDB, error) { - if len(trancheRows) > 0 { - return app.Config.SaveLadderRunWithTranchesAndSupersede(ctx, runRow, trancheRows) - } - return app.Config.SaveLadderRunWithTranches(ctx, runRow, trancheRows) -} - // ladderWithinCadenceWindow reports whether a new run should be skipped because // a recent run already covers the cadence window. It queries // LatestLadderRunStartedAt for the config and returns: diff --git a/internal/server/handler_ladder_test.go b/internal/server/handler_ladder_test.go index 31b7eba3c..92025cfcc 100644 --- a/internal/server/handler_ladder_test.go +++ b/internal/server/handler_ladder_test.go @@ -187,9 +187,9 @@ func (s *ladderTestStore) SaveLadderTranches(_ context.Context, tranches []confi return nil } -// SaveLadderRunWithTranches models the single-transaction persist (without the -// cancel-and-replace step). Kept for tests that exercise the non-superseding -// path directly. +// SaveLadderRunWithTranches models the L5 append-only persist: it captures the +// run and its new tranches and leaves existing tranches untouched (there is no +// supersede path anymore). func (s *ladderTestStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { if s.saveLadderRunErr != nil { return nil, s.saveLadderRunErr @@ -203,23 +203,6 @@ func (s *ladderTestStore) SaveLadderRunWithTranches(_ context.Context, run *conf return run, nil } -// SaveLadderRunWithTranchesAndSupersede models the L5 cancel-and-replace. It -// captures the same fields as SaveLadderRunWithTranches; a real DB would also -// mark prior scheduled tranches cancelled, but the unit test only needs to -// verify the handler wires the correct method. -func (s *ladderTestStore) SaveLadderRunWithTranchesAndSupersede(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { - if s.saveLadderRunErr != nil { - return nil, s.saveLadderRunErr - } - if s.saveLadderTranchesErr != nil { - return nil, s.saveLadderTranchesErr - } - s.savedRun = run - s.savedRuns = append(s.savedRuns, run) - s.savedTranches = tranches - return run, nil -} - // GetInFlightLadderCommitUSDHr returns the configurable in-flight value. // When inFlightUSDHr is nil, returns a pointer to 0.0 (no in-flight default). func (s *ladderTestStore) GetInFlightLadderCommitUSDHr(_ context.Context, _ string) (*float64, error) { @@ -291,9 +274,16 @@ const testBaselineLowWaterUSDHr = 10.0 // testBaseline returns a UsageBaseline with a non-nil LowWaterUSDPerHour. func testBaseline() pkgladder.UsageBaseline { + return testBaselineLowWater(testBaselineLowWaterUSDHr) +} + +// testBaselineLowWater returns a UsageBaseline with the given LowWaterUSDPerHour. +// The L5 drift-up tests need distinct low-water values across runs to exercise +// the netted gap re-opening, so the low-water is parameterized here. +func testBaselineLowWater(lowWater float64) pkgladder.UsageBaseline { return pkgladder.UsageBaseline{ - LowWaterUSDPerHour: nonZeroFloat64(testBaselineLowWaterUSDHr), - StableUSDPerHour: nonZeroFloat64(testBaselineLowWaterUSDHr * 0.9), + LowWaterUSDPerHour: nonZeroFloat64(lowWater), + StableUSDPerHour: nonZeroFloat64(lowWater * 0.9), LookbackDays: 30, Percentile: 5.0, } @@ -1141,23 +1131,20 @@ func TestRatToFloat64Ptr_PositiveValue(t *testing.T) { } // ============================================================ -// L5 convergence + conditional-supersede (handler-level, real pipeline) +// L5 netting + append-only convergence (handler-level, real pipeline) // ============================================================ // ladderStatefulStore is a faithful in-memory stand-in for the ladder tranche // persistence that executeLadderRun drives. Unlike ladderTestStore (which // returns a FIXED in-flight value), it tracks each persisted tranche's live -// status so the L5 netting + conditional-supersede interplay can be exercised -// end-to-end across multiple sequential runs. It mirrors PostgresStore -// semantics exactly: +// status so the L5 netting + append-only interplay can be exercised end-to-end +// across multiple sequential runs. It mirrors PostgresStore semantics exactly: // - GetInFlightLadderCommitUSDHr sums SCHEDULED tranches only, per config. -// - SaveLadderRunWithTranchesAndSupersede cancels the config's live scheduled -// tranches, then appends the run + the new generation (cancel-and-replace). -// - SaveLadderRunWithTranches appends the run + tranches WITHOUT cancelling -// (the Hold/preserve path). +// - SaveLadderRunWithTranches APPENDS the run + its new tranches and NEVER +// cancels existing scheduled tranches (there is no supersede path). // // The tests below drive the REAL executeLadderRun -> Allocate -> BuildTranches -// -> persistPlannedLadderRun pipeline; only the store and capability are fakes. +// -> persist pipeline; only the store and capability are fakes. type ladderStatefulStore struct { mockConfigStoreForHealth tranches []config.LadderTrancheDB // every persisted tranche, live status @@ -1196,32 +1183,19 @@ func (s *ladderStatefulStore) GetInFlightLadderCommitUSDHr(_ context.Context, co return &total, nil } +// SaveLadderRunWithTranches is APPEND-ONLY: it records the run and appends its +// new tranches without touching any existing tranche's status. This mirrors the +// production store, whose supersede/cancel path was removed. func (s *ladderStatefulStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { s.runs = append(s.runs, run) s.tranches = append(s.tranches, tranches...) return run, nil } -func (s *ladderStatefulStore) SaveLadderRunWithTranchesAndSupersede(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { - // Cancel-and-replace: flip the config's live scheduled tranches to - // cancelled, then append the new generation. - if run.ConfigID != nil { - for i := range s.tranches { - tr := &s.tranches[i] - if tr.ConfigID != nil && *tr.ConfigID == *run.ConfigID && tr.Status == pkgladder.TrancheStatusScheduled { - tr.Status = pkgladder.TrancheStatusCancelled - } - } - } - s.runs = append(s.runs, run) - s.tranches = append(s.tranches, tranches...) - return run, nil -} - // fullyDelayedRamp is a 2-step ramp with NO AfterDays==0 step, so ALL planned // commitment lands as SCHEDULED tranches (no buy-now). With a static E=0 fake // capability, that scheduled ramp is the only in-flight commitment, so the L5 -// netting alone decides whether a follow-up run Holds. +// netting alone decides whether a follow-up run Holds or tops up. func fullyDelayedRamp(t *testing.T) json.RawMessage { t.Helper() raw, err := json.Marshal(pkgladder.RampSchedule{ @@ -1237,13 +1211,13 @@ func fullyDelayedRamp(t *testing.T) json.RawMessage { // TestExecuteLadderRun_L5Convergence_PreservesRampAcrossHolds is the core L5 // regression: three sequential runs with CONSTANT usage must build the ramp // once (run1) and then PRESERVE it (run2, run3 Hold), never accumulating and -// never cancelling the live scheduled ramp. +// never discarding the live scheduled ramp. // -// FAILS on the pre-L5 code (unconditional supersede): run2 nets to a Hold but -// the old code still calls SaveLadderRunWithTranchesAndSupersede with an empty -// tranche set, cancelling the run1 ramp; the scheduled sum drops to 0, so this -// test's preservation assertions fail. PASSES after: run2/run3 take the plain -// preserve path. +// Guards the append-only + netting invariant: a Hold run must leave the config's +// existing scheduled tranches exactly as they were (same IDs, same total). A +// regression that discarded or re-planned the ramp on a Hold (the old +// cancel-and-replace bug) would drop the scheduled sum, failing these +// preservation assertions. func TestExecuteLadderRun_L5Convergence_PreservesRampAcrossHolds(t *testing.T) { ctx := testutil.TestContext(t) now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) @@ -1254,9 +1228,9 @@ func TestExecuteLadderRun_L5Convergence_PreservesRampAcrossHolds(t *testing.T) { store := &ladderStatefulStore{} app := &Application{Config: store} // Constant usage across all three runs: low-water 10 -> target 8 (80%). - cap := &fakeLadderCapability{t: t, baseline: testBaseline(10.0)} + capability := &fakeLadderCapability{t: t, baseline: testBaseline()} run := func(at time.Time) { - require.NoError(t, app.executeLadderRun(ctx, &dbCfg, cap, "123456789012", + require.NoError(t, app.executeLadderRun(ctx, &dbCfg, capability, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, at)) } @@ -1284,16 +1258,17 @@ func TestExecuteLadderRun_L5Convergence_PreservesRampAcrossHolds(t *testing.T) { assert.Equal(t, 0.0, store.runs[2].TotalHourlyCommit, "run3 must be a Hold (zero new commitment)") } -// TestExecuteLadderRun_L5DriftUp_CancelsAndReplaces verifies the drift-UP path: -// when usage rises so the netted gap re-opens, the run produces a NEW scheduled -// generation and cancel-and-replaces the old one, leaving exactly one live -// generation. +// TestExecuteLadderRun_L5DriftUp_AppendsToTargetAndConverges verifies the +// drift-UP path under append-only: when usage rises so the netted gap re-opens, +// the run APPENDS exactly the delta (target-E minus already-scheduled) and the +// prior generation is PRESERVED (not superseded), so the total scheduled +// commitment converges UP to the new target rather than being capped at the old +// ramp's size. // -// FAILS on the pre-netting code (which never nets and never cancels -// conditionally): the old generation would survive alongside the new, so the -// "old IDs cancelled" and "single live generation" assertions fail. PASSES -// after: the drift-up run supersedes. -func TestExecuteLadderRun_L5DriftUp_CancelsAndReplaces(t *testing.T) { +// This codifies the corrected behavior: the earlier cancel-and-replace design +// discarded the old 8 and inserted a new 8, capping coverage at 8 while target +// was 16 (coverage decay). Append-only reaches 16 = target. +func TestExecuteLadderRun_L5DriftUp_AppendsToTargetAndConverges(t *testing.T) { ctx := testutil.TestContext(t) now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) const cfgID = "cfg-l5-driftup" @@ -1304,33 +1279,79 @@ func TestExecuteLadderRun_L5DriftUp_CancelsAndReplaces(t *testing.T) { app := &Application{Config: store} // Run 1: low-water 10 -> target 8. Build a scheduled ramp summing 8. - lowCap := &fakeLadderCapability{t: t, baseline: testBaseline(10.0)} + lowCap := &fakeLadderCapability{t: t, baseline: testBaselineLowWater(10.0)} require.NoError(t, app.executeLadderRun(ctx, &dbCfg, lowCap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now)) gen1 := store.scheduledTrancheIDs(cfgID) require.NotEmpty(t, gen1, "run1 must build a scheduled ramp") - assert.InDelta(t, 8.0, store.scheduledSum(cfgID), 1e-9) + assert.InDelta(t, 8.0, store.scheduledSum(cfgID), 1e-9, "run1 ramp must sum to target-E (8)") // Run 2: usage DRIFTS UP (low-water 20 -> target 16). Netting subtracts the - // 8 already scheduled, leaving a fresh gap of 8 that re-opens allocation. - // This produces a NEW scheduled generation and cancel-and-replaces the old. - highCap := &fakeLadderCapability{t: t, baseline: testBaseline(20.0)} + // 8 already scheduled, leaving a fresh gap of 8 that APPENDS a new tranche + // generation on top of the preserved run1 generation. Total scheduled must + // converge to the new target (16), NOT stay capped at 8. + highCap := &fakeLadderCapability{t: t, baseline: testBaselineLowWater(20.0)} require.NoError(t, app.executeLadderRun(ctx, &dbCfg, highCap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now.Add(24*time.Hour))) - // The old generation must now be fully cancelled. + sched2 := store.scheduledTrancheIDs(cfgID) + // Run1's tranches must STILL be scheduled (append-only preserves them). for id := range gen1 { - assert.False(t, store.scheduledTrancheIDs(cfgID)[id], - "drift-up must cancel the old scheduled generation (id %s must no longer be scheduled)", id) + assert.True(t, sched2[id], "drift-up must PRESERVE the run1 tranche %s (append-only, no cancel)", id) } - // Exactly one new live generation remains, and it does not include any run1 ID. - gen2 := store.scheduledTrancheIDs(cfgID) - require.NotEmpty(t, gen2, "drift-up must produce a new scheduled generation") - for id := range gen2 { - assert.False(t, gen1[id], "the new generation must not reuse a cancelled run1 tranche ID") - } - // The new live scheduled ramp reflects the re-opened gap (8), not 8+8. - assert.InDelta(t, 8.0, store.scheduledSum(cfgID), 1e-9, - "only the new generation may be live; the cancelled old one must not add to the scheduled sum") + // The new generation adds tranches on top of run1's. + assert.Greater(t, len(sched2), len(gen1), "drift-up must append a new generation, not replace") + // Total scheduled commitment converges to the new target-E (16), not 8. + assert.InDelta(t, 16.0, store.scheduledSum(cfgID), 1e-9, + "append-only drift-up must converge scheduled commitment to target (16), not cap at the old ramp (8)") assert.Greater(t, store.runs[1].TotalHourlyCommit, 0.0, "drift-up run must plan positive new commitment") + + // Run 3: SAME (drifted-up) usage. Netting now sees 16 scheduled == target-E + // -> Hold. Total must stay at 16 (no further accumulation), ramp preserved. + require.NoError(t, app.executeLadderRun(ctx, &dbCfg, highCap, "123456789012", + pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now.Add(48*time.Hour))) + assert.InDelta(t, 16.0, store.scheduledSum(cfgID), 1e-9, + "run3 must Hold at target (16); netting prevents overshoot past the target") + assert.Equal(t, 0.0, store.runs[2].TotalHourlyCommit, "run3 must be a Hold (zero new commitment)") +} + +// TestExecuteLadderRun_L5OverCommitGuard_NettingCapsAppends proves the netting +// caps appends: under CONSTANT usage, three sequential runs must never let the +// total scheduled commitment exceed target-E, and Hold runs must append NO new +// tranches. Without netting, each run would re-plan the full gap and pile up +// (8 -> 16 -> 24); netting keeps it pinned at target-E after the first build. +func TestExecuteLadderRun_L5OverCommitGuard_NettingCapsAppends(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + const cfgID = "cfg-l5-overcommit" + // target-E = low-water(10) * 80% - 0 = 8. + const targetMinusE = 8.0 + dbCfg := validTestDBConfig(cfgID) + dbCfg.RampSchedule = fullyDelayedRamp(t) + + store := &ladderStatefulStore{} + app := &Application{Config: store} + capability := &fakeLadderCapability{t: t, baseline: testBaseline()} + + // Run 1: builds the ramp to target-E. + require.NoError(t, app.executeLadderRun(ctx, &dbCfg, capability, "123456789012", + pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now)) + assert.LessOrEqual(t, store.scheduledSum(cfgID), targetMinusE+1e-9, + "run1 scheduled commitment must not exceed target-E") + trancheCountAfterBuild := len(store.tranches) + require.Positive(t, trancheCountAfterBuild, "run1 must persist tranches") + + // Runs 2 and 3: constant usage -> Hold -> netting caps appends at target-E, + // and no new tranche rows are appended on a Hold. + for i, at := range []time.Time{now.Add(24 * time.Hour), now.Add(48 * time.Hour)} { + require.NoError(t, app.executeLadderRun(ctx, &dbCfg, capability, "123456789012", + pkgladder.Term1Year, pkgladder.PaymentNoUpfront, at)) + assert.LessOrEqual(t, store.scheduledSum(cfgID), targetMinusE+1e-9, + "run %d: netting must cap total scheduled commitment at target-E (no over-commit)", i+2) + assert.Equal(t, trancheCountAfterBuild, len(store.tranches), + "run %d Hold must append NO new tranches (netting closed the gap)", i+2) + } + // Final: still pinned exactly at target-E. + assert.InDelta(t, targetMinusE, store.scheduledSum(cfgID), 1e-9, + "after 3 constant-usage runs the scheduled commitment must equal target-E, never a multiple") } diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 57787c7f2..c01b9a6fb 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -341,9 +341,6 @@ func (m *mockConfigStoreForHealth) SaveLadderRun(_ context.Context, run *config. func (m *mockConfigStoreForHealth) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { return run, nil } -func (m *mockConfigStoreForHealth) SaveLadderRunWithTranchesAndSupersede(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { - return run, nil -} func (m *mockConfigStoreForHealth) GetInFlightLadderCommitUSDHr(_ context.Context, _ string) (*float64, error) { zero := 0.0 return &zero, nil diff --git a/pkg/ladder/allocate.go b/pkg/ladder/allocate.go index c09ad7481..e0f674754 100644 --- a/pkg/ladder/allocate.go +++ b/pkg/ladder/allocate.go @@ -146,8 +146,11 @@ func Allocate(in *AllocationInput) (*AllocateResult, error) { // The two-arm min is defensive: it guards against future target derivations // exceeding the observed floor (today target<=lowWater always holds because // TargetCoveragePct is validated to be <=100). Subtracting inFlight nets - // out scheduled/fired tranches that are already en route so repeated runs + // out scheduled-only tranches that are already en route so repeated runs // on the same day do not re-plan commitment that is already in flight. + // Only scheduled (not-yet-fired) tranches are netted here; fired tranches + // are already reflected in existingTotal (E), so netting them too would + // double-count and under-commit. preInflightGap := minRat(new(big.Rat).Sub(target, existingTotal), new(big.Rat).Sub(lowWater, existingTotal)) gap := new(big.Rat).Sub(preInflightGap, inFlight) if gap.Cmp(minAllocatableGap()) <= 0 {