From e64c030d0294bfa16f726c07d908325bf64baa56 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 10:58:39 +0200 Subject: [PATCH 1/2] feat(ladder): add per-config persist lock and PlanAndSaveLadderRun - PlanAndSaveLadderRun reads the config's in-flight commitment inside the persist transaction, after taking pg_advisory_xact_lock on hashtextextended('cudly:ladder-config:'+id), so two overlapping runs cannot both plan the same gap. - One shared helper (READ COMMITTED, empty config id rejected, lock_timeout applied to the advisory wait only) is used by the planner and by ClaimDueLadderTranches, so a claim cannot fire a row the planner's netting read had already excluded as expired. - Lock order documented next to LockPlanProvidersTx: ladder lock first. - Additive: the old interface methods stay until the handler switches. Refs #118 Co-Authored-By: Claude Sonnet 5.5 --- internal/analytics/collector_test.go | 4 + internal/config/interfaces.go | 20 +- internal/config/store_postgres_ladder.go | 39 +-- .../config/store_postgres_ladder_lifecycle.go | 8 +- internal/config/store_postgres_ladder_lock.go | 153 ++++++++++++ ...e_postgres_ladder_lock_integration_test.go | 233 ++++++++++++++++++ internal/mocks/stores.go | 21 ++ internal/server/test_helpers_test.go | 4 + 8 files changed, 460 insertions(+), 22 deletions(-) create mode 100644 internal/config/store_postgres_ladder_lock.go create mode 100644 internal/config/store_postgres_ladder_lock_integration_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 1beb9f70..15606677 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -857,6 +857,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 } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 1d8c96f0..ab5e3d19 100644 --- a/internal/config/interfaces.go +++ b/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. @@ -624,6 +628,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) diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index 168536f9..4b1ccd66 100644 --- a/internal/config/store_postgres_ladder.go +++ b/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 } diff --git a/internal/config/store_postgres_ladder_lifecycle.go b/internal/config/store_postgres_ladder_lifecycle.go index c2741089..75dcd47e 100644 --- a/internal/config/store_postgres_ladder_lifecycle.go +++ b/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 diff --git a/internal/config/store_postgres_ladder_lock.go b/internal/config/store_postgres_ladder_lock.go new file mode 100644 index 00000000..77d85f8f --- /dev/null +++ b/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 +} diff --git a/internal/config/store_postgres_ladder_lock_integration_test.go b/internal/config/store_postgres_ladder_lock_integration_test.go new file mode 100644 index 00000000..74bc7892 --- /dev/null +++ b/internal/config/store_postgres_ladder_lock_integration_test.go @@ -0,0 +1,233 @@ +//go:build integration + +package config + +// Real PostgreSQL tests for the per-config ladder lock. Waiting is observed in +// pg_locks (this config's advisory key, granted=false), never with sleeps. + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// ladderLockWaiters counts sessions blocked on this config's ladder lock. The +// bigint advisory key is split across pg_locks classid (high 32 bits) and +// objid (low 32 bits). +func ladderLockWaiters(ctx context.Context, t *testing.T, store *PostgresStore, cfgID string) int { + t.Helper() + var n int + require.NoError(t, store.db.QueryRow(ctx, ` + SELECT count(*) FROM pg_locks l, (SELECT hashtextextended($1, 0) AS k) h + WHERE l.locktype = 'advisory' AND l.objsubid = 1 AND NOT l.granted + AND l.classid = ((h.k >> 32) & 4294967295)::oid + AND l.objid = (h.k & 4294967295)::oid`, ladderConfigLockKey(cfgID)).Scan(&n)) + return n +} + +func waitForLadderWaiters(ctx context.Context, t *testing.T, store *PostgresStore, cfgID string, want int) { + t.Helper() + require.Eventually(t, func() bool { return ladderLockWaiters(ctx, t, store, cfgID) == want }, + 10*time.Second, 20*time.Millisecond, "expected %d waiter(s) on the ladder lock", want) +} + +// lockTestAsOf is the fixed run start shared by lockTestRun and the PlanAndSave +// calls (they must match). +var lockTestAsOf = time.Now().UTC().Truncate(time.Microsecond) + +func lockTestRun(cfgID string) *LadderRunDB { + return &LadderRunDB{ConfigID: &cfgID, StartedAt: lockTestAsOf, Status: ladder.RunStatusPlanned} +} + +// holdLadderLock runs PlanAndSaveLadderRun with a build that blocks until the +// returned release func is called, so the caller owns the lock meanwhile. +func holdLadderLock(ctx context.Context, t *testing.T, store *PostgresStore, cfgID string) (release func() error) { + t.Helper() + entered, gate, done := make(chan struct{}), make(chan struct{}), make(chan error, 1) + go func() { + _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + close(entered) + <-gate + return lockTestRun(cfgID), nil, nil + }) + done <- err + }() + select { + case <-entered: + case <-time.After(10 * time.Second): + t.Fatal("holder never entered build") + } + return func() error { close(gate); return <-done } +} + +func countRuns(ctx context.Context, t *testing.T, store *PostgresStore, cfgID string) int { + t.Helper() + var n int + require.NoError(t, store.db.QueryRow(ctx, `SELECT count(*) FROM ladder_runs WHERE config_id = $1`, cfgID).Scan(&n)) + return n +} + +func TestLadderLock_ClaimWaitsForPlanner(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + cfgID := seedLadderConfigWithExtID(ctx, t, store, "711111111111") + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, time.Now().UTC().Add(-time.Hour))})) + release := holdLadderLock(ctx, t, store, cfgID) + + var claimEntered atomic.Bool + inner := newExecCreator(ctx, t, store, new(atomic.Int64)) + claimDone := make(chan error, 1) + go func() { + _, err := store.ClaimDueLadderTranches(ctx, cfgID, time.Now().UTC(), 1, + func(c context.Context, tx pgx.Tx, tr LadderTrancheDB) (string, error) { + claimEntered.Store(true) + return inner(c, tx, tr) + }) + claimDone <- err + }() + waitForLadderWaiters(ctx, t, store, cfgID, 1) + assert.False(t, claimEntered.Load(), "claim must not reach its callback while the planner holds the lock") + require.NoError(t, release()) + require.NoError(t, <-claimDone) + assert.True(t, claimEntered.Load(), "claim proceeds once the planner releases") +} + +func TestLadderLock_IsPerConfig(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + a := seedLadderConfigWithExtID(ctx, t, store, "721111111111") + b := seedLadderConfigWithExtID(ctx, t, store, "722222222222") + release := holdLadderLock(ctx, t, store, a) + _, err := store.PlanAndSaveLadderRun(ctx, b, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + return lockTestRun(b), nil, nil + }) + require.NoError(t, err, "a held lock on config A must not block config B") + require.NoError(t, release()) +} + +func TestLadderLock_ReleasedOnBuildErrorPanicAndCancel(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + cfgID := seedLadderConfigWithExtID(ctx, t, store, "731111111111") + ok := func(float64) (*LadderRunDB, []LadderTrancheDB, error) { return lockTestRun(cfgID), nil, nil } //nolint:unparam // matches LadderPlanBuilder + plan := func(c context.Context, b LadderPlanBuilder) error { + _, err := store.PlanAndSaveLadderRun(c, cfgID, lockTestAsOf, b) + return err + } + + t.Run("build error", func(t *testing.T) { + err := plan(ctx, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { return nil, nil, errors.New("boom") }) + require.ErrorContains(t, err, "boom") + before := countRuns(ctx, t, store, cfgID) + require.NoError(t, plan(ctx, ok)) + assert.Equal(t, before+1, countRuns(ctx, t, store, cfgID)) + }) + t.Run("build panic", func(t *testing.T) { + assert.Panics(t, func() { + _ = plan(ctx, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { panic("boom") }) + }) + before := countRuns(ctx, t, store, cfgID) + require.NoError(t, plan(ctx, ok), "the lock must be free after a panic") + assert.Equal(t, before+1, countRuns(ctx, t, store, cfgID)) + }) + t.Run("ctx canceled while waiting", func(t *testing.T) { + before := countRuns(ctx, t, store, cfgID) + release := holdLadderLock(ctx, t, store, cfgID) + waitCtx, cancelWait := context.WithCancel(ctx) + errc := make(chan error, 1) + go func() { errc <- plan(waitCtx, ok) }() + waitForLadderWaiters(ctx, t, store, cfgID, 1) + cancelWait() + require.Error(t, <-errc) + waitForLadderWaiters(ctx, t, store, cfgID, 0) + require.NoError(t, release()) + assert.Equal(t, before+1, countRuns(ctx, t, store, cfgID), "only the holder's run exists; the canceled waiter wrote nothing") + require.NoError(t, plan(ctx, ok)) + }) +} + +func TestLadderLock_EmptyConfigIDRejected(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + built := false + _, err := store.PlanAndSaveLadderRun(ctx, "", time.Now().UTC(), func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + built = true + return nil, nil, nil + }) + require.Error(t, err) + assert.False(t, built) + _, err = store.ClaimDueLadderTranches(ctx, "", time.Now().UTC(), 1, nil) + require.Error(t, err) +} + +func TestLadderLock_RejectsNilRunAndStartedAtMismatch(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + cfgID := seedLadderConfigWithExtID(ctx, t, store, "751111111111") + _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { return nil, nil, nil }) + require.ErrorContains(t, err, "nil run") + _, err = store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf.Add(time.Second), func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + return lockTestRun(cfgID), nil, nil + }) + require.ErrorContains(t, err, "does not match asOf") + assert.Equal(t, 0, countRuns(ctx, t, store, cfgID)) +} + +// The lock_timeout bounds only the advisory wait, and the transaction is READ +// COMMITTED even when the database default is REPEATABLE READ. +func TestLadderLock_TimeoutBoundsOnlyTheWaitAndIsolationIsReadCommitted(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + store := setupLadderStore(ctx, t) + cfgID := seedLadderConfigWithExtID(ctx, t, store, "741111111111") + + old := ladderLockTimeout + ladderLockTimeout = 300 * time.Millisecond + defer func() { ladderLockTimeout = old }() + + release := holdLadderLock(ctx, t, store, cfgID) + _, err := store.PlanAndSaveLadderRun(ctx, cfgID, lockTestAsOf, func(float64) (*LadderRunDB, []LadderTrancheDB, error) { + return lockTestRun(cfgID), nil, nil + }) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr, "a stuck holder must fail the waiter loudly") + assert.Equal(t, "55P03", pgErr.Code) + require.NoError(t, release()) + + var dbName string + require.NoError(t, store.db.QueryRow(ctx, `SELECT current_database()`).Scan(&dbName)) + _, err = store.db.Exec(ctx, `ALTER DATABASE `+pgx.Identifier{dbName}.Sanitize()+` SET default_transaction_isolation = 'repeatable read'`) + require.NoError(t, err) + t.Cleanup(func() { + _, _ = store.db.Exec(context.Background(), `ALTER DATABASE `+pgx.Identifier{dbName}.Sanitize()+` RESET default_transaction_isolation`) + }) + pool := store.db.(interface{ Pool() *pgxpool.Pool }).Pool() + pool.Reset() // new connections pick up the database default + + past := time.Now().UTC().Add(-time.Hour) + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, past)})) + inner := newExecCreator(ctx, t, store, new(atomic.Int64)) + var lockTimeout, isolation string + _, err = store.ClaimDueLadderTranches(ctx, cfgID, time.Now().UTC(), 1, func(c context.Context, tx pgx.Tx, tr LadderTrancheDB) (string, error) { + require.NoError(t, tx.QueryRow(c, `SHOW lock_timeout`).Scan(&lockTimeout)) + require.NoError(t, tx.QueryRow(c, `SHOW transaction_isolation`).Scan(&isolation)) + return inner(c, tx, tr) + }) + require.NoError(t, err) + assert.Equal(t, "0", lockTimeout, "the wait timeout must not leak into later statements") + assert.Equal(t, "read committed", isolation) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index c24eccc9..6ef7a685 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1913,6 +1913,27 @@ func (m *MockConfigStore) SaveLadderRun(ctx context.Context, run *config.LadderR return v, args.Error(1) } +// PlanAndSaveLadderRun mocks the PlanAndSaveLadderRun operation. With no +// expectation registered it runs build with zero in-flight and returns the run +// it produced (transaction succeeds), so tests that only care about other +// calls stay green. +func (m *MockConfigStore) PlanAndSaveLadderRun(ctx context.Context, configID string, asOf time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { + m.record("PlanAndSaveLadderRun", ctx, configID, asOf) + if !isExpected(&m.Mock, "PlanAndSaveLadderRun") { + run, _, err := build(0) + return run, err + } + args := m.Called(ctx, configID, asOf) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).(*config.LadderRunDB) + if !ok { + panic(fmt.Sprintf("mock: expected *config.LadderRunDB, got %T", args.Get(0))) + } + return v, args.Error(1) +} + // SaveLadderRunWithTranches mocks the SaveLadderRunWithTranches operation. // Returns the run unchanged (transaction succeeds) when no expectation is // registered, so tests that only care about other calls stay green. diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 30364753..dd4934ab 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -412,6 +412,10 @@ func (m *mockConfigStoreForHealth) UpsertLadderConfig(_ context.Context, cfg *co func (m *mockConfigStoreForHealth) SaveLadderRun(_ context.Context, run *config.LadderRunDB) (*config.LadderRunDB, error) { return run, nil } +func (m *mockConfigStoreForHealth) PlanAndSaveLadderRun(_ context.Context, _ string, _ time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { + run, _, err := build(0) + return run, err +} func (m *mockConfigStoreForHealth) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { return run, nil } From dd870c026e33d7ba3b1c11d9d59acd1306b1103c Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 11:01:06 +0200 Subject: [PATCH 2/2] fix(ladder): plan the gap under the per-config lock - executeLadderRun now hands Allocate/BuildTranches to PlanAndSaveLadderRun, so the in-flight read, the plan and the tranche insert happen in one transaction under the config's lock. Provider reads stay outside the transaction so no lock is held across network I/O. - Remove SaveLadderRunWithTranches and GetInFlightLadderCommitUSDHr from the store interface and mocks; the handler had the only production caller. - Real-Postgres test: two overlapping runs for one config persist a total of the target, not twice it. Refs #118 Co-Authored-By: Claude Sonnet 5.5 --- internal/analytics/collector_test.go | 7 - internal/config/interfaces.go | 29 +-- internal/mocks/stores.go | 39 ---- internal/server/handler_ladder.go | 185 ++++++++---------- .../handler_ladder_lock_integration_test.go | 108 ++++++++++ internal/server/handler_ladder_test.go | 62 +++--- internal/server/test_helpers_test.go | 7 - 7 files changed, 222 insertions(+), 215 deletions(-) create mode 100644 internal/server/handler_ladder_lock_integration_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 15606677..f5d5576f 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -861,13 +861,6 @@ func (m *mockConfigStore) PlanAndSaveLadderRun(_ context.Context, _ string, _ ti run, _, err := build(0) return run, err } -func (m *mockConfigStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { - return run, nil -} -func (m *mockConfigStore) GetInFlightLadderCommitUSDHr(_ context.Context, _ string, _ time.Time) (*float64, error) { - zero := 0.0 - return &zero, nil -} func (m *mockConfigStore) GetLadderRun(_ context.Context, _ string) (*config.LadderRunDB, error) { return nil, nil } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index ab5e3d19..3939c6ae 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -604,30 +604,6 @@ type StoreInterface interface { // current status), so callers can distinguish a race from a hard error. // Statuses are typed ladder.RunStatus so callers cannot pass an arbitrary // string that would never match a stored status. - // SaveLadderRunWithTranches inserts the run row and its tranches in ONE - // transaction: a tranche-insert failure rolls back the run row too, so a - // status=planned run never persists without its tranches (which would let - // the cadence gate suppress the retry for a full window). - // - // L5 append-only: every ladder run persists its new scheduled tranches via - // this method and leaves any existing scheduled tranches untouched. In-flight - // netting (GetInFlightLadderCommitUSDHr) already subtracts existing - // scheduled tranches from the gap, so each run appends exactly the delta - // needed to reach target-E. Appending (never superseding) keeps prior - // tranches' original fire dates, converges to target on drift-up (run N - // tops up the gap the netting leaves), and cannot oscillate. - // - // GetInFlightLadderCommitUSDHr returns the total hourly USD commitment in - // flight for the given config: the sum of amount_usd_hr for tranches with - // status = 'scheduled' and scheduled_date after asOf ONLY (an elapsed - // tranche was never bought while no executor fires it, issue #118). - // Fired/completed tranches are executed - // purchases already reflected in the engine's ExistingUSDPerHour (the - // provider adapters fold payment-pending and active commitments into E), - // so summing them here too would double-count and under-purchase. Returns - // 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 @@ -635,12 +611,9 @@ type StoreInterface interface { // 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. + // ClaimDueLadderTranches. 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) GetLadderRun(ctx context.Context, id string) (*LadderRunDB, error) SaveLadderTranches(ctx context.Context, tranches []LadderTrancheDB) error LatestLadderRunStartedAt(ctx context.Context, configID string) (*time.Time, error) diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 6ef7a685..951e2dc6 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1934,25 +1934,6 @@ func (m *MockConfigStore) PlanAndSaveLadderRun(ctx context.Context, configID str return v, args.Error(1) } -// SaveLadderRunWithTranches mocks the SaveLadderRunWithTranches operation. -// Returns the run unchanged (transaction succeeds) when no expectation is -// registered, so tests that only care about other calls stay green. -func (m *MockConfigStore) SaveLadderRunWithTranches(ctx context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { - m.record("SaveLadderRunWithTranches", ctx, run, tranches) - if !isExpected(&m.Mock, "SaveLadderRunWithTranches") { - return run, nil - } - args := m.Called(ctx, run, tranches) - if args.Get(0) == nil { - return nil, args.Error(1) - } - v, ok := args.Get(0).(*config.LadderRunDB) - if !ok { - panic(fmt.Sprintf("mock: expected *config.LadderRunDB, got %T", args.Get(0))) - } - return v, args.Error(1) -} - // GetLadderRun mocks the GetLadderRun operation. // Returns (nil, nil) when no expectation is registered. func (m *MockConfigStore) GetLadderRun(ctx context.Context, id string) (*config.LadderRunDB, error) { @@ -1999,26 +1980,6 @@ func (m *MockConfigStore) LatestLadderRunStartedAt(ctx context.Context, configID return v, args.Error(1) } -// GetInFlightLadderCommitUSDHr mocks the GetInFlightLadderCommitUSDHr -// operation. Returns a pointer to zero (no in-flight commitment) when no -// expectation is registered, matching the happy-path default. -func (m *MockConfigStore) GetInFlightLadderCommitUSDHr(ctx context.Context, configID string, asOf time.Time) (*float64, error) { - m.record("GetInFlightLadderCommitUSDHr", ctx, configID, asOf) - if !isExpected(&m.Mock, "GetInFlightLadderCommitUSDHr") { - zero := 0.0 - return &zero, nil - } - args := m.Called(ctx, configID, asOf) - if args.Get(0) == nil { - return nil, args.Error(1) - } - v, ok := args.Get(0).(*float64) - if !ok { - panic(fmt.Sprintf("mock: expected *float64, got %T", args.Get(0))) - } - return v, args.Error(1) -} - // TransitionLadderRunStatus mocks the TransitionLadderRunStatus operation. // Returns (nil, nil) when no expectation is registered (CAS race-lost path). func (m *MockConfigStore) TransitionLadderRunStatus(ctx context.Context, id string, fromStatuses []ladder.RunStatus, toStatus ladder.RunStatus) (*config.LadderRunDB, error) { diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go index 83fb2390..b8210f19 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -317,156 +317,137 @@ func (app *Application) executeLadderRun( return fmt.Errorf("GetLayerStates: %w", err) } - // L5: Fetch in-flight commitment BEFORE Allocate so the engine can subtract - // it from the gap. In-flight = the config's scheduled tranches still within - // config.LadderFireGrace plus its fired (unresolved) tranches. Completed - // tranches are excluded because they are executed purchases already counted - // in ExistingUSDPerHour; netting them again would double-subtract. - // Including prior-run scheduled tranches lets - // the engine account for en-route commitment and produce a ~zero gap (Hold) - // when the scheduled ramp already covers the target. - inFlight, err := app.Config.GetInFlightLadderCommitUSDHr(ctx, dbCfg.ID, now) + // L5 netting runs inside the persist transaction (PlanAndSaveLadderRun): + // the store reads the config's in-flight commitment (scheduled tranches + // still within config.LadderFireGrace plus fired, unresolved ones; completed + // ones are already in ExistingUSDPerHour, so netting them would + // double-subtract) under the per-config lock and hands it to the closure, + // so two overlapping runs cannot both plan the same gap. The provider reads + // above stay outside the transaction: no lock is held across network I/O. + in := &ladderRunPlanner{ + dbCfg: dbCfg, engineCfg: engineCfg, baseline: baseline, layerStates: layerStates, + layers: capability.SupportedLayers(), term: term, paymentOpt: paymentOpt, now: now, + } + savedRun, err := app.Config.PlanAndSaveLadderRun(ctx, dbCfg.ID, now, in.plan) if err != nil { - return fmt.Errorf("GetInFlightLadderCommitUSDHr: %w", err) - } - if inFlight == nil { - // The store must return a non-nil value; nil signals an impossible - // query state (e.g. driver bug). Treat as fail-loud. - return fmt.Errorf("GetInFlightLadderCommitUSDHr returned nil for config %s", dbCfg.ID) + return fmt.Errorf("persist ladder run: %w", err) } + log.Printf("ladder_run: config %s: persisted run %s (status=%s, allocations=%d, tranches=%d, holds=%d)", + dbCfg.ID, savedRun.ID, savedRun.Status, in.allocations, in.tranches, in.holds) + return nil +} - supportedLayers := capability.SupportedLayers() +// ladderRunPlanner holds the read-side data of one run and turns the locked +// in-flight figure into the run row and tranche rows to persist. plan is pure +// compute (it runs while the per-config lock is held). +type ladderRunPlanner struct { + dbCfg *config.LadderConfigDB + engineCfg pkgladder.LadderConfig + baseline pkgladder.UsageBaseline + layerStates map[pkgladder.LayerType]pkgladder.LayerState + layers []pkgladder.LayerSpec + term pkgladder.Term + paymentOpt pkgladder.PaymentOption + now time.Time + + allocations, tranches, holds int // set by plan, for the log line +} - // Run Allocate with the in-flight figure so the gap is net of the config's - // prior scheduled (not-yet-fired) tranches. +func (p *ladderRunPlanner) plan(inFlight float64) (*config.LadderRunDB, []config.LadderTrancheDB, error) { allocResult, err := pkgladder.Allocate(&pkgladder.AllocationInput{ - Now: now, - Baseline: baseline, - LayerStates: layerStates, - Layers: supportedLayers, - Config: engineCfg, + Now: p.now, + Baseline: p.baseline, + LayerStates: p.layerStates, + Layers: p.layers, + Config: p.engineCfg, DataSources: []string{dataSourceAWSCostExplorer}, - InFlightUSDPerHour: inFlight, + InFlightUSDPerHour: &inFlight, }) if err != nil { - return fmt.Errorf("allocate: %w", err) + return nil, nil, fmt.Errorf("allocate: %w", err) } - // Run BuildTranches to produce the ramp schedule. runID := uuid.New().String() trancheResult, err := pkgladder.BuildTranches(&pkgladder.TrancheInput{ - Config: &engineCfg, + Config: &p.engineCfg, RunID: runID, - Term: term, - PaymentOption: paymentOpt, + Term: p.term, + PaymentOption: p.paymentOpt, NewID: func() string { return uuid.New().String() }, - Now: now, + Now: p.now, Allocations: allocResult.Allocations, }) if err != nil { - return fmt.Errorf("BuildTranches: %w", err) + return nil, nil, fmt.Errorf("BuildTranches: %w", err) } - // Assemble LadderPlan (Q5: interleaves purchases AND reshapes in one JSON). // The plan is display only: its purchase amounts are the immediate tranches - // that persistLadderRun writes, so it never authorizes more than the rows - // that netting sees. - immediate, err := immediateTrancheAmounts(trancheResult.Tranches, now) + // persisted below, so it never authorizes more than the rows netting sees. + immediate, err := immediateTrancheAmounts(trancheResult.Tranches, p.now) if err != nil { - return err + return nil, nil, err } - plan := assembleLadderPlan(scope, now, allocResult, immediate, layerStates, baseline, engineCfg.TargetCoveragePct, term, paymentOpt) - - // Validate and persist the run row + tranche audit rows. L5 append-only: - // SaveLadderRunWithTranches inserts the run and its NEW scheduled tranches - // and leaves any existing scheduled tranches untouched (see persistLadderRun). - return app.persistLadderRun(ctx, dbCfg, runID, now, plan, baseline, allocResult, trancheResult) + plan := assembleLadderPlan(p.engineCfg.Scope, p.now, allocResult, immediate, p.layerStates, p.baseline, p.engineCfg.TargetCoveragePct, p.term, p.paymentOpt) + p.allocations, p.tranches, p.holds = len(allocResult.Allocations), len(trancheResult.Tranches), len(allocResult.Holds) + return p.buildRows(runID, plan, trancheResult) } -// persistLadderRun validates the assembled plan and writes the ladder_runs row -// followed by its ladder_tranches audit rows. Split from executeLadderRun so -// each function stays within the project's cyclomatic-complexity limit (10). -// A failure at any step returns an error before the next write, so a config -// is never left with a half-persisted run. -func (app *Application) persistLadderRun( - ctx context.Context, - dbCfg *config.LadderConfigDB, - runID string, - now time.Time, - plan *pkgladder.LadderPlan, - baseline pkgladder.UsageBaseline, - allocResult *pkgladder.AllocateResult, - trancheResult *pkgladder.TrancheResult, -) error { +// buildRows validates the assembled plan and builds the ladder_runs row and its +// ladder_tranches rows. L5 APPEND-ONLY: the rows are INSERTED and never +// supersede upcoming tranches (the store only cancels already-overdue ones, +// #544). Netting has already removed the config's existing scheduled tranches +// from the gap, so the rows carry exactly the delta needed to reach target-E: +// - Hold (gap ~0): no tranche rows; only the run row is written and the +// existing scheduled ramp keeps its original fire dates. +// - Drift-up (gap re-opens): the rows are the top-up; the prior generation +// stays intact. +// +// Cancel-and-replace was removed because it fought the netting: on drift-up it +// discarded the amount the netting had subtracted, never converging. +// +// Drift-DOWN: if usage drops after a ramp is planned, the too-large scheduled +// ramp is preserved uncancelled (a Hold). That over-commitment is an accepted, +// separately-tracked follow-up (drift-down shrink), bounded by the low-water +// clamp in Allocate. +func (p *ladderRunPlanner) buildRows(runID string, plan *pkgladder.LadderPlan, trancheResult *pkgladder.TrancheResult) (*config.LadderRunDB, []config.LadderTrancheDB, error) { if err := plan.Validate(); err != nil { - return fmt.Errorf("LadderPlan.Validate: %w", err) + return nil, nil, fmt.Errorf("LadderPlan.Validate: %w", err) } - planJSON, err := marshalLadderPlan(plan) if err != nil { - return fmt.Errorf("marshal plan: %w", err) + return nil, nil, fmt.Errorf("marshal plan: %w", err) } - // Map *big.Rat monetary snapshot to *float64 (nil stays nil). - // These are stored in the run row for dashboards; the authoritative - // detail is in plan_json. - cfgID := dbCfg.ID - modeStr := dbCfg.Mode - cadenceStr := dbCfg.Cadence + // Map *big.Rat monetary snapshot to *float64 (nil stays nil). Stored in the + // run row for dashboards; the authoritative detail is in plan_json. + cfgID := p.dbCfg.ID + modeStr := p.dbCfg.Mode + cadenceStr := p.dbCfg.Cadence runRow := &config.LadderRunDB{ ID: runID, ConfigID: &cfgID, - StartedAt: now, + StartedAt: p.now, Status: pkgladder.RunStatusPlanned, Mode: &modeStr, Cadence: &cadenceStr, Plan: planJSON, - BaselineUSDHr: baseline.LowWaterUSDPerHour, + BaselineUSDHr: p.baseline.LowWaterUSDPerHour, TargetUSDHr: ratToFloat64Ptr(plan.TargetUSDPerHour), ExistingUSDHr: ratToFloat64Ptr(plan.ExistingUSDPerHour), GapUSDHr: ratToFloat64Ptr(plan.GapUSDPerHour), } - // Build the tranche rows (status=scheduled; nothing fires them yet). EVERY - // ramp step is a row, including the immediate one (scheduled_date = run - // start), so in-flight netting sees it. The run ID is known up front - // (runRow.ID), so tranches can reference it before the insert. + // EVERY ramp step is a row (status=scheduled; nothing fires them yet), + // including the immediate one (scheduled_date = run start), so in-flight + // netting sees it. trancheRows, err := buildTrancheDBRows(trancheResult.Tranches, runRow.ID, &cfgID) if err != nil { - return fmt.Errorf("buildTrancheDBRows: %w", err) + return nil, nil, fmt.Errorf("buildTrancheDBRows: %w", err) } // total_hourly_commit is what this run persists, not the allocation gap, so // it always equals the sum of the rows netting will read back. runRow.TotalHourlyCommit = sumTrancheRowAmounts(trancheRows) - - // L5 APPEND-ONLY: always persist via the plain path, which INSERTS the new - // scheduled tranches and never supersedes upcoming ones (it only cancels - // tranches that are already overdue, see #544). In-flight netting - // (GetInFlightLadderCommitUSDHr, subtracted from the gap in Allocate) has - // already removed the config's existing scheduled tranches from the gap, so - // trancheRows carries exactly the delta needed to reach target-E: - // - Hold (gap ~0): trancheRows is empty; only the run row is written and - // the existing scheduled ramp is preserved with its original fire dates. - // - Drift-up (gap re-opens): trancheRows is the top-up; appending it brings - // the total scheduled commitment to target-E while leaving (not re-dating - // or discarding) the prior generation intact. - // Cancel-and-replace was removed because it fought the netting: on drift-up - // it discarded the amount the netting had subtracted, capping coverage below - // target and never converging. - // - // Drift-DOWN note: if usage drops after a ramp is planned, the too-large - // scheduled ramp is preserved uncancelled (a Hold). That over-commitment is - // an accepted, separately-tracked follow-up (drift-down shrink); it is - // bounded by the low-water clamp in Allocate and not handled here. - savedRun, err := app.Config.SaveLadderRunWithTranches(ctx, runRow, trancheRows) - if err != nil { - return fmt.Errorf("persist ladder run: %w", err) - } - - log.Printf("ladder_run: config %s: persisted run %s (status=%s, allocations=%d, tranches=%d, holds=%d)", - dbCfg.ID, savedRun.ID, savedRun.Status, - len(allocResult.Allocations), len(trancheResult.Tranches), len(allocResult.Holds)) - return nil + return runRow, trancheRows, nil } // ladderWithinCadenceWindow reports whether a new run should be skipped because diff --git a/internal/server/handler_ladder_lock_integration_test.go b/internal/server/handler_ladder_lock_integration_test.go new file mode 100644 index 00000000..e3e4f4c2 --- /dev/null +++ b/internal/server/handler_ladder_lock_integration_test.go @@ -0,0 +1,108 @@ +//go:build integration + +package server + +// Real PostgreSQL test for the ladder persist lock: two overlapping +// executeLadderRun calls for one config must plan the gap once. The ladder +// capability is a fixture (labeled FIXTURE in newLadderB1Env); the store, the +// lock and the netting query are real. Waiting is observed in pg_locks. + +import ( + "context" + "sync/atomic" + "testing" + "time" + + pkgladder "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// hookedLadderStore wraps the real store and holds the FIRST plan builder +// inside the locked transaction until released. +type hookedLadderStore struct { + *config.PostgresStore + calls atomic.Int32 + entered chan struct{} // closed when the first builder is running under the lock + second chan struct{} // closed if a second builder runs while the first is held + gate chan struct{} +} + +func (h *hookedLadderStore) PlanAndSaveLadderRun(ctx context.Context, configID string, asOf time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { + return h.PostgresStore.PlanAndSaveLadderRun(ctx, configID, asOf, func(inFlight float64) (*config.LadderRunDB, []config.LadderTrancheDB, error) { + if h.calls.Add(1) == 1 { + close(h.entered) + <-h.gate + } else { + select { + case <-h.second: + default: + close(h.second) + } + } + return build(inFlight) + }) +} + +// waitersOnLadderLock counts sessions blocked on the config's ladder lock key +// (cudly:ladder-config:, split across pg_locks classid/objid). +func waitersOnLadderLock(ctx context.Context, t *testing.T, e *ladderB1Env) int { + t.Helper() + var n int + require.NoError(t, e.conn.QueryRow(ctx, ` + SELECT count(*) FROM pg_locks l, (SELECT hashtextextended('cudly:ladder-config:' || $1::text, 0) AS k) h + WHERE l.locktype = 'advisory' AND l.objsubid = 1 AND NOT l.granted + AND l.classid = ((h.k >> 32) & 4294967295)::oid + AND l.objid = (h.k & 4294967295)::oid`, e.cfg.ID).Scan(&n)) + return n +} + +func TestLadderLock_ConcurrentRunsPlanTheGapOnce(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + conn := ladderB1Conn(ctx, t) + e := newLadderB1Env(ctx, t, conn, ladderB1TwoStep) + hook := &hookedLadderStore{ + PostgresStore: config.NewPostgresStore(conn), + entered: make(chan struct{}), second: make(chan struct{}), gate: make(chan struct{}), + } + e.app.Config = hook + + runOnce := func() chan error { + errc := make(chan error, 1) + go func() { + errc <- e.app.executeLadderRun(ctx, e.cfg, e.cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, e.start) + }() + return errc + } + first := runOnce() + select { + case <-hook.entered: + case <-time.After(20 * time.Second): + t.Fatal("first run never reached the locked plan builder") + } + second := runOnce() + + // The second run must be parked on the lock, not planning alongside the first. + require.Eventually(t, func() bool { + select { + case <-hook.second: + return true // lock failed to serialize: stop waiting + default: + } + return waitersOnLadderLock(ctx, t, e) == 1 + }, 20*time.Second, 20*time.Millisecond) + select { + case <-hook.second: + t.Error("second run planned while the first still held the lock") + default: + } + + close(hook.gate) + require.NoError(t, <-first) + require.NoError(t, <-second) + + // Target is 10: the second run must have netted the first's rows (gap 0). + assert.InDelta(t, 10.0, e.live(ctx, t), 1e-9, "two concurrent runs must not both plan the gap") +} diff --git a/internal/server/handler_ladder_test.go b/internal/server/handler_ladder_test.go index 5285dda6..7774f276 100644 --- a/internal/server/handler_ladder_test.go +++ b/internal/server/handler_ladder_test.go @@ -123,8 +123,8 @@ type ladderTestStore struct { cloudAcctByID map[string]*config.CloudAccount cloudAcctErrByID map[string]error - // L5: configurable in-flight hourly commitment returned by - // GetInFlightLadderCommitUSDHr. Defaults to zero (nil -> 0.0 pointer). + // L5: configurable in-flight hourly commitment handed to the plan builder + // by PlanAndSaveLadderRun. Defaults to zero. inFlightUSDHr *float64 inFlightErr error @@ -187,10 +187,21 @@ func (s *ladderTestStore) SaveLadderTranches(_ context.Context, tranches []confi return nil } -// SaveLadderRunWithTranches models the L5 append-only persist: it captures the -// run and its new tranches and leaves existing tranches untouched (there is no -// supersede path anymore). -func (s *ladderTestStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { +// PlanAndSaveLadderRun models the L5 append-only persist: it hands build the +// configurable in-flight value, then captures the run and its new tranches and +// leaves existing tranches untouched (there is no supersede path anymore). +func (s *ladderTestStore) PlanAndSaveLadderRun(_ context.Context, _ string, _ time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { + if s.inFlightErr != nil { + return nil, s.inFlightErr + } + inFlight := 0.0 + if s.inFlightUSDHr != nil { + inFlight = *s.inFlightUSDHr + } + run, tranches, err := build(inFlight) + if err != nil { + return nil, err + } if s.saveLadderRunErr != nil { return nil, s.saveLadderRunErr } @@ -203,19 +214,6 @@ func (s *ladderTestStore) SaveLadderRunWithTranches(_ context.Context, run *conf return run, nil } -// GetInFlightLadderCommitUSDHr returns the configurable in-flight value. -// When inFlightUSDHr is nil, returns a pointer to 0.0 (no in-flight default). -func (s *ladderTestStore) GetInFlightLadderCommitUSDHr(_ context.Context, _ string, _ time.Time) (*float64, error) { - if s.inFlightErr != nil { - return nil, s.inFlightErr - } - if s.inFlightUSDHr != nil { - return s.inFlightUSDHr, nil - } - zero := 0.0 - return &zero, nil -} - // ============================================================ // Shared test helpers // ============================================================ @@ -1170,9 +1168,8 @@ func TestRatToFloat64Ptr_PositiveValue(t *testing.T) { // returns a FIXED in-flight value), it tracks each persisted tranche's live // status so the L5 netting + append-only interplay can be exercised end-to-end // across multiple sequential runs. It mirrors PostgresStore semantics exactly: -// - GetInFlightLadderCommitUSDHr sums SCHEDULED, not-yet-due tranches only, -// per config. -// - SaveLadderRunWithTranches APPENDS the run + its new tranches and NEVER +// - PlanAndSaveLadderRun nets SCHEDULED, not-yet-due tranches only, per +// config, then APPENDS the run + its new tranches and NEVER // cancels existing scheduled tranches (there is no supersede path). // // The tests below drive the REAL executeLadderRun -> Allocate -> BuildTranches @@ -1209,21 +1206,22 @@ func (s *ladderStatefulStore) scheduledTrancheIDs(configID string) map[string]bo return ids } -func (s *ladderStatefulStore) GetInFlightLadderCommitUSDHr(_ context.Context, configID string, asOf time.Time) (*float64, error) { - total := 0.0 +// PlanAndSaveLadderRun nets scheduled, not-yet-due tranches only, then is +// APPEND-ONLY: it records the run and appends its new tranches without touching +// any existing tranche's status. This mirrors the production store, whose +// supersede/cancel path was removed. +func (s *ladderStatefulStore) PlanAndSaveLadderRun(_ context.Context, configID string, asOf time.Time, build config.LadderPlanBuilder) (*config.LadderRunDB, error) { + inFlight := 0.0 for i := range s.tranches { tr := &s.tranches[i] if tr.ConfigID != nil && *tr.ConfigID == configID && tr.Status == pkgladder.TrancheStatusScheduled && tr.ScheduledDate.After(asOf) { - total += tr.AmountUSDHr + inFlight += tr.AmountUSDHr } } - return &total, nil -} - -// SaveLadderRunWithTranches is APPEND-ONLY: it records the run and appends its -// new tranches without touching any existing tranche's status. This mirrors the -// production store, whose supersede/cancel path was removed. -func (s *ladderStatefulStore) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, tranches []config.LadderTrancheDB) (*config.LadderRunDB, error) { + run, tranches, err := build(inFlight) + if err != nil { + return nil, err + } s.runs = append(s.runs, run) s.tranches = append(s.tranches, tranches...) return run, nil diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index dd4934ab..ef38b45b 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -416,13 +416,6 @@ func (m *mockConfigStoreForHealth) PlanAndSaveLadderRun(_ context.Context, _ str run, _, err := build(0) return run, err } -func (m *mockConfigStoreForHealth) SaveLadderRunWithTranches(_ context.Context, run *config.LadderRunDB, _ []config.LadderTrancheDB) (*config.LadderRunDB, error) { - return run, nil -} -func (m *mockConfigStoreForHealth) GetInFlightLadderCommitUSDHr(_ context.Context, _ string, _ time.Time) (*float64, error) { - zero := 0.0 - return &zero, nil -} func (m *mockConfigStoreForHealth) GetLadderRun(_ context.Context, _ string) (*config.LadderRunDB, error) { return nil, nil }