feat(ladder): stamp ladder_tranches.settled_at when a tranche completes

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
 	}))
 }

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)

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)

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)
 	}
 }

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)

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). 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
+	err := q.QueryRow(ctx, `
+		SELECT COALESCE(SUM(amount_usd_hr), 0)
+		FROM ladder_tranches
+		WHERE config_id = $1
+		  AND status = 'completed'
+		  AND 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
+}

internal/config/store_postgres_ladder_settled_integration_test.go

@@ -0,0 +1,153 @@
+//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"
+)
+
+// TestPostgresStoreDB_LadderSettledAt covers the stamp written by
+// TransitionLadderTrancheTx, the CHECK that rejects an unstamped completed
+// 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) {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	store, pool := settledTestStore(ctx, t)
+	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")
+		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")
+	})
+}
+
+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.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)
+		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 {
+	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)
+}

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}))
 

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; a CHECK requires it on every completed row.
+	SettledAt *time.Time `json:"settled_at,omitempty"`
 
 	// Required fields.
 	ID            string               `json:"id"`

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;

internal/database/postgres/migrations/000117_ladder_tranches_settled_at.up.sql

@@ -0,0 +1,33 @@
+-- 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.
+--
+-- 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 (
+        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);
+    END IF;
+END $$;