diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 5c7019925..70e7359e8 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/LeanerCloud/CUDly/internal/config" + "github.com/LeanerCloud/CUDly/pkg/ladder" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -747,6 +748,24 @@ func (m *mockConfigStore) GetLadderConfig(_ context.Context, _, _ string) (*conf func (m *mockConfigStore) UpsertLadderConfig(_ context.Context, cfg *config.LadderConfigDB) (*config.LadderConfigDB, error) { return cfg, nil } +func (m *mockConfigStore) SaveLadderRun(_ context.Context, run *config.LadderRunDB) (*config.LadderRunDB, error) { + return run, nil +} +func (m *mockConfigStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { + return run, nil +} +func (m *mockConfigStore) GetLadderRun(_ context.Context, _ string) (*config.LadderRunDB, error) { + return nil, nil +} +func (m *mockConfigStore) SaveLadderTranches(_ context.Context, _ []config.LadderTrancheDB) error { + return nil +} +func (m *mockConfigStore) LatestLadderRunStartedAt(_ context.Context, _ string) (*time.Time, error) { + return nil, nil +} +func (m *mockConfigStore) TransitionLadderRunStatus(_ context.Context, _ string, _ []ladder.RunStatus, _ ladder.RunStatus) (*config.LadderRunDB, error) { + return nil, nil +} func (m *mockConfigStore) UpdateGlobalConfigAtomic(_ context.Context, apply func(*config.GlobalConfig) error) (*config.GlobalConfig, error) { cfg := &config.GlobalConfig{} if err := apply(cfg); err != nil { diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 1c1067ef8..5af9a9942 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -5,6 +5,8 @@ import ( "time" "github.com/jackc/pgx/v5" + + "github.com/LeanerCloud/CUDly/pkg/ladder" ) // StoreInterface defines the methods required for configuration storage. @@ -329,4 +331,38 @@ type StoreInterface interface { GetLadderConfigs(ctx context.Context) ([]LadderConfigDB, error) GetLadderConfig(ctx context.Context, cloudAccountID, provider string) (*LadderConfigDB, error) UpsertLadderConfig(ctx context.Context, cfg *LadderConfigDB) (*LadderConfigDB, error) + + // Ladder run/tranche persistence (migration 000080/000081, PR-2). + // + // SaveLadderRun inserts a new ladder_runs row, returning the persisted row + // with all DB-stamped fields (id, created_at, updated_at) populated. + // If run.ID is empty, a new UUID is generated before the insert. + // + // GetLadderRun returns the row for the given id, or (nil, nil) when no + // row exists (mirrors GetLadderConfig semantics). + // + // SaveLadderTranches inserts a batch of ladder_tranches rows inside a + // single transaction. Each tranche must carry a non-empty ID; duplicate + // IDs within the batch are rejected at the DB UNIQUE constraint level. + // + // LatestLadderRunStartedAt returns the maximum started_at for the given + // config_id, or nil when no run has been recorded yet. Powers the per-cadence + // self-gate in the scheduler (Q6). + // + // TransitionLadderRunStatus atomically updates the status of a ladder_runs + // row from one of the fromStatuses to toStatus, returning the updated row. + // Returns (nil, nil) when zero rows are affected (CAS race lost or wrong + // current status), so callers can distinguish a race from a hard error. + // Statuses are typed ladder.RunStatus so callers cannot pass an arbitrary + // string that would never match a stored status. + // SaveLadderRunWithTranches inserts the run row and its tranches in ONE + // 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). + SaveLadderRun(ctx context.Context, run *LadderRunDB) (*LadderRunDB, error) + SaveLadderRunWithTranches(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, 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) + TransitionLadderRunStatus(ctx context.Context, id string, fromStatuses []ladder.RunStatus, toStatus ladder.RunStatus) (*LadderRunDB, error) } diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index 5d3290312..5a3097361 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -9,6 +9,8 @@ import ( "github.com/google/uuid" "github.com/jackc/pgx/v5" + + "github.com/LeanerCloud/CUDly/pkg/ladder" ) // ========================================== @@ -177,6 +179,277 @@ func scanLadderConfig(row scannable) (LadderConfigDB, error) { return cfg, nil } +// ========================================== +// LADDER RUN / TRANCHE STORE METHODS +// ========================================== + +// ladderRunInsertQuery INSERTs one ladder_runs row and RETURNs the full row for +// scanLadderRun. Shared by SaveLadderRun and SaveLadderRunWithTranches so the +// column list has a single definition. +const ladderRunInsertQuery = ` + INSERT INTO ladder_runs ( + id, config_id, started_at, completed_at, status, mode, cadence, + baseline_usd_hr, target_usd_hr, existing_usd_hr, gap_usd_hr, + plan, total_hourly_commit, total_upfront_cost, estimated_savings, + approval_token_hash, approval_token_expires_at, + approved_by, cancelled_by, fire_at, + created_at, updated_at + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7, + $8, $9, $10, $11, + $12, $13, $14, $15, + $16, $17, $18, $19, $20, + NOW(), NOW() + ) + RETURNING id, config_id, started_at, completed_at, status, mode, cadence, + baseline_usd_hr, target_usd_hr, existing_usd_hr, gap_usd_hr, + plan, total_hourly_commit, total_upfront_cost, estimated_savings, + approval_token_hash, approval_token_expires_at, + approved_by, cancelled_by, fire_at, + created_at, updated_at +` + +// ladderRunPlanJSON returns the plan blob to persist, defaulting an empty plan +// to the JSONB '{}' the schema expects (NOT NULL DEFAULT '{}'). +func ladderRunPlanJSON(run *LadderRunDB) json.RawMessage { + if len(run.Plan) == 0 { + return json.RawMessage(`{}`) + } + return run.Plan +} + +// ladderRunInsertArgs returns the positional args for ladderRunInsertQuery in +// column order. Nullable monetary fields stay as *float64 so NULL != $0. +func ladderRunInsertArgs(run *LadderRunDB, planJSON json.RawMessage) []any { + return []any{ + run.ID, + run.ConfigID, + run.StartedAt, + run.CompletedAt, + run.Status, + run.Mode, + run.Cadence, + run.BaselineUSDHr, + run.TargetUSDHr, + run.ExistingUSDHr, + run.GapUSDHr, + planJSON, + run.TotalHourlyCommit, + run.TotalUpfrontCost, + run.EstimatedSavings, + run.ApprovalTokenHash, + run.ApprovalTokenExpiresAt, + run.ApprovedBy, + run.CancelledBy, + run.FireAt, + } +} + +// SaveLadderRun inserts a new ladder_runs row. If run.ID is empty a fresh UUID +// is generated. Returns the persisted row with all DB-stamped fields populated. +func (s *PostgresStore) SaveLadderRun(ctx context.Context, run *LadderRunDB) (*LadderRunDB, error) { + if run.ID == "" { + run.ID = uuid.New().String() + } + row := s.db.QueryRow(ctx, ladderRunInsertQuery, ladderRunInsertArgs(run, ladderRunPlanJSON(run))...) + result, err := scanLadderRun(row) + if err != nil { + return nil, fmt.Errorf("failed to insert ladder_run id=%s: %w", run.ID, err) + } + return &result, nil +} + +// GetLadderRun returns the ladder_runs row for the given id, or (nil, nil) +// when no row exists. +func (s *PostgresStore) GetLadderRun(ctx context.Context, id string) (*LadderRunDB, error) { + query := ` + SELECT id, config_id, started_at, completed_at, status, mode, cadence, + baseline_usd_hr, target_usd_hr, existing_usd_hr, gap_usd_hr, + plan, total_hourly_commit, total_upfront_cost, estimated_savings, + approval_token_hash, approval_token_expires_at, + approved_by, cancelled_by, fire_at, + created_at, updated_at + FROM ladder_runs + WHERE id = $1 + ` + row := s.db.QueryRow(ctx, query, id) + result, err := scanLadderRun(row) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, fmt.Errorf("failed to get ladder_run id=%s: %w", id, err) + } + return &result, nil +} + +// insertLadderTranchesTx inserts every tranche within the given transaction. +// Each tranche must carry a non-empty ID; duplicate IDs are rejected at the DB +// PRIMARY KEY constraint. Shared by SaveLadderTranches and +// SaveLadderRunWithTranches so both go through identical insert logic. +func insertLadderTranchesTx(ctx context.Context, tx pgx.Tx, tranches []LadderTrancheDB) error { + for i := range tranches { + tr := &tranches[i] + if tr.ID == "" { + return fmt.Errorf("ladder_tranche at index %d has an empty ID; every tranche must carry a unique non-empty ID", i) + } + _, err := tx.Exec(ctx, ` + INSERT INTO ladder_tranches ( + id, config_id, run_id, layer_type, amount_usd_hr, + term, payment_option, scheduled_date, status, execution_id, + created_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NOW()) + `, + tr.ID, + tr.ConfigID, + tr.RunID, + string(tr.LayerType), + tr.AmountUSDHr, + string(tr.Term), + string(tr.PaymentOption), + tr.ScheduledDate, + string(tr.Status), + tr.ExecutionID, + ) + if err != nil { + return fmt.Errorf("failed to insert ladder_tranche id=%s run_id=%v: %w", tr.ID, tr.RunID, err) + } + } + return nil +} + +// SaveLadderTranches inserts a batch of ladder_tranches rows within a +// single transaction. Each tranche must carry a non-empty ID; duplicate +// IDs are rejected at the DB UNIQUE PRIMARY KEY constraint. An empty +// slice is a no-op. +func (s *PostgresStore) SaveLadderTranches(ctx context.Context, tranches []LadderTrancheDB) error { + if len(tranches) == 0 { + return nil + } + return s.WithTx(ctx, func(tx pgx.Tx) error { + return insertLadderTranchesTx(ctx, tx, tranches) + }) +} + +// SaveLadderRunWithTranches inserts the ladder_runs row AND its ladder_tranches +// rows in ONE transaction. If any tranche insert fails, the whole transaction +// (including the run row) is rolled back, so a run is never persisted without +// its tranches. This prevents a status=planned run with zero tranches, which +// the cadence self-gate (keyed on any run's started_at) would otherwise use to +// suppress the retry for the full cadence window. If run.ID is empty a fresh +// UUID is generated. Returns the persisted run row. +func (s *PostgresStore) SaveLadderRunWithTranches(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 { + 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) + } + 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). +func (s *PostgresStore) LatestLadderRunStartedAt(ctx context.Context, configID string) (*time.Time, error) { + query := `SELECT MAX(started_at) FROM ladder_runs WHERE config_id = $1` + var ts *time.Time + err := s.db.QueryRow(ctx, query, configID).Scan(&ts) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, fmt.Errorf("failed to get latest ladder_run started_at for config_id=%s: %w", configID, err) + } + return ts, nil +} + +// TransitionLadderRunStatus atomically transitions a ladder_runs row from +// one of fromStatuses to toStatus via a CAS UPDATE. Returns the updated row +// on success, or (nil, nil) when zero rows are affected (race lost or +// unexpected current status). A hard DB error is returned as a non-nil error. +func (s *PostgresStore) TransitionLadderRunStatus(ctx context.Context, id string, fromStatuses []ladder.RunStatus, toStatus ladder.RunStatus) (*LadderRunDB, error) { + // Convert the typed enum slice to the plain []string that pgx encodes for + // the ANY($3) text[] comparison. Keeping the public signature typed while + // converting here confines the stringly-typed shape to the DB boundary. + from := make([]string, len(fromStatuses)) + for i, st := range fromStatuses { + from[i] = string(st) + } + query := ` + UPDATE ladder_runs + SET status = $2, updated_at = NOW() + WHERE id = $1 AND status = ANY($3) + RETURNING id, config_id, started_at, completed_at, status, mode, cadence, + baseline_usd_hr, target_usd_hr, existing_usd_hr, gap_usd_hr, + plan, total_hourly_commit, total_upfront_cost, estimated_savings, + approval_token_hash, approval_token_expires_at, + approved_by, cancelled_by, fire_at, + created_at, updated_at + ` + row := s.db.QueryRow(ctx, query, id, string(toStatus), from) + result, err := scanLadderRun(row) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil // CAS race lost + } + return nil, fmt.Errorf("failed to transition ladder_run id=%s to status=%s: %w", id, toStatus, err) + } + return &result, nil +} + +// scanLadderRun scans a single ladder_runs row from a pgx.Row or pgx.Rows. +// Both implement the scannable interface (store_postgres_registrations.go). +func scanLadderRun(row scannable) (LadderRunDB, error) { + var r LadderRunDB + var planJSON []byte + + err := row.Scan( + &r.ID, + &r.ConfigID, + &r.StartedAt, + &r.CompletedAt, + &r.Status, + &r.Mode, + &r.Cadence, + &r.BaselineUSDHr, + &r.TargetUSDHr, + &r.ExistingUSDHr, + &r.GapUSDHr, + &planJSON, + &r.TotalHourlyCommit, + &r.TotalUpfrontCost, + &r.EstimatedSavings, + &r.ApprovalTokenHash, + &r.ApprovalTokenExpiresAt, + &r.ApprovedBy, + &r.CancelledBy, + &r.FireAt, + &r.CreatedAt, + &r.UpdatedAt, + ) + if err != nil { + return LadderRunDB{}, err + } + if planJSON != nil { + r.Plan = json.RawMessage(planJSON) + } + return r, nil +} + // marshalRampSchedule ensures the ramp_schedule value stored in the DB is // valid JSON. Nil/empty input is rejected; already-valid JSON is passed // through unchanged. Non-JSON input is rejected with a descriptive error. diff --git a/internal/config/store_postgres_ladder_test.go b/internal/config/store_postgres_ladder_test.go new file mode 100644 index 000000000..6bf329ad9 --- /dev/null +++ b/internal/config/store_postgres_ladder_test.go @@ -0,0 +1,361 @@ +//go:build integration +// +build integration + +package config + +// Real-DB integration tests for the ladder_runs / ladder_tranches store +// methods added in PR-2 (migrations 000080/000081). They run against a +// throwaway PostgreSQL container (testhelpers.SetupPostgresContainer), the +// same harness the other *_db_test.go / *_unassigned_test.go integration +// tests use, and are gated behind the `integration` build tag so the default +// `go test ./...` run stays hermetic. +// +// Coverage: +// - SaveLadderRun round-trips a run with all NULL monetary columns preserved +// as nil (never 0-coerced) AND a run with populated monetary columns. +// - LatestLadderRunStartedAt returns nil with no rows and the MAX(started_at) +// across multiple runs. +// - SaveLadderTranches batch-inserts every row in one transaction. +// - TransitionLadderRunStatus performs a CAS: it updates when the current +// status is in the from-set (win) and is a no-op returning (nil, nil) when +// it is not (lose). + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/CUDly/internal/database/postgres/migrations" + "github.com/LeanerCloud/CUDly/internal/database/postgres/testhelpers" + "github.com/LeanerCloud/CUDly/pkg/ladder" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// setupLadderStore starts a container, runs migrations, and returns a store. +// It skips (not fails) when Docker is unavailable so the suite degrades +// gracefully in environments without a container runtime. +func setupLadderStore(ctx context.Context, t *testing.T) *PostgresStore { + t.Helper() + container, err := testhelpers.SetupPostgresContainer(ctx, t) + if err != nil { + t.Skipf("Skipping integration test: cannot start PostgreSQL container: %v", err) + } + t.Cleanup(func() { container.Cleanup(context.Background()) }) + + if err := migrations.RunMigrations(ctx, container.DB.Pool(), getTestMigrationsPath(), "", ""); err != nil { + t.Skipf("Skipping integration test: migrations failed: %v", err) + } + return NewPostgresStore(container.DB) +} + +// seedLadderConfig inserts a cloud_account and a ladder_config, returning the +// ladder_config id for use as a run/tranche config_id FK. +func seedLadderConfig(ctx context.Context, t *testing.T, store *PostgresStore) 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', '123456789012', false, now(), now()) + `, acctID) + require.NoError(t, err, "seed cloud_accounts") + + 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") + require.NotEmpty(t, saved.ID) + return saved.ID +} + +func TestPostgresStore_SaveLadderRun_NullMonetaryPreserved(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + + now := time.Now().UTC().Truncate(time.Microsecond) + + t.Run("NULL monetary columns round-trip as nil", func(t *testing.T) { + run := &LadderRunDB{ + StartedAt: now, + Status: ladder.RunStatusPlanned, + // All monetary snapshot fields intentionally nil. + } + saved, err := store.SaveLadderRun(ctx, run) + require.NoError(t, err) + require.NotEmpty(t, saved.ID, "SaveLadderRun must generate an ID") + + got, err := store.GetLadderRun(ctx, saved.ID) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, ladder.RunStatusPlanned, got.Status) + // The core assertion: NULL != 0. Absent monetary values must stay nil. + assert.Nil(t, got.BaselineUSDHr, "baseline_usd_hr NULL must scan as nil, not 0") + assert.Nil(t, got.TargetUSDHr, "target_usd_hr NULL must scan as nil, not 0") + assert.Nil(t, got.ExistingUSDHr, "existing_usd_hr NULL must scan as nil, not 0") + assert.Nil(t, got.GapUSDHr, "gap_usd_hr NULL must scan as nil, not 0") + assert.Nil(t, got.ConfigID, "config_id NULL must scan as nil") + // NOT NULL accumulator totals default to 0. + assert.Equal(t, 0.0, got.TotalHourlyCommit) + // Plan defaults to '{}' JSONB. + assert.JSONEq(t, `{}`, string(got.Plan)) + }) + + t.Run("populated monetary columns round-trip by value", func(t *testing.T) { + baseline := 12.5 + target := 10.0 + existing := 3.0 + gap := 7.0 + run := &LadderRunDB{ + StartedAt: now, + Status: ladder.RunStatusPlanned, + BaselineUSDHr: &baseline, + TargetUSDHr: &target, + ExistingUSDHr: &existing, + GapUSDHr: &gap, + TotalHourlyCommit: 7.0, + Plan: []byte(`{"actions":[]}`), + } + saved, err := store.SaveLadderRun(ctx, run) + require.NoError(t, err) + + got, err := store.GetLadderRun(ctx, saved.ID) + require.NoError(t, err) + require.NotNil(t, got) + require.NotNil(t, got.BaselineUSDHr) + assert.InDelta(t, baseline, *got.BaselineUSDHr, 1e-6) + require.NotNil(t, got.GapUSDHr) + assert.InDelta(t, gap, *got.GapUSDHr, 1e-6) + assert.InDelta(t, 7.0, got.TotalHourlyCommit, 1e-6) + }) + + t.Run("GetLadderRun returns nil for a missing id", func(t *testing.T) { + got, err := store.GetLadderRun(ctx, uuid.New().String()) + require.NoError(t, err) + assert.Nil(t, got, "GetLadderRun must return (nil, nil) for a missing row") + }) +} + +func TestPostgresStore_LatestLadderRunStartedAt(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + configID := seedLadderConfig(ctx, t, store) + + t.Run("nil when no runs exist for the config", func(t *testing.T) { + latest, err := store.LatestLadderRunStartedAt(ctx, configID) + require.NoError(t, err) + assert.Nil(t, latest, "no runs -> nil (never a zero time)") + }) + + t.Run("returns MAX(started_at) across multiple runs", func(t *testing.T) { + older := time.Now().UTC().Add(-48 * time.Hour).Truncate(time.Microsecond) + newer := time.Now().UTC().Add(-2 * time.Hour).Truncate(time.Microsecond) + + for _, ts := range []time.Time{older, newer} { + ts := ts + cfgID := configID + _, err := store.SaveLadderRun(ctx, &LadderRunDB{ + ConfigID: &cfgID, + StartedAt: ts, + Status: ladder.RunStatusPlanned, + }) + require.NoError(t, err) + } + + latest, err := store.LatestLadderRunStartedAt(ctx, configID) + require.NoError(t, err) + require.NotNil(t, latest) + assert.WithinDuration(t, newer, *latest, time.Second, "must return the most recent started_at") + }) +} + +func TestPostgresStore_SaveLadderTranches_Batch(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + configID := seedLadderConfig(ctx, t, store) + + // A tranche run_id FK requires an existing run row. + cfgID := configID + run, err := store.SaveLadderRun(ctx, &LadderRunDB{ + ConfigID: &cfgID, + StartedAt: time.Now().UTC(), + Status: ladder.RunStatusPlanned, + }) + require.NoError(t, err) + + runID := run.ID + fire := time.Now().UTC().Add(7 * 24 * time.Hour).Truncate(time.Microsecond) + tranches := []LadderTrancheDB{ + { + ID: uuid.New().String(), + ConfigID: &cfgID, + RunID: &runID, + LayerType: ladder.LayerConvertibleRI, + Term: ladder.Term1Year, + PaymentOption: ladder.PaymentNoUpfront, + Status: ladder.TrancheStatusScheduled, + AmountUSDHr: 1.5, + ScheduledDate: fire, + }, + { + ID: uuid.New().String(), + ConfigID: &cfgID, + RunID: &runID, + LayerType: ladder.LayerEC2InstanceSP, + Term: ladder.Term1Year, + PaymentOption: ladder.PaymentNoUpfront, + Status: ladder.TrancheStatusScheduled, + AmountUSDHr: 2.5, + ScheduledDate: fire, + }, + } + + require.NoError(t, store.SaveLadderTranches(ctx, tranches)) + + // Verify both rows landed and reference the run. + var count int + require.NoError(t, store.db.QueryRow(ctx, + `SELECT count(*) FROM ladder_tranches WHERE run_id = $1`, runID).Scan(&count)) + assert.Equal(t, 2, count, "both tranches must be inserted in the batch") + + t.Run("empty slice is a no-op", func(t *testing.T) { + require.NoError(t, store.SaveLadderTranches(ctx, nil)) + }) + + t.Run("empty tranche ID is rejected", func(t *testing.T) { + err := store.SaveLadderTranches(ctx, []LadderTrancheDB{ + { + ID: "", // invalid + RunID: &runID, + LayerType: ladder.LayerConvertibleRI, + Term: ladder.Term1Year, + PaymentOption: ladder.PaymentNoUpfront, + Status: ladder.TrancheStatusScheduled, + AmountUSDHr: 1.0, + ScheduledDate: fire, + }, + }) + require.Error(t, err, "an empty tranche ID must be rejected") + }) +} + +func TestPostgresStore_TransitionLadderRunStatus_CAS(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + + run, err := store.SaveLadderRun(ctx, &LadderRunDB{ + StartedAt: time.Now().UTC(), + Status: ladder.RunStatusPlanned, + }) + require.NoError(t, err) + + t.Run("CAS win: current status is in the from-set", func(t *testing.T) { + updated, err := store.TransitionLadderRunStatus(ctx, run.ID, + []ladder.RunStatus{ladder.RunStatusPlanned}, ladder.RunStatusAwaitingApproval) + require.NoError(t, err) + require.NotNil(t, updated, "transition from the correct status must update the row") + assert.Equal(t, ladder.RunStatusAwaitingApproval, updated.Status) + }) + + t.Run("CAS lose: current status not in the from-set", func(t *testing.T) { + // The row is now awaiting_approval; a transition expecting planned must + // affect zero rows and signal the miss with (nil, nil). + updated, err := store.TransitionLadderRunStatus(ctx, run.ID, + []ladder.RunStatus{ladder.RunStatusPlanned}, ladder.RunStatusApproved) + require.NoError(t, err) + assert.Nil(t, updated, "CAS miss must return (nil, nil), not an error") + + // The stored status must be unchanged from the winning transition. + got, err := store.GetLadderRun(ctx, run.ID) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, ladder.RunStatusAwaitingApproval, got.Status, + "a lost CAS must not mutate the row") + }) +} + +// TestPostgresStore_SaveLadderRunWithTranches_Atomicity proves the run+tranche +// persist is transactional (B2): on success both land, and on a tranche-insert +// failure the run row is rolled back too (never a status=planned run with no +// tranches, which the cadence gate would then suppress a retry of). +func TestPostgresStore_SaveLadderRunWithTranches_Atomicity(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + configID := seedLadderConfig(ctx, t, store) + cfgID := configID + fire := time.Now().UTC().Add(7 * 24 * time.Hour).Truncate(time.Microsecond) + + newTranche := func(runID *string, id string) LadderTrancheDB { + return LadderTrancheDB{ + ID: id, + ConfigID: &cfgID, + RunID: runID, + LayerType: ladder.LayerConvertibleRI, + Term: ladder.Term1Year, + PaymentOption: ladder.PaymentNoUpfront, + Status: ladder.TrancheStatusScheduled, + AmountUSDHr: 1.5, + ScheduledDate: fire, + } + } + + t.Run("success: run and tranches both persist", func(t *testing.T) { + runID := uuid.New().String() + run := &LadderRunDB{ID: runID, ConfigID: &cfgID, StartedAt: time.Now().UTC(), Status: ladder.RunStatusPlanned} + tranches := []LadderTrancheDB{newTranche(&runID, uuid.New().String()), newTranche(&runID, uuid.New().String())} + + saved, err := store.SaveLadderRunWithTranches(ctx, run, tranches) + require.NoError(t, err) + require.NotNil(t, saved) + + got, err := store.GetLadderRun(ctx, runID) + require.NoError(t, err) + require.NotNil(t, got, "the run row must be persisted") + + var count int + require.NoError(t, store.db.QueryRow(ctx, + `SELECT count(*) FROM ladder_tranches WHERE run_id = $1`, runID).Scan(&count)) + assert.Equal(t, 2, count, "both tranches must be persisted in the same tx") + }) + + t.Run("rollback: a duplicate tranche ID rolls back the run row too", func(t *testing.T) { + runID := uuid.New().String() + run := &LadderRunDB{ID: runID, ConfigID: &cfgID, StartedAt: time.Now().UTC(), Status: ladder.RunStatusPlanned} + // Two tranches with the SAME id: the second insert violates the PK, + // failing the transaction after the run row was inserted. + dupID := uuid.New().String() + tranches := []LadderTrancheDB{newTranche(&runID, dupID), newTranche(&runID, dupID)} + + saved, err := store.SaveLadderRunWithTranches(ctx, run, tranches) + require.Error(t, err, "a duplicate tranche ID must fail the transaction") + assert.Nil(t, saved) + + // The run row must NOT exist: the whole tx rolled back. + got, err := store.GetLadderRun(ctx, runID) + require.NoError(t, err) + assert.Nil(t, got, "the run row must be rolled back when a tranche insert fails") + + var count int + require.NoError(t, store.db.QueryRow(ctx, + `SELECT count(*) FROM ladder_tranches WHERE run_id = $1`, runID).Scan(&count)) + assert.Equal(t, 0, count, "no tranche may survive a rolled-back transaction") + }) +} diff --git a/internal/config/types.go b/internal/config/types.go index 481e9586a..90f252877 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -1014,3 +1014,69 @@ type LadderConfigDB struct { TargetCoverage float64 `json:"target_coverage"` Enabled bool `json:"enabled"` } + +// LadderRunDB mirrors the ladder_runs table (migration 000080). +// Monetary snapshot columns are *float64 (nullable, NEVER 0-coerced: +// NULL means "not computed", not "$0"). Field order minimizes GC +// pointer-scan range: explicit pointer fields come before scalars. +type LadderRunDB struct { + // Nullable monetary snapshot: nil means "not computed", never $0. + BaselineUSDHr *float64 `json:"baseline_usd_hr,omitempty"` + TargetUSDHr *float64 `json:"target_usd_hr,omitempty"` + ExistingUSDHr *float64 `json:"existing_usd_hr,omitempty"` + GapUSDHr *float64 `json:"gap_usd_hr,omitempty"` + + // Nullable FK and optional text fields (all pointer types). + ConfigID *string `json:"config_id,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + ApprovalTokenHash *string `json:"approval_token_hash,omitempty"` + ApprovalTokenExpiresAt *time.Time `json:"approval_token_expires_at,omitempty"` + ApprovedBy *string `json:"approved_by,omitempty"` + CancelledBy *string `json:"cancelled_by,omitempty"` + FireAt *time.Time `json:"fire_at,omitempty"` + // Mode and Cadence are nullable in the DB (populated from LadderConfigDB + // at run creation time; nil only for legacy / partially-failed rows). + Mode *string `json:"mode,omitempty"` + Cadence *string `json:"cadence,omitempty"` + + // Plan JSON blob (JSONB, NOT NULL DEFAULT '{}'). + Plan json.RawMessage `json:"plan"` + + // Non-nullable accumulator totals (initialised to 0, not measurements; + // zero is a meaningful value for these counters unlike the monetary snapshot). + TotalHourlyCommit float64 `json:"total_hourly_commit"` + TotalUpfrontCost float64 `json:"total_upfront_cost"` + EstimatedSavings float64 `json:"estimated_savings"` + + // Required string / enum fields. + ID string `json:"id"` + Status ladder.RunStatus `json:"status"` + + // Required timestamps. + StartedAt time.Time `json:"started_at"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// LadderTrancheDB mirrors the ladder_tranches table (migration 000081). +// One row per ramp step per allocation; persisted as status=scheduled +// audit rows only in PR-2 (no firing sweep wired yet). Field order +// minimizes GC pointer-scan range. +type LadderTrancheDB struct { + // Nullable FK pointers. + ConfigID *string `json:"config_id,omitempty"` + RunID *string `json:"run_id,omitempty"` + ExecutionID *string `json:"execution_id,omitempty"` // references purchase_executions.execution_id + + // Required fields. + ID string `json:"id"` + LayerType ladder.LayerType `json:"layer_type"` + Term ladder.Term `json:"term"` + PaymentOption ladder.PaymentOption `json:"payment_option"` + Status ladder.TrancheStatus `json:"status"` + + // Monetary and timing. + AmountUSDHr float64 `json:"amount_usd_hr"` + ScheduledDate time.Time `json:"scheduled_date"` + CreatedAt time.Time `json:"created_at"` +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 9339192ae..f1bcbfdd3 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -7,6 +7,7 @@ import ( "github.com/LeanerCloud/CUDly/internal/auth" "github.com/LeanerCloud/CUDly/internal/config" + "github.com/LeanerCloud/CUDly/pkg/ladder" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/mock" ) @@ -1368,6 +1369,101 @@ func (m *MockConfigStore) UpsertLadderConfig(ctx context.Context, cfg *config.La return v, args.Error(1) } +// SaveLadderRun mocks the SaveLadderRun operation. +// Returns (nil, nil) when no expectation is registered. +func (m *MockConfigStore) SaveLadderRun(ctx context.Context, run *config.LadderRunDB) (*config.LadderRunDB, error) { + if !isExpected(&m.Mock, "SaveLadderRun") { + return nil, nil + } + args := m.Called(ctx, run) + 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) +} + +// SaveLadderRunWithTranches mocks the SaveLadderRunWithTranches operation. +// Returns the run unchanged (transaction succeeds) when no expectation is +// registered, so tests that only care about other calls stay green. +func (m *MockConfigStore) SaveLadderRunWithTranches(ctx context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { + if !isExpected(&m.Mock, "SaveLadderRunWithTranches") { + 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) +} + +// GetLadderRun mocks the GetLadderRun operation. +// Returns (nil, nil) when no expectation is registered. +func (m *MockConfigStore) GetLadderRun(ctx context.Context, id string) (*config.LadderRunDB, error) { + if !isExpected(&m.Mock, "GetLadderRun") { + return nil, nil + } + args := m.Called(ctx, id) + 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) +} + +// SaveLadderTranches mocks the SaveLadderTranches operation. +// Returns nil (no-op) when no expectation is registered. +func (m *MockConfigStore) SaveLadderTranches(ctx context.Context, tranches []config.LadderTrancheDB) error { + if !isExpected(&m.Mock, "SaveLadderTranches") { + return nil + } + return m.Called(ctx, tranches).Error(0) +} + +// LatestLadderRunStartedAt mocks the LatestLadderRunStartedAt operation. +// Returns (nil, nil) when no expectation is registered. +func (m *MockConfigStore) LatestLadderRunStartedAt(ctx context.Context, configID string) (*time.Time, error) { + if !isExpected(&m.Mock, "LatestLadderRunStartedAt") { + return nil, nil + } + args := m.Called(ctx, configID) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).(*time.Time) + if !ok { + panic(fmt.Sprintf("mock: expected *time.Time, 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) { + if !isExpected(&m.Mock, "TransitionLadderRunStatus") { + return nil, nil + } + args := m.Called(ctx, id, fromStatuses, toStatus) + 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) +} + // isExpected reports whether mock has any .On() expectation for method. func isExpected(mock *mock.Mock, method string) bool { for _, call := range mock.ExpectedCalls { diff --git a/internal/server/app.go b/internal/server/app.go index 17fa87fbe..e8ae2ccb5 100644 --- a/internal/server/app.go +++ b/internal/server/app.go @@ -29,7 +29,9 @@ import ( "github.com/LeanerCloud/CUDly/internal/scheduler" "github.com/LeanerCloud/CUDly/internal/secrets" "github.com/LeanerCloud/CUDly/internal/server/scheduledauth" + pkgladder "github.com/LeanerCloud/CUDly/pkg/ladder" "github.com/LeanerCloud/CUDly/pkg/logging" + awsladder "github.com/LeanerCloud/CUDly/providers/aws/ladder" "github.com/aws/aws-sdk-go-v2/aws" awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/sts" @@ -54,6 +56,20 @@ type Application struct { DB *database.Connection // PostgreSQL database connection TaskLocker TaskLocker // Advisory lock for scheduled tasks (defaults to DB) + // LadderCapabilityFactory constructs a LadderCapability for the given region + // and accountID. It is called once per ladder_run task invocation. + // Defaults to awsladder.NewFromAWSConfig in production; tests replace it + // with a fake factory that returns a hermetic LadderCapability. + LadderCapabilityFactory func(ctx context.Context, region, accountID string) (pkgladder.LadderCapability, error) + + // LadderAccountResolver resolves the Lambda's own AWS account ID and region + // for the single-account ladder gate (Q1). It MUST fail loud when the + // account cannot be determined: the account ID gates which configs run, so a + // transient STS failure must abort the whole ladder_run rather than silently + // skip every config as multi-account. Nil in production -> the default + // STS-backed resolver (defaultLadderAccountResolver); tests inject a stub. + LadderAccountResolver func(ctx context.Context) (accountID, region string, err error) + // Static file serving directory (from STATIC_DIR env var) staticDir string @@ -564,7 +580,12 @@ func NewApplication(ctx context.Context, version string) (*Application, error) { STSClient: stsClient, } - return NewApplicationFromDeps(ctx, cfg, deps) + app, err := NewApplicationFromDeps(ctx, cfg, deps) + if err != nil { + return nil, err + } + app.LadderCapabilityFactory = awsladder.NewFromAWSConfig + return app, nil } // ensureDB ensures the database connection is established (lazy initialization). diff --git a/internal/server/handler.go b/internal/server/handler.go index 46549cf09..d0f5631b5 100644 --- a/internal/server/handler.go +++ b/internal/server/handler.go @@ -53,6 +53,12 @@ const ( // record is eventually consistent without requiring the user to retry (which // would be rejected by Azure). See issue #290 Finding #6. TaskFinalizeRevocations ScheduledTaskType = "finalize_revocations" + // TaskLadderRun runs the commitment-laddering planner for every enabled + // ladder_config. It computes a plan (Allocate + BuildTranches), persists the + // ladder_run and ladder_tranches rows with status=planned, and stops. No + // purchases, no emails, no reshapes, no approval tokens are issued in this + // plan-only phase (PR-2). Execution arrives in a later PR. + TaskLadderRun ScheduledTaskType = "ladder_run" ) // scheduledEventActions maps a raw scheduled-event action string to its @@ -70,6 +76,7 @@ var scheduledEventActions = map[string]ScheduledTaskType{ "reap_stuck_purchases": TaskReapStuckPurchases, "fire_scheduled_purchases": TaskFireScheduledPurchases, "finalize_revocations": TaskFinalizeRevocations, + "ladder_run": TaskLadderRun, } // HandleScheduledTask processes a scheduled task by type. @@ -114,6 +121,7 @@ func (app *Application) dispatchTask(ctx context.Context, taskType ScheduledTask TaskReapStuckPurchases: func(c context.Context) (any, error) { return app.handleReapStuckPurchases(c) }, TaskFireScheduledPurchases: func(c context.Context) (any, error) { return app.handleFireScheduledPurchases(c) }, TaskFinalizeRevocations: func(c context.Context) (any, error) { return app.handleFinalizeRevocations(c) }, + TaskLadderRun: func(c context.Context) (any, error) { return app.handleLadderRun(c) }, } handler, ok := handlers[taskType] if !ok { diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go new file mode 100644 index 000000000..a7e74fcf6 --- /dev/null +++ b/internal/server/handler_ladder.go @@ -0,0 +1,706 @@ +package server + +import ( + "context" + "encoding/json" + "fmt" + "log" + "math/big" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/sts" + "github.com/google/uuid" + + "github.com/LeanerCloud/CUDly/internal/config" + pkgcommon "github.com/LeanerCloud/CUDly/pkg/common" + pkgladder "github.com/LeanerCloud/CUDly/pkg/ladder" +) + +// Cadence thresholds for the per-config self-gate (Q6). +// Each constant is a minimum elapsed time before the same config may run again. +// The 4-hour slack on each bound prevents skipped firings from drifting into +// the next cadence window: a daily at 23:55 still qualifies at 19:55 the next +// day even if the cron fires a few minutes early. +const ( + cadenceDailyMin = 20 * time.Hour // daily: 20h (24h - 4h slack) + cadenceWeeklyMin = 6*24*time.Hour + 20*time.Hour // weekly: 6d 20h (7d - 4h slack) +) + +// dataSourceAWSCostExplorer is the provenance tag stamped on every planned +// action in the PR-2 plan-only phase. The baseline is derived from AWS Cost +// Explorer once the CE adapter lands (PR-4); until then the tag documents the +// intended source. Named constant instead of a bare string literal so the +// provenance value has a single definition. +const dataSourceAWSCostExplorer = "aws-ce" + +// ladderConfigOutcome is the result of processing a single ladder_config entry +// in the handleLadderRun loop. Using a typed constant avoids bare strings on +// the outcome path. +type ladderConfigOutcome int + +const ( + outcomeSkippedDisabled ladderConfigOutcome = iota + outcomeSkippedMultiAccount // iota = 1 + outcomeSkippedCadence // iota = 2 + outcomeErrored // iota = 3 + outcomePlanned // iota = 4 +) + +// LadderRunResult is the aggregate outcome of one ladder_run task invocation. +// Each counter increments once per ladder_config entry processed. +type LadderRunResult struct { + Planned int `json:"planned"` + SkippedCadence int `json:"skipped_cadence"` + SkippedDisabled int `json:"skipped_disabled"` + SkippedMultiAccount int `json:"skipped_multi_account"` + Errored int `json:"errored"` +} + +// record increments the counter that corresponds to the given outcome. +func (r *LadderRunResult) record(o ladderConfigOutcome) { + switch o { + case outcomePlanned: + r.Planned++ + case outcomeSkippedDisabled: + r.SkippedDisabled++ + case outcomeSkippedMultiAccount: + r.SkippedMultiAccount++ + case outcomeSkippedCadence: + r.SkippedCadence++ + case outcomeErrored: + r.Errored++ + } +} + +// handleLadderRun is the top-level orchestrator for the ladder_run scheduled task. +// It iterates over all ladder_configs and runs the planning engine for each +// eligible config. The result struct carries per-outcome counts so the caller +// can surface them in the Lambda log and in future status pages. +// +// plan-only: no PurchaseLayer, ReshapeBuffer, email, or approval tokens. +func (app *Application) handleLadderRun(ctx context.Context) (*LadderRunResult, error) { + // Global kill-switch check. + globalCfg, err := app.Config.GetGlobalConfig(ctx) + if err != nil { + return nil, fmt.Errorf("ladder_run: failed to load global config: %w", err) + } + if !globalCfg.LadderingEnabled { + log.Println("ladder_run: laddering_enabled=false in global config, skipping all configs") + return &LadderRunResult{}, nil + } + + // Parse global-default term and payment option. Fail loud: money-path + // decisions must never proceed with an invalid/unknown option. + term, err := ladderTermFromYears(globalCfg.DefaultTerm) + if err != nil { + return nil, fmt.Errorf("ladder_run: invalid default_term in global config: %w", err) + } + paymentOpt, err := pkgladder.ParsePaymentOption(globalCfg.DefaultPayment) + if err != nil { + return nil, fmt.Errorf("ladder_run: invalid default_payment in global config: %w", err) + } + + // Load all ladder_config rows. GetLadderConfigs returns ALL rows; we + // filter Enabled=false here so the counter is visible (SkippedDisabled). + allConfigs, err := app.Config.GetLadderConfigs(ctx) + if err != nil { + return nil, fmt.Errorf("ladder_run: failed to load ladder configs: %w", err) + } + if len(allConfigs) == 0 { + log.Println("ladder_run: no ladder_config rows found, nothing to do") + return &LadderRunResult{}, nil + } + + // Q1: single-account only. Resolve the Lambda's own AWS account ID STRICTLY: + // this ID scopes which configs run, so an unresolved account must abort the + // whole task (fail loud) rather than fall through and skip every config as + // multi-account, which would report a false success on a money path. + ownAccountID, region, err := app.resolveLadderAccount(ctx) + if err != nil { + return nil, fmt.Errorf("ladder_run: could not resolve caller AWS account for the single-account gate: %w", err) + } + + now := time.Now().UTC() + result := app.runLadderConfigs(ctx, allConfigs, ownAccountID, region, term, paymentOpt, now) + + log.Printf("ladder_run done: planned=%d skipped_cadence=%d skipped_disabled=%d skipped_multi_account=%d errored=%d", + result.Planned, result.SkippedCadence, result.SkippedDisabled, result.SkippedMultiAccount, result.Errored) + return result, nil +} + +// resolveLadderAccount resolves the caller AWS account ID and region for the +// single-account gate. It delegates to the injected LadderAccountResolver (used +// by tests) or the default STS-backed resolver. Any error is returned to the +// caller so handleLadderRun can fail loud. +func (app *Application) resolveLadderAccount(ctx context.Context) (accountID, region string, err error) { + if app.LadderAccountResolver != nil { + return app.LadderAccountResolver(ctx) + } + return defaultLadderAccountResolver(ctx) +} + +// defaultLadderAccountResolver resolves the Lambda's own AWS account ID via STS +// STRICTLY: unlike resolveAccountID (which returns the "unknown" sentinel for +// audit-only use in the RI-exchange path), this returns an error when the +// account cannot be determined. The ladder path uses the account ID to SCOPE +// which configs run, so a fabricated/"unknown" value is unsafe: it would skip +// every config as multi-account and report a false success. +func defaultLadderAccountResolver(ctx context.Context) (string, string, error) { + awsCfg, err := awsconfig.LoadDefaultConfig(ctx) + if err != nil { + return "", "", fmt.Errorf("load AWS config: %w", err) + } + return resolveLadderIdentity(ctx, awsCfg) +} + +// resolveLadderIdentity validates the region and resolves the caller account ID +// from an already-loaded aws.Config. Split from defaultLadderAccountResolver so +// the region short-circuit is unit-testable without loading real credentials or +// calling STS. +// +// The region is validated FIRST: an empty region would make NewFromAWSConfig +// reject EVERY config later, turning the whole task into a silent all-Errored +// no-op that still returns success. Failing loud here aborts handleLadderRun +// before any config is processed (same silent-degradation class as the missing +// STS account ID below). +func resolveLadderIdentity(ctx context.Context, awsCfg aws.Config) (string, string, error) { + if awsCfg.Region == "" { + return "", "", fmt.Errorf("AWS region is empty; set AWS_REGION / AWS_DEFAULT_REGION or a region in the shared config") + } + stsClient := sts.NewFromConfig(awsCfg) + identity, err := stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{}) + if err != nil { + return "", "", fmt.Errorf("resolve caller account via STS: %w", err) + } + if identity.Account == nil || *identity.Account == "" { + return "", "", fmt.Errorf("STS GetCallerIdentity returned no account ID") + } + return *identity.Account, awsCfg.Region, nil +} + +// runLadderConfigs processes every ladder_config entry, isolating each one: +// an error on one config increments the Errored counter and continues to the +// next, so a single broken config never aborts the whole run. Extracted from +// handleLadderRun so the multi-config isolation behaviour is unit-testable +// without the AWS SDK account-resolution path in handleLadderRun. +func (app *Application) runLadderConfigs( + ctx context.Context, + configs []config.LadderConfigDB, + ownAccountID, region string, + term pkgladder.Term, + paymentOpt pkgladder.PaymentOption, + now time.Time, +) *LadderRunResult { + result := &LadderRunResult{} + for i := range configs { + result.record(app.processOneLadderConfig(ctx, &configs[i], ownAccountID, region, term, paymentOpt, now)) + } + return result +} + +// processOneLadderConfig applies all eligibility gates for a single +// ladder_config entry and runs the planning engine when eligible. It is +// extracted from handleLadderRun to keep both functions within the project's +// cyclomatic-complexity limit (10). +func (app *Application) processOneLadderConfig( + ctx context.Context, + dbCfg *config.LadderConfigDB, + ownAccountID, region string, + term pkgladder.Term, + paymentOpt pkgladder.PaymentOption, + now time.Time, +) ladderConfigOutcome { + if !dbCfg.Enabled { + log.Printf("ladder_run: config %s: enabled=false, skipping", dbCfg.ID) + return outcomeSkippedDisabled + } + + // Resolve the cloud account to get the 12-digit AWS account number (Q2). + cloudAcct, err := app.Config.GetCloudAccount(ctx, dbCfg.CloudAccountID) + if err != nil { + log.Printf("ladder_run: config %s: failed to get cloud account %s: %v", dbCfg.ID, dbCfg.CloudAccountID, err) + return outcomeErrored + } + if cloudAcct == nil { + log.Printf("ladder_run: config %s: cloud account %s not found in DB", dbCfg.ID, dbCfg.CloudAccountID) + return outcomeErrored + } + + // Q1: skip configs that belong to a different AWS account. + if cloudAcct.ExternalID != ownAccountID { + log.Printf("ladder_run: config %s: cloud account external_id=%q != lambda account=%q: skipped (multi_account_unsupported)", dbCfg.ID, cloudAcct.ExternalID, ownAccountID) + return outcomeSkippedMultiAccount + } + + // Cadence self-gate: skip if a run already started within the window. Fail + // CLOSED on a lookup error: if we cannot tell whether a recent run exists, + // running anyway risks a duplicate money action, so count the config Errored + // (visible in LadderRunResult.Errored) instead of proceeding silently. + within, reason, err := ladderWithinCadenceWindow(ctx, app.Config, dbCfg, now) + if err != nil { + log.Printf("ladder_run: config %s: cadence lookup failed, skipping to avoid a possible double-run: %v", dbCfg.ID, err) + return outcomeErrored + } + if within { + log.Printf("ladder_run: config %s: %s", dbCfg.ID, reason) + return outcomeSkippedCadence + } + + // Build the LadderCapability for this account. + if app.LadderCapabilityFactory == nil { + log.Printf("ladder_run: config %s: LadderCapabilityFactory is nil (not wired), erroring", dbCfg.ID) + return outcomeErrored + } + cap, err := app.LadderCapabilityFactory(ctx, region, cloudAcct.ExternalID) + if err != nil { + log.Printf("ladder_run: config %s: failed to build ladder capability: %v", dbCfg.ID, err) + return outcomeErrored + } + + // Run the plan engine and persist the result. + if err := app.executeLadderRun(ctx, dbCfg, cap, cloudAcct.ExternalID, term, paymentOpt, now); err != nil { + log.Printf("ladder_run: config %s: planning failed: %v", dbCfg.ID, err) + return outcomeErrored + } + return outcomePlanned +} + +// executeLadderRun runs the planning engine for a single ladder_config and +// persists the result as a ladder_runs row (+ ladder_tranches audit rows). +// +// It is split from handleLadderRun so tests can inject a hermetic +// LadderCapability (fake) and a mock store without touching AWS or PostgreSQL. +func (app *Application) executeLadderRun( + ctx context.Context, + dbCfg *config.LadderConfigDB, + cap pkgladder.LadderCapability, + accountID string, + term pkgladder.Term, + paymentOpt pkgladder.PaymentOption, + now time.Time, +) error { + // Convert the DB config row to the engine's typed LadderConfig. + engineCfg, err := ladderConfigToEngine(dbCfg, accountID) + if err != nil { + return fmt.Errorf("config conversion: %w", err) + } + scope := engineCfg.Scope + + // Collect read-side data. All three calls are independent; errors are + // fail-loud (not silently defaulted to empty). + baseline, err := cap.GetUsageBaseline(ctx, scope, engineCfg.LookbackDays, engineCfg.BaselinePercentile) + if err != nil { + return fmt.Errorf("GetUsageBaseline: %w", err) + } + + layerStates, err := cap.GetLayerStates(ctx, scope) + if err != nil { + return fmt.Errorf("GetLayerStates: %w", err) + } + + supportedLayers := cap.SupportedLayers() + + // Run Allocate. + allocResult, err := pkgladder.Allocate(&pkgladder.AllocationInput{ + Now: now, + Baseline: baseline, + LayerStates: layerStates, + Layers: supportedLayers, + Config: engineCfg, + DataSources: []string{dataSourceAWSCostExplorer}, + }) + if err != nil { + return fmt.Errorf("Allocate: %w", err) + } + + // Run BuildTranches to produce the ramp schedule. + runID := uuid.New().String() + trancheResult, err := pkgladder.BuildTranches(&pkgladder.TrancheInput{ + Config: &engineCfg, + RunID: runID, + Term: term, + PaymentOption: paymentOpt, + NewID: func() string { return uuid.New().String() }, + Now: now, + Allocations: allocResult.Allocations, + }) + if err != nil { + return fmt.Errorf("BuildTranches: %w", err) + } + + // 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. + return app.persistLadderRun(ctx, dbCfg, runID, now, plan, baseline, allocResult, trancheResult) +} + +// persistLadderRun validates the assembled plan and writes the ladder_runs row +// followed by its ladder_tranches audit rows. Split from executeLadderRun so +// each function stays within the project's cyclomatic-complexity limit (10). +// A failure at any step returns an error before the next write, so a config +// is never left with a half-persisted run. +func (app *Application) persistLadderRun( + ctx context.Context, + dbCfg *config.LadderConfigDB, + runID string, + now time.Time, + plan *pkgladder.LadderPlan, + baseline pkgladder.UsageBaseline, + allocResult *pkgladder.AllocateResult, + trancheResult *pkgladder.TrancheResult, +) error { + if err := plan.Validate(); err != nil { + return fmt.Errorf("LadderPlan.Validate: %w", err) + } + + planJSON, err := marshalLadderPlan(plan) + if err != nil { + return fmt.Errorf("marshal plan: %w", err) + } + + // Map *big.Rat monetary snapshot to *float64 (nil stays nil). + // These are stored in the run row for dashboards; the authoritative + // detail is in plan_json. + cfgID := dbCfg.ID + modeStr := dbCfg.Mode + cadenceStr := dbCfg.Cadence + runRow := &config.LadderRunDB{ + ID: runID, + ConfigID: &cfgID, + StartedAt: now, + Status: pkgladder.RunStatusPlanned, + Mode: &modeStr, + Cadence: &cadenceStr, + Plan: planJSON, + BaselineUSDHr: baseline.LowWaterUSDPerHour, + TargetUSDHr: ratToFloat64Ptr(plan.TargetUSDPerHour), + ExistingUSDHr: ratToFloat64Ptr(plan.ExistingUSDPerHour), + GapUSDHr: ratToFloat64Ptr(plan.GapUSDPerHour), + TotalHourlyCommit: allocTotalHourlyCommit(allocResult.Allocations), + } + + // Build the tranche audit rows (status=scheduled; no firing in PR-2). The + // run ID is known up front (runRow.ID), so tranches can reference it before + // the insert. + trancheRows, err := buildTrancheDBRows(trancheResult.Tranches, runRow.ID, &cfgID) + if err != nil { + 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) + if err != nil { + return fmt.Errorf("SaveLadderRunWithTranches: %w", err) + } + + log.Printf("ladder_run: config %s: persisted run %s (status=%s, allocations=%d, tranches=%d, holds=%d)", + dbCfg.ID, savedRun.ID, savedRun.Status, + len(allocResult.Allocations), len(trancheResult.Tranches), len(allocResult.Holds)) + return nil +} + +// 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: +// - (true, reason, nil) when a recent run is within the window (skip), +// - (false, "", nil) when the config is eligible to run, +// - (false, "", err) when the lookup itself failed. +// +// The lookup error is propagated (NOT swallowed) so the caller can fail CLOSED: +// if we cannot determine whether a recent run exists, proceeding risks a +// duplicate money action, so the caller counts the config Errored rather than +// running it. +func ladderWithinCadenceWindow(ctx context.Context, store config.StoreInterface, dbCfg *config.LadderConfigDB, now time.Time) (bool, string, error) { + latest, err := store.LatestLadderRunStartedAt(ctx, dbCfg.ID) + if err != nil { + return false, "", fmt.Errorf("LatestLadderRunStartedAt: %w", err) + } + if latest == nil { + return false, "", nil // no previous run, always eligible + } + elapsed := now.Sub(*latest) + switch dbCfg.Cadence { + case string(pkgladder.CadenceDaily): + if elapsed < cadenceDailyMin { + return true, fmt.Sprintf("cadence=daily: last run %v ago (threshold %v), skipping", elapsed.Round(time.Minute), cadenceDailyMin), nil + } + case string(pkgladder.CadenceWeekly): + if elapsed < cadenceWeeklyMin { + return true, fmt.Sprintf("cadence=weekly: last run %v ago (threshold %v), skipping", elapsed.Round(time.Minute), cadenceWeeklyMin), nil + } + default: + // Unknown cadence: log but do not block the run. The LadderConfig + // validator will surface this as an error during Allocate. + log.Printf("ladder_run: config %s: unknown cadence=%q, proceeding without cadence gate", dbCfg.ID, dbCfg.Cadence) + } + return false, "", nil +} + +// ladderConfigToEngine converts a LadderConfigDB row to a pkg/ladder LadderConfig. +// It parses typed enums at the boundary and fails loud on any unknown value. +func ladderConfigToEngine(dbCfg *config.LadderConfigDB, accountID string) (pkgladder.LadderConfig, error) { + mode, err := pkgladder.ParseLadderMode(dbCfg.Mode) + if err != nil { + return pkgladder.LadderConfig{}, fmt.Errorf("mode: %w", err) + } + cadence, err := pkgladder.ParseLadderCadence(dbCfg.Cadence) + if err != nil { + return pkgladder.LadderConfig{}, fmt.Errorf("cadence: %w", err) + } + + var ramp pkgladder.RampSchedule + if err := json.Unmarshal(dbCfg.RampSchedule, &ramp); err != nil { + return pkgladder.LadderConfig{}, fmt.Errorf("ramp_schedule: %w", err) + } + + return pkgladder.LadderConfig{ + Scope: pkgladder.Scope{ + Provider: pkgcommon.ProviderAWS, + AccountID: accountID, + }, + Mode: mode, + Cadence: cadence, + Ramp: ramp, + TargetCoveragePct: dbCfg.TargetCoverage, + BufferFraction: dbCfg.BufferFraction, + BaselinePercentile: dbCfg.BaselinePercentile, + LookbackDays: dbCfg.LookbackDays, + MaxActionsPerRun: dbCfg.MaxActionsPerRun, + BufferUtilizationThresholdPct: dbCfg.BufferUtilizationThreshold, + MaxHourlyCommitPerRun: dbCfg.MaxHourlyCommitPerRun, + }, nil +} + +// assembleLadderPlan builds a LadderPlan from the Allocate result and the +// collected read-side data. Q5: Actions interleaves both purchases +// (Allocations) and Reshapes in one authoritative plan JSON. Term and +// PaymentOption are stamped on every ActionPurchase entry; they are required +// by PlannedAction.Validate for purchase actions. +// +// The monetary snapshot is derived to stay consistent with the authoritative +// allocation rather than recomputed independently: +// - target = baseline low-water * targetCoveragePct/100 (the coverage goal), +// - existing = sum of per-layer ExistingUSDPerHour, +// - gap = sum of the planned allocation gaps (what is actually being placed +// this run), NOT target - existing. +// +// When the baseline low-water is nil the run is a Hold-only no-op (Allocate +// returns no allocations), so target and gap stay nil (never 0-coerced); +// existing is still reported from the observed layer states. +func assembleLadderPlan( + scope pkgladder.Scope, + now time.Time, + allocResult *pkgladder.AllocateResult, + layerStates map[pkgladder.LayerType]pkgladder.LayerState, + baseline pkgladder.UsageBaseline, + targetCoveragePct float64, + term pkgladder.Term, + paymentOpt pkgladder.PaymentOption, +) *pkgladder.LadderPlan { + var actions []pkgladder.PlannedAction + // Purchase allocations come first so approval-email ordering is stable. + for _, alloc := range allocResult.Allocations { + actions = append(actions, pkgladder.PlannedAction{ + Action: pkgladder.ActionPurchase, + Layer: alloc.Layer, + AmountUSDPerHour: alloc.GapUSDPerHour, + Term: term, + PaymentOption: paymentOpt, + Rationale: alloc.Rationale, + DataSources: alloc.DataSources, + }) + } + // Reshapes follow; they carry no amount (AmountUSDPerHour must be nil). + actions = append(actions, allocResult.Reshapes...) + // Holds are informational; append last. + actions = append(actions, allocResult.Holds...) + + // existing = sum of per-layer existing commitment (always reportable). + var existingTotal float64 + for _, ls := range layerStates { + if ls.ExistingUSDPerHour != nil { + existingTotal += *ls.ExistingUSDPerHour + } + } + existing := ratFromFloat(existingTotal) + + // target and gap are only meaningful when the baseline is available; nil + // stays nil (a Hold-only no-op has no purchase target/gap to report). + var target, gap *big.Rat + if baseline.LowWaterUSDPerHour != nil { + if low := ratFromFloat(*baseline.LowWaterUSDPerHour); low != nil { + if pctFrac := ratFromFloat(targetCoveragePct / 100.0); pctFrac != nil { + target = new(big.Rat).Mul(low, pctFrac) + } + } + // gap is what is actually being placed this run (Σ allocation gaps), + // consistent with total_hourly_commit, not target - existing. + gap = sumAllocationGaps(allocResult.Allocations) + } + + return &pkgladder.LadderPlan{ + Scope: scope, + GeneratedAt: now, + TargetUSDPerHour: target, + ExistingUSDPerHour: existing, + GapUSDPerHour: gap, + Actions: actions, + Baseline: baseline, + } +} + +// ratFromFloat converts a float64 to *big.Rat. It returns nil when +// big.Rat.SetFloat64 returns nil (non-finite inputs like NaN or Inf). +// This mirrors the project-wide convention for converting float64 to exact +// rational arithmetic at the boundary. +func ratFromFloat(f float64) *big.Rat { + r := new(big.Rat) + if r.SetFloat64(f) == nil { + return nil + } + return r +} + +// ratToFloat64Ptr converts a *big.Rat to a *float64 suitable for storage in +// a nullable DB column. nil stays nil (never 0-coerced). +func ratToFloat64Ptr(r *big.Rat) *float64 { + if r == nil { + return nil + } + f, _ := r.Float64() + return &f +} + +// sumAllocationGaps sums the GapUSDPerHour across all allocations as an exact +// *big.Rat. It returns a zero (non-nil) rat for an empty allocation set: a +// Hold-only run plans exactly $0 of new commitment, which is a meaningful +// value here (what is being placed), not "not computed". +func sumAllocationGaps(allocs []pkgladder.Allocation) *big.Rat { + sum := new(big.Rat) + for _, a := range allocs { + if a.GapUSDPerHour != nil { + sum.Add(sum, a.GapUSDPerHour) + } + } + return sum +} + +// allocTotalHourlyCommit sums the GapUSDPerHour across all allocations, +// converting to float64. This is the non-nullable total_hourly_commit stored +// in the run row; it starts at 0 for a no-op (Hold-only) plan, which is the +// correct semantic (zero new commitment, not "not computed"). +func allocTotalHourlyCommit(allocs []pkgladder.Allocation) float64 { + f, _ := sumAllocationGaps(allocs).Float64() + return f +} + +// buildTrancheDBRows converts pkg/ladder Tranche rows to config.LadderTrancheDB +// rows ready for SaveLadderTranches. It fails loud on a malformed +// AmountUSDPerHour (money path): rather than persist a silently 0-coerced +// $0 tranche, it returns an error so the whole run is counted Errored and +// nothing partial is written. +func buildTrancheDBRows(tranches []pkgladder.Tranche, runID string, configID *string) ([]config.LadderTrancheDB, error) { + rows := make([]config.LadderTrancheDB, 0, len(tranches)) + for _, tr := range tranches { + // AmountUSDHr: parse the RatString back to float64. A missing or + // malformed RatString means BuildTranches produced a broken tranche; + // that should never happen after Validate() passes, so surface it as + // an error instead of writing a $0 row. + r, ok := new(big.Rat).SetString(tr.AmountUSDPerHour) + if !ok || r == nil { + return nil, fmt.Errorf("tranche %s: malformed amount_usd_per_hour %q (not a parseable rational)", tr.ID, tr.AmountUSDPerHour) + } + amountUSDHr, _ := r.Float64() + runIDCopy := runID + rows = append(rows, config.LadderTrancheDB{ + ID: tr.ID, + ConfigID: configID, + RunID: &runIDCopy, + LayerType: tr.Layer, + Term: tr.Term, + PaymentOption: tr.PaymentOption, + Status: pkgladder.TrancheStatusScheduled, + AmountUSDHr: amountUSDHr, + ScheduledDate: tr.FireAfter, + }) + } + return rows, nil +} + +// ladderPlanJSONAction is a JSON-serializable form of pkgladder.PlannedAction. +// PlannedAction.AmountUSDPerHour is *big.Rat (not JSON-friendly); this DTO +// converts it to *float64 at the boundary. +type ladderPlanJSONAction struct { + Action string `json:"action"` + Layer string `json:"layer"` + AmountUSDHr *float64 `json:"amount_usd_hr,omitempty"` // nil for Hold/Reshape + Term string `json:"term,omitempty"` + PaymentOption string `json:"payment_option,omitempty"` + Rationale string `json:"rationale"` + DataSources []string `json:"data_sources,omitempty"` +} + +// ladderPlanJSONDTO is a JSON-serializable form of pkgladder.LadderPlan. +// The *big.Rat monetary fields are converted to *float64 so they round-trip +// cleanly through JSONB without binary encoding. +type ladderPlanJSONDTO struct { + Scope pkgladder.Scope `json:"scope"` + GeneratedAt time.Time `json:"generated_at"` + TargetUSDHr *float64 `json:"target_usd_hr,omitempty"` + ExistingUSDHr *float64 `json:"existing_usd_hr,omitempty"` + GapUSDHr *float64 `json:"gap_usd_hr,omitempty"` + Actions []ladderPlanJSONAction `json:"actions"` + Baseline pkgladder.UsageBaseline `json:"baseline"` +} + +// marshalLadderPlan serializes a LadderPlan to JSON for storage in the +// plan JSONB column. *big.Rat fields are converted to *float64 at this +// boundary; nil stays nil. +func marshalLadderPlan(plan *pkgladder.LadderPlan) (json.RawMessage, error) { + actions := make([]ladderPlanJSONAction, 0, len(plan.Actions)) + for _, a := range plan.Actions { + action := ladderPlanJSONAction{ + Action: string(a.Action), + Layer: string(a.Layer), + AmountUSDHr: ratToFloat64Ptr(a.AmountUSDPerHour), + Term: string(a.Term), + PaymentOption: string(a.PaymentOption), + Rationale: a.Rationale, + DataSources: a.DataSources, + } + actions = append(actions, action) + } + dto := ladderPlanJSONDTO{ + Scope: plan.Scope, + GeneratedAt: plan.GeneratedAt, + TargetUSDHr: ratToFloat64Ptr(plan.TargetUSDPerHour), + ExistingUSDHr: ratToFloat64Ptr(plan.ExistingUSDPerHour), + GapUSDHr: ratToFloat64Ptr(plan.GapUSDPerHour), + Actions: actions, + Baseline: plan.Baseline, + } + b, err := json.Marshal(dto) + if err != nil { + return nil, err + } + return json.RawMessage(b), nil +} + +// ladderTermFromYears converts a GlobalConfig.DefaultTerm (integer years) to +// a typed pkgladder.Term. Fails loud on any unrecognized value so money-path +// decisions never proceed with an unknown commitment term. +func ladderTermFromYears(years int) (pkgladder.Term, error) { + switch years { + case 1: + return pkgladder.Term1Year, nil + case 3: + return pkgladder.Term3Year, nil + default: + return "", fmt.Errorf("unsupported default_term=%d years (allowed: 1, 3)", years) + } +} diff --git a/internal/server/handler_ladder_test.go b/internal/server/handler_ladder_test.go new file mode 100644 index 000000000..77a206245 --- /dev/null +++ b/internal/server/handler_ladder_test.go @@ -0,0 +1,991 @@ +package server + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/CUDly/internal/config" + "github.com/LeanerCloud/CUDly/internal/testutil" + pkgcommon "github.com/LeanerCloud/CUDly/pkg/common" + pkgladder "github.com/LeanerCloud/CUDly/pkg/ladder" +) + +// ============================================================ +// Fake LadderCapability +// ============================================================ + +// fakeLadderCapability is an in-process fake implementing pkgladder.LadderCapability. +// PurchaseLayer and ReshapeBuffer call t.Fatal so that any test which exercises them +// fails immediately, asserting the ABSENCE of those calls in the plan-only phase. +type fakeLadderCapability struct { + baseline pkgladder.UsageBaseline + baselineErr error + layerStates map[pkgladder.LayerType]pkgladder.LayerState + layerStatesErr error + supportedLayers []pkgladder.LayerSpec + t *testing.T +} + +func (f *fakeLadderCapability) Provider() pkgcommon.ProviderType { + return pkgcommon.ProviderAWS +} + +func (f *fakeLadderCapability) SupportedLayers() []pkgladder.LayerSpec { + if f.supportedLayers != nil { + return f.supportedLayers + } + // Default: single RI layer as base. + return []pkgladder.LayerSpec{ + { + Type: pkgladder.LayerConvertibleRI, + Roles: []pkgladder.LayerRole{pkgladder.RoleBase}, + }, + { + Type: pkgladder.LayerEC2InstanceSP, + Roles: []pkgladder.LayerRole{pkgladder.RoleFlex}, + }, + } +} + +func (f *fakeLadderCapability) ListCommitments(_ context.Context, _ pkgladder.Scope) ([]pkgcommon.Commitment, error) { + return nil, nil +} + +func (f *fakeLadderCapability) GetLayerStates(_ context.Context, _ pkgladder.Scope) (map[pkgladder.LayerType]pkgladder.LayerState, error) { + if f.layerStatesErr != nil { + return nil, f.layerStatesErr + } + if f.layerStates != nil { + return f.layerStates, nil + } + // Default: both layers have zero existing commitment (explicit zero, not nil). + zero := 0.0 + return map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerConvertibleRI: { + Layer: pkgladder.LayerConvertibleRI, + ExistingUSDPerHour: &zero, + ExpiringUSDPerHour: &zero, + }, + pkgladder.LayerEC2InstanceSP: { + Layer: pkgladder.LayerEC2InstanceSP, + ExistingUSDPerHour: &zero, + ExpiringUSDPerHour: &zero, + }, + }, nil +} + +func (f *fakeLadderCapability) GetUsageBaseline(_ context.Context, _ pkgladder.Scope, _ int, _ float64) (pkgladder.UsageBaseline, error) { + return f.baseline, f.baselineErr +} + +// PurchaseLayer must NOT be called in the plan-only phase. +func (f *fakeLadderCapability) PurchaseLayer(_ context.Context, _ pkgladder.LayerType, _ pkgcommon.Recommendation, _ pkgcommon.PurchaseOptions) (pkgcommon.PurchaseResult, error) { + f.t.Fatal("PurchaseLayer called in plan-only PR-2: purchases must never fire during planning") + return pkgcommon.PurchaseResult{}, nil +} + +// ReshapeBuffer must NOT be called in the plan-only phase. +func (f *fakeLadderCapability) ReshapeBuffer(_ context.Context, _ pkgladder.Scope, _ pkgladder.BufferReshapeConfig) (pkgladder.ReshapeSummary, error) { + f.t.Fatal("ReshapeBuffer called in plan-only PR-2: reshapes must never execute during planning") + return pkgladder.ReshapeSummary{}, nil +} + +// ============================================================ +// Inline ladder-aware mock store +// ============================================================ + +// ladderTestStore extends mockConfigStoreForHealth with capturable ladder run methods. +// Capture fields (saveLadderRunFn, saveLadderTranchesFn) let individual tests +// assert on what was persisted without importing a testify mock. +type ladderTestStore struct { + mockConfigStoreForHealth + globalCfg *config.GlobalConfig + globalCfgErr error + ladderConfigs []config.LadderConfigDB + ladderConfigsErr error + cloudAcct *config.CloudAccount + cloudAcctErr error + latestRunAt *time.Time + latestRunAtErr error + + // Per-cloud-account-id overrides, keyed by LadderConfigDB.CloudAccountID. + // When a config's CloudAccountID is present here, the mapped account/error + // takes precedence over the single cloudAcct/cloudAcctErr fields. This lets + // a multi-config test give each config a distinct account outcome. + cloudAcctByID map[string]*config.CloudAccount + cloudAcctErrByID map[string]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. + savedRun *config.LadderRunDB + savedRuns []*config.LadderRunDB + savedTranches []config.LadderTrancheDB + saveLadderRunErr error + saveLadderTranchesErr error +} + +func (s *ladderTestStore) GetGlobalConfig(_ context.Context) (*config.GlobalConfig, error) { + if s.globalCfgErr != nil { + return nil, s.globalCfgErr + } + if s.globalCfg != nil { + return s.globalCfg, nil + } + return &config.GlobalConfig{}, nil +} + +func (s *ladderTestStore) GetLadderConfigs(_ context.Context) ([]config.LadderConfigDB, error) { + return s.ladderConfigs, s.ladderConfigsErr +} + +func (s *ladderTestStore) GetCloudAccount(_ context.Context, id string) (*config.CloudAccount, error) { + if s.cloudAcctErrByID != nil { + if err, ok := s.cloudAcctErrByID[id]; ok { + return nil, err + } + } + if s.cloudAcctByID != nil { + if acct, ok := s.cloudAcctByID[id]; ok { + return acct, nil + } + } + return s.cloudAcct, s.cloudAcctErr +} + +func (s *ladderTestStore) LatestLadderRunStartedAt(_ context.Context, _ string) (*time.Time, error) { + return s.latestRunAt, s.latestRunAtErr +} + +func (s *ladderTestStore) SaveLadderRun(_ context.Context, run *config.LadderRunDB) (*config.LadderRunDB, error) { + if s.saveLadderRunErr != nil { + return nil, s.saveLadderRunErr + } + s.savedRun = run + s.savedRuns = append(s.savedRuns, run) + return run, nil +} + +func (s *ladderTestStore) SaveLadderTranches(_ context.Context, tranches []config.LadderTrancheDB) error { + if s.saveLadderTranchesErr != nil { + return s.saveLadderTranchesErr + } + s.savedTranches = tranches + 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). +func (s *ladderTestStore) SaveLadderRunWithTranches(_ 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 +} + +// ============================================================ +// Shared test helpers +// ============================================================ + +// validTestDBConfig returns a LadderConfigDB with all required fields set to +// valid values. Tests override specific fields as needed. +func validTestDBConfig(id string) config.LadderConfigDB { + rampJSON, _ := json.Marshal(pkgladder.RampSchedule{ + Steps: []pkgladder.RampStep{{AfterDays: 0, Fraction: 1.0}}, + }) + return config.LadderConfigDB{ + ID: id, + CloudAccountID: "cloud-acct-uuid", + Provider: "aws", + Enabled: true, + Mode: "email_approval", + Cadence: "daily", + TargetCoverage: 80.0, + BufferFraction: 0.0, // no buffer role needed for base/flex-only fake + BaselinePercentile: 5.0, + LookbackDays: 30, + MaxActionsPerRun: 5, + BufferUtilizationThreshold: 50.0, + RampSchedule: rampJSON, + } +} + +// validTestGlobalCfg returns a GlobalConfig that enables laddering and has a +// known default term and payment option. +func validTestGlobalCfg() *config.GlobalConfig { + return &config.GlobalConfig{ + LadderingEnabled: true, + DefaultTerm: 1, + DefaultPayment: "no-upfront", + } +} + +// validTestCloudAccount returns a CloudAccount whose ExternalID matches the +// given AWS account ID (simulating resolveAccountID). +func validTestCloudAccount(externalID string) *config.CloudAccount { + return &config.CloudAccount{ + ID: "cloud-acct-uuid", + Provider: "aws", + ExternalID: externalID, + Enabled: true, + } +} + +// nonZeroFloat64 is a helper to create a *float64 from a literal. +func nonZeroFloat64(f float64) *float64 { return &f } + +// testBaseline returns a UsageBaseline with a non-nil LowWaterUSDPerHour. +func testBaseline(lowWater float64) pkgladder.UsageBaseline { + return pkgladder.UsageBaseline{ + LowWaterUSDPerHour: nonZeroFloat64(lowWater), + StableUSDPerHour: nonZeroFloat64(lowWater * 0.9), + LookbackDays: 30, + Percentile: 5.0, + } +} + +// ============================================================ +// ParseScheduledEvent registration +// ============================================================ + +func TestHandleLadderRun_ParseScheduledEvent(t *testing.T) { + taskType, err := ParseScheduledEvent([]byte(`{"action":"ladder_run"}`)) + require.NoError(t, err) + assert.Equal(t, TaskLadderRun, taskType) +} + +// ============================================================ +// handleLadderRun: global kill-switch +// ============================================================ + +func TestHandleLadderRun_GlobalDisabled(t *testing.T) { + ctx := testutil.TestContext(t) + store := &ladderTestStore{ + globalCfg: &config.GlobalConfig{LadderingEnabled: false}, + } + app := &Application{Config: store} + + result, err := app.handleLadderRun(ctx) + + require.NoError(t, err) + require.NotNil(t, result) + // Nothing should be planned or skipped when the global flag is off. + assert.Equal(t, 0, result.Planned) + assert.Equal(t, 0, result.SkippedCadence) + assert.Equal(t, 0, result.SkippedDisabled) + // SaveLadderRun must not be called. + assert.Nil(t, store.savedRun, "SaveLadderRun must not be called when laddering is globally disabled") +} + +func TestHandleLadderRun_GlobalConfigError(t *testing.T) { + ctx := testutil.TestContext(t) + store := &ladderTestStore{ + globalCfgErr: errors.New("db error"), + } + app := &Application{Config: store} + + _, err := app.handleLadderRun(ctx) + + require.Error(t, err) + assert.Contains(t, err.Error(), "db error") +} + +// B1: a strict caller-account resolution failure (e.g. transient STS error) +// must fail the WHOLE task loud. It must NOT fall through and skip every config +// as multi-account (which would return success and let EventBridge see a +// false no-op on a money path). Nothing may be persisted or marked skipped. +func TestHandleLadderRun_AccountResolutionFailure(t *testing.T) { + ctx := testutil.TestContext(t) + dbCfg := validTestDBConfig("cfg-1") + store := &ladderTestStore{ + globalCfg: validTestGlobalCfg(), + ladderConfigs: []config.LadderConfigDB{dbCfg}, + } + app := &Application{ + Config: store, + LadderAccountResolver: func(_ context.Context) (string, string, error) { + return "", "", errors.New("STS GetCallerIdentity failed") + }, + LadderCapabilityFactory: func(_ context.Context, _, _ string) (pkgladder.LadderCapability, error) { + t.Fatal("factory must not be called when account resolution fails") + return nil, nil + }, + } + + result, err := app.handleLadderRun(ctx) + + require.Error(t, err, "an unresolved caller account must fail the task loud") + assert.Contains(t, err.Error(), "resolve caller AWS account") + assert.Nil(t, result, "no result struct on a hard failure (not a false success)") + assert.Nil(t, store.savedRun, "nothing may be persisted when the account is unresolved") + assert.Empty(t, store.savedRuns) +} + +// resolveLadderIdentity must fail loud on an empty region BEFORE calling STS: +// an empty region would make NewFromAWSConfig reject every config later, turning +// the task into a silent all-Errored no-op that still reports success. The +// region check short-circuits, so this exercises the new guard with no +// credentials/STS involved. +func TestResolveLadderIdentity_EmptyRegion_FailsLoud(t *testing.T) { + ctx := testutil.TestContext(t) + + accountID, region, err := resolveLadderIdentity(ctx, aws.Config{Region: ""}) + + require.Error(t, err, "an empty region must fail loud, not fall through to STS") + assert.Contains(t, err.Error(), "region") + assert.Empty(t, accountID, "no account ID may be returned on the empty-region failure") + assert.Empty(t, region, "no region may be returned on the empty-region failure") +} + +// B1 (region): an empty-region resolution failure must abort the whole +// handleLadderRun task loud, mirroring TestHandleLadderRun_AccountResolutionFailure. +// Nothing may be persisted and no config may be marked skipped. +func TestHandleLadderRun_EmptyRegionResolution_FailsLoud(t *testing.T) { + ctx := testutil.TestContext(t) + dbCfg := validTestDBConfig("cfg-1") + store := &ladderTestStore{ + globalCfg: validTestGlobalCfg(), + ladderConfigs: []config.LadderConfigDB{dbCfg}, + } + app := &Application{ + Config: store, + // The default resolver would return this exact error for an empty region; + // the stub keeps the test off STS while asserting the fail-loud contract. + LadderAccountResolver: func(c context.Context) (string, string, error) { + return resolveLadderIdentity(c, aws.Config{Region: ""}) + }, + LadderCapabilityFactory: func(_ context.Context, _, _ string) (pkgladder.LadderCapability, error) { + t.Fatal("factory must not be called when the region is unresolved") + return nil, nil + }, + } + + result, err := app.handleLadderRun(ctx) + + require.Error(t, err, "an empty region must fail the whole task loud") + assert.Contains(t, err.Error(), "region") + assert.Nil(t, result, "no result struct on a hard failure (not a false success)") + assert.Nil(t, store.savedRun, "nothing may be persisted when the region is unresolved") + assert.Empty(t, store.savedRuns) +} + +// ============================================================ +// handleLadderRun: per-config disabled/multi-account skipping +// ============================================================ + +func TestHandleLadderRun_AllConfigsDisabled(t *testing.T) { + ctx := testutil.TestContext(t) + dbCfg := validTestDBConfig("cfg-1") + dbCfg.Enabled = false + + store := &ladderTestStore{ + globalCfg: validTestGlobalCfg(), + ladderConfigs: []config.LadderConfigDB{dbCfg}, + } + app := &Application{ + Config: store, + // Inject a stub resolver so the strict account gate does not hit STS. + LadderAccountResolver: func(_ context.Context) (string, string, error) { + return "123456789012", "us-east-1", nil + }, + // No LadderCapabilityFactory: must not be reached. + LadderCapabilityFactory: func(_ context.Context, _, _ string) (pkgladder.LadderCapability, error) { + t.Fatal("factory called on a disabled config") + return nil, nil + }, + } + + result, err := app.handleLadderRun(ctx) + + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, 0, result.Planned) + assert.Equal(t, 1, result.SkippedDisabled) + assert.Nil(t, store.savedRun) +} + +// ============================================================ +// executeLadderRun: healthy single-config run +// ============================================================ + +func TestExecuteLadderRun_HealthyRun(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-healthy") + // Two-step ramp with a DELAYED step: an AfterDays==0 step becomes a BuyNow + // action, only AfterDays>0 steps become persisted tranches. This ensures + // SaveLadderTranches receives a non-empty batch so the assertion below + // exercises real tranche persistence. + dbCfg.RampSchedule, _ = json.Marshal(pkgladder.RampSchedule{ + Steps: []pkgladder.RampStep{ + {AfterDays: 0, Fraction: 0.5}, + {AfterDays: 7, Fraction: 0.5}, + }, + }) + + store := &ladderTestStore{} + app := &Application{Config: store} + + cap := &fakeLadderCapability{ + t: t, + baseline: testBaseline(10.0), + } + + err := app.executeLadderRun(ctx, &dbCfg, cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now) + require.NoError(t, err) + + // SaveLadderRun must have been called exactly once. + require.NotNil(t, store.savedRun, "SaveLadderRun must be called for a healthy run") + run := store.savedRun + assert.Equal(t, pkgladder.RunStatusPlanned, run.Status, "run status must be planned (plan-only milestone)") + assert.Equal(t, "cfg-healthy", *run.ConfigID) + assert.Equal(t, "email_approval", *run.Mode) + assert.Equal(t, "daily", *run.Cadence) + + // Monetary snapshot: baseline-derived fields must be non-nil (we have a valid baseline). + assert.NotNil(t, run.BaselineUSDHr, "baseline_usd_hr must be non-nil when baseline is available") + require.NotNil(t, run.TargetUSDHr, "target_usd_hr must be non-nil when baseline is available") + // target = low-water (10) * target_coverage (80%) = 8.0, derived consistently + // with TargetCoveragePct rather than the raw low-water. + assert.InDelta(t, 8.0, *run.TargetUSDHr, 1e-9, "target must be low-water * coverage%%, not raw low-water") + require.NotNil(t, run.GapUSDHr, "gap_usd_hr must be non-nil when baseline is available") + // gap (Σ planned allocation gaps) must equal total_hourly_commit exactly. + assert.InDelta(t, run.TotalHourlyCommit, *run.GapUSDHr, 1e-9, "gap must equal total_hourly_commit (both Σ allocation gaps)") + assert.Greater(t, run.TotalHourlyCommit, 0.0, "a healthy run below target must plan a positive commitment") + + // Plan JSON must be valid non-empty JSON. + require.True(t, len(run.Plan) > 2, "plan JSON must be non-empty") + var planDTO ladderPlanJSONDTO + require.NoError(t, json.Unmarshal(run.Plan, &planDTO), "plan JSON must unmarshal cleanly") + assert.Equal(t, "aws", string(planDTO.Scope.Provider)) + assert.Equal(t, "123456789012", planDTO.Scope.AccountID) + assert.False(t, planDTO.GeneratedAt.IsZero(), "plan generated_at must not be zero") + + // Plan must contain at least one purchase action stamped with term/payment. + var purchaseActions int + for _, a := range planDTO.Actions { + if a.Action == string(pkgladder.ActionPurchase) { + purchaseActions++ + assert.Equal(t, string(pkgladder.Term1Year), a.Term, "purchase action must carry the term") + assert.Equal(t, string(pkgladder.PaymentNoUpfront), a.PaymentOption, "purchase action must carry the payment option") + require.NotNil(t, a.AmountUSDHr, "purchase action must carry a non-nil amount") + } + } + assert.Positive(t, purchaseActions, "a healthy below-target run must produce purchase actions") + + // SaveLadderTranches must have been called with the ramp tranches produced + // from the allocations (non-empty for a below-target run). + require.NotNil(t, store.savedTranches, "SaveLadderTranches must be called for a healthy run") + assert.NotEmpty(t, store.savedTranches, "healthy below-target run must persist tranche audit rows") + for _, tr := range store.savedTranches { + assert.Equal(t, pkgladder.TrancheStatusScheduled, tr.Status, "PR-2 tranches are scheduled-only (no firing)") + assert.Greater(t, tr.AmountUSDHr, 0.0, "each tranche must carry a positive amount") + } + // No panic means PurchaseLayer and ReshapeBuffer were not called. +} + +// ============================================================ +// executeLadderRun: GetUsageBaseline ERROR -> no run persisted +// ============================================================ + +// TestExecuteLadderRun_BaselineError_NoRunPersisted covers the case where +// GetUsageBaseline itself returns an ERROR (e.g. the CE adapter is not wired). +// This is distinct from a nil-baseline-with-no-error (which the engine treats +// as a Hold, see TestExecuteLadderRun_NilBaseline_HoldPlan). On a baseline +// error the run must fail loud and persist NOTHING. +func TestExecuteLadderRun_BaselineError_NoRunPersisted(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-baseline-err") + + store := &ladderTestStore{} + app := &Application{Config: store} + + cap := &fakeLadderCapability{ + t: t, + baselineErr: errors.New("CE on-demand series not yet wired"), + } + + err := app.executeLadderRun(ctx, &dbCfg, cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now) + require.Error(t, err) + assert.Contains(t, err.Error(), "GetUsageBaseline") + // Fail-loud: nothing may be persisted when the baseline lookup errors. + assert.Nil(t, store.savedRun, "no ladder_runs row may be persisted on a baseline error") + assert.Nil(t, store.savedTranches, "no tranches may be persisted on a baseline error") +} + +// TestExecuteLadderRun_NilBaseline_HoldPlan covers GetUsageBaseline returning a +// baseline with a nil LowWaterUSDPerHour and NO error. The engine treats this +// as a Hold (explainable no-op), so executeLadderRun must persist a +// status=planned run with NULL target/gap monetary cols and no purchase +// actions -- NOT an error. +func TestExecuteLadderRun_NilBaseline_HoldPlan(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-nilbaseline") + + store := &ladderTestStore{} + app := &Application{Config: store} + + cap := &fakeLadderCapability{ + t: t, + // Baseline present but LowWaterUSDPerHour nil -> engine holds. + baseline: pkgladder.UsageBaseline{ + LowWaterUSDPerHour: nil, + LookbackDays: 30, + Percentile: 5.0, + }, + } + + err := app.executeLadderRun(ctx, &dbCfg, cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now) + require.NoError(t, err, "nil baseline with no error must be a Hold, not an error") + + require.NotNil(t, store.savedRun, "a Hold run must still be persisted") + run := store.savedRun + assert.Equal(t, pkgladder.RunStatusPlanned, run.Status, "Hold run status must be planned") + // Monetary snapshot: baseline/target/gap must be NULL (nil), never 0-coerced. + assert.Nil(t, run.BaselineUSDHr, "baseline_usd_hr must be nil when low-water is nil") + assert.Nil(t, run.TargetUSDHr, "target_usd_hr must be nil when baseline is unavailable") + assert.Nil(t, run.GapUSDHr, "gap_usd_hr must be nil when baseline is unavailable") + assert.Equal(t, 0.0, run.TotalHourlyCommit, "a Hold run plans zero new commitment") + + // Plan must contain no purchase actions (Hold-only). + var planDTO ladderPlanJSONDTO + require.NoError(t, json.Unmarshal(run.Plan, &planDTO)) + for _, a := range planDTO.Actions { + assert.NotEqual(t, string(pkgladder.ActionPurchase), a.Action, "a Hold-only plan must contain no purchase actions") + } + // No tranches produced from an empty allocation set. + assert.Empty(t, store.savedTranches, "a Hold-only run produces no tranches") +} + +// TestExecuteLadderRun_ZeroGap_HoldPlan covers a run where existing commitment +// already meets or exceeds the target, so the gap is <= the minimum and the +// engine returns a Hold. The run must persist as status=planned with no +// purchase actions. +func TestExecuteLadderRun_ZeroGap_HoldPlan(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-zerogap") + + store := &ladderTestStore{} + app := &Application{Config: store} + + // Existing base commitment (10) already meets the low-water (10); with + // target coverage 80% the target (8) is below existing, so gap <= 0 -> Hold. + tenUSD := 10.0 + zero := 0.0 + cap := &fakeLadderCapability{ + t: t, + baseline: testBaseline(10.0), + layerStates: map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerConvertibleRI: { + Layer: pkgladder.LayerConvertibleRI, + ExistingUSDPerHour: &tenUSD, + ExpiringUSDPerHour: &zero, + }, + pkgladder.LayerEC2InstanceSP: { + Layer: pkgladder.LayerEC2InstanceSP, + ExistingUSDPerHour: &zero, + ExpiringUSDPerHour: &zero, + }, + }, + } + + err := app.executeLadderRun(ctx, &dbCfg, cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now) + require.NoError(t, err) + + require.NotNil(t, store.savedRun, "a zero-gap Hold run must still be persisted") + run := store.savedRun + assert.Equal(t, pkgladder.RunStatusPlanned, run.Status) + assert.Equal(t, 0.0, run.TotalHourlyCommit, "zero-gap run plans zero new commitment") + + var planDTO ladderPlanJSONDTO + require.NoError(t, json.Unmarshal(run.Plan, &planDTO)) + for _, a := range planDTO.Actions { + assert.NotEqual(t, string(pkgladder.ActionPurchase), a.Action, "zero-gap plan must contain no purchase actions") + } + assert.Empty(t, store.savedTranches, "a zero-gap Hold run produces no tranches") +} + +// TestExecuteLadderRun_MaxActionsExceeded covers Allocate erroring because the +// split produces more allocations than MaxActionsPerRun allows. executeLadderRun +// must fail loud and persist nothing. +func TestExecuteLadderRun_MaxActionsExceeded(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-maxactions") + dbCfg.MaxActionsPerRun = 1 // base+flex split yields 2 allocations > 1 + + store := &ladderTestStore{} + app := &Application{Config: store} + + // Stable (10) is far below the target (80% of low-water 100 = 80), so the + // base layer only absorbs up to stable and the remainder spills to flex, + // producing TWO allocations. With MaxActionsPerRun=1 the engine must error. + cap := &fakeLadderCapability{ + t: t, + baseline: pkgladder.UsageBaseline{ + LowWaterUSDPerHour: nonZeroFloat64(100.0), + StableUSDPerHour: nonZeroFloat64(10.0), + LookbackDays: 30, + Percentile: 5.0, + }, + } + + err := app.executeLadderRun(ctx, &dbCfg, cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now) + require.Error(t, err, "Allocate must error when the split exceeds max_actions_per_run") + assert.Contains(t, err.Error(), "max_actions_per_run") + assert.Nil(t, store.savedRun, "no run may be persisted when Allocate errors") + assert.Nil(t, store.savedTranches, "no tranches may be persisted when Allocate errors") +} + +// ============================================================ +// cadence gate +// ============================================================ + +func TestLadderWithinCadenceWindow_Daily_TooRecent(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-cadence") + dbCfg.Cadence = "daily" + + recent := now.Add(-19 * time.Hour) // 19h ago < 20h threshold + store := &ladderTestStore{latestRunAt: &recent} + + skip, reason, err := ladderWithinCadenceWindow(ctx, store, &dbCfg, now) + + require.NoError(t, err) + assert.True(t, skip, "must skip a daily config run within 20h of the last run") + assert.Contains(t, reason, "cadence=daily") +} + +func TestLadderWithinCadenceWindow_Daily_Eligible(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-cadence") + dbCfg.Cadence = "daily" + + lastRun := now.Add(-25 * time.Hour) // 25h ago > 20h threshold + store := &ladderTestStore{latestRunAt: &lastRun} + + skip, _, err := ladderWithinCadenceWindow(ctx, store, &dbCfg, now) + + require.NoError(t, err) + assert.False(t, skip, "must run a daily config when last run was > 20h ago") +} + +func TestLadderWithinCadenceWindow_Weekly_TooRecent(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-cadence-w") + dbCfg.Cadence = "weekly" + + recent := now.Add(-5 * 24 * time.Hour) // 5 days ago < 6d20h threshold + store := &ladderTestStore{latestRunAt: &recent} + + skip, reason, err := ladderWithinCadenceWindow(ctx, store, &dbCfg, now) + + require.NoError(t, err) + assert.True(t, skip, "must skip a weekly config run within 6d20h of the last run") + assert.Contains(t, reason, "cadence=weekly") +} + +func TestLadderWithinCadenceWindow_NoHistory_AlwaysEligible(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-cadence-new") + dbCfg.Cadence = "daily" + + store := &ladderTestStore{latestRunAt: nil} // no previous run + + skip, _, err := ladderWithinCadenceWindow(ctx, store, &dbCfg, now) + + require.NoError(t, err) + assert.False(t, skip, "must always run when there is no previous ladder run") +} + +// B4: a LatestLadderRunStartedAt lookup error must fail CLOSED (propagate the +// error) so the caller can count the config Errored instead of double-running. +func TestLadderWithinCadenceWindow_DBError_FailsClosed(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + dbCfg := validTestDBConfig("cfg-cadence-dberr") + dbCfg.Cadence = "daily" + + store := &ladderTestStore{latestRunAtErr: errors.New("db unavailable")} + + skip, _, err := ladderWithinCadenceWindow(ctx, store, &dbCfg, now) + + require.Error(t, err, "a lookup error must be propagated, not swallowed") + assert.False(t, skip, "must not signal skip on a lookup error") + assert.Contains(t, err.Error(), "LatestLadderRunStartedAt") +} + +// B4 (handler level): a config whose cadence lookup errors must be counted +// Errored with no run persisted (fail closed, per-config isolation). +func TestProcessOneLadderConfig_CadenceDBError_Errored(t *testing.T) { + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + const ownAccount = "123456789012" + dbCfg := validTestDBConfig("cfg-cadence-dberr") + + store := &ladderTestStore{ + cloudAcct: validTestCloudAccount(ownAccount), + latestRunAtErr: errors.New("db unavailable"), + } + app := &Application{ + Config: store, + LadderCapabilityFactory: func(_ context.Context, _, _ string) (pkgladder.LadderCapability, error) { + t.Fatal("factory must not be called when the cadence gate fails closed") + return nil, nil + }, + } + + result := app.runLadderConfigs(ctx, []config.LadderConfigDB{dbCfg}, ownAccount, "us-east-1", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now) + + assert.Equal(t, 1, result.Errored, "a cadence lookup error must count the config Errored") + assert.Equal(t, 0, result.Planned) + assert.Equal(t, 0, result.SkippedCadence, "a lookup error is not a cadence skip") + assert.Nil(t, store.savedRun, "no run may be persisted when the cadence gate fails closed") +} + +// ============================================================ +// ladderConfigToEngine: parse errors fail loud +// ============================================================ + +func TestLadderConfigToEngine_BadMode(t *testing.T) { + dbCfg := validTestDBConfig("cfg-bad-mode") + dbCfg.Mode = "invalid_mode" + + _, err := ladderConfigToEngine(&dbCfg, "123456789012") + + require.Error(t, err) + assert.Contains(t, err.Error(), "mode") +} + +func TestLadderConfigToEngine_BadCadence(t *testing.T) { + dbCfg := validTestDBConfig("cfg-bad-cadence") + dbCfg.Cadence = "monthly" + + _, err := ladderConfigToEngine(&dbCfg, "123456789012") + + require.Error(t, err) + assert.Contains(t, err.Error(), "cadence") +} + +func TestLadderConfigToEngine_BadRampJSON(t *testing.T) { + dbCfg := validTestDBConfig("cfg-bad-ramp") + dbCfg.RampSchedule = []byte(`not-json`) + + _, err := ladderConfigToEngine(&dbCfg, "123456789012") + + require.Error(t, err) + assert.Contains(t, err.Error(), "ramp_schedule") +} + +func TestLadderConfigToEngine_ValidConfig(t *testing.T) { + dbCfg := validTestDBConfig("cfg-valid") + engineCfg, err := ladderConfigToEngine(&dbCfg, "123456789012") + + require.NoError(t, err) + assert.Equal(t, pkgcommon.ProviderAWS, engineCfg.Scope.Provider) + assert.Equal(t, "123456789012", engineCfg.Scope.AccountID) + assert.Equal(t, pkgladder.ModeEmailApproval, engineCfg.Mode) + assert.Equal(t, pkgladder.CadenceDaily, engineCfg.Cadence) + assert.Equal(t, 80.0, engineCfg.TargetCoveragePct) + assert.Equal(t, 5.0, engineCfg.BaselinePercentile) + assert.Equal(t, 30, engineCfg.LookbackDays) +} + +// ============================================================ +// ladderTermFromYears +// ============================================================ + +func TestLadderTermFromYears(t *testing.T) { + tests := []struct { + years int + want pkgladder.Term + wantErr bool + }{ + {1, pkgladder.Term1Year, false}, + {3, pkgladder.Term3Year, false}, + {2, "", true}, + {0, "", true}, + } + for _, tc := range tests { + tc := tc + t.Run("", func(t *testing.T) { + got, err := ladderTermFromYears(tc.years) + if tc.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} + +// ============================================================ +// Multi-config isolation: one error does not abort others +// ============================================================ + +func TestHandleLadderRun_MultiConfigIsolation(t *testing.T) { + // runLadderConfigs is the loop that handleLadderRun runs after the AWS SDK + // account-resolution step. Testing it directly (rather than handleLadderRun) + // avoids awsconfig.LoadDefaultConfig / STS while still proving the isolation + // guarantee: an error on one config must not abort the others. + ctx := testutil.TestContext(t) + now := time.Date(2026, 7, 1, 12, 0, 0, 0, time.UTC) + const ownAccount = "123456789012" + + // Config A: healthy -> plans a run. + cfgHealthy := validTestDBConfig("cfg-healthy") + cfgHealthy.CloudAccountID = "acct-ok" + // Config B: its cloud account lookup errors -> counted Errored, must not + // prevent config A from planning. + cfgBroken := validTestDBConfig("cfg-broken") + cfgBroken.CloudAccountID = "acct-bad" + + store := &ladderTestStore{ + cloudAcctByID: map[string]*config.CloudAccount{ + "acct-ok": validTestCloudAccount(ownAccount), + }, + cloudAcctErrByID: map[string]error{ + "acct-bad": errors.New("cloud account lookup failed"), + }, + } + app := &Application{ + Config: store, + LadderCapabilityFactory: func(_ context.Context, _, _ string) (pkgladder.LadderCapability, error) { + return &fakeLadderCapability{t: t, baseline: testBaseline(10.0)}, nil + }, + } + + // Order the broken config first to prove a leading failure does not abort + // the healthy config that follows. + configs := []config.LadderConfigDB{cfgBroken, cfgHealthy} + result := app.runLadderConfigs(ctx, configs, ownAccount, "us-east-1", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now) + + require.NotNil(t, result) + assert.Equal(t, 1, result.Planned, "the healthy config must still be planned despite the broken one") + assert.Equal(t, 1, result.Errored, "the broken config must be counted Errored") + assert.Equal(t, 0, result.SkippedDisabled) + assert.Equal(t, 0, result.SkippedMultiAccount) + + // Exactly one run persisted (the healthy config); it must be the healthy one. + require.Len(t, store.savedRuns, 1, "only the healthy config persists a run") + require.NotNil(t, store.savedRuns[0].ConfigID) + assert.Equal(t, "cfg-healthy", *store.savedRuns[0].ConfigID) +} + +// ============================================================ +// buildTrancheDBRows: malformed amount fails loud (no $0 row) +// ============================================================ + +func TestBuildTrancheDBRows_MalformedAmount_Errors(t *testing.T) { + runID := "run-1" + cfgID := "cfg-1" + tranches := []pkgladder.Tranche{ + { + ID: "tr-good", + RunID: runID, + Layer: pkgladder.LayerConvertibleRI, + Term: pkgladder.Term1Year, + PaymentOption: pkgladder.PaymentNoUpfront, + Status: pkgladder.TrancheStatusScheduled, + AmountUSDPerHour: "3/2", + FireAfter: time.Now(), + }, + { + ID: "tr-bad", + RunID: runID, + Layer: pkgladder.LayerConvertibleRI, + Term: pkgladder.Term1Year, + PaymentOption: pkgladder.PaymentNoUpfront, + Status: pkgladder.TrancheStatusScheduled, + AmountUSDPerHour: "not-a-number", + FireAfter: time.Now(), + }, + } + + rows, err := buildTrancheDBRows(tranches, runID, &cfgID) + require.Error(t, err, "a malformed amount must fail loud, not write a $0 tranche") + assert.Contains(t, err.Error(), "tr-bad") + assert.Nil(t, rows, "no rows may be returned when any tranche amount is malformed") +} + +// ============================================================ +// ratToFloat64Ptr: nil stays nil, non-nil converts correctly +// ============================================================ + +func TestRatToFloat64Ptr_Nil(t *testing.T) { + assert.Nil(t, ratToFloat64Ptr(nil), "nil *big.Rat must produce nil *float64") +} + +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. + allocResult, err := pkgladder.Allocate(&pkgladder.AllocationInput{ + Now: time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC), + Baseline: pkgladder.UsageBaseline{ + LowWaterUSDPerHour: nonZeroFloat64(5.0), + StableUSDPerHour: nonZeroFloat64(4.5), + LookbackDays: 30, + Percentile: 5.0, + }, + LayerStates: map[pkgladder.LayerType]pkgladder.LayerState{ + pkgladder.LayerConvertibleRI: { + Layer: pkgladder.LayerConvertibleRI, + ExistingUSDPerHour: nonZeroFloat64(0.0), + ExpiringUSDPerHour: nonZeroFloat64(0.0), + }, + pkgladder.LayerEC2InstanceSP: { + Layer: pkgladder.LayerEC2InstanceSP, + ExistingUSDPerHour: nonZeroFloat64(0.0), + ExpiringUSDPerHour: nonZeroFloat64(0.0), + }, + }, + Layers: []pkgladder.LayerSpec{ + {Type: pkgladder.LayerConvertibleRI, Roles: []pkgladder.LayerRole{pkgladder.RoleBase}}, + {Type: pkgladder.LayerEC2InstanceSP, Roles: []pkgladder.LayerRole{pkgladder.RoleFlex}}, + }, + Config: pkgladder.LadderConfig{ + Scope: pkgladder.Scope{Provider: pkgcommon.ProviderAWS, AccountID: "123456789012"}, + Mode: pkgladder.ModeEmailApproval, + Cadence: pkgladder.CadenceDaily, + TargetCoveragePct: 100.0, + BufferFraction: 0.0, + BaselinePercentile: 5.0, + LookbackDays: 30, + MaxActionsPerRun: 10, + BufferUtilizationThresholdPct: 50.0, + Ramp: pkgladder.RampSchedule{Steps: []pkgladder.RampStep{{AfterDays: 0, Fraction: 1.0}}}, + }, + }) + require.NoError(t, err) + require.NotEmpty(t, allocResult.Allocations, "should have allocations when existing=0 < baseline") + + gap := allocResult.Allocations[0].GapUSDPerHour + require.NotNil(t, gap) + + ptr := ratToFloat64Ptr(gap) + require.NotNil(t, ptr, "non-nil big.Rat must produce non-nil *float64") + assert.Greater(t, *ptr, 0.0, "gap must be positive") +} diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 3f4301321..e37552e89 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -6,6 +6,7 @@ import ( "github.com/LeanerCloud/CUDly/internal/config" "github.com/LeanerCloud/CUDly/internal/database" + "github.com/LeanerCloud/CUDly/pkg/ladder" "github.com/jackc/pgx/v5" ) @@ -329,6 +330,24 @@ func (m *mockConfigStoreForHealth) GetLadderConfig(_ context.Context, _, _ strin func (m *mockConfigStoreForHealth) UpsertLadderConfig(_ context.Context, cfg *config.LadderConfigDB) (*config.LadderConfigDB, error) { return cfg, nil } +func (m *mockConfigStoreForHealth) SaveLadderRun(_ context.Context, run *config.LadderRunDB) (*config.LadderRunDB, error) { + return run, nil +} +func (m *mockConfigStoreForHealth) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { + return run, nil +} +func (m *mockConfigStoreForHealth) GetLadderRun(_ context.Context, _ string) (*config.LadderRunDB, error) { + return nil, nil +} +func (m *mockConfigStoreForHealth) SaveLadderTranches(_ context.Context, _ []config.LadderTrancheDB) error { + return nil +} +func (m *mockConfigStoreForHealth) LatestLadderRunStartedAt(_ context.Context, _ string) (*time.Time, error) { + return nil, nil +} +func (m *mockConfigStoreForHealth) TransitionLadderRunStatus(_ context.Context, _ string, _ []ladder.RunStatus, _ ladder.RunStatus) (*config.LadderRunDB, error) { + return nil, nil +} func (m *mockConfigStoreForHealth) UpdateGlobalConfigAtomic(_ context.Context, apply func(*config.GlobalConfig) error) (*config.GlobalConfig, error) { cfg := &config.GlobalConfig{} if err := apply(cfg); err != nil { diff --git a/providers/aws/ladder/factory.go b/providers/aws/ladder/factory.go new file mode 100644 index 000000000..7a1df3c9e --- /dev/null +++ b/providers/aws/ladder/factory.go @@ -0,0 +1,102 @@ +package ladder + +import ( + "context" + "fmt" + + pkgladder "github.com/LeanerCloud/CUDly/pkg/ladder" + "github.com/LeanerCloud/CUDly/providers/aws/recommendations" + ec2svc "github.com/LeanerCloud/CUDly/providers/aws/services/ec2" +) + +// NewFromAWSConfig constructs an AWSLadder for the given region and accountID. +// All read-side data-source adapters that are not yet wired in PR-2 are replaced +// by no-op stubs: +// - RI and SP listers return empty slices (no existing commitments is safe for +// a plan-only run: the engine correctly shows the full gap as unhedged). +// - RI coverage and utilization sources return empty results. +// - The on-demand series source returns an explicit error so GetUsageBaseline +// fails loud; executeLadderRun returns before persisting anything, so the +// config is counted Errored for the run (no ladder_runs row is written). +// Real wiring arrives in PR-4 (Cost Explorer data source). +// +// The write side (PurchaseLayer / ReshapeBuffer) is not wired here; all writes +// are rejected with errWriteNotWired, which is correct for the plan-only phase. +// +// NewFromAWSConfig matches the LadderCapabilityFactory signature on Application +// so it can be assigned directly: +// +// app.LadderCapabilityFactory = awsladder.NewFromAWSConfig +func NewFromAWSConfig(_ context.Context, region, accountID string) (pkgladder.LadderCapability, error) { + if region == "" { + return nil, fmt.Errorf("awsladder.NewFromAWSConfig: region must not be empty") + } + if accountID == "" { + return nil, fmt.Errorf("awsladder.NewFromAWSConfig: accountID must not be empty") + } + cfg := Config{ + Region: region, + AccountID: accountID, + } + l, err := New( + cfg, + noopRILister{}, + noopSPLister{}, + noopRICoverageSource{}, + noopOnDemandSeriesSource{}, + noopUtilizationSource{}, + nil, // spCoverageSource: wired when PR-4 lands + nil, // spUtilizationSource: wired when PR-4 lands + ) + if err != nil { + return nil, fmt.Errorf("awsladder.NewFromAWSConfig: %w", err) + } + return l, nil +} + +// noopRILister satisfies riLister by reporting no active convertible RIs. +// Returning an empty slice is safe: the engine counts ExistingUSDPerHour for +// RI layers as zero and plans the full gap. +type noopRILister struct{} + +func (noopRILister) ListConvertibleReservedInstances(_ context.Context) ([]ec2svc.ConvertibleRI, error) { + return []ec2svc.ConvertibleRI{}, nil +} + +// noopSPLister satisfies spLister by reporting no active Savings Plans. +// Returning an empty slice is safe for the same reason as noopRILister. +type noopSPLister struct{} + +func (noopSPLister) ListActiveSPs(_ context.Context) ([]ActiveSP, error) { + return []ActiveSP{}, nil +} + +// noopRICoverageSource satisfies riCoverageSource with an empty coverage map. +// The engine sets CoveragePct to nil for all RI pools when no coverage data is +// available, which is treated as "not yet measured" (not "zero coverage"). +type noopRICoverageSource struct{} + +func (noopRICoverageSource) GetRICoverageMap(_ context.Context, _ int, _ []string) (recommendations.PoolCoverageMap, error) { + return recommendations.PoolCoverageMap{}, nil +} + +// noopOnDemandSeriesSource satisfies onDemandSeriesSource by returning an +// explicit error. GetUsageBaseline requires a non-empty daily spend series to +// compute the low-water baseline; without it the engine cannot produce a +// meaningful plan. executeLadderRun returns early on the GetUsageBaseline +// error, so the config is counted Errored and no ladder_runs row is persisted. +// Real wiring (Cost Explorer adapter) arrives in PR-4. +type noopOnDemandSeriesSource struct{} + +func (noopOnDemandSeriesSource) GetOnDemandSeries(_ context.Context, _ string, _ int) ([]float64, error) { + return nil, fmt.Errorf("on-demand series source not yet wired: the Cost Explorer adapter is connected in PR-4; until then plan runs will be recorded as failed") +} + +// noopUtilizationSource satisfies utilizationSource by returning an empty +// slice. The engine treats nil/empty utilization as "not yet measured" and +// leaves UtilizationPct nil for RI layers. +type noopUtilizationSource struct{} + +func (noopUtilizationSource) GetRIUtilization(_ context.Context, _ int) ([]recommendations.RIUtilization, error) { + return []recommendations.RIUtilization{}, nil +}