diff --git a/.changeset/latest-replan-wins.md b/.changeset/latest-replan-wins.md new file mode 100644 index 00000000..079d81ff --- /dev/null +++ b/.changeset/latest-replan-wins.md @@ -0,0 +1,7 @@ +--- +"ftw": patch +--- + +Keep the newest MPC replan when solves finish out of order. Each request now +keeps its mode and reason together, and an older result cannot replace a plan +started after it. diff --git a/go/internal/api/api_loadpoint_schedule_test.go b/go/internal/api/api_loadpoint_schedule_test.go index a00909ba..02588a8a 100644 --- a/go/internal/api/api_loadpoint_schedule_test.go +++ b/go/internal/api/api_loadpoint_schedule_test.go @@ -18,10 +18,8 @@ import ( // what the handlers do: PUT stores and rolls, DELETE clears, and both // force a replan tagged with the schedule-change reason. -// newScheduleServer wires a manager and an MPC service whose store is -// an empty temp db: ReplanWithReason records its reason and then -// returns at "no prices available yet", which is all a replan -// assertion needs. +// newScheduleServer wires a manager and an MPC service with enough input for +// the route-triggered replan to publish its plan and reason together. func newScheduleServer(t *testing.T) (*Server, *loadpoint.Manager, *mpc.Service) { t.Helper() mgr := loadpoint.NewManager() @@ -31,7 +29,26 @@ func newScheduleServer(t *testing.T) (*Server, *loadpoint.Manager, *mpc.Service) t.Fatalf("opening state store: %v", err) } t.Cleanup(func() { st.Close() }) - svc := &mpc.Service{Store: st, Zone: "SE4"} + start := time.Now().UTC().Truncate(15 * time.Minute) + prices := make([]state.PricePoint, 4) + for i := range prices { + prices[i] = state.PricePoint{ + Zone: "SE4", SlotTsMs: start.Add(time.Duration(i) * 15 * time.Minute).UnixMilli(), + SlotLenMin: 15, SpotOreKwh: 50, TotalOreKwh: 100, + Source: "test", FetchedAtMs: start.UnixMilli(), + } + } + if err := st.SavePrices(prices); err != nil { + t.Fatalf("saving prices: %v", err) + } + svc := mpc.New(st, nil, "SE4", mpc.Params{ + Mode: mpc.ModeSelfConsumption, SoCLevels: 11, ActionLevels: 5, + CapacityWh: 10000, InitialSoCPct: 50, SoCMinPct: 10, SoCMaxPct: 95, + MaxChargeW: 3000, MaxDischargeW: 3000, + ChargeEfficiency: 0.95, DischargeEfficiency: 0.95, + }) + svc.Horizon = time.Hour + svc.BaseLoad = 500 return New(&Deps{Loadpoints: mgr, MPC: svc}), mgr, svc } diff --git a/go/internal/mpc/reactive_test.go b/go/internal/mpc/reactive_test.go index bf882bdb..32c0dc16 100644 --- a/go/internal/mpc/reactive_test.go +++ b/go/internal/mpc/reactive_test.go @@ -10,6 +10,31 @@ import ( "github.com/srcfl/ftw/go/internal/telemetry" ) +func seedReactivePrices(t *testing.T, st *state.Store) { + t.Helper() + start := time.Now().UTC().Truncate(15 * time.Minute) + prices := make([]state.PricePoint, 4) + for i := range prices { + prices[i] = state.PricePoint{ + Zone: "SE3", SlotTsMs: start.Add(time.Duration(i) * 15 * time.Minute).UnixMilli(), + SlotLenMin: 15, SpotOreKwh: 50, TotalOreKwh: 100, + Source: "test", FetchedAtMs: start.UnixMilli(), + } + } + if err := st.SavePrices(prices); err != nil { + t.Fatalf("save prices: %v", err) + } +} + +func reactiveTestParams() Params { + return Params{ + Mode: ModeSelfConsumption, SoCLevels: 11, ActionLevels: 5, + CapacityWh: 10000, InitialSoCPct: 50, SoCMinPct: 10, SoCMaxPct: 95, + MaxChargeW: 3000, MaxDischargeW: 3000, + ChargeEfficiency: 0.95, DischargeEfficiency: 0.95, + } +} + // buildTestService spins up a minimal Service with one cached plan // covering the current time, so checkDivergence has something to // compare against. @@ -19,6 +44,7 @@ func buildTestService(t *testing.T, planPV, planLoad float64) (*Service, *teleme t.Fatal(err) } t.Cleanup(func() { st.Close() }) + seedReactivePrices(t, st) tel := telemetry.NewStore() tel.DriverHealthMut("site").RecordSuccess() tel.DriverHealthMut("inverter").RecordSuccess() @@ -32,6 +58,9 @@ func buildTestService(t *testing.T, planPV, planLoad float64) (*Service, *teleme MinReplanGap: time.Millisecond, PVDivergenceWh: 500, LoadDivergenceWh: 400, + Horizon: time.Hour, + Defaults: reactiveTestParams(), + BaseLoad: 500, } now := time.Now() s.last = &Plan{ @@ -175,13 +204,16 @@ func buildDefaultTestService(t *testing.T, planPV, planLoad float64) (*Service, t.Fatal(err) } t.Cleanup(func() { st.Close() }) + seedReactivePrices(t, st) tel := telemetry.NewStore() tel.DriverHealthMut("site").RecordSuccess() tel.DriverHealthMut("inverter").RecordSuccess() // Mirror New()'s defaults so this test exercises the production // reactive-trigger numbers. - s := New(st, tel, "SE3", Params{}) + s := New(st, tel, "SE3", reactiveTestParams()) + s.Horizon = time.Hour + s.BaseLoad = 500 s.SiteMeter = "site" s.ReactiveInterval = 10 * time.Millisecond now := time.Now() @@ -255,8 +287,11 @@ func twinDriftService(t *testing.T) *Service { t.Fatal(err) } t.Cleanup(func() { st.Close() }) + seedReactivePrices(t, st) tel := telemetry.NewStore() - s := New(st, tel, "SE3", Params{}) + s := New(st, tel, "SE3", reactiveTestParams()) + s.Horizon = time.Hour + s.BaseLoad = 500 s.ReactiveInterval = 10 * time.Millisecond // Make sure cooldown doesn't suppress the first trigger. s.lastReplanAt = time.Now().Add(-time.Hour) @@ -282,9 +317,8 @@ func TestTwinDriftReplanFiresOnLargePVShift(t *testing.T) { // A live PV predictor that now returns 1500 W per slot — RMSE = 500 W, // well past the 250 W threshold. s.PV = func(time.Time, float64) float64 { return 1500 } - // Stub a minimal plan so replan() (called on trigger) doesn't panic — - // it'll bail with "no prices available yet" but lastReason has been - // set on the service before that call, which is what we assert. + // Stub a minimal active plan. The seeded prices let the triggered replan + // commit its reason with the replacement plan. s.last = &Plan{GeneratedAtMs: now.UnixMilli()} s.checkTwinDrift(context.Background()) diff --git a/go/internal/mpc/service.go b/go/internal/mpc/service.go index 8e913fa1..6038b27e 100644 --- a/go/internal/mpc/service.go +++ b/go/internal/mpc/service.go @@ -105,8 +105,9 @@ type Service struct { Loadpoint LoadpointProbe // optional — when non-nil, the DP extends its state with EV dimensions Loadpoints LoadpointsProbe - // SaveDiag is called synchronously after every successful replan - // with the same Diagnostic the /api/mpc/diagnose endpoint would + // SaveDiag is called synchronously after every successful replan that + // remains the newest request, with the same Diagnostic the + // /api/mpc/diagnose endpoint would // return + the trigger reason ("scheduled" / "reactive-pv" / // "reactive-load" / "manual"). Nil disables persistence — the // in-memory diagnose still works. Wired in main.go against @@ -175,7 +176,12 @@ type Service struct { MaxExportW float64 lastReplanAt time.Time - lastReason string // "scheduled" | "reactive-pv" | "reactive-load" | "manual" + lastReason string // reason paired with the currently published plan + // latestReplanGeneration identifies the newest requested solve. A solve + // may run after a newer request starts, but it cannot publish over it. + // This guard does not cancel solver work; transport deadlines and worker + // queue ownership remain separate concerns. + latestReplanGeneration uint64 // guarded by mu // ExportBonusOreKwh and ExportFeeOreKwh flow in from config.Price. // Used to compute default ExportOrePerKWh when Params doesn't set it. @@ -224,6 +230,15 @@ type plannedPredictions struct { builtAt time.Time } +// replanRequest is an immutable snapshot of the caller's intent. In +// particular, mode and reason must stay paired while a slower solve runs. +type replanRequest struct { + generation uint64 + params Params + fleet []BatteryFleetMember + reason string +} + func (s *Service) driverOnline(name string) bool { if s == nil || s.Tele == nil { return false @@ -528,6 +543,10 @@ func (s *Service) SlotAt(now time.Time) (string, float64, bool) { } s.mu.RLock() p := s.last + params := s.lastParams + if params.Mode == "" { + params = s.Defaults + } s.mu.RUnlock() if p == nil { return "", 0, false @@ -539,7 +558,7 @@ func (s *Service) SlotAt(now time.Time) (string, float64, bool) { for _, a := range p.Actions { end := a.SlotStartMs + int64(a.SlotLenMin)*60*1000 if nowMs >= a.SlotStartMs && nowMs < end { - return actionToSlot(a, s.Defaults.Mode) + return actionToSlot(a, params.Mode) } } return "", 0, false @@ -587,8 +606,9 @@ func (s *Service) SetMode(ctx context.Context, mode Mode) { } s.mu.Lock() s.Defaults.Mode = mode + request := s.beginReplanLocked("mode_changed") s.mu.Unlock() - s.replan(ctx) + s.runReplan(ctx, request) } // Start runs the planner in a goroutine. Does an initial plan immediately. @@ -613,8 +633,7 @@ func (s *Service) Stop() { func (s *Service) loop(ctx context.Context) { defer close(s.done) - s.lastReason = "scheduled" - s.replan(ctx) + s.replan(ctx, "scheduled") t := time.NewTicker(s.Interval) defer t.Stop() var reactiveTick <-chan time.Time @@ -630,8 +649,7 @@ func (s *Service) loop(ctx context.Context) { case <-ctx.Done(): return case <-t.C: - s.lastReason = "scheduled" - s.replan(ctx) + s.replan(ctx, "scheduled") case <-reactiveTick: s.observeShadow(time.Now()) s.checkDivergence(ctx) @@ -753,13 +771,12 @@ func (s *Service) checkDivergence(ctx context.Context) { "pv_err_wh", pvInt, "loadint_wh", loadInt, "pv_w_now", pvW, "plan_pv_w", slot.PVW, "load_w_now", loadW, "plan_load_w", slot.LoadW) - s.lastReason = reason // Reset integrals after triggering so we don't immediately re-fire. s.mu.Lock() s.pvErrIntWh = 0 s.loadErrIntWh = 0 s.mu.Unlock() - s.replan(ctx) + s.replan(ctx, reason) } // snapshotPredictions samples the PV + load twins at the build-time slot @@ -895,15 +912,12 @@ func (s *Service) checkTwinDrift(ctx context.Context) { if reason == "" { return } - s.mu.Lock() - s.lastReason = reason - s.mu.Unlock() - s.replan(ctx) + s.replan(ctx, reason) } // Replan recomputes the plan once using current prices + forecast + SoC. // Exposed for tests and API triggers. -func (s *Service) Replan(ctx context.Context) *Plan { return s.replan(ctx) } +func (s *Service) Replan(ctx context.Context) *Plan { return s.replan(ctx, "manual") } // ReplanWithReason is Replan with an explicit reason string that lands // in slog + the diagnose snapshot. Use it when an external event (API @@ -913,15 +927,36 @@ func (s *Service) Replan(ctx context.Context) *Plan { return s.replan(ctx) } // 12:34?". Reasons should be short kebab-style, e.g. // "surplus_only_disabled", "target_soc_changed", "mode_changed". func (s *Service) ReplanWithReason(ctx context.Context, reason string) *Plan { - if reason != "" { - s.mu.Lock() - s.lastReason = reason - s.mu.Unlock() + return s.replan(ctx, reason) +} + +func (s *Service) replan(ctx context.Context, reason string) *Plan { + return s.runReplan(ctx, s.beginReplan(reason)) +} + +func (s *Service) beginReplan(reason string) replanRequest { + s.mu.Lock() + defer s.mu.Unlock() + return s.beginReplanLocked(reason) +} + +// beginReplanLocked assigns the generation and snapshots the effective +// defaults under the same lock. Callers that change Defaults first, such as +// SetMode, use this form so no old solve can commit in between those actions. +func (s *Service) beginReplanLocked(reason string) replanRequest { + if reason == "" { + reason = "manual" + } + s.latestReplanGeneration++ + return replanRequest{ + generation: s.latestReplanGeneration, + params: s.Defaults, + fleet: append([]BatteryFleetMember(nil), s.BatteryFleet...), + reason: reason, } - return s.replan(ctx) } -func (s *Service) replan(ctx context.Context) *Plan { +func (s *Service) runReplan(ctx context.Context, request replanRequest) *Plan { now := time.Now() untilMs := now.Add(s.Horizon).UnixMilli() sinceMs := now.UnixMilli() - 15*60*1000 // small margin — slot starting ≤15min ago still in-flight @@ -974,10 +1009,8 @@ func (s *Service) replan(ctx context.Context) *Plan { clampSlotGridLimits(slots, s.FuseMaxW, s.MaxExportW) clampSlotGridLimits(fallbackSlots, s.FuseMaxW, s.MaxExportW) - s.mu.RLock() - p := s.Defaults - fleet := append([]BatteryFleetMember(nil), s.BatteryFleet...) - s.mu.RUnlock() + p := request.params + fleet := request.fleet if len(fleet) > 0 { var ok bool p, ok = s.onlineFleetParams(p, fleet) @@ -1093,6 +1126,9 @@ func (s *Service) replan(ctx context.Context) *Plan { "loadpoint_id", loadpointID, ) var plan Plan + var shadowRecoursePlan *Plan + var shadowError string + publishShadow := false if s.Optimizer == nil { slots = fallbackSlots plan = Optimize(slots, p) @@ -1133,10 +1169,10 @@ func (s *Service) replan(ctx context.Context) *Plan { candidate.DPShadow.FirstAction.EMSMode = mode } - var recoursePlan *Plan if s.EnableRecourseShadow { + publishShadow = true if len(p.activeLoadpoints()) > 0 { - s.ensureShadowEvaluator().SetError("recourse shadow skipped while flexible loads are active", now) + shadowError = "recourse shadow skipped while flexible loads are active" } else { policy := s.ChallengerPolicy if policy == "" { @@ -1162,9 +1198,9 @@ func (s *Service) replan(ctx context.Context) *Plan { } if recourseErr != nil { slog.Warn("mpc: stochastic challenger failed", "policy", policy, "err", recourseErr) - s.ensureShadowEvaluator().SetError(recourseErr.Error(), now) + shadowError = recourseErr.Error() } else { - recoursePlan = &recourse + shadowRecoursePlan = &recourse candidate.RecourseShadow = compareDPShadow(candidate, recourse) candidate.RecourseShadow.ForecastBasis = "same stochastic scenario input; conditional decisions after non-anticipative prefix" candidate.RecourseShadow.Solver = recourse.Solver @@ -1176,9 +1212,6 @@ func (s *Service) replan(ctx context.Context) *Plan { } } } - evaluator := s.ensureShadowEvaluator() - evaluator.SetPlans(&candidate, recoursePlan, slots, p, time.Now()) - candidate.ShadowEvaluation = evaluator.Snapshot() } optimizerSolveMs := 0.0 if candidate.Solver != nil { @@ -1237,17 +1270,36 @@ func (s *Service) replan(ctx context.Context) *Plan { pp := s.snapshotPredictions(slots, forecasts) s.mu.Lock() + if request.generation != s.latestReplanGeneration { + latest := s.latestReplanGeneration + s.mu.Unlock() + slog.Info("mpc: discarded superseded replan", + "generation", request.generation, + "latest_generation", latest, + "mode", p.Mode, + "reason", request.reason) + return s.Latest() + } + if publishShadow { + if s.shadowEvaluator == nil { + s.shadowEvaluator = newStatefulShadowEvaluator() + } + if shadowError != "" { + s.shadowEvaluator.SetError(shadowError, now) + } + s.shadowEvaluator.SetPlans(&plan, shadowRecoursePlan, slots, p, time.Now()) + plan.ShadowEvaluation = s.shadowEvaluator.Snapshot() + } s.last = &plan s.lastSlots = slots s.lastParams = p s.lastLoadpointID = loadpointID s.lastReplanAt = time.Now() s.plannedPredictions = pp - reason := s.lastReason - if reason == "" { - reason = "manual" - } + s.lastReason = request.reason + reason := request.reason replanAtMs := s.lastReplanAt.UnixMilli() + saveDiag := s.SaveDiag s.mu.Unlock() // Horizon statistics — surfaced in logs so operators can // reconstruct "what did the DP know?" without pulling the full @@ -1285,15 +1337,15 @@ func (s *Service) replan(ctx context.Context) *Plan { // this replan later. Best-effort: errors log and continue so a // flaky disk never blocks planning. // - // Critically: build from the LOCAL plan/slots/p we just computed, - // not from s.last via Diagnose(). A concurrent replan could have - // swapped s.last between our unlock and the Diagnose() call, - // which would pair a different plan with OUR reason — writing a - // corrupt snapshot. Using the locals keeps (plan, reason) - // atomically consistent even under concurrent replans. - if s.SaveDiag != nil { + // Build from the local plan/slots/p accepted by the generation check, + // not from s.last via Diagnose(). This keeps each persisted plan paired + // with the params and reason from the same request. Once accepted, every + // active plan gets a historical diagnostic even if another request starts + // before this disk write completes. A never-active plan does not reach this + // hook, and the service mutex is not held across disk I/O. + if saveDiag != nil { if d := buildDiagnostic(&plan, slots, p, s.Zone, replanAtMs, reason); d != nil { - if err := s.SaveDiag(d, reason); err != nil { + if err := saveDiag(d, reason); err != nil { slog.Warn("mpc: persist diagnostic failed", "err", err) } } @@ -1301,15 +1353,6 @@ func (s *Service) replan(ctx context.Context) *Plan { return &plan } -func (s *Service) ensureShadowEvaluator() *StatefulShadowEvaluator { - s.mu.Lock() - defer s.mu.Unlock() - if s.shadowEvaluator == nil { - s.shadowEvaluator = newStatefulShadowEvaluator() - } - return s.shadowEvaluator -} - // observeShadow samples realized exogenous power for closed-loop scoring. It // deliberately derives house load without battery or vehicle power so both // virtual policies receive the same uncontrollable input. diff --git a/go/internal/mpc/service_latest_wins_test.go b/go/internal/mpc/service_latest_wins_test.go new file mode 100644 index 00000000..9a33962e --- /dev/null +++ b/go/internal/mpc/service_latest_wins_test.go @@ -0,0 +1,140 @@ +package mpc + +import ( + "context" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/state" +) + +type blockingFirstOptimizer struct { + calls atomic.Int32 + firstMode chan Mode + secondMode chan Mode + releaseFirst chan struct{} +} + +func (o *blockingFirstOptimizer) Optimize(ctx context.Context, slots []Slot, p Params) (Plan, error) { + call := o.calls.Add(1) + if call == 1 { + o.firstMode <- p.Mode + select { + case <-o.releaseFirst: + case <-ctx.Done(): + return Plan{}, ctx.Err() + } + } else if call == 2 { + o.secondMode <- p.Mode + } + + plan := Optimize(slots, p) + status := "old-self-consumption" + if call == 2 { + status = "new-arbitrage" + } + plan.Solver = &SolverInfo{ + Engine: "test", Backend: "blocking", Status: status, + Formulation: "deterministic", + } + return plan, nil +} + +func (*blockingFirstOptimizer) Close() error { return nil } + +func TestReplanNewestRequestWinsWhenOlderSolveFinishesLast(t *testing.T) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatalf("open state: %v", err) + } + defer st.Close() + + now := time.Now().UTC().Truncate(time.Hour) + prices := make([]state.PricePoint, 4) + for i := range prices { + prices[i] = state.PricePoint{ + Zone: "SE3", SlotTsMs: now.Add(time.Duration(i) * time.Hour).UnixMilli(), + SlotLenMin: 60, SpotOreKwh: float64(40 + i*20), TotalOreKwh: float64(90 + i*20), + Source: "test", FetchedAtMs: now.UnixMilli(), + } + } + if err := st.SavePrices(prices); err != nil { + t.Fatalf("save prices: %v", err) + } + + optimizer := &blockingFirstOptimizer{ + firstMode: make(chan Mode, 1), + secondMode: make(chan Mode, 1), + releaseFirst: make(chan struct{}), + } + svc := New(st, nil, "SE3", Params{ + Mode: ModeSelfConsumption, SoCLevels: 11, ActionLevels: 5, + CapacityWh: 10000, InitialSoCPct: 50, SoCMinPct: 10, SoCMaxPct: 95, + MaxChargeW: 3000, MaxDischargeW: 3000, + ChargeEfficiency: 0.95, DischargeEfficiency: 0.95, + }) + svc.Horizon = 4 * time.Hour + svc.BaseLoad = 500 + svc.Optimizer = optimizer + + type savedDiagnostic struct { + mode Mode + reason string + } + saved := make(chan savedDiagnostic, 2) + svc.SaveDiag = func(d *Diagnostic, reason string) error { + saved <- savedDiagnostic{mode: d.Params.Mode, reason: reason} + return nil + } + + oldDone := make(chan *Plan, 1) + go func() { + oldDone <- svc.ReplanWithReason(context.Background(), "old-self-consumption") + }() + + if mode := <-optimizer.firstMode; mode != ModeSelfConsumption { + t.Fatalf("first solve mode = %q, want %q", mode, ModeSelfConsumption) + } + + // The second solve starts after the mode change and finishes while the + // first solve remains blocked. + svc.SetMode(context.Background(), ModeArbitrage) + if mode := <-optimizer.secondMode; mode != ModeArbitrage { + t.Fatalf("second solve mode = %q, want %q", mode, ModeArbitrage) + } + + published := svc.Latest() + if published == nil || published.Solver == nil || published.Solver.Status != "new-arbitrage" { + t.Fatalf("newer plan was not published: %+v", published) + } + if got := <-saved; got.mode != ModeArbitrage || got.reason != "mode_changed" { + t.Fatalf("saved diagnostic = %+v, want arbitrage/mode_changed", got) + } + + close(optimizer.releaseFirst) + if got := <-oldDone; got != published { + t.Fatalf("superseded caller returned an unpublished plan: got=%p published=%p", got, published) + } + + svc.mu.RLock() + lastMode := svc.lastParams.Mode + lastReason := svc.lastReason + lastGeneration := svc.latestReplanGeneration + svc.mu.RUnlock() + if lastMode != ModeArbitrage || lastReason != "mode_changed" || lastGeneration != 2 { + t.Fatalf("published state = mode %q reason %q generation %d", lastMode, lastReason, lastGeneration) + } + if latest := svc.Latest(); latest != published { + t.Fatal("older solve replaced the newer published plan") + } + if d := svc.Diagnose(); d == nil || d.Params.Mode != ModeArbitrage || d.LastReason != "mode_changed" { + t.Fatalf("diagnostic was replaced by the older solve: %+v", d) + } + select { + case extra := <-saved: + t.Fatalf("superseded solve persisted a diagnostic: %+v", extra) + default: + } +}