feat(ladder): reconcile fired tranches from execution outcomes (#118 PR-C2)

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

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

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

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,28 +235,28 @@ 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")
 
 	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))
+}

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