diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index adcd046eb..292b6138e 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -759,6 +759,10 @@ 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) 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..21605e430 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -378,8 +378,27 @@ 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). + // + // 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 + // 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) + 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..929721bf8 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -361,6 +361,34 @@ 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 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) { + var total float64 + err := s.db.QueryRow(ctx, ` + SELECT COALESCE(SUM(amount_usd_hr), 0) + FROM ladder_tranches + WHERE config_id = $1 + AND status = $2 + `, configID, string(ladder.TrancheStatusScheduled)).Scan(&total) + if err != nil { + return nil, fmt.Errorf("GetInFlightLadderCommitUSDHr config_id=%s: %w", configID, err) + } + return &total, nil +} + // LatestLadderRunStartedAt returns the maximum started_at for the given // config_id, or nil when no run has been recorded yet. Drives the per-cadence // self-gate in handleLadderRun (Q6). diff --git a/internal/config/store_postgres_ladder_test.go b/internal/config/store_postgres_ladder_test.go index 6bf329ad9..2a993727f 100644 --- a/internal/config/store_postgres_ladder_test.go +++ b/internal/config/store_postgres_ladder_test.go @@ -359,3 +359,159 @@ 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 scheduled tranches exist. +// - 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) + 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) // 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) + + // 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 only", func(t *testing.T) { + result, err := store.GetInFlightLadderCommitUSDHr(ctx, configID) + require.NoError(t, err) + require.NotNil(t, result) + // 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) { + 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_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) + configID := seedLadderConfigWithExtID(ctx, t, store, "333333333333") + cfgID := configID + fire := time.Now().UTC().Add(7 * 24 * time.Hour).Truncate(time.Microsecond) + + 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 + } + + // 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) + + // 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") + + // 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 cc7deb90c..c31e4b7cf 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1444,6 +1444,25 @@ 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) +} + // 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..4de3b406a 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -272,6 +272,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 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. +// +// 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, @@ -300,16 +315,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 = 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) + } + 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 the config's + // prior scheduled (not-yet-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) @@ -333,7 +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. + // 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) } @@ -390,12 +426,27 @@ 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). + // 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("SaveLadderRunWithTranches: %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)", diff --git a/internal/server/handler_ladder_test.go b/internal/server/handler_ladder_test.go index da04c670b..92025cfcc 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 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 @@ -198,6 +203,19 @@ func (s *ladderTestStore) SaveLadderRunWithTranches(_ context.Context, run *conf 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 // ============================================================ @@ -256,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, } @@ -1055,8 +1080,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), @@ -1102,3 +1129,229 @@ 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 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 + 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. +// - 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 +// -> persist 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 +} + +// 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 +} + +// 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 or tops up. +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 discarding the live scheduled ramp. +// +// 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) + 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%). + capability := &fakeLadderCapability{t: t, baseline: testBaseline()} + run := func(at time.Time) { + require.NoError(t, app.executeLadderRun(ctx, &dbCfg, capability, "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_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. +// +// 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" + 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: 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, "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 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))) + + sched2 := store.scheduledTrancheIDs(cfgID) + // Run1's tranches must STILL be scheduled (append-only preserves them). + for id := range gen1 { + assert.True(t, sched2[id], "drift-up must PRESERVE the run1 tranche %s (append-only, no cancel)", id) + } + // 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 6a628bde7..c01b9a6fb 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -341,6 +341,10 @@ 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) 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..e0f674754 100644 --- a/pkg/ladder/allocate.go +++ b/pkg/ladder/allocate.go @@ -35,6 +35,18 @@ func minAllocatableGap() *big.Rat { // and causes Allocate to return an error (fail loud: incomplete provider data // aborts the run). // +// 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 // and PlannedAction produced by Allocate so that approval emails and audit @@ -42,12 +54,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 +138,23 @@ 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-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 { - 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 +200,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 +442,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()