From 873f24fdb2d6623e01106367c58e4a5701702524 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 12:23:49 +0200 Subject: [PATCH 1/2] feat(ladder): stamp ladder_tranches.settled_at when a tranche completes Adds the nullable settled_at column (migration 000117), set to clock_timestamp() in the same UPDATE that moves a tranche to completed (TransitionLadderTrancheTx), a NOT VALID CHECK that every completed row carries it, and settledLadderCommit: the sum of a config's completed tranches settled strictly after an explicit anchor. Legacy completed rows with a NULL stamp are counted whatever the anchor (fail closed); failed and canceled rows are not. Inert: nothing calls settledLadderCommit yet. PR-B (amend ceiling) and PR-C (#843) consume it. Refs #842 #843 #118 Co-Authored-By: Claude Sonnet 5.5 --- internal/config/store_postgres_ladder.go | 5 +- .../config/store_postgres_ladder_lifecycle.go | 7 +- ...tgres_ladder_lifecycle_integration_test.go | 9 +- .../config/store_postgres_ladder_reconcile.go | 4 +- .../config/store_postgres_ladder_settled.go | 28 +++ ...ostgres_ladder_settled_integration_test.go | 189 ++++++++++++++++++ internal/config/store_postgres_ladder_test.go | 6 + internal/config/types.go | 3 + ...000117_ladder_tranches_settled_at.down.sql | 4 + .../000117_ladder_tranches_settled_at.up.sql | 23 +++ 10 files changed, 269 insertions(+), 9 deletions(-) create mode 100644 internal/config/store_postgres_ladder_settled.go create mode 100644 internal/config/store_postgres_ladder_settled_integration_test.go create mode 100644 internal/database/postgres/migrations/000117_ladder_tranches_settled_at.down.sql create mode 100644 internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index 4b1ccd66..2933f801 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -297,8 +297,8 @@ func insertLadderTranchesTx(ctx context.Context, tx pgx.Tx, tranches []LadderTra 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()) + settled_at, created_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, NOW()) `, tr.ID, tr.ConfigID, @@ -310,6 +310,7 @@ func insertLadderTranchesTx(ctx context.Context, tx pgx.Tx, tranches []LadderTra tr.ScheduledDate, string(tr.Status), tr.ExecutionID, + tr.SettledAt, ) if err != nil { return fmt.Errorf("failed to insert ladder_tranche id=%s run_id=%v: %w", tr.ID, tr.RunID, err) diff --git a/internal/config/store_postgres_ladder_lifecycle.go b/internal/config/store_postgres_ladder_lifecycle.go index 75dcd47e..53fb5cde 100644 --- a/internal/config/store_postgres_ladder_lifecycle.go +++ b/internal/config/store_postgres_ladder_lifecycle.go @@ -75,7 +75,8 @@ func (s *PostgresStore) TransitionLadderTrancheTx(ctx context.Context, tx pgx.Tx } tag, err := tx.Exec(ctx, ` UPDATE ladder_tranches - SET status = $3, status_reason = NULLIF($4, '') + SET status = $3, status_reason = NULLIF($4, ''), + settled_at = CASE WHEN $3 = 'completed' THEN clock_timestamp() ELSE settled_at END WHERE id = $1 AND status = $2 AND ($2 <> 'scheduled' OR execution_id IS NULL) @@ -144,7 +145,7 @@ func (s *PostgresStore) ClaimDueLadderTranches( func selectDueLadderTranchesTx(ctx context.Context, tx pgx.Tx, configID string, asOf time.Time, limit int) ([]LadderTrancheDB, error) { rows, err := tx.Query(ctx, ` SELECT id, config_id, run_id, layer_type, amount_usd_hr, term, - payment_option, scheduled_date, status, created_at + payment_option, scheduled_date, status, settled_at, created_at FROM ladder_tranches WHERE config_id = $1 AND status = 'scheduled' @@ -163,7 +164,7 @@ func selectDueLadderTranchesTx(ctx context.Context, tx pgx.Tx, configID string, for rows.Next() { var tr LadderTrancheDB if err := rows.Scan(&tr.ID, &tr.ConfigID, &tr.RunID, &tr.LayerType, &tr.AmountUSDHr, &tr.Term, - &tr.PaymentOption, &tr.ScheduledDate, &tr.Status, &tr.CreatedAt); err != nil { + &tr.PaymentOption, &tr.ScheduledDate, &tr.Status, &tr.SettledAt, &tr.CreatedAt); err != nil { return nil, fmt.Errorf("failed to scan due ladder_tranche: %w", err) } out = append(out, tr) diff --git a/internal/config/store_postgres_ladder_lifecycle_integration_test.go b/internal/config/store_postgres_ladder_lifecycle_integration_test.go index 42bb21ac..b446deab 100644 --- a/internal/config/store_postgres_ladder_lifecycle_integration_test.go +++ b/internal/config/store_postgres_ladder_lifecycle_integration_test.go @@ -22,11 +22,16 @@ import ( ) func lifecycleTranche(cfgID string, amount float64, status ladder.TrancheStatus, at time.Time) LadderTrancheDB { - return LadderTrancheDB{ + tr := LadderTrancheDB{ ID: uuid.New().String(), ConfigID: &cfgID, LayerType: ladder.LayerComputeSP, Term: ladder.Term1Year, PaymentOption: ladder.PaymentNoUpfront, Status: status, AmountUSDHr: amount, ScheduledDate: at, } + if status == ladder.TrancheStatusCompleted { + settled := at + tr.SettledAt = &settled + } + return tr } func trancheStatus(ctx context.Context, t *testing.T, store *PostgresStore, id string) ladder.TrancheStatus { @@ -352,7 +357,7 @@ func TestClaimVsCancelAndOverdueSweep_ExactlyOneOutcome(t *testing.T) { t.Fatalf("iteration %d: unexpected final status %s", i, final) } // Retire the row so the next iteration's limit-1 claim sees only its own tranche. - _, err := store.db.Exec(ctx, `UPDATE ladder_tranches SET status = 'completed' WHERE id = $1`, tr.ID) + _, err := store.db.Exec(ctx, `UPDATE ladder_tranches SET status = 'completed', settled_at = now() WHERE id = $1`, tr.ID) require.NoError(t, err) } } diff --git a/internal/config/store_postgres_ladder_reconcile.go b/internal/config/store_postgres_ladder_reconcile.go index 301d5604..3e28ba39 100644 --- a/internal/config/store_postgres_ladder_reconcile.go +++ b/internal/config/store_postgres_ladder_reconcile.go @@ -37,7 +37,7 @@ func (s *PostgresStore) ListFiredLadderTranches(ctx context.Context, limit int) } rows, err := s.db.Query(ctx, ` SELECT id, config_id, run_id, execution_id, layer_type, amount_usd_hr, term, - payment_option, scheduled_date, status, created_at + payment_option, scheduled_date, status, settled_at, created_at FROM ladder_tranches WHERE status = 'fired' AND execution_id IS NOT NULL ORDER BY scheduled_date, id @@ -51,7 +51,7 @@ func (s *PostgresStore) ListFiredLadderTranches(ctx context.Context, limit int) for rows.Next() { var tr LadderTrancheDB if err := rows.Scan(&tr.ID, &tr.ConfigID, &tr.RunID, &tr.ExecutionID, &tr.LayerType, &tr.AmountUSDHr, &tr.Term, - &tr.PaymentOption, &tr.ScheduledDate, &tr.Status, &tr.CreatedAt); err != nil { + &tr.PaymentOption, &tr.ScheduledDate, &tr.Status, &tr.SettledAt, &tr.CreatedAt); err != nil { return nil, fmt.Errorf("failed to scan fired ladder_tranche: %w", err) } out = append(out, tr) diff --git a/internal/config/store_postgres_ladder_settled.go b/internal/config/store_postgres_ladder_settled.go new file mode 100644 index 00000000..909a4c32 --- /dev/null +++ b/internal/config/store_postgres_ladder_settled.go @@ -0,0 +1,28 @@ +package config + +import ( + "context" + "fmt" + "time" +) + +// settledLadderCommit sums the config's completed tranches that settled after +// anchor (strictly: settled_at == anchor is not counted). A completed row with +// a NULL settled_at predates the column and is counted whatever the anchor, so +// the answer errs toward more commitment, never less. failed and canceled +// rows are never counted. Consumers add it to the in-flight sum when the +// existing commitment they compare against was read at or before anchor. +func settledLadderCommit(ctx context.Context, q queryRower, configID string, anchor time.Time) (float64, error) { + var total float64 + err := q.QueryRow(ctx, ` + SELECT COALESCE(SUM(amount_usd_hr), 0) + FROM ladder_tranches + WHERE config_id = $1 + AND status = 'completed' + AND (settled_at IS NULL OR settled_at > $2) + `, configID, anchor).Scan(&total) + if err != nil { + return 0, fmt.Errorf("settled ladder commit config_id=%s: %w", configID, err) + } + return total, nil +} diff --git a/internal/config/store_postgres_ladder_settled_integration_test.go b/internal/config/store_postgres_ladder_settled_integration_test.go new file mode 100644 index 00000000..c2811e42 --- /dev/null +++ b/internal/config/store_postgres_ladder_settled_integration_test.go @@ -0,0 +1,189 @@ +//go:build integration +// +build integration + +package config + +import ( + "context" + "net/url" + "os" + "strings" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const settledAtMigration = 117 + +// TestPostgresStoreDB_LadderSettledAt runs against a dedicated database named +// pr842a_* (LADDER_SETTLED_TEST_DATABASE_URL). It covers migration 000117 +// (down, up, idempotent re-run, legacy rows keep NULL), the stamp written by +// TransitionLadderTrancheTx, the CHECK that rejects an unstamped completed +// row, and the boundaries of settledLadderCommit. +func TestPostgresStoreDB_LadderSettledAt(t *testing.T) { + dsn := os.Getenv("LADDER_SETTLED_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("set LADDER_SETTLED_TEST_DATABASE_URL to a dedicated pr842a_* test database") + } + parsed, err := url.Parse(dsn) + require.NoError(t, err) + require.True(t, strings.HasPrefix(strings.TrimPrefix(parsed.Path, "/"), "pr842a_"), "database must be dedicated to PR-A verification") + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + pool, err := pgxpool.New(ctx, dsn) + require.NoError(t, err) + defer pool.Close() + path := getTestMigrationsPath() + require.NoError(t, migrations.RunMigrations(ctx, pool, path, "", "")) + _, err = pool.Exec(ctx, `TRUNCATE ladder_tranches, ladder_runs, ladder_configs, cloud_accounts CASCADE`) + require.NoError(t, err, "the dedicated database is reset so the test is re-runnable") + store := &PostgresStore{db: pool} + cfgID := seedLadderConfig(ctx, t, store) + + insert := func(status ladder.TrancheStatus, amount float64, settled *time.Time) string { + t.Helper() + tr := lifecycleTranche(cfgID, amount, status, time.Now().UTC().Add(time.Hour)) + tr.SettledAt = settled + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) + return tr.ID + } + settledAt := func(id string) *time.Time { + t.Helper() + var at *time.Time + require.NoError(t, pool.QueryRow(ctx, `SELECT settled_at FROM ladder_tranches WHERE id = $1`, id).Scan(&at)) + return at + } + + t.Run("transition stamps only completed", func(t *testing.T) { + completed, failed, skipped := insert(ladder.TrancheStatusFired, 1, nil), insert(ladder.TrancheStatusFired, 1, nil), insert(ladder.TrancheStatusScheduled, 1, nil) + before := time.Now().Add(-time.Second) + moved, err := store.TransitionLadderTranche(ctx, completed, ladder.TrancheStatusFired, ladder.TrancheStatusCompleted, "commitment_visible") + require.NoError(t, err) + require.True(t, moved) + after := time.Now().Add(time.Second) + at := settledAt(completed) + require.NotNil(t, at, "completed must be stamped") + assert.True(t, at.After(before) && at.Before(after), "stamp %s must be the transition time", at) + + _, err = store.TransitionLadderTranche(ctx, failed, ladder.TrancheStatusFired, ladder.TrancheStatusFailed, "boom") + require.NoError(t, err) + _, err = store.TransitionLadderTranche(ctx, skipped, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled, "no") + require.NoError(t, err) + assert.Nil(t, settledAt(failed)) + assert.Nil(t, settledAt(skipped)) + }) + + t.Run("a lost compare-and-set leaves the stamp alone", func(t *testing.T) { + id := insert(ladder.TrancheStatusFired, 1, nil) + moved, err := store.TransitionLadderTranche(ctx, id, ladder.TrancheStatusFired, ladder.TrancheStatusCompleted, "x") + require.NoError(t, err) + require.True(t, moved) + first := settledAt(id) + moved, err = store.TransitionLadderTranche(ctx, id, ladder.TrancheStatusFired, ladder.TrancheStatusCompleted, "x") + require.NoError(t, err) + assert.False(t, moved) + assert.True(t, first.Equal(*settledAt(id))) + }) + + t.Run("constraint rejects an unstamped completed row", func(t *testing.T) { + tr := lifecycleTranche(cfgID, 1, ladder.TrancheStatusCompleted, time.Now().UTC()) + tr.SettledAt = nil + err := store.SaveLadderTranches(ctx, []LadderTrancheDB{tr}) + requireCheckViolation(t, err) + fired := insert(ladder.TrancheStatusFired, 1, nil) + _, err = pool.Exec(ctx, `UPDATE ladder_tranches SET status = 'completed' WHERE id = $1`, fired) + requireCheckViolation(t, err) + }) + + t.Run("settledLadderCommit boundaries", func(t *testing.T) { + otherCfg := seedLadderConfigWithExtID(ctx, t, store, "333333333333") + anchor := time.Now().UTC().Truncate(time.Microsecond) + add := func(status ladder.TrancheStatus, amount float64, settled *time.Time) { + tr := lifecycleTranche(otherCfg, amount, status, anchor.Add(time.Hour)) + tr.SettledAt = settled + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) + } + before, equal, after := anchor.Add(-time.Microsecond), anchor, anchor.Add(time.Microsecond) + sum := func() float64 { + v, err := settledLadderCommit(ctx, pool, otherCfg, anchor) + require.NoError(t, err) + return v + } + assert.InDelta(t, 0, sum(), 1e-9, "no rows") + add(ladder.TrancheStatusCompleted, 1, &before) + add(ladder.TrancheStatusCompleted, 2, &equal) + assert.InDelta(t, 0, sum(), 1e-9, "settled at or before the anchor is already in the existing commitment") + add(ladder.TrancheStatusCompleted, 4, &after) + assert.InDelta(t, 4, sum(), 1e-9, "one microsecond after counts") + add(ladder.TrancheStatusFailed, 8, nil) + add(ladder.TrancheStatusCancelled, 16, nil) + add(ladder.TrancheStatusFired, 32, nil) + add(ladder.TrancheStatusScheduled, 64, nil) + assert.InDelta(t, 4, sum(), 1e-9, "only completed rows count") + assert.InDelta(t, 0, mustSettled(ctx, t, pool, cfgID, time.Now().Add(time.Hour)), 1e-9, "other config's completions after the anchor only") + }) + + t.Run("migration down and up", func(t *testing.T) { + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, settledAtMigration-1)) + assert.Equal(t, 0, settledColumns(ctx, t, pool), "down drops the column") + assert.Equal(t, 0, settledConstraints(ctx, t, pool), "down drops the constraint") + + legacy := lifecycleTranche(cfgID, 5, ladder.TrancheStatusCompleted, time.Now().UTC()) + _, err := pool.Exec(ctx, `INSERT INTO ladder_tranches (id, config_id, layer_type, amount_usd_hr, term, payment_option, scheduled_date, status) + VALUES ($1, $2, $3, 5, $4, $5, $6, 'completed')`, legacy.ID, cfgID, string(legacy.LayerType), string(legacy.Term), string(legacy.PaymentOption), legacy.ScheduledDate) + require.NoError(t, err) + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, settledAtMigration)) + assert.Equal(t, 1, settledColumns(ctx, t, pool)) + var validated bool + require.NoError(t, pool.QueryRow(ctx, `SELECT convalidated FROM pg_constraint WHERE conname = 'ladder_tranches_completed_has_settled_at'`).Scan(&validated)) + assert.False(t, validated, "the constraint is NOT VALID so legacy rows stay") + assert.Nil(t, settledAt(legacy.ID), "a legacy completed row keeps NULL") + got, err := settledLadderCommit(ctx, pool, cfgID, time.Now().Add(24*time.Hour)) + require.NoError(t, err) + assert.GreaterOrEqual(t, got, 5.0, "a NULL legacy completed row is counted whatever the anchor") + + up, err := os.ReadFile(getTestMigrationsPath() + "/000117_ladder_tranches_settled_at.up.sql") + require.NoError(t, err) + _, err = pool.Exec(ctx, string(up)) + require.NoError(t, err, "the up statement must be idempotent") + assert.Equal(t, 1, settledConstraints(ctx, t, pool), "a re-run adds no second constraint") + }) +} + +func mustSettled(ctx context.Context, t *testing.T, pool *pgxpool.Pool, cfgID string, anchor time.Time) float64 { + t.Helper() + v, err := settledLadderCommit(ctx, pool, cfgID, anchor) + require.NoError(t, err) + return v +} + +func requireCheckViolation(t *testing.T, err error) { + t.Helper() + require.Error(t, err) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr) + assert.Equal(t, "23514", pgErr.Code) + assert.Equal(t, "ladder_tranches_completed_has_settled_at", pgErr.ConstraintName) +} + +func settledColumns(ctx context.Context, t *testing.T, pool *pgxpool.Pool) int { + t.Helper() + var n int + require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM information_schema.columns WHERE table_name = 'ladder_tranches' AND column_name = 'settled_at'`).Scan(&n)) + return n +} + +func settledConstraints(ctx context.Context, t *testing.T, pool *pgxpool.Pool) int { + t.Helper() + var n int + require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM pg_constraint WHERE conname = 'ladder_tranches_completed_has_settled_at'`).Scan(&n)) + return n +} diff --git a/internal/config/store_postgres_ladder_test.go b/internal/config/store_postgres_ladder_test.go index 1fd4d27f..36c86b6c 100644 --- a/internal/config/store_postgres_ladder_test.go +++ b/internal/config/store_postgres_ladder_test.go @@ -423,6 +423,10 @@ func TestPostgresStore_GetInFlightLadderCommitUSDHr(t *testing.T) { AmountUSDHr: amount, ScheduledDate: scheduledDate, } + if status == ladder.TrancheStatusCompleted { + settled := asOf + tr.SettledAt = &settled + } require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) } insertTranche := func(t *testing.T, cID *string, amount float64, status ladder.TrancheStatus) { @@ -582,6 +586,8 @@ func TestPostgresStore_SaveLadderRunWithTranches_CancelsOverdue(t *testing.T) { `INSERT INTO purchase_executions (plan_id, scheduled_date) VALUES ($1, NOW()) RETURNING execution_id`, planID).Scan(&execID)) fired, completed, linked := newTranche(cfgID, "", start.Add(-24*time.Hour)), newTranche(cfgID, "", start.Add(-24*time.Hour)), newTranche(cfgID, "", start.Add(-24*time.Hour)) fired.RunID, completed.RunID, linked.RunID = nil, nil, nil + settledAt := start + completed.SettledAt = &settledAt fired.Status, completed.Status, linked.ExecutionID = ladder.TrancheStatusFired, ladder.TrancheStatusCompleted, &execID require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{fired, completed, linked})) diff --git a/internal/config/types.go b/internal/config/types.go index 56a83547..73939fb3 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -1317,6 +1317,9 @@ type LadderTrancheDB struct { ConfigID *string `json:"config_id,omitempty"` RunID *string `json:"run_id,omitempty"` ExecutionID *string `json:"execution_id,omitempty"` // references purchase_executions.execution_id + // SettledAt is stamped by TransitionLadderTrancheTx when the tranche + // becomes completed. NULL on a completed row means it predates the column. + SettledAt *time.Time `json:"settled_at,omitempty"` // Required fields. ID string `json:"id"` diff --git a/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.down.sql b/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.down.sql new file mode 100644 index 00000000..92a75549 --- /dev/null +++ b/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.down.sql @@ -0,0 +1,4 @@ +ALTER TABLE ladder_tranches + DROP CONSTRAINT IF EXISTS ladder_tranches_completed_has_settled_at; +ALTER TABLE ladder_tranches + DROP COLUMN IF EXISTS settled_at; diff --git a/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql b/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql new file mode 100644 index 00000000..f4460884 --- /dev/null +++ b/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql @@ -0,0 +1,23 @@ +-- Migration 000117 (platform#842/#843): when a ladder tranche became completed. +-- Netting and the amend ceiling need the completions that landed after a given +-- instant, because a completed tranche leaves the in-flight sum before it shows +-- in the existing commitment the run read. Nullable and additive. +-- TransitionLadderTrancheTx stamps it in the same UPDATE that sets completed. +-- The CHECK is NOT VALID: rows completed before this migration keep a NULL +-- settled_at (consumers count them as settled-after-any-anchor, fail closed), +-- while every new or updated row must carry the stamp. Idempotent. +ALTER TABLE ladder_tranches + ADD COLUMN IF NOT EXISTS settled_at TIMESTAMPTZ; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'ladder_tranches_completed_has_settled_at' + AND conrelid = 'ladder_tranches'::regclass + ) THEN + ALTER TABLE ladder_tranches + ADD CONSTRAINT ladder_tranches_completed_has_settled_at + CHECK (status <> 'completed' OR settled_at IS NOT NULL) NOT VALID; + END IF; +END $$; From 5c3c72f5a26e713f7824d90d9ff88b5355aee0ec Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 12:41:12 +0200 Subject: [PATCH 2/2] fix(ladder): backfill settled_at and enforce the CHECK so legacy completed rows stay deletable A NOT VALID CHECK is still evaluated on every UPDATE of a row, including the ON DELETE SET NULL updates of config_id/run_id/execution_id, so a legacy completed row with NULL settled_at made deleting its account, config or execution fail with 23514. Backfill completed rows with now() (never earlier than the real settle time, so sums over settled_at > anchor err toward counting more), then add the CHECK validated. Drop the now-unreachable NULL branch from settledLadderCommit. Adds a migration test (runs in the CI integration job) covering backfill, unrelated updates and the three deletes on a legacy completed row, the CHECK, idempotence and down/up. Moves the stamp/boundary test onto the shared container helper so it runs in CI too, and asserts config scoping. Fixes the retry-ladder test fixture that wrote completed without a stamp. Refs #842 #843 #118 Co-Authored-By: Claude Sonnet 5.5 --- ...purchases_retry_ladder_integration_test.go | 2 +- .../config/store_postgres_ladder_settled.go | 10 +- ...ostgres_ladder_settled_integration_test.go | 86 ++++--------- internal/config/types.go | 2 +- .../000117_ladder_tranches_settled_at.up.sql | 18 ++- .../000117_ladder_tranches_settled_at_test.go | 113 ++++++++++++++++++ 6 files changed, 159 insertions(+), 72 deletions(-) create mode 100644 internal/database/postgres/migrations/000117_ladder_tranches_settled_at_test.go diff --git a/internal/api/handler_purchases_retry_ladder_integration_test.go b/internal/api/handler_purchases_retry_ladder_integration_test.go index f7c5b7db..96237c75 100644 --- a/internal/api/handler_purchases_retry_ladder_integration_test.go +++ b/internal/api/handler_purchases_retry_ladder_integration_test.go @@ -24,7 +24,7 @@ import ( func (r *ladderRetryRig) setTrancheStatus(t *testing.T, status string) { require.NoError(t, r.store.WithTx(context.Background(), func(tx pgx.Tx) error { - _, err := tx.Exec(context.Background(), `UPDATE ladder_tranches SET status = $2 WHERE id = $1`, r.trancheID, status) + _, err := tx.Exec(context.Background(), `UPDATE ladder_tranches SET status = $2, settled_at = CASE WHEN $2 = 'completed' THEN now() END WHERE id = $1`, r.trancheID, status) return err })) } diff --git a/internal/config/store_postgres_ladder_settled.go b/internal/config/store_postgres_ladder_settled.go index 909a4c32..7b4bf8ee 100644 --- a/internal/config/store_postgres_ladder_settled.go +++ b/internal/config/store_postgres_ladder_settled.go @@ -7,10 +7,10 @@ import ( ) // settledLadderCommit sums the config's completed tranches that settled after -// anchor (strictly: settled_at == anchor is not counted). A completed row with -// a NULL settled_at predates the column and is counted whatever the anchor, so -// the answer errs toward more commitment, never less. failed and canceled -// rows are never counted. Consumers add it to the in-flight sum when the +// anchor (strictly: settled_at == anchor is not counted). Rows completed before +// the column existed were backfilled with the migration time, so they are +// counted for any anchor older than it. failed and canceled rows are +// never counted. Consumers add it to the in-flight sum when the // existing commitment they compare against was read at or before anchor. func settledLadderCommit(ctx context.Context, q queryRower, configID string, anchor time.Time) (float64, error) { var total float64 @@ -19,7 +19,7 @@ func settledLadderCommit(ctx context.Context, q queryRower, configID string, anc FROM ladder_tranches WHERE config_id = $1 AND status = 'completed' - AND (settled_at IS NULL OR settled_at > $2) + AND settled_at > $2 `, configID, anchor).Scan(&total) if err != nil { return 0, fmt.Errorf("settled ladder commit config_id=%s: %w", configID, err) diff --git a/internal/config/store_postgres_ladder_settled_integration_test.go b/internal/config/store_postgres_ladder_settled_integration_test.go index c2811e42..3c73cf14 100644 --- a/internal/config/store_postgres_ladder_settled_integration_test.go +++ b/internal/config/store_postgres_ladder_settled_integration_test.go @@ -19,32 +19,15 @@ import ( "github.com/stretchr/testify/require" ) -const settledAtMigration = 117 - -// TestPostgresStoreDB_LadderSettledAt runs against a dedicated database named -// pr842a_* (LADDER_SETTLED_TEST_DATABASE_URL). It covers migration 000117 -// (down, up, idempotent re-run, legacy rows keep NULL), the stamp written by +// TestPostgresStoreDB_LadderSettledAt covers the stamp written by // TransitionLadderTrancheTx, the CHECK that rejects an unstamped completed -// row, and the boundaries of settledLadderCommit. +// row, and the boundaries of settledLadderCommit. Migration behavior is in +// the migrations package (000117). Set LADDER_SETTLED_TEST_DATABASE_URL to a +// dedicated pr842a_* database to run it without Docker. func TestPostgresStoreDB_LadderSettledAt(t *testing.T) { - dsn := os.Getenv("LADDER_SETTLED_TEST_DATABASE_URL") - if dsn == "" { - t.Skip("set LADDER_SETTLED_TEST_DATABASE_URL to a dedicated pr842a_* test database") - } - parsed, err := url.Parse(dsn) - require.NoError(t, err) - require.True(t, strings.HasPrefix(strings.TrimPrefix(parsed.Path, "/"), "pr842a_"), "database must be dedicated to PR-A verification") - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) defer cancel() - pool, err := pgxpool.New(ctx, dsn) - require.NoError(t, err) - defer pool.Close() - path := getTestMigrationsPath() - require.NoError(t, migrations.RunMigrations(ctx, pool, path, "", "")) - _, err = pool.Exec(ctx, `TRUNCATE ladder_tranches, ladder_runs, ladder_configs, cloud_accounts CASCADE`) - require.NoError(t, err, "the dedicated database is reset so the test is re-runnable") - store := &PostgresStore{db: pool} + store, pool := settledTestStore(ctx, t) cfgID := seedLadderConfig(ctx, t, store) insert := func(status ladder.TrancheStatus, amount float64, settled *time.Time) string { @@ -127,35 +110,30 @@ func TestPostgresStoreDB_LadderSettledAt(t *testing.T) { add(ladder.TrancheStatusFired, 32, nil) add(ladder.TrancheStatusScheduled, 64, nil) assert.InDelta(t, 4, sum(), 1e-9, "only completed rows count") - assert.InDelta(t, 0, mustSettled(ctx, t, pool, cfgID, time.Now().Add(time.Hour)), 1e-9, "other config's completions after the anchor only") + mine := lifecycleTranche(cfgID, 128, ladder.TrancheStatusCompleted, anchor) + mine.SettledAt = &after + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{mine})) + assert.InDelta(t, 4, sum(), 1e-9, "another config's completed rows are not counted") + assert.InDelta(t, 128, mustSettled(ctx, t, pool, cfgID, anchor), 1e-9, "scoped to its own config") }) +} - t.Run("migration down and up", func(t *testing.T) { - require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, settledAtMigration-1)) - assert.Equal(t, 0, settledColumns(ctx, t, pool), "down drops the column") - assert.Equal(t, 0, settledConstraints(ctx, t, pool), "down drops the constraint") - - legacy := lifecycleTranche(cfgID, 5, ladder.TrancheStatusCompleted, time.Now().UTC()) - _, err := pool.Exec(ctx, `INSERT INTO ladder_tranches (id, config_id, layer_type, amount_usd_hr, term, payment_option, scheduled_date, status) - VALUES ($1, $2, $3, 5, $4, $5, $6, 'completed')`, legacy.ID, cfgID, string(legacy.LayerType), string(legacy.Term), string(legacy.PaymentOption), legacy.ScheduledDate) +func settledTestStore(ctx context.Context, t *testing.T) (*PostgresStore, *pgxpool.Pool) { + t.Helper() + if dsn := os.Getenv("LADDER_SETTLED_TEST_DATABASE_URL"); dsn != "" { + parsed, err := url.Parse(dsn) require.NoError(t, err) - - require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, settledAtMigration)) - assert.Equal(t, 1, settledColumns(ctx, t, pool)) - var validated bool - require.NoError(t, pool.QueryRow(ctx, `SELECT convalidated FROM pg_constraint WHERE conname = 'ladder_tranches_completed_has_settled_at'`).Scan(&validated)) - assert.False(t, validated, "the constraint is NOT VALID so legacy rows stay") - assert.Nil(t, settledAt(legacy.ID), "a legacy completed row keeps NULL") - got, err := settledLadderCommit(ctx, pool, cfgID, time.Now().Add(24*time.Hour)) + require.True(t, strings.HasPrefix(strings.TrimPrefix(parsed.Path, "/"), "pr842a_"), "database must be dedicated to PR-A verification") + pool, err := pgxpool.New(ctx, dsn) require.NoError(t, err) - assert.GreaterOrEqual(t, got, 5.0, "a NULL legacy completed row is counted whatever the anchor") - - up, err := os.ReadFile(getTestMigrationsPath() + "/000117_ladder_tranches_settled_at.up.sql") - require.NoError(t, err) - _, err = pool.Exec(ctx, string(up)) - require.NoError(t, err, "the up statement must be idempotent") - assert.Equal(t, 1, settledConstraints(ctx, t, pool), "a re-run adds no second constraint") - }) + t.Cleanup(pool.Close) + require.NoError(t, migrations.RunMigrations(ctx, pool, getTestMigrationsPath(), "", "")) + _, err = pool.Exec(ctx, `TRUNCATE ladder_tranches, ladder_runs, ladder_configs, cloud_accounts CASCADE`) + require.NoError(t, err, "the dedicated database is reset so the test is re-runnable") + return &PostgresStore{db: pool}, pool + } + store := setupLadderStore(ctx, t) + return store, store.db.(interface{ Pool() *pgxpool.Pool }).Pool() } func mustSettled(ctx context.Context, t *testing.T, pool *pgxpool.Pool, cfgID string, anchor time.Time) float64 { @@ -173,17 +151,3 @@ func requireCheckViolation(t *testing.T, err error) { assert.Equal(t, "23514", pgErr.Code) assert.Equal(t, "ladder_tranches_completed_has_settled_at", pgErr.ConstraintName) } - -func settledColumns(ctx context.Context, t *testing.T, pool *pgxpool.Pool) int { - t.Helper() - var n int - require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM information_schema.columns WHERE table_name = 'ladder_tranches' AND column_name = 'settled_at'`).Scan(&n)) - return n -} - -func settledConstraints(ctx context.Context, t *testing.T, pool *pgxpool.Pool) int { - t.Helper() - var n int - require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM pg_constraint WHERE conname = 'ladder_tranches_completed_has_settled_at'`).Scan(&n)) - return n -} diff --git a/internal/config/types.go b/internal/config/types.go index 73939fb3..a34e03b9 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -1318,7 +1318,7 @@ type LadderTrancheDB struct { RunID *string `json:"run_id,omitempty"` ExecutionID *string `json:"execution_id,omitempty"` // references purchase_executions.execution_id // SettledAt is stamped by TransitionLadderTrancheTx when the tranche - // becomes completed. NULL on a completed row means it predates the column. + // becomes completed; a CHECK requires it on every completed row. SettledAt *time.Time `json:"settled_at,omitempty"` // Required fields. diff --git a/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql b/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql index f4460884..12925a89 100644 --- a/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql +++ b/internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql @@ -3,12 +3,22 @@ -- instant, because a completed tranche leaves the in-flight sum before it shows -- in the existing commitment the run read. Nullable and additive. -- TransitionLadderTrancheTx stamps it in the same UPDATE that sets completed. --- The CHECK is NOT VALID: rows completed before this migration keep a NULL --- settled_at (consumers count them as settled-after-any-anchor, fail closed), --- while every new or updated row must carry the stamp. Idempotent. +-- +-- Rows completed before this migration are backfilled with now(). That is never +-- earlier than the real settle time, so any "settled after anchor" sum counts +-- them for every anchor older than this migration: it errs toward counting +-- more commitment, never less. The backfill must come before the CHECK is +-- enforced: a CHECK, even NOT VALID, is re-evaluated on every UPDATE of a row, +-- including the ON DELETE SET NULL updates of config_id/run_id/execution_id, +-- so an unstamped completed row would make deleting its account, config or +-- execution fail. Idempotent. ALTER TABLE ladder_tranches ADD COLUMN IF NOT EXISTS settled_at TIMESTAMPTZ; +UPDATE ladder_tranches +SET settled_at = now() +WHERE status = 'completed' AND settled_at IS NULL; + DO $$ BEGIN IF NOT EXISTS ( @@ -18,6 +28,6 @@ BEGIN ) THEN ALTER TABLE ladder_tranches ADD CONSTRAINT ladder_tranches_completed_has_settled_at - CHECK (status <> 'completed' OR settled_at IS NOT NULL) NOT VALID; + CHECK (status <> 'completed' OR settled_at IS NOT NULL); END IF; END $$; diff --git a/internal/database/postgres/migrations/000117_ladder_tranches_settled_at_test.go b/internal/database/postgres/migrations/000117_ladder_tranches_settled_at_test.go new file mode 100644 index 00000000..72d809d8 --- /dev/null +++ b/internal/database/postgres/migrations/000117_ladder_tranches_settled_at_test.go @@ -0,0 +1,113 @@ +//go:build integration +// +build integration + +package migrations_test + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestMigration_LadderTranchesSettledAt pins migration 000117 against a real +// PostgreSQL: completed rows that predate it are backfilled, the CHECK is +// enforced, deleting the account, config or execution of a legacy completed +// row still works (the ON DELETE SET NULL updates re-evaluate the CHECK), and +// the up statement is idempotent. +func TestMigration_LadderTranchesSettledAt(t *testing.T) { + ctx := context.Background() + path := getMigrationsPath() + container, err := testhelpers.SetupPostgresContainer(ctx, t) + require.NoError(t, err) + defer container.Cleanup(ctx) + pool := container.DB.Pool() + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, previousMigrationVersion(t, 117))) + ids := seedLegacyCompletedTranches(ctx, t, pool) + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 117)) + for name, id := range ids.tranches { + var settled bool + require.NoError(t, pool.QueryRow(ctx, `SELECT settled_at IS NOT NULL FROM ladder_tranches WHERE id = $1`, id).Scan(&settled)) + assert.True(t, settled, "legacy completed row %s must be backfilled", name) + } + var validated bool + require.NoError(t, pool.QueryRow(ctx, `SELECT convalidated FROM pg_constraint WHERE conname = 'ladder_tranches_completed_has_settled_at'`).Scan(&validated)) + assert.True(t, validated) + + _, err = pool.Exec(ctx, `UPDATE ladder_tranches SET status_reason = 'x' WHERE id = $1`, ids.tranches["account"]) + require.NoError(t, err, "an unrelated update of a legacy completed row must succeed") + _, err = pool.Exec(ctx, `DELETE FROM cloud_accounts WHERE id = $1`, ids.account) + require.NoError(t, err, "deleting the account of a legacy completed tranche") + _, err = pool.Exec(ctx, `DELETE FROM ladder_configs WHERE id = $1`, ids.config) + require.NoError(t, err, "deleting the config of a legacy completed tranche") + _, err = pool.Exec(ctx, `DELETE FROM purchase_executions WHERE execution_id = $1`, ids.execution) + require.NoError(t, err, "deleting the execution of a legacy completed tranche") + + _, err = pool.Exec(ctx, `INSERT INTO ladder_tranches (layer_type, amount_usd_hr, term, payment_option, scheduled_date, status) + VALUES ('compute-sp', 1, '1y', 'no_upfront', NOW(), 'completed')`) + require.Error(t, err, "an unstamped completed row must be rejected") + + up, err := os.ReadFile(filepath.Join(path, "000117_ladder_tranches_settled_at.up.sql")) + require.NoError(t, err) + _, err = pool.Exec(ctx, string(up)) + require.NoError(t, err, "the up statement must be idempotent") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, previousMigrationVersion(t, 117))) + var cols, cons int + require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM information_schema.columns WHERE table_name = 'ladder_tranches' AND column_name = 'settled_at'`).Scan(&cols)) + require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM pg_constraint WHERE conname = 'ladder_tranches_completed_has_settled_at'`).Scan(&cons)) + assert.Equal(t, 0, cols+cons, "down drops the column and the constraint") + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 117), "up after down re-backfills completed rows") +} + +type legacyTranches struct { + account, config, execution string + tranches map[string]string +} + +func seedLegacyCompletedTranches(ctx context.Context, t *testing.T, pool *pgxpool.Pool) legacyTranches { + t.Helper() + l := legacyTranches{tranches: map[string]string{}} + account := func(ext string) (acct, cfg string) { + require.NoError(t, pool.QueryRow(ctx, `INSERT INTO cloud_accounts (name, enabled, provider, external_id, aws_is_org_root) + VALUES ($1, true, 'aws', $1, false) RETURNING id`, ext).Scan(&acct)) + require.NoError(t, pool.QueryRow(ctx, `INSERT INTO ladder_configs + (cloud_account_id, provider, mode, cadence, target_coverage, buffer_fraction, baseline_percentile, + lookback_days, buffer_utilization_threshold, max_actions_per_run, ramp_schedule) + VALUES ($1, 'aws', 'auto_approve', 'daily', 80, 0.1, 5, 30, 50, 5, '{"steps":[{"after_days":0,"fraction":1}]}') + RETURNING id`, acct).Scan(&cfg)) + return acct, cfg + } + tranche := func(name, cfg, exec string) { + var cfgArg, execArg any + if cfg != "" { + cfgArg = cfg + } + if exec != "" { + execArg = exec + } + var id string + require.NoError(t, pool.QueryRow(ctx, `INSERT INTO ladder_tranches + (config_id, execution_id, layer_type, amount_usd_hr, term, payment_option, scheduled_date, status) + VALUES ($1, $2, 'compute-sp', 1, '1y', 'no_upfront', NOW(), 'completed') RETURNING id`, cfgArg, execArg).Scan(&id)) + l.tranches[name] = id + } + acct1, cfg1 := account("111111111111") + _, cfg2 := account("222222222222") + l.account, l.config = acct1, cfg2 + tranche("account", cfg1, "") + tranche("config", cfg2, "") + var plan string + require.NoError(t, pool.QueryRow(ctx, `INSERT INTO purchase_plans (name) VALUES ('legacy') RETURNING id`).Scan(&plan)) + require.NoError(t, pool.QueryRow(ctx, `INSERT INTO purchase_executions (plan_id, scheduled_date) VALUES ($1, NOW()) RETURNING execution_id`, plan).Scan(&l.execution)) + tranche("execution", "", l.execution) + return l +}