feat(ladder): per-config persist lock and PlanAndSaveLadderRun

internal/analytics/collector_test.go

@@ -861,6 +861,10 @@ func (m *mockConfigStore) UpsertLadderConfig(_ context.Context, cfg *config.Ladd
 func (m *mockConfigStore) SaveLadderRun(_ context.Context, run *config.LadderRunDB) (*config.LadderRunDB, error) {
 	return run, nil
 }
+func (m *mockConfigStore) PlanAndSaveLadderRun(_ context.Context, _ string, _ time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) {
+	run, _, err := build(0)
+	return run, err
+}
 func (m *mockConfigStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) {
 	return run, nil
 }

internal/config/interfaces.go

@@ -75,9 +75,13 @@ type StoreInterface interface {
 	// LockPlanProvidersTx takes the transaction-scoped advisory lock of every
 	// named provider, in sorted order, so two plan steps that resolve the same
 	// provider's recommendations cannot both attach the same capacity. Lock
-	// order is advisory lock, then LockPurchasePlanTx; nothing else takes these
-	// locks, so it cannot deadlock. Duplicates are taken once; an empty list
-	// and an empty provider name are errors.
+	// order is advisory lock, then LockPurchasePlanTx. The ladder config lock
+	// (PlanAndSaveLadderRun, ClaimDueLadderTranches) is always taken BEFORE both:
+	// ClaimDueLadderTranches's create callback may take these locks, so they
+	// come after the ladder lock; code holding an execution or tranche row
+	// (retry linker, reconciler) must never take the ladder lock. No cycle.
+	// Duplicates are taken once; an empty list and an empty provider name are
+	// errors.
 	LockPlanProvidersTx(ctx context.Context, tx pgx.Tx, providers []string) error
 	// OccupiedRampStepsInRangeTx returns the steps in [from, to] of planID that
 	// already have a fan-out unit which bought or is still working, ascending.
@@ -628,6 +632,16 @@ type StoreInterface interface {
 	// a non-nil pointer (zero when no scheduled tranches exist) so callers can
 	// pass it directly to AllocationInput.InFlightUSDPerHour. Never returns
 	// nil without an error.
+	// PlanAndSaveLadderRun is the ladder planning persist: in ONE READ COMMITTED
+	// transaction under the config's advisory lock it reads the in-flight
+	// commitment (scheduled rows within LadderFireGrace plus fired rows), calls
+	// build with it, then inserts the run, sweeps overdue tranches and inserts
+	// the new ones (append-only, L5). Overlapping runs for one config serialize:
+	// the second nets the first's committed rows. build must be pure compute
+	// (no I/O); run.StartedAt must equal asOf. The lock is also taken first by
+	// ClaimDueLadderTranches. SaveLadderRunWithTranches below bypasses it and
+	// must not be used for planning.
+	PlanAndSaveLadderRun(ctx context.Context, configID string, asOf time.Time, build LadderPlanBuilder) (*LadderRunDB, error)
 	SaveLadderRun(ctx context.Context, run *LadderRunDB) (*LadderRunDB, error)
 	SaveLadderRunWithTranches(ctx context.Context, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error)
 	GetInFlightLadderCommitUSDHr(ctx context.Context, configID string, asOf time.Time) (*float64, error)

internal/config/store_postgres_ladder.go

@@ -372,18 +372,11 @@ func (s *PostgresStore) SaveLadderRunWithTranches(ctx context.Context, run *Ladd
 	}
 	var result LadderRunDB
 	err := s.WithTx(ctx, func(tx pgx.Tx) error {
-		row := tx.QueryRow(ctx, ladderRunInsertQuery, ladderRunInsertArgs(run, ladderRunPlanJSON(run))...)
-		scanned, err := scanLadderRun(row)
+		saved, err := insertLadderRunWithTranchesTx(ctx, tx, run, tranches)
 		if err != nil {
-			return fmt.Errorf("failed to insert ladder_run id=%s: %w", run.ID, err)
-		}
-		if err := cancelOverdueLadderTranchesTx(ctx, tx, run.ConfigID, run.StartedAt); err != nil {
-			return err
-		}
-		if err := insertLadderTranchesTx(ctx, tx, tranches); err != nil {
 			return err
 		}
-		result = scanned
+		result = *saved
 		return nil
 	})
 	if err != nil {
@@ -392,6 +385,24 @@ func (s *PostgresStore) SaveLadderRunWithTranches(ctx context.Context, run *Ladd
 	return &result, nil
 }
 
+// insertLadderRunWithTranchesTx is the shared write half of
+// SaveLadderRunWithTranches and PlanAndSaveLadderRun: run row, overdue sweep,
+// then the new tranches. run.ID must be set.
+func insertLadderRunWithTranchesTx(ctx context.Context, tx pgx.Tx, run *LadderRunDB, tranches []LadderTrancheDB) (*LadderRunDB, error) {
+	row := tx.QueryRow(ctx, ladderRunInsertQuery, ladderRunInsertArgs(run, ladderRunPlanJSON(run))...)
+	scanned, err := scanLadderRun(row)
+	if err != nil {
+		return nil, fmt.Errorf("failed to insert ladder_run id=%s: %w", run.ID, err)
+	}
+	if err := cancelOverdueLadderTranchesTx(ctx, tx, run.ConfigID, run.StartedAt); err != nil {
+		return nil, err
+	}
+	if err := insertLadderTranchesTx(ctx, tx, tranches); err != nil {
+		return nil, err
+	}
+	return &scanned, nil
+}
+
 // GetInFlightLadderCommitUSDHr returns the total hourly USD commitment already
 // in flight for the given config: the SUM of amount_usd_hr for ladder_tranches
 // rows where config_id=$1 and either
@@ -410,15 +421,9 @@ func (s *PostgresStore) SaveLadderRunWithTranches(ctx context.Context, run *Ladd
 // nil without a non-nil error, so callers can pass it directly to
 // AllocationInput.InFlightUSDPerHour without an extra nil-guard.
 func (s *PostgresStore) GetInFlightLadderCommitUSDHr(ctx context.Context, configID string, asOf time.Time) (*float64, error) {
-	var total float64
-	err := s.db.QueryRow(ctx, `
-		SELECT COALESCE(SUM(amount_usd_hr), 0)
-		FROM ladder_tranches
-		WHERE config_id = $1
-		  AND (status = $2 OR (status = $3 AND scheduled_date > $4))
-	`, configID, string(ladder.TrancheStatusFired), string(ladder.TrancheStatusScheduled), ladderGraceCutoff(asOf)).Scan(&total)
+	total, err := inFlightLadderCommit(ctx, s.db, configID, asOf)
 	if err != nil {
-		return nil, fmt.Errorf("GetInFlightLadderCommitUSDHr config_id=%s: %w", configID, err)
+		return nil, err
 	}
 	return &total, nil
 }

internal/config/store_postgres_ladder_lifecycle.go

@@ -94,7 +94,11 @@ func (s *PostgresStore) TransitionLadderTrancheTx(ctx context.Context, tx pgx.Tx
 // returns an error the whole transaction rolls back and nothing is claimed, so
 // a claim and its execution row can never disagree after a crash. Rows locked
 // by a concurrent claim are skipped (FOR UPDATE SKIP LOCKED), so exactly one
-// caller wins each tranche.
+// caller wins each tranche. The whole claim runs under the config's ladder
+// lock (taken first), including the create callback: with a caller asOf earlier
+// than the planner's, a claim could otherwise fire a row the planner's netting
+// read had already excluded as expired, and the row would be bought AND
+// re-planned. A callback slower than ladderLockTimeout makes planners fail loudly.
 func (s *PostgresStore) ClaimDueLadderTranches(
 	ctx context.Context,
 	configID string,
@@ -106,7 +110,7 @@ func (s *PostgresStore) ClaimDueLadderTranches(
 		return nil, fmt.Errorf("ClaimDueLadderTranches limit must be positive, got %d", limit)
 	}
 	var claimed []LadderTrancheDB
-	err := s.WithTx(ctx, func(tx pgx.Tx) error {
+	err := s.withLadderConfigLock(ctx, configID, func(tx pgx.Tx) error {
 		due, err := selectDueLadderTranchesTx(ctx, tx, configID, asOf, limit)
 		if err != nil {
 			return err

internal/config/store_postgres_ladder_lock.go

@@ -0,0 +1,153 @@
+package config
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/ladder"
+)
+
+// ladderConfigLockPrefix namespaces the per-config ladder advisory lock. It is
+// a distinct text input to the same hashtextextended(text, 0) bigint key space
+// as planStepProviderLockPrefix and the purchase-submit lock; a 64-bit hash
+// collision could only cause a spurious wait, never a correctness break.
+const ladderConfigLockPrefix = "cudly:ladder-config:"
+
+// ladderLockTimeout bounds only the advisory-lock wait so a stuck holder fails
+// loudly instead of queueing every later run. Package var so a test can shrink it.
+var ladderLockTimeout = 30 * time.Second
+
+// LadderPlanBuilder computes a run and its new tranches from the config's
+// in-flight commitment. It runs while the per-config lock is held inside the
+// persist transaction: it must be pure compute (no provider or network I/O).
+type LadderPlanBuilder func(inFlightUSDHr float64) (*LadderRunDB, []LadderTrancheDB, error)
+
+func ladderConfigLockKey(configID string) string { return ladderConfigLockPrefix + configID }
+
+// withLadderConfigLock runs fn in a READ COMMITTED transaction that holds the
+// config's ladder lock. It is the only way planning (PlanAndSaveLadderRun) and
+// claiming (ClaimDueLadderTranches) serialize on one config.
+//
+// Lock order: this lock is taken FIRST. Whatever fn takes afterwards (provider
+// plan-step locks, LockPurchasePlanTx, row locks) comes after it; code holding
+// an execution or tranche row lock must never take this lock.
+//
+// Isolation is set explicitly: under REPEATABLE READ the snapshot is fixed at
+// the first statement, before the wait, so the in-flight read would be stale.
+func (s *PostgresStore) withLadderConfigLock(ctx context.Context, configID string, fn func(tx pgx.Tx) error) error {
+	if configID == "" {
+		return errors.New("ladder config lock: empty config id")
+	}
+	tx, err := s.db.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer func() {
+		// ErrTxClosed after Commit is expected; the deferred call also covers a panic in fn.
+		if rbErr := tx.Rollback(ctx); rbErr != nil && !errors.Is(rbErr, pgx.ErrTxClosed) {
+			log.Printf("ladder config lock %s: rollback: %v", configID, rbErr)
+		}
+	}()
+
+	if err := acquireLadderConfigLock(ctx, tx, configID); err != nil {
+		return err
+	}
+	if err := fn(tx); err != nil {
+		return err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("failed to commit transaction: %w", err)
+	}
+	return nil
+}
+
+// acquireLadderConfigLock must be the first thing run in tx: the isolation
+// level can only be set before any query.
+func acquireLadderConfigLock(ctx context.Context, tx pgx.Tx, configID string) error {
+	if _, err := tx.Exec(ctx, "SET TRANSACTION ISOLATION LEVEL READ COMMITTED"); err != nil {
+		return fmt.Errorf("ladder config lock %s: set isolation: %w", configID, err)
+	}
+	if _, err := tx.Exec(ctx, fmt.Sprintf("SET LOCAL lock_timeout = %d", ladderLockTimeout.Milliseconds())); err != nil {
+		return fmt.Errorf("ladder config lock %s: set lock_timeout: %w", configID, err)
+	}
+	if _, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", ladderConfigLockKey(configID)); err != nil {
+		return fmt.Errorf("ladder config lock %s: %w", configID, err)
+	}
+	// The timeout was for the advisory wait only; do not leak it into fn's statements.
+	if _, err := tx.Exec(ctx, "SET LOCAL lock_timeout TO DEFAULT"); err != nil {
+		return fmt.Errorf("ladder config lock %s: reset lock_timeout: %w", configID, err)
+	}
+	return nil
+}
+
+// PlanAndSaveLadderRun reads the config's in-flight commitment, lets build turn
+// it into a run plus tranches, and persists both, all in ONE transaction under
+// the config's ladder lock. Two overlapping runs for one config therefore
+// cannot both plan the same gap: the second waits, then reads the first's
+// committed tranches as in-flight. build must be pure compute. The run's
+// ConfigID must equal configID and its StartedAt must equal asOf, which bounds
+// both the netting read and the overdue sweep.
+func (s *PostgresStore) PlanAndSaveLadderRun(ctx context.Context, configID string, asOf time.Time, build LadderPlanBuilder) (*LadderRunDB, error) {
+	var result LadderRunDB
+	err := s.withLadderConfigLock(ctx, configID, func(tx pgx.Tx) error {
+		inFlight, err := inFlightLadderCommit(ctx, tx, configID, asOf)
+		if err != nil {
+			return err
+		}
+		run, tranches, err := build(inFlight)
+		if err != nil {
+			return err
+		}
+		if run == nil {
+			return errors.New("ladder plan builder returned a nil run")
+		}
+		if run.ConfigID == nil || *run.ConfigID != configID {
+			return fmt.Errorf("ladder run config_id does not match locked config %s", configID)
+		}
+		// The netting read above and the overdue sweep in the insert must use
+		// the same grace cutoff, or a row could be counted and canceled (or neither).
+		if !run.StartedAt.Equal(asOf) {
+			return fmt.Errorf("ladder run started_at %s does not match asOf %s", run.StartedAt, asOf)
+		}
+		if run.ID == "" {
+			run.ID = uuid.New().String()
+		}
+		saved, err := insertLadderRunWithTranchesTx(ctx, tx, run, tranches)
+		if err != nil {
+			return err
+		}
+		result = *saved
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &result, nil
+}
+
+// queryRower is satisfied by pgx.Tx and the pool connection.
+type queryRower interface {
+	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
+}
+
+// inFlightLadderCommit sums the config's in-flight commitment; see
+// GetInFlightLadderCommitUSDHr for the row classes counted and excluded.
+func inFlightLadderCommit(ctx context.Context, q queryRower, configID string, asOf 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 = $2 OR (status = $3 AND scheduled_date > $4))
+	`, configID, string(ladder.TrancheStatusFired), string(ladder.TrancheStatusScheduled), ladderGraceCutoff(asOf)).Scan(&total)
+	if err != nil {
+		return 0, fmt.Errorf("in-flight ladder commit config_id=%s: %w", configID, err)
+	}
+	return total, nil
+}