From 452115a7b0b7201901056a88480eaf18f71cfb81 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 03:04:31 +0200 Subject: [PATCH] feat(ladder): reconcile fired tranches from execution outcomes (#118 PR-C2) Add the tranche reconciler and the store pieces it needs. Nothing calls it yet (no scheduled task, no executor, flags off), so no tranche can reach fired in a running system. - migration 000114: ladder_tranches.status_reason (additive); the overdue sweep stamps 'expired'; TransitionLadderTranche takes a reason (required for failed and canceled) and gains a Tx variant. - ListFiredLadderTranches and LockLadderTrancheChainTx: lock the tranche row, follow retry_execution_id (depth cap 20, which also stops a cycle), then lock and re-read the tail execution row, the same row LinkRetryExecutionAtomic updates. - ReconcileLadderTranches: failed only when every node of the chain is failed with no_submission (or the tail is canceled/expired with proven predecessors); completed only when the commitment is visible; partially completed, unproven failures and unresolved chains stay fired and keep netting; unresolved over 6h is logged as an error. The proof is read only from failed rows. Refs #118 Co-Authored-By: Claude Sonnet 5.5 --- ...ladder_reconcile_retry_integration_test.go | 133 +++++++++++ internal/config/store_postgres_ladder.go | 2 +- .../config/store_postgres_ladder_lifecycle.go | 26 ++- ...tgres_ladder_lifecycle_integration_test.go | 47 +++- .../store_postgres_ladder_lifecycle_test.go | 15 +- .../config/store_postgres_ladder_reconcile.go | 115 ++++++++++ ...114_ladder_tranches_status_reason.down.sql | 2 + ...00114_ladder_tranches_status_reason.up.sql | 5 + ...0114_ladder_tranches_status_reason_test.go | 53 +++++ internal/purchase/ladder_reconcile.go | 161 ++++++++++++++ .../ladder_reconcile_integration_test.go | 210 ++++++++++++++++++ internal/purchase/ladder_reconcile_test.go | 55 +++++ 12 files changed, 809 insertions(+), 15 deletions(-) create mode 100644 internal/api/handler_ladder_reconcile_retry_integration_test.go create mode 100644 internal/config/store_postgres_ladder_reconcile.go create mode 100644 internal/database/postgres/migrations/000114_ladder_tranches_status_reason.down.sql create mode 100644 internal/database/postgres/migrations/000114_ladder_tranches_status_reason.up.sql create mode 100644 internal/database/postgres/migrations/000114_ladder_tranches_status_reason_test.go create mode 100644 internal/purchase/ladder_reconcile.go create mode 100644 internal/purchase/ladder_reconcile_integration_test.go create mode 100644 internal/purchase/ladder_reconcile_test.go diff --git a/internal/api/handler_ladder_reconcile_retry_integration_test.go b/internal/api/handler_ladder_reconcile_retry_integration_test.go new file mode 100644 index 00000000..eb0d79ee --- /dev/null +++ b/internal/api/handler_ladder_reconcile_retry_integration_test.go @@ -0,0 +1,133 @@ +//go:build integration + +package api + +// Real PostgreSQL: the ladder reconciler against the real Retry write path +// (persistRetrySuccessorTx -> LinkRetryExecutionAtomic), issue #118 PR-C2. + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/purchase" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// FIXTURE: the provider layer listing. +type visibleNever struct{} + +func (visibleNever) CommitmentVisible(context.Context, config.LadderTrancheDB, string) (bool, error) { + return false, nil +} + +// pausedChainStore holds the reconciler's transaction open right after it has +// locked the chain (tranche row + tail execution row), until release is closed. +type pausedChainStore struct { + *config.PostgresStore + locked chan struct{} + release chan struct{} +} + +func (s *pausedChainStore) LockLadderTrancheChainTx(ctx context.Context, tx pgx.Tx, id string) ([]config.LadderChainNode, error) { + chain, err := s.PostgresStore.LockLadderTrancheChainTx(ctx, tx, id) + close(s.locked) + <-s.release + return chain, err +} + +type ladderRetryRig struct { + store *config.PostgresStore + trancheID string + failedExec *config.PurchaseExecution +} + +func newLadderRetryRig(t *testing.T) *ladderRetryRig { + pg, store := setupSubmitDB(t) + ctx := context.Background() + pool := pg.DB.Pool() + acct := uuid.New().String() + _, err := pool.Exec(ctx, `INSERT INTO cloud_accounts (id, name, enabled, provider, external_id, aws_is_org_root, created_at, updated_at) + VALUES ($1, 'c2', true, 'aws', '123456789012', false, now(), now())`, acct) + require.NoError(t, err) + cfg, err := store.UpsertLadderConfig(ctx, &config.LadderConfigDB{CloudAccountID: acct, Provider: "aws", Enabled: true, + Mode: "email_approval", Cadence: "daily", TargetCoverage: 80, BufferFraction: 0.1, BaselinePercentile: 5, LookbackDays: 30, + BufferUtilizationThreshold: 50, MaxActionsPerRun: 5, RampSchedule: []byte(`{"steps":[{"after_days":0,"fraction":1.0}]}`)}) + require.NoError(t, err) + tr := config.LadderTrancheDB{ID: uuid.New().String(), ConfigID: &cfg.ID, LayerType: ladder.LayerComputeSP, + Term: ladder.Term1Year, PaymentOption: ladder.PaymentNoUpfront, Status: ladder.TrancheStatusScheduled, + AmountUSDHr: 5, ScheduledDate: time.Now().Add(-time.Hour)} + require.NoError(t, store.SaveLadderTranches(ctx, []config.LadderTrancheDB{tr})) + var failedID string + _, err = store.ClaimDueLadderTranches(ctx, cfg.ID, time.Now(), 1, + func(ctx context.Context, tx pgx.Tx, _ config.LadderTrancheDB) (string, error) { + insErr := tx.QueryRow(ctx, `INSERT INTO purchase_executions (scheduled_date, status, no_submission, error) + VALUES (NOW(), 'failed', true, '') RETURNING execution_id`).Scan(&failedID) + return failedID, insErr + }) + require.NoError(t, err) + failed, err := store.GetExecutionByID(ctx, failedID) + require.NoError(t, err) + return &ladderRetryRig{store: store, trancheID: tr.ID, failedExec: failed} +} + +func (r *ladderRetryRig) trancheStatus(t *testing.T) string { + var s string + require.NoError(t, r.store.WithTx(context.Background(), func(tx pgx.Tx) error { + return tx.QueryRow(context.Background(), `SELECT status FROM ladder_tranches WHERE id = $1`, r.trancheID).Scan(&s) + })) + return s +} + +func (r *ladderRetryRig) successor() *config.PurchaseExecution { + return &config.PurchaseExecution{Status: "pending", ScheduledDate: time.Now(), RetryAttemptN: 1} +} + +// A retry that links first leaves the tranche fired: the reconciler sees the +// linked tail and does not conclude the chain is proven failed. +func TestLadderReconcile_RetryLinkedFirstKeepsTrancheFired(t *testing.T) { + r := newLadderRetryRig(t) + h := &Handler{config: r.store} + require.NoError(t, h.persistRetrySuccessorTx(context.Background(), r.failedExec, r.successor(), nil)) + + res, err := purchase.ReconcileLadderTranches(context.Background(), r.store, visibleNever{}, time.Now()) + require.NoError(t, err) + assert.Equal(t, 1, res.Unresolved) + assert.Equal(t, "fired", r.trancheStatus(t)) +} + +// While the reconciler holds the tail row, the real Retry write cannot link a +// successor to it: persistRetrySuccessorTx blocks until the reconciler's +// transaction ends. Without the tail lock the retry would link mid-decision. +func TestLadderReconcile_TailLockBlocksRealRetryLink(t *testing.T) { + r := newLadderRetryRig(t) + ctx := context.Background() + paused := &pausedChainStore{PostgresStore: r.store, locked: make(chan struct{}), release: make(chan struct{})} + + reconciled := make(chan error, 1) + go func() { + _, err := purchase.ReconcileLadderTranches(ctx, paused, visibleNever{}, time.Now()) + reconciled <- err + }() + <-paused.locked + + h := &Handler{config: r.store} + retried := make(chan error, 1) + go func() { retried <- h.persistRetrySuccessorTx(ctx, r.failedExec, r.successor(), nil) }() + + select { + case err := <-retried: + t.Fatalf("retry linked while the reconciler held the tail row (err=%v)", err) + case <-time.After(1500 * time.Millisecond): + } + + close(paused.release) + require.NoError(t, <-reconciled) + require.NoError(t, <-retried) + assert.Equal(t, "failed", r.trancheStatus(t), "decided on the chain it locked") +} diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index a6067a56..168536f9 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -332,7 +332,7 @@ func cancelOverdueLadderTranchesTx(ctx context.Context, tx pgx.Tx, configID *str } _, err := tx.Exec(ctx, ` UPDATE ladder_tranches - SET status = $1 + SET status = $1, status_reason = 'expired' WHERE config_id = $2 AND status = $3 AND execution_id IS NULL diff --git a/internal/config/store_postgres_ladder_lifecycle.go b/internal/config/store_postgres_ladder_lifecycle.go index 95fb1832..c2741089 100644 --- a/internal/config/store_postgres_ladder_lifecycle.go +++ b/internal/config/store_postgres_ladder_lifecycle.go @@ -49,21 +49,37 @@ func validateLadderTrancheEdge(from, to ladder.TrancheStatus) error { // failure. An illegal edge is an error. scheduled -> fired is not allowed here // because a claim must create the execution in the same transaction; use // ClaimDueLadderTranches. A scheduled tranche that already has an execution -// cannot be canceled. -func (s *PostgresStore) TransitionLadderTranche(ctx context.Context, id string, from, to ladder.TrancheStatus) (bool, error) { +// cannot be canceled. reason is stored in status_reason and is required for +// the failed and canceled outcomes, so a terminal tranche always says why. +func (s *PostgresStore) TransitionLadderTranche(ctx context.Context, id string, from, to ladder.TrancheStatus, reason string) (bool, error) { + var moved bool + err := s.WithTx(ctx, func(tx pgx.Tx) error { + var txErr error + moved, txErr = s.TransitionLadderTrancheTx(ctx, tx, id, from, to, reason) + return txErr + }) + return moved, err +} + +// TransitionLadderTrancheTx is TransitionLadderTranche inside the caller's +// transaction, for callers that must re-check other rows under lock first. +func (s *PostgresStore) TransitionLadderTrancheTx(ctx context.Context, tx pgx.Tx, id string, from, to ladder.TrancheStatus, reason string) (bool, error) { if err := validateLadderTrancheEdge(from, to); err != nil { return false, err } if from == ladder.TrancheStatusScheduled && to == ladder.TrancheStatusFired { return false, fmt.Errorf("ladder tranche %s: scheduled -> fired must go through ClaimDueLadderTranches", id) } - tag, err := s.db.Exec(ctx, ` + if reason == "" && (to == ladder.TrancheStatusFailed || to == ladder.TrancheStatusCancelled) { + return false, fmt.Errorf("ladder tranche %s: a %s transition needs a status reason", id, to) + } + tag, err := tx.Exec(ctx, ` UPDATE ladder_tranches - SET status = $3 + SET status = $3, status_reason = NULLIF($4, '') WHERE id = $1 AND status = $2 AND ($2 <> 'scheduled' OR execution_id IS NULL) - `, id, string(from), string(to)) + `, id, string(from), string(to), reason) if err != nil { return false, fmt.Errorf("failed to transition ladder_tranche id=%s %s -> %s: %w", id, from, to, err) } diff --git a/internal/config/store_postgres_ladder_lifecycle_integration_test.go b/internal/config/store_postgres_ladder_lifecycle_integration_test.go index a733911a..42bb21ac 100644 --- a/internal/config/store_postgres_ladder_lifecycle_integration_test.go +++ b/internal/config/store_postgres_ladder_lifecycle_integration_test.go @@ -223,10 +223,10 @@ func TestTransitionLadderTranche_EveryLegalEdgeAndCompareAndSet(t *testing.T) { require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{canceled, toCompleted, toFailed})) require.Equal(t, 7.0, inFlight(ctx, t, store, cfgID, asOf)) - ok, err := store.TransitionLadderTranche(ctx, canceled.ID, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled) + ok, err := store.TransitionLadderTranche(ctx, canceled.ID, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled, "test") require.NoError(t, err) assert.True(t, ok) - ok, err = store.TransitionLadderTranche(ctx, canceled.ID, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled) + ok, err = store.TransitionLadderTranche(ctx, canceled.ID, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled, "test") require.NoError(t, err) assert.False(t, ok, "second apply loses the compare-and-set") assert.Equal(t, 6.0, inFlight(ctx, t, store, cfgID, asOf), "canceled releases netting") @@ -235,14 +235,14 @@ func TestTransitionLadderTranche_EveryLegalEdgeAndCompareAndSet(t *testing.T) { require.NoError(t, err) assert.Equal(t, 6.0, inFlight(ctx, t, store, cfgID, asOf), "fired still nets") - ok, err = store.TransitionLadderTranche(ctx, toCompleted.ID, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled) + ok, err = store.TransitionLadderTranche(ctx, toCompleted.ID, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled, "test") require.NoError(t, err) assert.False(t, ok, "a claimed tranche can no longer be canceled from scheduled") - ok, err = store.TransitionLadderTranche(ctx, toCompleted.ID, ladder.TrancheStatusFired, ladder.TrancheStatusCompleted) + ok, err = store.TransitionLadderTranche(ctx, toCompleted.ID, ladder.TrancheStatusFired, ladder.TrancheStatusCompleted, "test") require.NoError(t, err) assert.True(t, ok) assert.Equal(t, 4.0, inFlight(ctx, t, store, cfgID, asOf), "completed releases netting (now counted in E)") - ok, err = store.TransitionLadderTranche(ctx, toFailed.ID, ladder.TrancheStatusFired, ladder.TrancheStatusFailed) + ok, err = store.TransitionLadderTranche(ctx, toFailed.ID, ladder.TrancheStatusFired, ladder.TrancheStatusFailed, "test") require.NoError(t, err) assert.True(t, ok) assert.Equal(t, 0.0, inFlight(ctx, t, store, cfgID, asOf), "failed releases netting") @@ -250,13 +250,13 @@ func TestTransitionLadderTranche_EveryLegalEdgeAndCompareAndSet(t *testing.T) { for _, id := range []string{canceled.ID, toCompleted.ID, toFailed.ID} { for _, to := range allLadderTrancheStatuses { st := trancheStatus(ctx, t, store, id) - ok, err = store.TransitionLadderTranche(ctx, id, st, to) + ok, err = store.TransitionLadderTranche(ctx, id, st, to, "test") assert.Error(t, err, "terminal %s -> %s", st, to) assert.False(t, ok) } } - ok, err = store.TransitionLadderTranche(ctx, uuid.New().String(), ladder.TrancheStatusFired, ladder.TrancheStatusFailed) + ok, err = store.TransitionLadderTranche(ctx, uuid.New().String(), ladder.TrancheStatusFired, ladder.TrancheStatusFailed, "test") require.NoError(t, err) assert.False(t, ok, "unknown id is a lost race, not an error") } @@ -330,7 +330,7 @@ func TestClaimVsCancelAndOverdueSweep_ExactlyOneOutcome(t *testing.T) { }() go func() { defer wg.Done() - cancelWon, cancelErr = store.TransitionLadderTranche(ctx, tr.ID, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled) + cancelWon, cancelErr = store.TransitionLadderTranche(ctx, tr.ID, ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled, "test") }() go func() { defer wg.Done() @@ -356,3 +356,34 @@ func TestClaimVsCancelAndOverdueSweep_ExactlyOneOutcome(t *testing.T) { require.NoError(t, err) } } + +// A transition stores its reason; the overdue sweep stamps 'expired'. +func TestLadderTranche_StatusReasonIsStored(t *testing.T) { + ctx := context.Background() + store := setupLadderStore(ctx, t) + cfgID := seedLadderConfig(ctx, t, store) + now := time.Now().UTC().Truncate(time.Microsecond) + reasonOf := func(id string) *string { + var r *string + require.NoError(t, store.db.QueryRow(ctx, `SELECT status_reason FROM ladder_tranches WHERE id = $1`, id).Scan(&r)) + return r + } + + fired := lifecycleTranche(cfgID, 1, ladder.TrancheStatusFired, now) + overdue := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, now.Add(-LadderFireGrace-time.Hour)) + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{fired, overdue})) + + _, err := store.TransitionLadderTranche(ctx, fired.ID, ladder.TrancheStatusFired, ladder.TrancheStatusFailed, "") + require.Error(t, err, "a failed transition needs a reason") + ok, err := store.TransitionLadderTranche(ctx, fired.ID, ladder.TrancheStatusFired, ladder.TrancheStatusFailed, "no_submission") + require.NoError(t, err) + require.True(t, ok) + require.NotNil(t, reasonOf(fired.ID)) + assert.Equal(t, "no_submission", *reasonOf(fired.ID)) + + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + return cancelOverdueLadderTranchesTx(ctx, tx, &cfgID, now) + })) + require.NotNil(t, reasonOf(overdue.ID)) + assert.Equal(t, "expired", *reasonOf(overdue.ID)) +} diff --git a/internal/config/store_postgres_ladder_lifecycle_test.go b/internal/config/store_postgres_ladder_lifecycle_test.go index 4e02b338..5dc7c1c0 100644 --- a/internal/config/store_postgres_ladder_lifecycle_test.go +++ b/internal/config/store_postgres_ladder_lifecycle_test.go @@ -43,7 +43,7 @@ func TestTransitionLadderTranche_RejectsIllegalEdgesWithoutQuerying(t *testing.T if validateLadderTrancheEdge(from, to) == nil && (from != ladder.TrancheStatusScheduled || to != ladder.TrancheStatusFired) { continue } - applied, err := s.TransitionLadderTranche(context.Background(), "id", from, to) + applied, err := s.TransitionLadderTrancheTx(context.Background(), nil, "id", from, to, "test") assert.Error(t, err, "%s -> %s", from, to) assert.False(t, applied) } @@ -54,3 +54,16 @@ func TestLadderGraceCutoff_IsExactComplementOfNettingBound(t *testing.T) { asOf := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) assert.Equal(t, asOf.Add(-24*time.Hour), ladderGraceCutoff(asOf)) } + +// A failed or canceled transition without a reason is rejected before any SQL. +func TestTransitionLadderTranche_RequiresReasonForFailedAndCanceled(t *testing.T) { + s := &PostgresStore{} + for _, c := range []struct{ from, to ladder.TrancheStatus }{ + {ladder.TrancheStatusFired, ladder.TrancheStatusFailed}, + {ladder.TrancheStatusScheduled, ladder.TrancheStatusCancelled}, + } { + applied, err := s.TransitionLadderTrancheTx(context.Background(), nil, "id", c.from, c.to, "") + assert.ErrorContains(t, err, "needs a status reason", "%s -> %s", c.from, c.to) + assert.False(t, applied) + } +} diff --git a/internal/config/store_postgres_ladder_reconcile.go b/internal/config/store_postgres_ladder_reconcile.go new file mode 100644 index 00000000..87b68ad3 --- /dev/null +++ b/internal/config/store_postgres_ladder_reconcile.go @@ -0,0 +1,115 @@ +package config + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" +) + +// LadderChainMaxDepth caps how many executions the reconciler follows along a +// retry chain (head = the execution the tranche was claimed with). A longer +// chain is returned truncated, which the reconciler treats as unresolved. +const LadderChainMaxDepth = 20 + +// ErrLadderTrancheNotFired means the tranche row is gone or no longer fired +// when the reconciler locked it: another writer won, nothing to do. +var ErrLadderTrancheNotFired = errors.New("ladder tranche is not fired") + +// LadderChainNode is the part of one purchase_executions row the reconciler +// reads. NoSubmission is only meaningful while Status is failed. +type LadderChainNode struct { + ExecutionID string + Status string + NoSubmission bool + RetryExecutionID *string + CreatedAt time.Time +} + +// ListFiredLadderTranches returns fired tranches that have an execution, +// oldest first, for the reconciler to examine one by one. +func (s *PostgresStore) ListFiredLadderTranches(ctx context.Context, limit int) ([]LadderTrancheDB, error) { + if limit <= 0 { + return nil, fmt.Errorf("ListFiredLadderTranches limit must be positive, got %d", limit) + } + 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 + FROM ladder_tranches + WHERE status = 'fired' AND execution_id IS NOT NULL + ORDER BY scheduled_date, id + LIMIT $1 + `, limit) + if err != nil { + return nil, fmt.Errorf("failed to list fired ladder_tranches: %w", err) + } + defer rows.Close() + var out []LadderTrancheDB + 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 { + return nil, fmt.Errorf("failed to scan fired ladder_tranche: %w", err) + } + out = append(out, tr) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to read fired ladder_tranches: %w", err) + } + return out, nil +} + +// LockLadderTrancheChainTx locks a fired tranche, follows retry_execution_id +// from its execution to the tail (at most LadderChainMaxDepth nodes, which also +// stops a retry cycle), then locks the tail execution row and re-reads it. +// LinkRetryExecutionAtomic updates that same row, so once this returns a retry +// cannot link a successor until the caller's transaction ends. The chain is +// returned head first. A chain that was cut short by the depth cap (a cycle hits it) or +// a tail that changed between the walk and the lock ends with a node whose +// RetryExecutionID is set; the caller must treat that as unresolved. +func (s *PostgresStore) LockLadderTrancheChainTx(ctx context.Context, tx pgx.Tx, trancheID string) ([]LadderChainNode, error) { + var headID *string + err := tx.QueryRow(ctx, + `SELECT execution_id FROM ladder_tranches WHERE id = $1 AND status = 'fired' FOR UPDATE`, trancheID).Scan(&headID) + if errors.Is(err, pgx.ErrNoRows) || (err == nil && headID == nil) { + return nil, fmt.Errorf("%w: %s", ErrLadderTrancheNotFired, trancheID) + } + if err != nil { + return nil, fmt.Errorf("lock ladder_tranche %s: %w", trancheID, err) + } + var chain []LadderChainNode + next := headID + for next != nil && len(chain) < LadderChainMaxDepth { + node, nodeErr := readChainNode(ctx, tx, *next, false) + if nodeErr != nil { + return nil, nodeErr + } + chain = append(chain, node) + next = node.RetryExecutionID + } + tail, err := readChainNode(ctx, tx, chain[len(chain)-1].ExecutionID, true) + if err != nil { + return nil, err + } + chain[len(chain)-1] = tail + return chain, nil +} + +func readChainNode(ctx context.Context, tx pgx.Tx, executionID string, forUpdate bool) (LadderChainNode, error) { + q := `SELECT execution_id::text, status, no_submission, retry_execution_id::text, created_at + FROM purchase_executions WHERE execution_id = $1` + if forUpdate { + q += ` FOR UPDATE` + } + var n LadderChainNode + err := tx.QueryRow(ctx, q, executionID).Scan(&n.ExecutionID, &n.Status, &n.NoSubmission, &n.RetryExecutionID, &n.CreatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return n, fmt.Errorf("%w: execution %s", ErrNotFound, executionID) + } + if err != nil { + return n, fmt.Errorf("read execution %s: %w", executionID, err) + } + return n, nil +} diff --git a/internal/database/postgres/migrations/000114_ladder_tranches_status_reason.down.sql b/internal/database/postgres/migrations/000114_ladder_tranches_status_reason.down.sql new file mode 100644 index 00000000..8f0fb306 --- /dev/null +++ b/internal/database/postgres/migrations/000114_ladder_tranches_status_reason.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE ladder_tranches + DROP COLUMN IF EXISTS status_reason; diff --git a/internal/database/postgres/migrations/000114_ladder_tranches_status_reason.up.sql b/internal/database/postgres/migrations/000114_ladder_tranches_status_reason.up.sql new file mode 100644 index 00000000..32d63462 --- /dev/null +++ b/internal/database/postgres/migrations/000114_ladder_tranches_status_reason.up.sql @@ -0,0 +1,5 @@ +-- Migration 000114 (platform#118 PR-C2): why a ladder tranche reached a +-- terminal status ('expired', 'no_submission', 'approval_canceled', ...). +-- Nullable and additive; old code ignores it. Idempotent. +ALTER TABLE ladder_tranches + ADD COLUMN IF NOT EXISTS status_reason TEXT; diff --git a/internal/database/postgres/migrations/000114_ladder_tranches_status_reason_test.go b/internal/database/postgres/migrations/000114_ladder_tranches_status_reason_test.go new file mode 100644 index 00000000..007a3786 --- /dev/null +++ b/internal/database/postgres/migrations/000114_ladder_tranches_status_reason_test.go @@ -0,0 +1,53 @@ +//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/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestMigration_LadderTranchesStatusReason pins migration 000114 against a real +// PostgreSQL: a pre-existing tranche gets a NULL reason, the up statement is +// idempotent, and the down migration drops the column. +func TestMigration_LadderTranchesStatusReason(t *testing.T) { + ctx := context.Background() + migrationsPath := 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, migrationsPath, previousMigrationVersion(t, 114))) + var trancheID string + require.NoError(t, pool.QueryRow(ctx, ` + INSERT INTO ladder_tranches (layer_type, term, payment_option, status, amount_usd_hr, scheduled_date) + VALUES ('compute-sp', '1y', 'no_upfront', 'scheduled', 1, NOW()) + RETURNING id`).Scan(&trancheID)) + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 114)) + var reason *string + require.NoError(t, pool.QueryRow(ctx, `SELECT status_reason FROM ladder_tranches WHERE id = $1`, trancheID).Scan(&reason)) + assert.Nil(t, reason, "an existing tranche has no reason") + + up, err := os.ReadFile(filepath.Join(migrationsPath, "000114_ladder_tranches_status_reason.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, migrationsPath, 113)) + var cols int + require.NoError(t, pool.QueryRow(ctx, ` + SELECT count(*) FROM information_schema.columns + WHERE table_name = 'ladder_tranches' AND column_name = 'status_reason'`).Scan(&cols)) + assert.Equal(t, 0, cols, "down migration must drop the column") +} diff --git a/internal/purchase/ladder_reconcile.go b/internal/purchase/ladder_reconcile.go new file mode 100644 index 00000000..50978324 --- /dev/null +++ b/internal/purchase/ladder_reconcile.go @@ -0,0 +1,161 @@ +package purchase + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/jackc/pgx/v5" +) + +// LadderFiredStaleAfter is how long a fired tranche may stay unresolved before +// the reconciler logs an alert for it. It never changes the tranche: an +// unresolved fired tranche keeps netting, so the ladder cannot over-buy. +const LadderFiredStaleAfter = 6 * time.Hour + +const ladderReconcileBatch = 200 + +// Reasons stored in ladder_tranches.status_reason by the reconciler. +const ( + ladderReasonNoSubmission = "no_submission" + ladderReasonApprovalCanceled = "approval_canceled" + ladderReasonApprovalExpired = "approval_expired" + ladderReasonCommitmentVisible = "commitment_visible" +) + +// CommitmentVisibility reports whether the commitment bought by a completed +// execution already shows up in the provider's layer listing. The production +// adapter lands with the executor; an error or false keeps the tranche fired. +type CommitmentVisibility interface { + CommitmentVisible(ctx context.Context, tranche config.LadderTrancheDB, executionID string) (bool, error) +} + +// LadderReconcileStore is the slice of the Postgres store the reconciler uses. +type LadderReconcileStore interface { + ListFiredLadderTranches(ctx context.Context, limit int) ([]config.LadderTrancheDB, error) + WithTx(ctx context.Context, fn func(tx pgx.Tx) error) error + LockLadderTrancheChainTx(ctx context.Context, tx pgx.Tx, trancheID string) ([]config.LadderChainNode, error) + TransitionLadderTrancheTx(ctx context.Context, tx pgx.Tx, id string, from, to ladder.TrancheStatus, reason string) (bool, error) +} + +// LadderReconcileResult counts what one sweep did. +type LadderReconcileResult struct { + Checked, Completed, Failed, Unresolved, Stale, LostRace int +} + +type ladderVerdict struct { + to ladder.TrancheStatus // empty: leave the tranche fired + reason string + note string // why it stays fired (for the log) +} + +// decideFiredTranche maps a locked retry chain (head first) to a verdict. The +// no_submission flag is read only from nodes whose status is failed. A tranche +// becomes failed only when nothing in the whole chain can have reached the +// provider; anything else stays fired, because fired keeps netting and failed +// does not. +func decideFiredTranche(chain []config.LadderChainNode, visible func(executionID string) (bool, error)) ladderVerdict { + tail := chain[len(chain)-1] + if tail.RetryExecutionID != nil { + return ladderVerdict{note: fmt.Sprintf("retry chain unresolved after %d executions", len(chain))} + } + switch tail.Status { + case config.StatusCompleted: + ok, err := visible(tail.ExecutionID) + if err != nil { + return ladderVerdict{note: fmt.Sprintf("commitment visibility check failed: %v", err)} + } + if !ok { + return ladderVerdict{note: "completed, commitment not yet visible"} + } + return ladderVerdict{to: ladder.TrancheStatusCompleted, reason: ladderReasonCommitmentVisible} + case config.StatusPartiallyCompleted: + return ladderVerdict{note: "partially completed: some recommendations may have been bought"} + case config.StatusFailed: + if !tail.NoSubmission { + return ladderVerdict{note: "failed without a no_submission proof (outcome unknown)"} + } + return failedIfPredecessorsProven(chain, ladderReasonNoSubmission) + case config.StatusCanceled, config.LegacyStatusCanceled: + return failedIfPredecessorsProven(chain, ladderReasonApprovalCanceled) + case config.StatusExpired: + return failedIfPredecessorsProven(chain, ladderReasonApprovalExpired) + default: + return ladderVerdict{note: "execution is " + tail.Status} + } +} + +func failedIfPredecessorsProven(chain []config.LadderChainNode, reason string) ladderVerdict { + for _, n := range chain[:len(chain)-1] { + if n.Status != config.StatusFailed || !n.NoSubmission { + return ladderVerdict{note: fmt.Sprintf("earlier attempt %s (%s) has no no_submission proof", n.ExecutionID, n.Status)} + } + } + return ladderVerdict{to: ladder.TrancheStatusFailed, reason: reason} +} + +// ReconcileLadderTranches moves fired tranches to completed or failed from the +// state of their executions. Each tranche is decided inside one transaction +// that holds the tranche row and the retry-chain tail row, so a concurrent +// Retry cannot link a successor between the decision and the write. Tranches +// it cannot prove stay fired and keep netting; those older than +// LadderFiredStaleAfter are logged as errors. Not scheduled anywhere yet. +func ReconcileLadderTranches(ctx context.Context, store LadderReconcileStore, vis CommitmentVisibility, now time.Time) (LadderReconcileResult, error) { + var res LadderReconcileResult + tranches, err := store.ListFiredLadderTranches(ctx, ladderReconcileBatch) + if err != nil { + return res, err + } + for i := range tranches { + res.Checked++ + if err := reconcileOne(ctx, store, vis, now, tranches[i], &res); err != nil { + return res, err + } + } + return res, nil +} + +func reconcileOne(ctx context.Context, store LadderReconcileStore, vis CommitmentVisibility, now time.Time, tr config.LadderTrancheDB, res *LadderReconcileResult) error { + var verdict ladderVerdict + var moved bool + var head time.Time + err := store.WithTx(ctx, func(tx pgx.Tx) error { + chain, err := store.LockLadderTrancheChainTx(ctx, tx, tr.ID) + if err != nil { + return err + } + head = chain[0].CreatedAt + verdict = decideFiredTranche(chain, func(execID string) (bool, error) { + return vis.CommitmentVisible(ctx, tr, execID) + }) + if verdict.to == "" { + return nil + } + moved, err = store.TransitionLadderTrancheTx(ctx, tx, tr.ID, ladder.TrancheStatusFired, verdict.to, verdict.reason) + return err + }) + switch { + case errors.Is(err, config.ErrLadderTrancheNotFired): + res.LostRace++ + return nil + case err != nil: + return fmt.Errorf("reconcile ladder_tranche %s: %w", tr.ID, err) + case verdict.to == "": + res.Unresolved++ + if now.Sub(head) > LadderFiredStaleAfter { + res.Stale++ + logging.Errorf("ladder tranche %s fired %s ago is unresolved: %s", tr.ID, now.Sub(head).Round(time.Minute), verdict.note) + } + case !moved: + res.LostRace++ + case verdict.to == ladder.TrancheStatusCompleted: + res.Completed++ + default: + res.Failed++ + } + return nil +} diff --git a/internal/purchase/ladder_reconcile_integration_test.go b/internal/purchase/ladder_reconcile_integration_test.go new file mode 100644 index 00000000..bc0bf52a --- /dev/null +++ b/internal/purchase/ladder_reconcile_integration_test.go @@ -0,0 +1,210 @@ +//go:build integration + +package purchase + +// Real PostgreSQL tests for the ladder tranche reconciler (issue #118, PR-C2). +// Executions, tranches and the claim are real rows; only the commitment +// visibility check is a stub (labeled FIXTURE). + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// FIXTURE: stands in for the provider layer listing. +type stubVisibility struct{ visible bool } + +func (s stubVisibility) CommitmentVisible(context.Context, config.LadderTrancheDB, string) (bool, error) { + return s.visible, nil +} + +type ladderRig struct { + ctx context.Context + t *testing.T + store *config.PostgresStore + pool *pgxpool.Pool + cfgID string + planID string +} + +func newLadderRig(t *testing.T) *ladderRig { + ctx := context.Background() + store, pool := rampStepStore(ctx, t) + r := &ladderRig{ctx: ctx, t: t, store: store, pool: pool} + acct := uuid.New().String() + _, err := pool.Exec(ctx, `INSERT INTO cloud_accounts (id, name, enabled, provider, external_id, aws_is_org_root, created_at, updated_at) + VALUES ($1, 'c2', true, 'aws', '123456789012', false, now(), now())`, acct) + require.NoError(t, err) + saved, err := store.UpsertLadderConfig(ctx, &config.LadderConfigDB{ + CloudAccountID: acct, Provider: "aws", Enabled: true, Mode: "email_approval", Cadence: "daily", + TargetCoverage: 80, BufferFraction: 0.1, BaselinePercentile: 5, LookbackDays: 30, + BufferUtilizationThreshold: 50, MaxActionsPerRun: 5, RampSchedule: []byte(`{"steps":[{"after_days":0,"fraction":1.0}]}`)}) + require.NoError(t, err) + r.cfgID = saved.ID + require.NoError(t, pool.QueryRow(ctx, `INSERT INTO purchase_plans (name) VALUES ('c2') RETURNING id`).Scan(&r.planID)) + return r +} + +// provenFailedExecution inserts a failed row carrying the no_submission proof. +func (r *ladderRig) provenFailedExecution() string { + var id string + require.NoError(r.t, r.pool.QueryRow(r.ctx, ` + INSERT INTO purchase_executions (plan_id, scheduled_date, status, no_submission, created_at) + VALUES ($1, NOW(), 'failed', true, NOW()) RETURNING execution_id`, + r.planID).Scan(&id)) + return id +} + +func (r *ladderRig) link(from, to string) { + _, err := r.pool.Exec(r.ctx, `UPDATE purchase_executions SET retry_execution_id = $2 WHERE execution_id = $1`, from, to) + require.NoError(r.t, err) +} + +// firedTranche claims a real scheduled tranche with a real execution row. +func (r *ladderRig) firedTranche(headStatus string, headProof bool, age time.Duration) (trancheID, headExec string) { + tr := config.LadderTrancheDB{ID: uuid.New().String(), ConfigID: &r.cfgID, LayerType: ladder.LayerComputeSP, + Term: ladder.Term1Year, PaymentOption: ladder.PaymentNoUpfront, Status: ladder.TrancheStatusScheduled, + AmountUSDHr: 5, ScheduledDate: time.Now().Add(-time.Hour)} + require.NoError(r.t, r.store.SaveLadderTranches(r.ctx, []config.LadderTrancheDB{tr})) + claimed, err := r.store.ClaimDueLadderTranches(r.ctx, r.cfgID, time.Now(), 10, + func(ctx context.Context, tx pgx.Tx, _ config.LadderTrancheDB) (string, error) { + var id string + err := tx.QueryRow(ctx, `INSERT INTO purchase_executions (plan_id, scheduled_date, status, no_submission, created_at) + VALUES ($1, NOW(), $2, $3, NOW() - $4::interval) RETURNING execution_id`, r.planID, headStatus, headProof, age.String()).Scan(&id) + return id, err + }) + require.NoError(r.t, err) + require.Len(r.t, claimed, 1) + return tr.ID, *claimed[0].ExecutionID +} + +func (r *ladderRig) reconcile(visible bool) LadderReconcileResult { + res, err := ReconcileLadderTranches(r.ctx, r.store, stubVisibility{visible}, time.Now()) + require.NoError(r.t, err) + return res +} + +func (r *ladderRig) state(trancheID string) (status string, reason *string) { + require.NoError(r.t, r.pool.QueryRow(r.ctx, `SELECT status, status_reason FROM ladder_tranches WHERE id = $1`, trancheID).Scan(&status, &reason)) + return +} + +func (r *ladderRig) inFlight() float64 { + v, err := r.store.GetInFlightLadderCommitUSDHr(r.ctx, r.cfgID, time.Now()) + require.NoError(r.t, err) + require.NotNil(r.t, v) + return *v +} + +func TestReconcileLadderTranches_Mappings(t *testing.T) { + type tc struct { + name string + head string + headProof bool + visible bool + wantStatus string + wantReason string + } + cases := []tc{ + {"failed with proof", "failed", true, true, "failed", "no_submission"}, + {"failed without proof (reaper / outcome unknown) keeps netting", "failed", false, true, "fired", ""}, + {"proof flag on a running row is ignored", "running", true, true, "fired", ""}, + {"canceled approval", "canceled", false, true, "failed", "approval_canceled"}, + {"legacy canceled spelling", config.LegacyStatusCanceled, false, true, "failed", "approval_canceled"}, + {"expired approval", "expired", false, true, "failed", "approval_expired"}, + {"partially completed stays fired", "partially_completed", false, true, "fired", ""}, + {"completed and visible", "completed", false, true, "completed", "commitment_visible"}, + {"completed, commitment not visible yet", "completed", false, false, "fired", ""}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + r := newLadderRig(t) + id, _ := r.firedTranche(c.head, c.headProof, time.Minute) + require.InDelta(t, 5, r.inFlight(), 1e-9) + r.reconcile(c.visible) + status, reason := r.state(id) + assert.Equal(t, c.wantStatus, status) + if c.wantReason == "" { + assert.Nil(t, reason) + } else { + require.NotNil(t, reason) + assert.Equal(t, c.wantReason, *reason) + } + if c.wantStatus == "fired" { + assert.InDelta(t, 5, r.inFlight(), 1e-9, "an unresolved tranche must keep netting") + } + if c.wantStatus == "failed" { + assert.InDelta(t, 0, r.inFlight(), 1e-9, "a proven-failed tranche stops netting") + } + r.reconcile(c.visible) // idempotent + again, _ := r.state(id) + assert.Equal(t, c.wantStatus, again) + }) + } +} + +// First attempt was reaped (failed, no proof), then a retry failed before +// submission. The first attempt may have bought, so the tranche stays fired. +func TestReconcileLadderTranches_RetryChainNeedsProofOnEveryNode(t *testing.T) { + r := newLadderRig(t) + id, e1 := r.firedTranche("failed", false, time.Minute) + e2 := r.provenFailedExecution() + e3 := r.provenFailedExecution() + r.link(e1, e2) + r.link(e2, e3) + r.reconcile(true) + status, _ := r.state(id) + assert.Equal(t, "fired", status) + + // With proof on the first node too, the whole chain is proven. + _, err := r.pool.Exec(r.ctx, `UPDATE purchase_executions SET no_submission = true WHERE execution_id = $1`, e1) + require.NoError(t, err) + r.reconcile(true) + status, reason := r.state(id) + assert.Equal(t, "failed", status) + require.NotNil(t, reason) + assert.Equal(t, "no_submission", *reason) +} + +func TestReconcileLadderTranches_CycleAndDepthCapStayFired(t *testing.T) { + r := newLadderRig(t) + cyc, a := r.firedTranche("failed", true, time.Minute) + b := r.provenFailedExecution() + r.link(a, b) + r.link(b, a) + + long, head := r.firedTranche("failed", true, time.Minute) + prev := head + for i := 0; i < config.LadderChainMaxDepth+1; i++ { // one past the cap, every node proven + next := r.provenFailedExecution() + r.link(prev, next) + prev = next + } + r.reconcile(true) + for name, id := range map[string]string{"cycle": cyc, "too long": long} { + status, _ := r.state(id) + assert.Equal(t, "fired", status, name) + } +} + +func TestReconcileLadderTranches_StaleUnresolvedIsAlertedNotChanged(t *testing.T) { + r := newLadderRig(t) + old, _ := r.firedTranche("failed", false, LadderFiredStaleAfter+time.Hour) + fresh, _ := r.firedTranche("failed", false, time.Minute) + res := r.reconcile(true) + assert.Equal(t, 2, res.Unresolved) + assert.Equal(t, 1, res.Stale) + for _, id := range []string{old, fresh} { + status, _ := r.state(id) + assert.Equal(t, "fired", status) + } +} diff --git a/internal/purchase/ladder_reconcile_test.go b/internal/purchase/ladder_reconcile_test.go new file mode 100644 index 00000000..dae1b24c --- /dev/null +++ b/internal/purchase/ladder_reconcile_test.go @@ -0,0 +1,55 @@ +package purchase + +import ( + "errors" + "testing" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" +) + +func chainNode(id, status string, proof bool) config.LadderChainNode { + return config.LadderChainNode{ExecutionID: id, Status: status, NoSubmission: proof} +} + +func visibleAlways(string) (bool, error) { return true, nil } + +// Pure mapping (no DB). The real-database behavior is in +// ladder_reconcile_integration_test.go. +func TestDecideFiredTranche(t *testing.T) { + linked := chainNode("e1", "failed", true) + next := "e2" + linked.RetryExecutionID = &next + cases := []struct { + name string + chain []config.LadderChainNode + visible func(string) (bool, error) + want ladder.TrancheStatus + reason string + }{ + {"failed with proof", []config.LadderChainNode{chainNode("e1", "failed", true)}, visibleAlways, ladder.TrancheStatusFailed, "no_submission"}, + {"failed without proof (reaped or outcome unknown)", []config.LadderChainNode{chainNode("e1", "failed", false)}, visibleAlways, "", ""}, + {"first attempt reaped, later attempts proven", []config.LadderChainNode{chainNode("e1", "failed", false), chainNode("e2", "failed", true), chainNode("e3", "failed", true)}, visibleAlways, "", ""}, + {"every attempt proven", []config.LadderChainNode{chainNode("e1", "failed", true), chainNode("e2", "failed", true)}, visibleAlways, ladder.TrancheStatusFailed, "no_submission"}, + {"canceled tail", []config.LadderChainNode{chainNode("e1", "canceled", false)}, visibleAlways, ladder.TrancheStatusFailed, "approval_canceled"}, + {"legacy spelling tail", []config.LadderChainNode{chainNode("e1", config.LegacyStatusCanceled, false)}, visibleAlways, ladder.TrancheStatusFailed, "approval_canceled"}, + {"expired tail after unproven attempt", []config.LadderChainNode{chainNode("e1", "failed", false), chainNode("e2", "expired", false)}, visibleAlways, "", ""}, + {"proof flag ignored on a non-failed row", []config.LadderChainNode{chainNode("e1", "running", true)}, visibleAlways, "", ""}, + {"partially completed", []config.LadderChainNode{chainNode("e1", "partially_completed", false)}, visibleAlways, "", ""}, + {"completed and visible", []config.LadderChainNode{chainNode("e1", "completed", false)}, visibleAlways, ladder.TrancheStatusCompleted, "commitment_visible"}, + {"completed not visible", []config.LadderChainNode{chainNode("e1", "completed", false)}, func(string) (bool, error) { return false, nil }, "", ""}, + {"completed visibility error", []config.LadderChainNode{chainNode("e1", "completed", false)}, func(string) (bool, error) { return false, errors.New("boom") }, "", ""}, + {"chain cut short (tail still linked)", []config.LadderChainNode{linked}, visibleAlways, "", ""}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + v := decideFiredTranche(c.chain, c.visible) + assert.Equal(t, c.want, v.to) + assert.Equal(t, c.reason, v.reason) + if c.want == "" { + assert.NotEmpty(t, v.note, "a tranche left fired must say why") + } + }) + } +}