diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 3a61e057..7f6dcb7a 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -865,13 +865,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 04aa5276..d6d0cd88 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -608,30 +608,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 @@ -639,12 +615,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 c9d2b4eb..b96dd4fe 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1949,25 +1949,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) { @@ -2014,26 +1995,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 cc06c0d6..44988128 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -420,13 +420,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 }