From 42ae5008f2f8b90760d9a008fa83a0fb9981514a Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 13:07:28 +0200 Subject: [PATCH] fix(ladder): net tranches completed between the E read and the locked in-flight sum PlanAndSaveLadderRun now adds completed tranches settled after ladderSettleAnchor(asOf) to the locked in-flight figure. asOf is captured before the layer-state (E) read, so a tranche completing in between is netted (counted twice if E already has it: a safe under-buy). The anchor is asOf minus ladderSettleSkewMargin to cover app/DB clock skew. Safe only once the visibility adapter (executor PR5a) lands. Closes #843 Co-Authored-By: Claude Sonnet 5.5 --- internal/config/interfaces.go | 9 ++- internal/config/store_postgres_ladder.go | 7 +- internal/config/store_postgres_ladder_lock.go | 43 ++++++++++++- ...es_ladder_settled_plan_integration_test.go | 64 +++++++++++++++++++ internal/server/handler_ladder.go | 12 ++-- ...ndler_ladder_immediate_integration_test.go | 50 ++++++++++++++- 6 files changed, 174 insertions(+), 11 deletions(-) create mode 100644 internal/config/store_postgres_ladder_settled_plan_integration_test.go diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 6ed9bfe6..a951b0de 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -619,8 +619,13 @@ type StoreInterface interface { // 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. + // (no I/O); run.StartedAt must equal asOf. asOf MUST be captured before the + // caller reads the existing commitment (E): tranches completed after asOf + // are added to the in-flight figure, so one that completes between the E + // read and the locked read is netted, never missed (a double count is a + // safe under-buy). This holds only if a tranche is marked completed after + // its commitment is visible through E's source (executor PR5a). The lock is + // also taken first by ClaimDueLadderTranches. PlanAndSaveLadderRun(ctx context.Context, configID string, asOf time.Time, build LadderPlanBuilder) (*LadderRunDB, error) SaveLadderRun(ctx context.Context, run *LadderRunDB) (*LadderRunDB, error) GetLadderRun(ctx context.Context, id string) (*LadderRunDB, error) diff --git a/internal/config/store_postgres_ladder.go b/internal/config/store_postgres_ladder.go index 2933f801..4f6414c8 100644 --- a/internal/config/store_postgres_ladder.go +++ b/internal/config/store_postgres_ladder.go @@ -413,8 +413,11 @@ func insertLadderRunWithTranchesTx(ctx context.Context, tx pgx.Tx, run *LadderRu // pending, in progress, or outcome unknown). Fired nets regardless of age; // only a transition to completed or failed releases it. // -// Excluded: completed (counted in the engine's ExistingUSDPerHour; summing it -// too would double-subtract), failed and cancelled (never bought; counting +// Excluded: completed (counted in the engine's ExistingUSDPerHour, but only if +// that figure was read after the tranche became visible: a caller that read +// ExistingUSDPerHour before this call must also add settledLadderCommit with +// ladderSettleAnchor(asOf), as PlanAndSaveLadderRun does, or a tranche that +// completes in between is netted nowhere), failed and cancelled (never bought; counting // them would silently under-buy), and scheduled rows past the grace window // (cancelled by cancelOverdueLadderTranchesTx with the complementary bound). // diff --git a/internal/config/store_postgres_ladder_lock.go b/internal/config/store_postgres_ladder_lock.go index 77d85f8f..2e7ee207 100644 --- a/internal/config/store_postgres_ladder_lock.go +++ b/internal/config/store_postgres_ladder_lock.go @@ -86,6 +86,19 @@ func acquireLadderConfigLock(ctx context.Context, tx pgx.Tx, configID string) er return nil } +// ladderSettleSkewMargin widens the settled-tranche window backwards because +// settled_at is stamped with the DB clock while asOf comes from the app clock. +// A skew of up to this size cannot make a tranche that settled after the +// caller's existing-commitment read fall outside the window; the cost is a +// bounded double count (a safe under-buy) that heals on the next run. It must +// stay far below the shortest cadence (cadenceDailyMin, 20h), so consecutive +// runs' windows never overlap. +const ladderSettleSkewMargin = time.Minute + +// ladderSettleAnchor is the anchor every caller passes to settledLadderCommit +// for an existing-commitment read taken at or after asOf. +func ladderSettleAnchor(asOf time.Time) time.Time { return asOf.Add(-ladderSettleSkewMargin) } + // 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 @@ -93,10 +106,19 @@ func acquireLadderConfigLock(ctx context.Context, tx pgx.Tx, configID string) er // 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. +// +// asOf MUST be captured before the caller reads the existing commitment (E, +// GetLayerStates), which happens outside this transaction. Tranches that +// completed after asOf are added to the in-flight figure, so one completing +// between the E read and the locked read is netted (counted twice if E already +// has it: a safe under-buy, never a miss). This relies on a tranche being +// marked completed only after its commitment is visible through the same source +// as E, and on visibility never reverting; it is safe only once the visibility +// adapter (executor PR5a) provides that. 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) + inFlight, err := lockedLadderCommit(ctx, tx, configID, asOf) if err != nil { return err } @@ -131,6 +153,25 @@ func (s *PostgresStore) PlanAndSaveLadderRun(ctx context.Context, configID strin return &result, nil } +// lockedLadderCommit is the figure PlanAndSaveLadderRun hands to build: +// in-flight plus tranches settled after the skew-widened anchor. +// +// Order is load-bearing: in-flight FIRST, settled SECOND. Each statement takes +// a fresh READ COMMITTED snapshot, so a tranche fired at the first and +// completed before the second is counted by both (safe); the reverse order +// could miss one completing between the two. +func lockedLadderCommit(ctx context.Context, q queryRower, configID string, asOf time.Time) (float64, error) { + inFlight, err := inFlightLadderCommit(ctx, q, configID, asOf) + if err != nil { + return 0, err + } + settled, err := settledLadderCommit(ctx, q, configID, ladderSettleAnchor(asOf)) + if err != nil { + return 0, err + } + return inFlight + settled, nil +} + // queryRower is satisfied by pgx.Tx and the pool connection. type queryRower interface { QueryRow(ctx context.Context, sql string, args ...any) pgx.Row diff --git a/internal/config/store_postgres_ladder_settled_plan_integration_test.go b/internal/config/store_postgres_ladder_settled_plan_integration_test.go new file mode 100644 index 00000000..01cdd267 --- /dev/null +++ b/internal/config/store_postgres_ladder_settled_plan_integration_test.go @@ -0,0 +1,64 @@ +//go:build integration +// +build integration + +package config + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestPostgresStoreDB_PlanAndSaveSettledTerm checks what build receives from +// PlanAndSaveLadderRun: in-flight plus completed tranches settled after +// ladderSettleAnchor(asOf). Run with LADDER_SETTLED_TEST_DATABASE_URL (see +// TestPostgresStoreDB_LadderSettledAt). +func TestPostgresStoreDB_PlanAndSaveSettledTerm(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + store, _ := settledTestStore(ctx, t) + cfgID := seedLadderConfig(ctx, t, store) + asOf := time.Now().UTC().Truncate(time.Microsecond) + anchor := ladderSettleAnchor(asOf) + + add := func(status ladder.TrancheStatus, amount float64, settled *time.Time) { + t.Helper() + tr := lifecycleTranche(cfgID, amount, status, asOf.Add(time.Hour)) + tr.SettledAt = settled + require.NoError(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{tr})) + } + planned := func() float64 { + t.Helper() + var got float64 + _, err := store.PlanAndSaveLadderRun(ctx, cfgID, asOf, func(inFlight float64) (*LadderRunDB, []LadderTrancheDB, error) { + got = inFlight + return lockTestRunAt(cfgID, asOf), nil, nil + }) + require.NoError(t, err) + return got + } + at := func(d time.Duration) *time.Time { v := anchor.Add(d); return &v } + + assert.InDelta(t, 0, planned(), 1e-9, "empty") + add(ladder.TrancheStatusCompleted, 1, at(-time.Microsecond)) + add(ladder.TrancheStatusCompleted, 2, at(0)) + assert.InDelta(t, 0, planned(), 1e-9, "settled at or before the anchor is already in E") + add(ladder.TrancheStatusCompleted, 4, at(time.Microsecond)) + assert.InDelta(t, 4, planned(), 1e-9, "just after the anchor counts") + // Inside the skew margin, before asOf itself: a margin of 0 would miss it. + add(ladder.TrancheStatusCompleted, 8, at(ladderSettleSkewMargin-time.Second)) + assert.InDelta(t, 12, planned(), 1e-9, "settled within the skew margin before asOf counts") + add(ladder.TrancheStatusFailed, 16, nil) + add(ladder.TrancheStatusCancelled, 32, nil) + assert.InDelta(t, 12, planned(), 1e-9, "failed and canceled never count") + add(ladder.TrancheStatusFired, 64, nil) + assert.InDelta(t, 76, planned(), 1e-9, "fired still counts as in-flight, added to the settled term") +} + +func lockTestRunAt(cfgID string, at time.Time) *LadderRunDB { + return &LadderRunDB{ConfigID: &cfgID, StartedAt: at, Status: ladder.RunStatusPlanned} +} diff --git a/internal/server/handler_ladder.go b/internal/server/handler_ladder.go index b8210f19..2614db1f 100644 --- a/internal/server/handler_ladder.go +++ b/internal/server/handler_ladder.go @@ -319,11 +319,13 @@ func (app *Application) executeLadderRun( // 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. + // still within config.LadderFireGrace plus fired, unresolved ones) under + // the per-config lock, plus tranches completed after `now` (settled term). + // `now` was captured before the GetLayerStates read above, so a tranche that + // completed after that read is in neither E nor in-flight without the + // settled term; one already in E is counted twice (a safe under-buy). 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, diff --git a/internal/server/handler_ladder_immediate_integration_test.go b/internal/server/handler_ladder_immediate_integration_test.go index 94c34e73..3580ae5c 100644 --- a/internal/server/handler_ladder_immediate_integration_test.go +++ b/internal/server/handler_ladder_immediate_integration_test.go @@ -63,6 +63,7 @@ type ladderB1Env struct { app *Application cfg *config.LadderConfigDB cap *fakeLadderCapability + cap2 pkgladder.LadderCapability // when set, run uses it instead of cap start time.Time } @@ -91,7 +92,11 @@ func newLadderB1Env(ctx context.Context, t *testing.T, conn *database.Connection func (e *ladderB1Env) run(ctx context.Context, t *testing.T, now time.Time) string { t.Helper() - require.NoError(t, e.app.executeLadderRun(ctx, e.cfg, e.cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now)) + var capability pkgladder.LadderCapability = e.cap + if e.cap2 != nil { + capability = e.cap2 + } + require.NoError(t, e.app.executeLadderRun(ctx, e.cfg, capability, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now)) var id string require.NoError(t, e.conn.QueryRow(ctx, `SELECT id FROM ladder_runs WHERE config_id=$1 ORDER BY started_at DESC, created_at DESC LIMIT 1`, e.cfg.ID).Scan(&id)) return id @@ -235,3 +240,46 @@ func TestLadderB1_DailyRerunNeverExceedsTarget(t *testing.T) { } assert.InDelta(t, ladderB1Target, e.live(ctx, t), 1e-6) } + +// completeBeforeStaleE completes a tranche inside GetLayerStates and then +// returns the stale E that predates the completion: the #843 race. +type completeBeforeStaleE struct { + *fakeLadderCapability + complete func() +} + +func (c *completeBeforeStaleE) GetLayerStates(ctx context.Context, scope pkgladder.Scope) (map[pkgladder.LayerType]pkgladder.LayerState, error) { + c.complete() + return c.fakeLadderCapability.GetLayerStates(ctx, scope) +} + +// A tranche that goes fired->completed between the E read and the locked +// in-flight read is in neither E (stale) nor the in-flight sum (not fired any +// more). Run 1 plans 10 as 5 immediate + 5 in 30 days; the immediate one is +// fired, then completes during run 2's E read. Run 2 must net both rows (10) +// and plan nothing; before the fix it re-planned the completed 5. +// FIXTURE capability; real store, real schema. +func TestLadderB1_TrancheCompletingBetweenEReadAndLockedSumIsNetted(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + e := newLadderB1Env(ctx, t, ladderB1Conn(ctx, t), ladderB1TwoStep) + first := e.run(ctx, t, e.start) + immediate := e.rows(ctx, t, first)[0] + require.InDelta(t, 5.0, immediate.amount, 1e-9) + var trancheID string + require.NoError(t, e.conn.QueryRow(ctx, `UPDATE ladder_tranches SET status='fired' WHERE run_id=$1 AND scheduled_date=$2 RETURNING id`, first, immediate.at).Scan(&trancheID)) + + now2 := time.Now().UTC().Truncate(time.Microsecond) // captured before the E read, as the handler does + e.cap2 = &completeBeforeStaleE{fakeLadderCapability: e.cap, complete: func() { + moved, err := config.NewPostgresStore(e.conn).TransitionLadderTranche(ctx, trancheID, pkgladder.TrancheStatusFired, pkgladder.TrancheStatusCompleted, "commitment_visible") + require.NoError(t, err) + require.True(t, moved) + }} + second := e.run(ctx, t, now2) + + var sum float64 + for _, r := range e.rows(ctx, t, second) { + sum += r.amount + } + assert.InDelta(t, 0, sum, 1e-9, "the completed tranche must be netted, not re-planned") +}