fix(ladder): net tranches completed between the E read and the locked in-flight sum

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)

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).
 //

internal/config/store_postgres_ladder_lock.go

@@ -86,17 +86,39 @@ 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
 // 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.
+//
+// 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

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

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,

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