Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions internal/analytics/collector_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
19 changes: 19 additions & 0 deletions internal/config/interfaces.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
28 changes: 28 additions & 0 deletions internal/config/store_postgres_ladder.go
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down
156 changes: 156 additions & 0 deletions internal/config/store_postgres_ladder_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)")
}
19 changes: 19 additions & 0 deletions internal/mocks/stores.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
75 changes: 63 additions & 12 deletions internal/server/handler_ladder.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand All @@ -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)
}

Expand Down Expand Up @@ -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)",
Expand Down
Loading
Loading