From 0806ffc30af3c41039eb413d06c2b32e9a3d497c Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sun, 18 Oct 2026 20:52:45 +0200 Subject: [PATCH 1/2] fix(history): balance household Load from interval source energy --- .changeset/history-source-load.md | 5 + docs/roadmap.md | 7 + docs/site-convention.md | 12 +- go/cmd/ftw/energy_history_test.go | 18 +- go/cmd/ftw/forecast_measurements.go | 3 + go/cmd/ftw/history_disabled_battery_test.go | 12 +- go/cmd/ftw/history_source_load_test.go | 77 ++++ go/cmd/ftw/main.go | 63 ++- go/internal/api/api.go | 17 +- go/internal/api/api_savings.go | 18 +- go/internal/api/api_source_history_test.go | 96 +++++ go/internal/state/energy_ledger.go | 11 +- go/internal/state/history_dashboard.go | 68 +++- go/internal/state/history_dashboard_cost.go | 46 +++ .../state/history_dashboard_retention.go | 21 +- go/internal/state/history_load_estimate.go | 5 +- go/internal/state/history_plain.go | 1 + go/internal/state/history_schema.go | 3 + go/internal/state/history_sources.go | 334 ++++++++++++++++ go/internal/state/history_sources_test.go | 376 ++++++++++++++++++ go/internal/state/history_sqlite.go | 30 +- go/internal/state/ledger_flows.go | 10 + go/internal/state/store.go | 65 ++- go/internal/telemetry/forecast.go | 3 + 24 files changed, 1252 insertions(+), 49 deletions(-) create mode 100644 .changeset/history-source-load.md create mode 100644 go/cmd/ftw/history_source_load_test.go create mode 100644 go/internal/api/api_source_history_test.go create mode 100644 go/internal/state/history_sources.go create mode 100644 go/internal/state/history_sources_test.go diff --git a/.changeset/history-source-load.md b/.changeset/history-source-load.md new file mode 100644 index 000000000..e3775a55a --- /dev/null +++ b/.changeset/history-source-load.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Calculate recorded household Load from source energy over the same interval in Core. Different meter and charger reporting rates no longer depend on a valid instantaneous Load sample. The app, web history and daily totals use the same source balance. Keep signed source energy through aggregation, preserve real gaps, and leave live display and control unchanged. diff --git a/docs/roadmap.md b/docs/roadmap.md index c308a5a49..2d78cdc4c 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -58,6 +58,13 @@ using an estimate in planning. A model prediction must never verify itself or raise a safety limit. These are acceptance requirements, not shipped support. Necessary safety, security, recovery and support fixes continue throughout. + +Household Load history must balance source energy over the same interval in +Core. Live updates remain immediate. Different reporting rates must not +require a valid instantaneous Load sample before energy can be recorded. +Verify the app, web, daily totals, tariff boundaries, restart and missing +source cases against the same stored evidence. + Correct misleading value labels when their scope is known; that need not wait for the new counterfactual model. Reading and analysis access for agents can also support the evidence work before agents receive control authority. diff --git a/docs/site-convention.md b/docs/site-convention.md index dac47d325..77b9d80d7 100644 --- a/docs/site-convention.md +++ b/docs/site-convention.md @@ -46,9 +46,19 @@ this one thing were the only thing running. Balance equation (all signed): ``` -grid_w = load_w + bat_w + pv_w +grid_w = load_w + bat_w + pv_w + ev_w + v2x_w ``` +Live household Load uses the current fresh site-side powers. Historical Load +uses source energy over a common interval, with EV charging excluded. Core +holds each qualified snapshot forward until the next safe snapshot, at most +15 seconds later, and stores signed source energy by UTC minute. It balances +the sums when reading a chart interval, day or tariff period. A negative +minute may contribute to a valid longer interval; it is never clipped first. +Chart means need complete interval coverage. Missing sources, writer gaps, +process stops and topology changes cannot become measured zero. This is an +estimate from received measurements, not a correction of device clocks. + In self-consumption mode, the controller drives the site meter toward the grid target (normally 0 W): it charges from live surplus and may discharge to cover local load. It must not intentionally export via the battery; export diff --git a/go/cmd/ftw/energy_history_test.go b/go/cmd/ftw/energy_history_test.go index e52123cc0..ded25ba2d 100644 --- a/go/cmd/ftw/energy_history_test.go +++ b/go/cmd/ftw/energy_history_test.go @@ -186,8 +186,14 @@ func TestPersistTelemetryTickUsesPersistenceFreshness(t *testing.T) { if (len(history) == 1) != tc.wantSaved { t.Fatalf("history = %+v, want saved=%v", history, tc.wantSaved) } - if tc.wantSaved && (history[0].TsMs != tickMS || history[0].LoadW != 1200) { - t.Fatalf("tick timestamp or household power changed: %+v", history[0]) + if tc.wantSaved && (history[0].TsMs != tickMS || history[0].GridW != 1200 || history[0].Missing&state.MissingLoad == 0) { + t.Fatal("fresh grid receipt lost or instantaneous Load entered history") + } + if tc.wantSaved { + p, _ := buildHistoryPoint(tel, ctrl, tickMS, time.Minute) + if p.Sources.AtMS < sampleAt.UnixMilli() || p.Sources.Missing != 0 { + t.Fatal("fresh source snapshot predates receipt or became unknown") + } } assets, err := st.EnergyAssets() if err != nil { @@ -199,8 +205,8 @@ func TestPersistTelemetryTickUsesPersistenceFreshness(t *testing.T) { consumerSaved = asset.LastSeenMS == sampleAt.UnixMilli() } } - if consumerSaved != tc.wantSaved { - t.Fatalf("consumer ledger assets = %+v, want saved=%v", assets, tc.wantSaved) + if consumerSaved { + t.Fatal("instantaneous consumer observation entered interval history") } }) } @@ -288,14 +294,14 @@ func TestStaleMeterTickKeepsSamplesAndIndependentLedgerWithoutDispatch(t *testin tel.Update("solar", telemetry.DerPV, -400, nil, json.RawMessage(`{"generation_wh":110}`)) tel.Get("solar", telemetry.DerPV).UpdatedAt = now tel.RecordDriverSuccess("solar") - if _, err := persistTelemetryTick(st, tel, ctrl, now.Add(time.Second).UnixMilli(), time.Minute, testEnergyIdentity(st)); err != nil { + if _, err := persistTelemetryTick(st, tel, ctrl, time.Now().UnixMilli(), time.Minute, testEnergyIdentity(st)); err != nil { t.Fatal(err) } if err := st.FlushHistory(context.Background()); err != nil { t.Fatal(err) } - history, err := st.LoadHistory(base.UnixMilli(), now.Add(time.Second).UnixMilli(), 0) + history, err := st.LoadHistory(base.UnixMilli(), time.Now().Add(time.Second).UnixMilli(), 0) if err != nil { t.Fatal(err) } diff --git a/go/cmd/ftw/forecast_measurements.go b/go/cmd/ftw/forecast_measurements.go index f55b82638..861548dd2 100644 --- a/go/cmd/ftw/forecast_measurements.go +++ b/go/cmd/ftw/forecast_measurements.go @@ -31,6 +31,9 @@ func historyMeasurementOptions(cfg *config.Config, catalog []drivers.CatalogEntr } } opts.ExpectedFlows = flows + if cfg != nil { + opts.HistoryRevision = cfg.Revision + } return opts } diff --git a/go/cmd/ftw/history_disabled_battery_test.go b/go/cmd/ftw/history_disabled_battery_test.go index 51a54fcc5..3784c7d0c 100644 --- a/go/cmd/ftw/history_disabled_battery_test.go +++ b/go/cmd/ftw/history_disabled_battery_test.go @@ -37,20 +37,22 @@ func TestHistoryDisabledBatteryKeepsP1Load(t *testing.T) { tel.Update("solar", telemetry.DerPV, -2000, nil, nil) tel.RecordDriverSuccess("solar") opts := historyMeasurementOptions(cfg, catalog) - at := time.Now().Add(-2 * time.Second).UnixMilli() - for i := int64(0); i < 2; i++ { + at := time.Now().Add(-10 * time.Minute).Truncate(5 * time.Minute).UnixMilli() + for i := int64(0); i <= 30; i++ { for _, kind := range []telemetry.DerType{telemetry.DerMeter, telemetry.DerPV} { name := map[telemetry.DerType]string{telemetry.DerMeter: "p1", telemetry.DerPV: "solar"}[kind] - tel.Get(name, kind).UpdatedAt = time.UnixMilli(at + i*1000) + tel.Get(name, kind).UpdatedAt = time.Now() } - if _, err := persistTelemetryTick(st, tel, tickPersistControl{SiteMeterDriver: "p1"}, at+i*1000, time.Minute, nil, opts); err != nil { + p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "p1"}, at+i*10000, time.Minute, opts) + p.Sources.AtMS = at + i*10000 + if err := st.RecordTickWithEnergy(p, nil, nil); err != nil { t.Fatal(err) } } if err := st.FlushHistory(context.Background()); err != nil { t.Fatal(err) } - points, err := st.LoadHistory(at, at+1000, 0) + points, err := st.LoadHistory(at, at+300000, 5) if err != nil { t.Fatal(err) } diff --git a/go/cmd/ftw/history_source_load_test.go b/go/cmd/ftw/history_source_load_test.go new file mode 100644 index 000000000..bfa7df3e9 --- /dev/null +++ b/go/cmd/ftw/history_source_load_test.go @@ -0,0 +1,77 @@ +package main + +import ( + "context" + "math" + "path/filepath" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +func TestHistoryLoadIndependentReportingCadence(t *testing.T) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + tel := telemetry.NewStore() + tel.Update("grid", telemetry.DerMeter, 4000, nil, nil) + tel.RecordDriverSuccess("grid") + tel.Update("charger", telemetry.DerEV, 3000, nil, nil) + tel.RecordDriverSuccess("charger") + // Both powers are fresh. The charger reports more slowly than the meter. + tel.Get("charger", telemetry.DerEV).UpdatedAt = time.Now().Add(-45 * time.Second) + base := time.Now().Add(-10 * time.Minute).Truncate(5 * time.Minute).UnixMilli() + for i := int64(0); i <= 30; i++ { + p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "grid"}, base+i*10000, 90*time.Second) + if p.Sources.Missing != 0 { + t.Fatal("fresh charger rejected by historical source qualification") + } + p.Sources.AtMS = base + i*10000 + if err := st.RecordTickWithEnergy(p, nil, nil); err != nil { + t.Fatal(err) + } + } + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + d, err := st.DailyEnergy(base, base+300000) + if err != nil { + t.Fatal(err) + } + if d.Missing&state.MissingLoad != 0 || math.Abs(d.LoadWh-1000.0/12) > 0.001 { + t.Fatal("fresh asynchronous sources lost historical household energy") + } + native, err := (&appHistory{st: st}).Window(context.Background(), []string{"load_w"}, 300000, base, base+300000) + if err != nil { + t.Fatal(err) + } + if len(native["load_w"]) != 1 || math.Abs(native["load_w"][0].W-1000) > 0.001 { + t.Fatal("native history did not receive the interval household balance") + } +} + +func TestHistorySourceQualificationDoesNotInventMissingFlows(t *testing.T) { + for _, tc := range []struct { + reason string + watts float64 + known bool + want state.HistoryMissing + }{ + {"missing_config", 0, true, state.MissingLoad}, + {"invalid_balance", 1000, true, 0}, + {"nonfinite_sum", math.Inf(1), false, state.MissingLoad | state.MissingEV}, + } { + known := map[telemetry.DerType]bool{telemetry.DerMeter: true, telemetry.DerPV: true, telemetry.DerBattery: true, telemetry.DerEV: tc.known, telemetry.DerV2X: true} + p := historySources(telemetry.ForecastReading{At: time.Now(), Known: known, Reason: tc.reason, EVW: tc.watts}) + if p.Missing != tc.want || math.IsNaN(p.EVW) || math.IsInf(p.EVW, 0) { + t.Fatal("history source qualification lost missing evidence or rejected unrelated telemetry") + } + } +} diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index d1652ce78..e2b154523 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -14,6 +14,7 @@ import ( "io" "io/fs" "log/slog" + "math" "net" "net/http" "net/url" @@ -22,6 +23,7 @@ import ( "path/filepath" "reflect" "runtime" + "sort" "strconv" "strings" "sync" @@ -3910,6 +3912,20 @@ func snapshotTickPersistControl(ctrl *control.State, ctrlMu *sync.Mutex) tickPer func persistTelemetryTick(st *state.Store, tel *telemetry.Store, ctrl tickPersistControl, nowMs int64, historyMaxAge time.Duration, identity energyIdentityLookup, options ...telemetry.ForecastOptions) (int, error) { hp, historyAvailable := buildHistoryPoint(tel, ctrl, nowMs, historyMaxAge, options...) + if hp.Sources != nil { + // Live Load is not an energy observation. The source interval writer + // owns historical Load; the chart placeholder stays explicitly unknown. + hp.LoadW = 0 + hp.Missing |= state.MissingLoad + var detail map[string]any + if json.Unmarshal([]byte(hp.JSON), &detail) == nil { + detail["load_house_w"] = nil + detail["forecast_measurement_quality"] = "source_interval_pending" + if raw, err := json.Marshal(detail); err == nil { + hp.JSON = string(raw) + } + } + } samples := tel.FlushSamples() stSamples := make([]state.Sample, len(samples)) for i, sm := range samples { @@ -3919,9 +3935,19 @@ func persistTelemetryTick(st *state.Store, tel *telemetry.Store, ctrl tickPersis } } energyObservations := buildEnergyObservations(st, tel, ctrl, hp, identity) + if hp.Sources != nil { + var devices []string + for _, o := range energyObservations { + if o.DeviceID != "" { + devices = append(devices, string(o.AssetKind)+":"+o.DeviceID) + } + } + sort.Strings(devices) + hp.Sources.Topology += "\ndevices:" + strings.Join(devices, "\n") + } filtered := energyObservations[:0] for _, observation := range energyObservations { - if (!historyAvailable || hp.Missing&state.MissingLoad != 0) && observation.AssetKind == state.AssetObservedConsumer { + if (!historyAvailable || hp.Sources != nil || hp.Missing&state.MissingLoad != 0) && observation.AssetKind == state.AssetObservedConsumer { continue } if observation.AssetKind != state.AssetObservedConsumer { @@ -4022,9 +4048,44 @@ func buildHistoryPoint(tel *telemetry.Store, ctrl tickPersistControl, nowMs int6 } blob, _ := json.Marshal(map[string]any{"drivers": perDriver, "targets": targets, "ev_w": p.Missing.Value(state.MissingEV, balance.EVW), "v2x_w": p.Missing.Value(state.MissingV2X, balance.V2XW), "load_house_w": p.LoadW, "forecast_measurement_quality": quality, "history_reason": balance.Reason}) p.JSON = string(blob) + liveSources := tel.LiveMeasurementNow(ctrl.SiteMeterDriver, opts) + p.Sources = historySources(liveSources) + p.Sources.Topology += "\nsettings:" + strconv.FormatInt(opts.HistoryRevision, 10) + for _, source := range liveSources.Sources { + hasPower = hasPower || source.PowerW != nil + } return p, hasPower } +func historySources(reading telemetry.ForecastReading) *state.HistorySources { + p := &state.HistorySources{AtMS: reading.At.UnixMilli(), GridW: reading.GridW, PVW: reading.PVW, BatteryW: reading.BatteryW, EVW: reading.EVW, V2XW: reading.V2XW} + var flows []string + for _, source := range reading.Sources { + flows = append(flows, source.Flow.Driver+":"+source.Flow.DerType.String()+":"+source.Flow.FlowID) + } + sort.Strings(flows) + p.Topology = strings.Join(flows, "\n") + for kind, flag := range map[telemetry.DerType]state.HistoryMissing{telemetry.DerMeter: state.MissingGrid, telemetry.DerPV: state.MissingPV, telemetry.DerBattery: state.MissingBattery, telemetry.DerEV: state.MissingEV, telemetry.DerV2X: state.MissingV2X} { + if !reading.Known[kind] { + p.Missing |= flag + } + } + // A negative instantaneous residual does not invalidate its independent + // sources. Missing configuration and duplicate electrical flows do. + if reading.Reason != "" && reading.Reason != "invalid_balance" { + p.Missing |= state.MissingLoad + } + for flag, w := range map[state.HistoryMissing]*float64{state.MissingGrid: &p.GridW, state.MissingPV: &p.PVW, state.MissingBattery: &p.BatteryW, state.MissingEV: &p.EVW, state.MissingV2X: &p.V2XW} { + if math.IsNaN(*w) || math.IsInf(*w, 0) { + p.Missing |= flag + } + if p.Missing&flag != 0 { + *w = 0 + } + } + return p +} + func restoreLatestMPCDiagnostic(st *state.Store, svc *mpc.Service, now time.Time) { if st == nil || svc == nil { return diff --git a/go/internal/api/api.go b/go/internal/api/api.go index 42a06f614..3d7b29341 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -2504,6 +2504,11 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { now := time.Now() loc := now.Location() todayMidnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc) + sourceThrough, sourceMode, err := s.deps.State.HistorySourceProgress(r.Context()) + if err != nil { + http.Error(w, "history load failed", http.StatusInternalServerError) + return + } // Walk day-by-day: past days hit the immutable-day cache, today always // recomputes (in-progress). Each cache miss / today fire is one SQL @@ -2513,9 +2518,13 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { for i := days - 1; i >= 0; i-- { dayStart := todayMidnight.AddDate(0, 0, -i) dayKey := dayStart.Format("2006-01-02") + cacheKey := dayKey + if sourceMode { + cacheKey += ":source_interval_balance_v1" + } // A bounded balance episode can close just after midnight. Let its // final intervals arrive before treating the previous day as immutable. - cacheReady := dailyEnergyCacheReady(dayStart, now) + cacheReady := dailyEnergyCacheReady(dayStart, now) && (!sourceMode || sourceThrough >= dayStart.AddDate(0, 0, 1).UnixMilli()) var de state.DayEnergy if !cacheReady { @@ -2534,14 +2543,14 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { // per-day DailyEnergy SQL pass; with the table populated // the same query reduces to N PK lookups. s.dailyCacheMu.Lock() - cached, ok := s.dailyCache[dayKey] + cached, ok := s.dailyCache[cacheKey] s.dailyCacheMu.Unlock() if ok { de = cached } else if persisted, present, err := s.deps.State.LoadDailyEnergy(dayKey); err == nil && present { de = persisted s.dailyCacheMu.Lock() - s.dailyCache[dayKey] = de + s.dailyCache[cacheKey] = de s.dailyCacheMu.Unlock() } else { dayEnd := dayStart.AddDate(0, 0, 1) @@ -2553,7 +2562,7 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { } de = d s.dailyCacheMu.Lock() - s.dailyCache[dayKey] = de + s.dailyCache[cacheKey] = de s.dailyCacheMu.Unlock() // Persist for next restart. Closed days only — // today is excluded via the isToday branch above. diff --git a/go/internal/api/api_savings.go b/go/internal/api/api_savings.go index b0da60119..00435a67a 100644 --- a/go/internal/api/api_savings.go +++ b/go/internal/api/api_savings.go @@ -182,6 +182,11 @@ func (s *Server) handleSavingsDaily(w http.ResponseWriter, r *http.Request) { } s.ensureSavingsCache() + sourceThrough, sourceMode, err := s.deps.State.HistorySourceProgress(r.Context()) + if err != nil { + http.Error(w, "savings load failed", http.StatusInternalServerError) + return + } out := make([]map[string]any, 0, days) var tImpWh, tExpWh, tLoadWh, tActual, tBaseline, tSaved float64 @@ -190,11 +195,16 @@ func (s *Server) handleSavingsDaily(w http.ResponseWriter, r *http.Request) { for i := days - 1; i >= 0; i-- { dayStart := todayMidnight.AddDate(0, 0, -i) dayKey := dayStart.Format("2006-01-02") + cacheKey := dayKey + if sourceMode { + cacheKey += ":source_interval_balance_v1" + } isToday := i == 0 + dayEnd := dayStart.AddDate(0, 0, 1) var ds daySavings - if isToday { - b, err := s.deps.State.DailyCostBreakdownContext(r.Context(), dayStart.UnixMilli(), now.UnixMilli(), zone, ep) + if isToday || (sourceMode && sourceThrough < dayEnd.UnixMilli()) { + b, err := s.deps.State.DailyCostBreakdownContext(r.Context(), dayStart.UnixMilli(), min(now.UnixMilli(), dayEnd.UnixMilli()), zone, ep) if err != nil { slog.Error("handleSavingsDaily: DailyCostBreakdown failed", "err", err, "day", dayKey) http.Error(w, "savings load failed", http.StatusInternalServerError) @@ -203,7 +213,7 @@ func (s *Server) handleSavingsDaily(w http.ResponseWriter, r *http.Request) { ds = fromBreakdown(b, resolutionFor(b)) } else { s.savingsCacheMu.Lock() - cached, ok := s.savingsCache[dayKey] + cached, ok := s.savingsCache[cacheKey] s.savingsCacheMu.Unlock() if ok { ds = cached @@ -217,7 +227,7 @@ func (s *Server) handleSavingsDaily(w http.ResponseWriter, r *http.Request) { } ds = fromBreakdown(b, resolutionFor(b)) s.savingsCacheMu.Lock() - s.savingsCache[dayKey] = ds + s.savingsCache[cacheKey] = ds s.savingsCacheMu.Unlock() } } diff --git a/go/internal/api/api_source_history_test.go b/go/internal/api/api_source_history_test.go new file mode 100644 index 000000000..b5ce8c3a0 --- /dev/null +++ b/go/internal/api/api_source_history_test.go @@ -0,0 +1,96 @@ +package api + +import ( + "context" + "encoding/json" + "math" + "net/http" + "net/http/httptest" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/state" +) + +func TestSourceHistoryDayCachesWaitForDurableBoundary(t *testing.T) { + for _, firstBatch := range []bool{false, true} { + t.Run(map[bool]string{false: "existing method", true: "first source batch"}[firstBatch], func(t *testing.T) { testSourceHistoryDayCaches(t, firstBatch) }) + } +} + +func testSourceHistoryDayCaches(t *testing.T, firstBatch bool) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + srv := New(&Deps{State: st, Cfg: &config.Config{Price: &config.Price{Zone: "SE3"}}, CfgMu: &sync.RWMutex{}}) + now := time.Now() + midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).UnixMilli() + base := midnight - 60000 + tick := func(at int64) { + t.Helper() + p := state.HistoryPoint{TsMs: at, Missing: state.MissingLoad, JSON: `{}`, Sources: &state.HistorySources{AtMS: at, Topology: "grid", GridW: 1000}} + if err := st.RecordTickWithEnergy(p, nil, nil); err != nil { + t.Fatal(err) + } + } + read := func(path string) float64 { + t.Helper() + rr := httptest.NewRecorder() + srv.Handler().ServeHTTP(rr, httptest.NewRequest(http.MethodGet, path, nil)) + if rr.Code != http.StatusOK { + t.Fatal("history API failed", rr.Code) + } + var result struct { + Days []struct { + Load *float64 `json:"load_wh"` + } `json:"days"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &result); err != nil { + t.Fatal(err) + } + if len(result.Days) != 2 { + t.Fatal("missing day windows") + } + if result.Days[0].Load == nil { + return 0 + } + return *result.Days[0].Load + } + if firstBatch { + for _, path := range []string{"/api/energy/daily?days=2", "/api/savings/daily?days=2"} { + _ = read(path) + } + } + for i := int64(0); i < 6; i++ { + tick(base + i*10000) + } + if !firstBatch { + for _, path := range []string{"/api/energy/daily?days=2", "/api/savings/daily?days=2"} { + _ = read(path) + } + day := time.UnixMilli(base).In(now.Location()).Format("2006-01-02") + if _, ok := srv.dailyCache[day]; ok { + t.Fatal("energy cached an uncommitted closed day") + } + if _, ok := srv.savingsCache[day]; ok { + t.Fatal("savings cached an uncommitted closed day") + } + } + tick(midnight) + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + for _, path := range []string{"/api/energy/daily?days=2", "/api/savings/daily?days=2"} { + if got := read(path); math.Abs(got-1000.0/60) > 1e-8 { + t.Fatal("same API server retained Load from before final commit") + } + } +} diff --git a/go/internal/state/energy_ledger.go b/go/internal/state/energy_ledger.go index aab2916ac..7ab464869 100644 --- a/go/internal/state/energy_ledger.go +++ b/go/internal/state/energy_ledger.go @@ -466,7 +466,16 @@ func (s *Store) LoadEnergyHistoryContext(ctx context.Context, q EnergyHistoryQue } out = append(out, p) } - return out, truncated, rows.Err() + if err := rows.Err(); err != nil { + return nil, false, err + } + rows.Close() + var sourceTruncated bool + out, sourceTruncated, err = s.applySourceLoadLedger(ctx, q, out) + if err != nil { + return nil, false, err + } + return out, truncated || sourceTruncated, nil } // PruneEnergyLedger keeps five-minute detail for 30 days, hourly totals for diff --git a/go/internal/state/history_dashboard.go b/go/internal/state/history_dashboard.go index d21b48b94..01230129f 100644 --- a/go/internal/state/history_dashboard.go +++ b/go/internal/state/history_dashboard.go @@ -24,6 +24,9 @@ const siteEnergyCursorSchema = `CREATE TABLE IF NOT EXISTS history_site_cursor ( // A missing usable site snapshot breaks integration even if it recovers before // the maximum gap. The cursor and chart/energy rows share the tick receipt. func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoint) error { + if err := s.recordSourceEnergyTx(ctx, tx, p); err != nil { + return err + } // Older schema-7 Cores read only the original tables. Keep partial rows // separate, and clear their cursor so rollback cannot bridge a gap. if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO history_partial_cursor SELECT key,value FROM history_site_cursor`); err != nil { @@ -71,7 +74,7 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi } } from = p.TsMs - if previous > 0 && p.TsMs-previous <= maxCostIntegrationGap.Milliseconds() { + if p.Sources == nil && previous > 0 && p.TsMs-previous <= maxCostIntegrationGap.Milliseconds() { from = previous } else { previous = 0 @@ -132,6 +135,13 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi return err } } + if p.Sources != nil { + // Older writers must not integrate their first receipt backwards over + // a source-accounted period after rollback. Keep its saved prefix intact. + if _, err := tx.ExecContext(ctx, `UPDATE history_partial_cursor SET value=0 WHERE key IN ('last','residual_eligible')`); err != nil { + return err + } + } table, kind := "history_dashboard", "dashboard" if p.Missing != 0 { table, kind = "history_dashboard_partial", "partial_dashboard" @@ -212,6 +222,8 @@ func siteEVWh(p *HistoryPoint, hours float64) float64 { } type siteEnergyRow struct { + evCharge, evDischarge float64 + source bool estimateLoad float64 estimateCovered int64 missing HistoryMissing @@ -225,16 +237,35 @@ type siteEnergyRow struct { func (s *Store) walkSiteEnergy(ctx context.Context, from, to int64, visit func(siteEnergyRow) error) error { ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - rows, err := s.history.QueryContext(ctx, `WITH energy AS (SELECT *, 'energy' AS kind FROM history_site_energy UNION ALL SELECT *, 'partial_energy' AS kind FROM history_site_energy_partial) SELECT e.start_ms,e.resolution_ms,e.first_ms,e.last_ms,e.covered_ms,e.n,e.import_wh,e.export_wh,e.pv_wh,e.bat_charge_wh,e.bat_discharge_wh,e.load_wh,e.ev_wh,COALESCE(q.missing,0),CASE WHEN e.kind='partial_energy' AND (q.missing & ~16) IN (136,40) THEN COALESCE((SELECT SUM(load_wh) FROM history_load_estimates x WHERE x.start_ms>=e.start_ms AND x.start_ms=e.start_ms AND x.start_ms=? AND e.start_ms=? AND last_ms<=? ORDER BY e.start_ms`, bucketStart(from, EnergyLedgerDailyBucketMS), to, from, to) + rows, err := s.history.QueryContext(ctx, `WITH energy AS ( + SELECT *, 0 AS ev_charge_wh,0 AS ev_discharge_wh,'energy' AS kind FROM history_site_energy e WHERE NOT EXISTS(SELECT 1 FROM history_source_periods sp WHERE sp.start_mse.first_ms) + UNION ALL SELECT *,0 AS ev_charge_wh,0 AS ev_discharge_wh,'partial_energy' AS kind FROM history_site_energy_partial e WHERE NOT EXISTS(SELECT 1 FROM history_source_periods sp WHERE sp.start_mse.first_ms) + UNION ALL SELECT *, 'source_energy' AS kind FROM history_source_energy) + SELECT e.start_ms,e.resolution_ms,e.first_ms,e.last_ms,e.covered_ms,e.n,e.import_wh,e.export_wh,e.pv_wh,e.bat_charge_wh,e.bat_discharge_wh,e.load_wh,e.ev_wh,COALESCE(q.missing,0), + CASE WHEN e.kind='partial_energy' AND (q.missing & ~16) IN (136,40) THEN COALESCE((SELECT SUM(load_wh) FROM history_load_estimates x WHERE x.start_ms>=e.start_ms AND x.start_ms+x.resolution_ms<=e.start_ms+e.resolution_ms),0) ELSE 0 END, + CASE WHEN e.kind='partial_energy' AND (q.missing & ~16) IN (136,40) THEN COALESCE((SELECT SUM(covered_ms) FROM history_load_estimates x WHERE x.start_ms>=e.start_ms AND x.start_ms+x.resolution_ms<=e.start_ms+e.resolution_ms),0) ELSE 0 END,e.kind,e.ev_charge_wh,e.ev_discharge_wh + FROM energy e LEFT JOIN history_quality q ON q.kind=e.kind AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms + WHERE e.start_ms>=? AND e.start_ms=? AND e.last_ms<=? ORDER BY e.first_ms`, bucketStart(from, EnergyLedgerDailyBucketMS), to, from, to) if err != nil { return err } defer rows.Close() for rows.Next() { var r siteEnergyRow - if err := rows.Scan(&r.start, &r.width, &r.first, &r.last, &r.covered, &r.n, &r.imp, &r.exp, &r.pv, &r.charge, &r.discharge, &r.load, &r.ev, &r.missing, &r.estimateLoad, &r.estimateCovered); err != nil { + var kind string + if err := rows.Scan(&r.start, &r.width, &r.first, &r.last, &r.covered, &r.n, &r.imp, &r.exp, &r.pv, &r.charge, &r.discharge, &r.load, &r.ev, &r.missing, &r.estimateLoad, &r.estimateCovered, &kind, &r.evCharge, &r.evDischarge); err != nil { return err } + r.source = kind == "source_energy" + if r.source { + r.load = r.imp - r.exp + r.pv - r.charge + r.discharge - r.ev + r.ev = r.evCharge + if r.missing&MissingPower != 0 { + r.load = 0 + } else { + r.missing |= EstimatedLoad + } + } if err := visit(r); err != nil { return err } @@ -247,7 +278,13 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner var originalLoad float64 var originalMissing HistoryMissing var estimatedMS, coveredMS int64 + var sourceMS int64 + var first, last int64 err := s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { + if first == 0 || r.first < first { + first = r.first + } + last = max(last, r.last) out.ImportWh += r.imp out.ExportWh += r.exp out.PVWh += r.pv @@ -256,6 +293,9 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner originalLoad += r.load originalMissing |= r.missing coveredMS += r.covered + if r.source { + sourceMS += r.covered + } if r.missing&^MissingSoC == MissingLoad|BalanceMismatch || r.missing&^MissingSoC == MissingLoad|MissingEV { if r.estimateCovered > 0 && r.estimateCovered == r.covered { r.load, r.missing = r.estimateLoad, (r.missing&MissingSoC)|EstimatedLoad @@ -270,9 +310,29 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner } return nil }) - if out.LoadWh < 0 || (estimatedMS > 120000 && estimatedMS > coveredMS/20) { + if sourceMS == 0 && (out.LoadWh < 0 || (estimatedMS > 120000 && estimatedMS > coveredMS/20)) { out.LoadWh = originalLoad out.Missing = originalMissing } + if out.LoadWh < 0 { + out.LoadWh = 0 + out.Missing |= MissingLoad + } + cut, hasSources, coverageErr := s.sourceEnergyFrom(ctx) + if coverageErr != nil { + return out, coverageErr + } + if hasSources && to > cut { + expected, err := s.sourcePeriodCoverage(ctx, max(from, cut), to) + if err != nil { + return out, err + } + if sourceMS < expected { + out.Missing |= MissingLoad + } + if coveredMS < last-first { + out.Missing |= MissingLoad + } + } return out, err } diff --git a/go/internal/state/history_dashboard_cost.go b/go/internal/state/history_dashboard_cost.go index 926a84218..f5fd43ae9 100644 --- a/go/internal/state/history_dashboard_cost.go +++ b/go/internal/state/history_dashboard_cost.go @@ -35,10 +35,42 @@ func (s *Store) integrateHistoryRange(ctx context.Context, since, until int64, s out = addDayCost(out, old) } idx := 0 + sourcePrices := make(map[int]siteEnergyRow) + var sourceLoad float64 + var sourceCovered int64 err = s.walkSiteEnergy(ctx, max(since, from), until, func(r siteEnergyRow) error { out.ImportWh += r.imp out.ExportWh += r.exp out.LoadWh += r.load + if r.source { + if r.missing&MissingLoad == 0 { + sourceLoad += r.load + sourceCovered += r.covered + } + if r.missing&MissingGrid != 0 { + return nil + } + for idx < len(slots) && slots[idx].EndMs <= r.first { + idx++ + } + if idx < len(slots) && slots[idx].StartMs <= r.first && slots[idx].EndMs >= r.last { + slot := slots[idx] + // Measured grid cost does not depend on a household residual. + out.ImportCostOre += r.imp * slot.TotalOre / 1000 + out.ExportRevenueOre += r.exp * gridcost.ExportPriceOre(slot.SpotOreKwh, ep) / 1000 + a := sourcePrices[idx] + if a.covered == 0 { + a.first = r.first + } + a.last = r.last + a.load += r.load + a.ev += r.ev + a.covered += r.covered + a.missing |= r.missing + sourcePrices[idx] = a + } + return nil + } if r.missing&(MissingGrid|MissingLoad) != 0 { return nil } @@ -58,6 +90,20 @@ func (s *Store) integrateHistoryRange(ctx context.Context, since, until int64, s } return nil }) + if sourceLoad >= 0 { + out.HistoryCoveredMs += sourceCovered + } + for idx, r := range sourcePrices { + if r.missing&MissingLoad != 0 || r.load < 0 || r.covered != r.last-r.first { + continue + } + out.PricedCoveredMs += r.covered + out.BaselineHouseOre += r.load * slots[idx].TotalOre / 1000 + out.EVWh += r.ev + } + if out.LoadWh < 0 { + out.LoadWh = 0 + } return out, err } diff --git a/go/internal/state/history_dashboard_retention.go b/go/internal/state/history_dashboard_retention.go index 34d7e71d2..2cf1f7f59 100644 --- a/go/internal/state/history_dashboard_retention.go +++ b/go/internal/state/history_dashboard_retention.go @@ -183,10 +183,13 @@ func (s *Store) rollSiteEnergy(ctx context.Context, cutoff int64) error { return err } } - return nil + return s.rollSiteEnergyStorage(ctx, cutoff, sourceEnergyTable, "source_energy") } func (s *Store) rollSiteEnergyTable(ctx context.Context, cutoff int64, partial bool) error { table, kind := energyStorage(partial) + return s.rollSiteEnergyStorage(ctx, cutoff, table, kind) +} +func (s *Store) rollSiteEnergyStorage(ctx context.Context, cutoff int64, table, kind string) error { width := int64((15 * time.Minute).Milliseconds()) cutoff = bucketStart(cutoff, width) for { @@ -206,9 +209,21 @@ func (s *Store) rollSiteEnergyTable(ctx context.Context, cutoff int64, partial b if err := tx.QueryRowContext(ctx, `SELECT COALESCE(`+historyMissingAggregate+`,0) FROM history_quality WHERE kind='`+kind+`' AND resolution_ms=? AND start_ms>=? AND start_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND last_ms<=?`, from, to, from, to).Scan(&covered) + err := s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { covered += r.covered; return nil }) return covered, err } diff --git a/go/internal/state/history_plain.go b/go/internal/state/history_plain.go index 2e1656a27..004eeb973 100644 --- a/go/internal/state/history_plain.go +++ b/go/internal/state/history_plain.go @@ -33,6 +33,7 @@ const ( // readers still need these frozen tiers. Import cursors and Parquet receipts // travel with the hours they produced so restart cannot count samples twice. var plainHistoryTables = []string{ + sourceEnergyTable, "history_source_cursor", "history_source_periods", "history_hot", "history_warm", "history_cold", "ts_drivers", "ts_metrics", "ts_latest", "energy_daily", "energy_ledger_meta", "energy_assets", "energy_ledger_entries", "energy_ledger_cursors", diff --git a/go/internal/state/history_schema.go b/go/internal/state/history_schema.go index 19d5a048c..1c4e02e73 100644 --- a/go/internal/state/history_schema.go +++ b/go/internal/state/history_schema.go @@ -5,6 +5,9 @@ import "strings" var historySchema = append(append([]string{}, sqliteLegacyHistoryStmts...), `CREATE TABLE IF NOT EXISTS ts_latest (driver_id INTEGER NOT NULL,metric_id INTEGER NOT NULL,ts_ms INTEGER NOT NULL,value REAL NOT NULL,PRIMARY KEY(driver_id,metric_id)) WITHOUT ROWID`, loadEstimateSchema, loadEstimatePendingSchema, dashboardSchema, siteEnergySchema, siteEnergyCursorSchema, historyQualitySchema, historyDailyQualitySchema, + sourceCursorSchema, sourcePeriodsSchema, strings.Replace(strings.Replace(siteEnergySchema, "history_site_energy", sourceEnergyTable, 1), "ev_wh REAL NOT NULL,", "ev_wh REAL NOT NULL,ev_charge_wh REAL NOT NULL,ev_discharge_wh REAL NOT NULL,", 1), + `CREATE INDEX IF NOT EXISTS idx_source_energy_age ON history_source_energy(resolution_ms,start_ms)`, + `CREATE INDEX IF NOT EXISTS idx_source_period_end ON history_source_periods(last_ms,start_ms)`, strings.Replace(dashboardSchema, "history_dashboard", "history_dashboard_partial", 1), strings.Replace(siteEnergySchema, "history_site_energy", "history_site_energy_partial", 1), strings.Replace(siteEnergyCursorSchema, "history_site_cursor", "history_partial_cursor", 1), `CREATE TABLE IF NOT EXISTS energy_daily_partial(day TEXT PRIMARY KEY,import_wh REAL NOT NULL,export_wh REAL NOT NULL,pv_wh REAL NOT NULL,bat_charged_wh REAL NOT NULL,bat_discharged_wh REAL NOT NULL,load_wh REAL NOT NULL,computed_at_ms INTEGER NOT NULL) STRICT`, `CREATE INDEX IF NOT EXISTS idx_partial_dashboard_last ON history_dashboard_partial(last_ms)`, diff --git a/go/internal/state/history_sources.go b/go/internal/state/history_sources.go new file mode 100644 index 000000000..6786b3351 --- /dev/null +++ b/go/internal/state/history_sources.go @@ -0,0 +1,334 @@ +package state + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "math" + "sort" + "time" +) + +// HistorySources carries independently fresh site-side powers, never the live +// Load result. Topology identifies the electrical flows included in the sum. +type HistorySources struct { + AtMS int64 + Topology string + Missing HistoryMissing + GridW, PVW, BatteryW, EVW, V2XW float64 +} + +const sourcePeriodsSchema = `CREATE TABLE IF NOT EXISTS history_source_periods(start_ms INTEGER PRIMARY KEY,last_ms INTEGER NOT NULL) WITHOUT ROWID` + +const sourceCursorSchema = `CREATE TABLE IF NOT EXISTS history_source_cursor(key TEXT PRIMARY KEY,value TEXT NOT NULL) WITHOUT ROWID` + +// This uses the site energy columns so tariff and day readers share exactly +// the same evidence. load_wh stays zero: Load is derived after interval sums. +// ev_wh is the signed EV+V2X energy, including discharge. +const sourceEnergyTable = "history_source_energy" +const maxSourceEnergyWidth int64 = 900000 + +func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *HistoryPoint) error { + if p == nil || p.Sources == nil { + _, err := tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key='last'`) + return err + } + at := p.Sources.AtMS + if at <= 0 { + return errors.New("history source snapshot has no observation time") + } + var last struct { + At int64 + Start int64 + Sources HistorySources + } + var raw string + err := tx.QueryRowContext(ctx, `SELECT value FROM history_source_cursor WHERE key='last'`).Scan(&raw) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if raw != "" { + if err := json.Unmarshal([]byte(raw), &last); err != nil { + return err + } + } + var through int64 + if err := tx.QueryRowContext(ctx, `SELECT COALESCE((SELECT CAST(value AS INTEGER) FROM history_source_cursor WHERE key='through'),0)`).Scan(&through); err != nil { + return err + } + if at <= through { + if at < through { + _, err = tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key='last'`) + } + return err + } + if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO history_source_cursor VALUES('from',?)`, at); err != nil { + return err + } + if last.At == 0 { + last.Start = at + } + if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_periods VALUES(?,?) ON CONFLICT(start_ms) DO UPDATE SET last_ms=excluded.last_ms`, last.Start, at); err != nil { + return err + } + // Hold the previous receipt forward, bounded by the next safe tick. Do + // not assign a newly arrived charger value to time before its receipt. + if last.At > 0 && at-last.At <= maxLoadEstimateTickMS && last.Sources.Topology == p.Sources.Topology { + // A delayed writer may finish a just-closed local day. Invalidate only + // the nearby cached dates; source rows and old history stay intact. + low := time.UnixMilli(last.At).UTC().AddDate(0, 0, -1).Format("2006-01-02") + high := time.UnixMilli(at).UTC().AddDate(0, 0, 1).Format("2006-01-02") + for _, table := range []string{"energy_daily", "energy_daily_partial", "history_daily_quality"} { + if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE day>=? AND day<=?`, low, high); err != nil { + return err + } + } + v := last.Sources + missing := (v.Missing | p.Sources.Missing) & MissingPower + if missing != 0 { + missing |= MissingLoad + } + for from := last.At; from < at; { + start := bucketStart(from, ArchiveResolutionMS) + to := min(at, start+ArchiveResolutionMS) + ms := to - from + h := float64(ms) / 3600000 + if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_energy VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(start_ms,resolution_ms) DO UPDATE SET first_ms=MIN(first_ms,excluded.first_ms),last_ms=MAX(last_ms,excluded.last_ms),covered_ms=covered_ms+excluded.covered_ms,n=n+excluded.n, + import_wh=import_wh+excluded.import_wh,export_wh=export_wh+excluded.export_wh,pv_wh=pv_wh+excluded.pv_wh,bat_charge_wh=bat_charge_wh+excluded.bat_charge_wh,bat_discharge_wh=bat_discharge_wh+excluded.bat_discharge_wh,ev_wh=ev_wh+excluded.ev_wh,ev_charge_wh=ev_charge_wh+excluded.ev_charge_wh,ev_discharge_wh=ev_discharge_wh+excluded.ev_discharge_wh`, + start, ArchiveResolutionMS, from, to, ms, 1, + math.Max(0, v.GridW)*h, math.Max(0, -v.GridW)*h, -v.PVW*h, + math.Max(0, v.BatteryW)*h, math.Max(0, -v.BatteryW)*h, 0, (v.EVW+v.V2XW)*h, (math.Max(0, v.EVW)+math.Max(0, v.V2XW))*h, (math.Max(0, -v.EVW)+math.Max(0, -v.V2XW))*h); err != nil { + return err + } + if err := putHistoryQuality(ctx, tx, "source_energy", start, ArchiveResolutionMS, missing); err != nil { + return err + } + from = to + } + } + last.At, last.Sources = at, *p.Sources + b, err := json.Marshal(last) + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, string(b)); err != nil { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('through',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, at) + return err +} + +func (s *Store) sourceEnergyFrom(ctx context.Context) (int64, bool, error) { + var from int64 + err := s.history.QueryRowContext(ctx, `SELECT CAST(value AS INTEGER) FROM history_source_cursor WHERE key='from'`).Scan(&from) + if errors.Is(err, sql.ErrNoRows) { + return 0, false, nil + } + return from, err == nil, err +} + +// HistorySourceProgress is the last committed receipt time. Once it reaches +// a day boundary, later source integration cannot change that closed day. +func (s *Store) HistorySourceProgress(ctx context.Context) (int64, bool, error) { + var through int64 + err := s.history.QueryRowContext(ctx, `SELECT CAST(value AS INTEGER) FROM history_source_cursor WHERE key='through'`).Scan(&through) + if errors.Is(err, sql.ErrNoRows) { + return 0, false, nil + } + return through, err == nil, err +} + +func (s *Store) sourcePeriodCoverage(ctx context.Context, from, to int64) (int64, error) { + var ms int64 + err := s.history.QueryRowContext(ctx, `SELECT COALESCE(SUM(MIN(last_ms,?)-MAX(start_ms,?)),0) FROM history_source_periods WHERE last_ms>? AND start_ms=? ORDER BY start_ms`, to, from) + if err != nil { + return nil, err + } + defer rows.Close() + var out [][2]int64 + for rows.Next() { + var v [2]int64 + if err := rows.Scan(&v[0], &v[1]); err != nil { + return nil, err + } + out = append(out, v) + if len(out) > maxRawSeriesPoints { + return nil, ErrHistoryQueryLimit + } + } + return out, rows.Err() +} + +type sourceLoadBucket struct { + start, width, first, last, covered, n int64 + wh float64 + missing HistoryMissing +} + +// Whole source minutes only: never prorate a stored bucket across a gap or a +// price boundary. Signed minutes survive until this requested interval sums. +func (s *Store) sourceLoadBuckets(ctx context.Context, from, to, width int64) ([]sourceLoadBucket, error) { + rows, err := s.history.QueryContext(ctx, `SELECT e.start_ms,e.resolution_ms,e.first_ms,e.last_ms,e.covered_ms,e.n, + e.import_wh-e.export_wh+e.pv_wh-e.bat_charge_wh+e.bat_discharge_wh-e.ev_wh,COALESCE(q.missing,0) + FROM history_source_energy e LEFT JOIN history_quality q ON q.kind='source_energy' AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms + WHERE e.start_ms>=? AND e.start_ms? ORDER BY e.start_ms`, from-maxSourceEnergyWidth, to, from) + if err != nil { + return nil, err + } + defer rows.Close() + byStart := map[int64]*sourceLoadBucket{} + for rows.Next() { + var r sourceLoadBucket + if err := rows.Scan(&r.start, &r.width, &r.first, &r.last, &r.covered, &r.n, &r.wh, &r.missing); err != nil { + return nil, err + } + start := from + (r.start-from)/width*width + if r.start < from { + start = from + } + w := max(width, r.width) + b := byStart[start] + if b == nil { + b = &sourceLoadBucket{start: start, width: w, first: r.first, last: r.last} + byStart[start] = b + } + if r.first < from || r.last > to || r.first < start || r.last > start+w { + b.missing |= MissingLoad + continue + } + b.first = min(b.first, r.first) + b.last = max(b.last, r.last) + b.covered += r.covered + b.n += r.n + b.wh += r.wh + b.missing |= r.missing + } + if err := rows.Err(); err != nil { + return nil, err + } + out := make([]sourceLoadBucket, 0, len(byStart)) + for _, b := range byStart { + // Only a complete common interval is a chart mean. A partial tail + // stays live-only, never diluted by padding unobserved time with zero. + if b.covered != b.width || b.first != b.start || b.last != b.start+b.width || b.wh < 0 || !finiteHistoryValue(b.wh) { + b.missing |= MissingLoad + } + out = append(out, *b) + } + sort.Slice(out, func(i, j int) bool { return out[i].start < out[j].start }) + return out, nil +} + +func finiteHistoryValue(v float64) bool { return !math.IsNaN(v) && !math.IsInf(v, 0) } + +func (s *Store) applySourceLoadHistory(ctx context.Context, points []HistoryPoint, from, to, width int64) error { + cut, ok, err := s.sourceEnergyFrom(ctx) + if err != nil || !ok || to <= cut { + return err + } + buckets, err := s.sourceLoadBuckets(ctx, from, to, width) + if err != nil { + return err + } + periods, err := s.sourcePeriods(ctx, from, to) + if err != nil { + return err + } + byStart := map[int64]sourceLoadBucket{} + for _, b := range buckets { + byStart[b.start] = b + } + for i := range points { + p := &points[i] + if p.TsMs < cut { + continue + } + inPeriod := false + for _, v := range periods { + if v[0] <= p.TsMs && p.TsMs <= v[1] { + inPeriod = true + break + } + } + if !inPeriod { + continue + } + start := from + (p.TsMs-from)/width*width + p.Missing &^= MissingLoad | BalanceMismatch | EstimatedLoad + p.LoadW = 0 + b, exists := byStart[start] + if !exists || b.covered == 0 || b.missing&MissingPower != 0 { + p.Missing |= MissingLoad + } else { + p.LoadW = b.wh * 3600000 / float64(b.width) + p.Missing |= EstimatedLoad + } + // JSON must not retain the old instantaneous Load alongside the column. + var detail map[string]any + if json.Unmarshal([]byte(p.JSON), &detail) == nil && detail != nil { + detail["load_house_w"] = p.Missing.Value(MissingLoad, p.LoadW) + detail["history_load_method"] = "source_interval_balance_v1" + if raw, err := json.Marshal(detail); err == nil { + p.JSON = string(raw) + } + } + } + return nil +} + +func (s *Store) applySourceLoadLedger(ctx context.Context, q EnergyHistoryQuery, points []EnergyLedgerPoint) ([]EnergyLedgerPoint, bool, error) { + if q.AssetID != "" { + return points, false, nil + } + cut, ok, err := s.sourceEnergyFrom(ctx) + if err != nil || !ok || q.UntilMS <= cut { + return points, false, err + } + readTo := q.UntilMS + if q.BucketMS < (q.UntilMS-q.SinceMS)/int64(q.Limit+1) { + readTo = q.SinceMS + int64(q.Limit+1)*q.BucketMS + } + buckets, err := s.sourceLoadBuckets(ctx, q.SinceMS, readTo, q.BucketMS) + if err != nil { + return nil, false, err + } + periods, err := s.sourcePeriods(ctx, q.SinceMS, q.UntilMS) + if err != nil { + return nil, false, err + } + out := points[:0] + for _, p := range points { + overlaps := false + if p.Flow == FlowConsumerUse { + for _, v := range periods { + if v[0] < p.BucketStartMS+p.BucketLenMS && v[1] > p.BucketStartMS { + overlaps = true + break + } + } + } + if overlaps { + continue + } + out = append(out, p) + } + truncated := false + for _, b := range buckets { + if b.covered == 0 || b.missing&MissingPower != 0 { + continue + } + if (b.start-q.SinceMS)/q.BucketMS >= int64(q.Limit) { + truncated = true + continue + } + out = append(out, EnergyLedgerPoint{SchemaVersion: EnergyLedgerSchemaVersion, AssetID: "system", Flow: FlowConsumerUse, BucketStartMS: b.start, BucketLenMS: b.width, EnergyWh: b.wh, Source: "power_telemetry", Quality: "integrated", Provenance: "source_interval_balance_v1", SampleCount: b.n}) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].BucketStartMS < out[j].BucketStartMS }) + return out, truncated, nil +} diff --git a/go/internal/state/history_sources_test.go b/go/internal/state/history_sources_test.go new file mode 100644 index 000000000..87ba1e151 --- /dev/null +++ b/go/internal/state/history_sources_test.go @@ -0,0 +1,376 @@ +package state + +import ( + "context" + "database/sql" + "fmt" + "path/filepath" + "testing" + "time" +) + +func sourceTick(t *testing.T, s *Store, at int64, v HistorySources) { + t.Helper() + v.AtMS = at + if v.Topology == "" { + v.Topology = "grid+charger" + } + // Deliberately contradictory instantaneous Load: readers must use sources. + p := HistoryPoint{TsMs: at, LoadW: 999999, Missing: MissingLoad | BalanceMismatch, JSON: `{"ev_w":0,"v2x_w":0}`, Sources: &v} + dashboardTick(t, s, fmt.Sprint(at), &p) +} + +func sourceStore(t *testing.T) *Store { + t.Helper() + s := freshStore(t) + if err := s.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + return s +} + +func sourceSeries(t *testing.T, s *Store, base int64, minutes []float64) { + t.Helper() + for i := int64(0); i <= int64(len(minutes))*6; i++ { + minute := min(int(i/6), len(minutes)-1) + sourceTick(t, s, base+i*10000, HistorySources{GridW: minutes[minute] + 2000, EVW: 2000}) + } +} + +func TestSourceLoadBalancesAfterAggregation(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + sourceSeries(t, s, base, []float64{-1000, 3000, 1000, 1000, 1000}) + ctx := context.Background() + minute, err := s.sourceLoadBuckets(ctx, base, base+300000, 60000) + if err != nil || len(minute) != 5 { + t.Fatal("minute evidence unavailable", err) + } + if minute[0].missing&MissingLoad == 0 { + t.Fatal("negative entire minute became valid Load") + } + check := func(s *Store) { + t.Helper() + b, err := s.sourceLoadBuckets(ctx, base, base+300000, 300000) + if err != nil || len(b) != 1 || b[0].missing&MissingLoad != 0 { + t.Fatal("signed interval balance missing", err) + } + near(t, "interval Wh", b[0].wh, 1000.0/12) + d, err := s.DailyEnergy(base, base+300000) + if err != nil || d.Missing&MissingLoad != 0 { + t.Fatal("daily balance missing", err) + } + near(t, "daily Wh", d.LoadWh, b[0].wh) + web, err := s.LoadHistory(base, base+300000, 1) + if err != nil || len(web) != 1 || web[0].Missing&MissingLoad != 0 { + t.Fatal("web balance missing", err) + } + near(t, "web W", web[0].LoadW, 1000) + native, _, err := s.LoadEnergyHistory(EnergyHistoryQuery{SinceMS: base, UntilMS: base + 300000, BucketMS: 300000, Limit: 10}) + if err != nil { + t.Fatal(err) + } + var wh float64 + var n int + for _, p := range native { + if p.Flow == FlowConsumerUse { + wh += p.EnergyWh + n++ + } + } + if n != 1 { + t.Fatal("native household missing or duplicated") + } + near(t, "native Wh", wh, b[0].wh) + } + check(s) + path := s.mainDBPath + if err := s.Close(); err != nil { + t.Fatal(err) + } + reopened, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + check(reopened) +} + +func TestSourceLoadKeepsUnknownDistinctFromZero(t *testing.T) { + for _, kind := range []string{"zero", "missing EV", "write gap", "topology", "partial tail", "time reversal", "duplicate"} { + t.Run(kind, func(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for i := int64(0); i <= 6; i++ { + if kind == "write gap" && i == 3 { + continue + } + if kind == "partial tail" && i > 1 { + break + } + v := HistorySources{} + if kind == "missing EV" && i == 3 { + v.Missing = MissingEV + } + if kind == "topology" && i >= 3 { + v.Topology = "new meter+charger" + } + if kind == "time reversal" && i == 3 { + sourceTick(t, s, base+5000, v) + } + sourceTick(t, s, base+i*10000, v) + if kind == "duplicate" { + sourceTick(t, s, base+i*10000, v) + } + } + b, err := s.sourceLoadBuckets(context.Background(), base, base+60000, 60000) + if err != nil || len(b) != 1 { + t.Fatal(err) + } + known := b[0].missing&MissingLoad == 0 + if known != (kind == "zero" || kind == "duplicate") { + t.Fatal("missing evidence classified as measured zero", kind) + } + if known { + near(t, "zero Wh", b[0].wh, 0) + } + }) + } +} + +func TestSourceLoadUsesSnapshotTimeAndDoesNotBridgeRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.db") + s, err := Open(path) + if err != nil { + t.Fatal(err) + } + if err := s.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for i := int64(0); i < 3; i++ { + p := HistoryPoint{TsMs: base + i*10000, JSON: `{}`, Sources: &HistorySources{AtMS: base + 20000 + i*10000, Topology: "grid", GridW: 1000}} + dashboardTick(t, s, fmt.Sprint(i), &p) + } + var first, covered int64 + if err := s.history.QueryRow(`SELECT MIN(first_ms),SUM(covered_ms) FROM history_source_energy`).Scan(&first, &covered); err != nil { + t.Fatal(err) + } + if first != base+20000 || covered != 20000 { + t.Fatal("source power assigned before observation") + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + s, err = Open(path) + if err != nil { + t.Fatal(err) + } + defer s.Close() + if err := s.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + sourceTick(t, s, base+50000, HistorySources{Topology: "grid", GridW: 1000}) + if err := s.history.QueryRow(`SELECT SUM(covered_ms) FROM history_source_energy`).Scan(&covered); err != nil { + t.Fatal(err) + } + if covered != 20000 { + t.Fatal("integrated across stopped process") + } +} + +func TestSourceLoadKeepsOlderWriterPeriod(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + sourceSeries(t, s, base, []float64{1000}) + // Simulate rollback: its writer has no Sources field. A later new writer + // must keep this separately recorded minute without double-counting it. + for i := int64(0); i <= 6; i++ { + p := HistoryPoint{TsMs: base + 120000 + i*10000, GridW: 1000, LoadW: 1000, JSON: `{"ev_w":0,"v2x_w":0}`} + dashboardTick(t, s, "old-"+fmt.Sprint(i), &p) + } + sourceSeries(t, s, base+240000, []float64{1000}) + d, err := s.DailyEnergy(base, base+300000) + if err != nil { + t.Fatal(err) + } + // Two source minutes and the older writer's minute (its opening interval + // follows the older integration policy and is deliberately queried alone). + old, err := s.DailyEnergy(base+120000, base+180000) + if err != nil { + t.Fatal(err) + } + near(t, "older writer Wh", old.LoadWh, 1000.0/60) + if d.LoadWh < 3*1000.0/60-1e-8 { + t.Fatal("rollback history hidden by source method") + } +} + +func TestSourceLoadTariffBalanceAndIndependentGridCost(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + sourceSeries(t, s, base, []float64{-1000, 3000}) + slots := []priceSlot{{StartMs: base, EndMs: base + 60000, TotalOre: 100}, {StartMs: base + 60000, EndMs: base + 120000, TotalOre: 200}} + c, err := s.integrateHistoryRange(context.Background(), base, base+120000, slots, ExportPricing{}) + if err != nil { + t.Fatal(err) + } + near(t, "valid tariff baseline", c.BaselineHouseOre, 50*200.0/1000) + near(t, "independent import cost", c.ImportCostOre, (1000.0/60)*.1+(5000.0/60)*.2) + if c.PricedCoveredMs != 60000 { + t.Fatal("negative tariff balance counted as priced household use") + } + whole := []priceSlot{{StartMs: base, EndMs: base + 120000, TotalOre: 100}} + c, err = s.integrateHistoryRange(context.Background(), base, base+120000, whole, ExportPricing{}) + if err != nil { + t.Fatal(err) + } + near(t, "common tariff balance", c.BaselineHouseOre, (2000.0/60)*.1) +} + +func TestSourceLoadRetentionBackupAndDayBoundary(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).Add(-15 * time.Minute).UnixMilli() + minutes := make([]float64, 30) + for i := range minutes { + minutes[i] = 1000 + } + sourceSeries(t, s, base, minutes) + // A closed-day cache can exist while the last delayed receipt commits. + day := time.UnixMilli(base).UTC().Format("2006-01-02") + if err := s.SaveDailyEnergy(day, DayEnergy{LoadWh: 1, Intervals: 1}); err != nil { + t.Fatal(err) + } + sourceTick(t, s, base+1800000+10000, HistorySources{GridW: 3000, EVW: 2000}) + if _, hit, err := s.LoadDailyEnergy(day); err != nil || hit { + t.Fatal("affected closed-day cache retained stale Load", err) + } + for _, bounds := range [][2]int64{{base, base + 900000}, {base + 900000, base + 1800000}} { + d, err := s.DailyEnergy(bounds[0], bounds[1]) + if err != nil { + t.Fatal(err) + } + near(t, "day boundary Wh", d.LoadWh, 250) + } + if err := s.rollSiteEnergy(context.Background(), base+1800000); err != nil { + t.Fatal(err) + } + b, err := s.sourceLoadBuckets(context.Background(), base, base+1800000, 900000) + if err != nil || len(b) != 2 { + t.Fatal("retained source quarters missing", err) + } + for _, v := range b { + if v.missing&MissingLoad != 0 { + t.Fatal("retention lost coverage") + } + near(t, "retained Wh", v.wh, 250) + } + backup := filepath.Join(t.TempDir(), "restored.db") + if err := s.copyStateForBackup(context.Background(), backup); err != nil { + t.Fatal(err) + } + if err := s.exportHistoryToSQLite(context.Background(), backup); err != nil { + t.Fatal(err) + } + db, err := openRaw(backup) + if err != nil { + t.Fatal(err) + } + defer db.Close() + for _, table := range []string{sourceEnergyTable, "history_source_cursor", "history_source_periods"} { + a, na, err := sourceTableDigest(s.history, table) + if err != nil { + t.Fatal(err) + } + b, nb, err := sourceTableDigest(db, table) + if err != nil { + t.Fatal(err) + } + if na != nb || a != b { + t.Fatal("backup changed source evidence") + } + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + restored, err := Open(backup) + if err != nil { + t.Fatal(err) + } + defer restored.Close() + d, err := restored.DailyEnergy(base, base+1800000) + if err != nil { + t.Fatal(err) + } + near(t, "restored household Wh", d.LoadWh, 500) +} + +func sourceTableDigest(db *sql.DB, table string) (string, int64, error) { + rows, err := db.Query(`SELECT * FROM ` + table + historyOrder(table)) + if err != nil { + return "", 0, err + } + defer rows.Close() + return hashHistoryRows(rows) +} + +func TestSourceLoadBackupRefusesIncompleteMethod(t *testing.T) { + s := sourceStore(t) + if _, err := s.history.Exec(`DROP TABLE history_source_periods`); err != nil { + t.Fatal(err) + } + if err := s.exportHistoryToSQLite(context.Background(), filepath.Join(t.TempDir(), "history.db")); err == nil { + t.Fatal("export accepted incomplete source method") + } +} + +func TestSourceLoadTruncationWithoutLegacyLedger(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + minutes := make([]float64, 15) + for i := range minutes { + minutes[i] = 1000 + } + sourceSeries(t, s, base, minutes) + p, truncated, err := s.LoadEnergyHistory(EnergyHistoryQuery{SinceMS: base, UntilMS: base + 900000, BucketMS: 300000, Limit: 1}) + if err != nil || !truncated || len(p) != 1 { + t.Fatal("source-only query lost truncation metadata", err) + } +} + +func TestSourceLoadMidMinuteTransitionKeepsLegacyPrefix(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for i := int64(0); i <= 3; i++ { + p := HistoryPoint{TsMs: base + i*10000, GridW: 1000, LoadW: 1000, Missing: MissingSoC, JSON: `{}`} + dashboardTick(t, s, "legacy-prefix-"+fmt.Sprint(i), &p) + } + for i := int64(3); i <= 12; i++ { + sourceTick(t, s, base+i*10000, HistorySources{GridW: 1000}) + } + d, err := s.DailyEnergy(base, base+120000) + if err != nil { + t.Fatal(err) + } + near(t, "source and legacy Wh", d.LoadWh, 1000.0/30) + near(t, "source and legacy import", d.ImportWh, 1000.0/30) + var legacyLoad float64 + if err := s.history.QueryRow(`SELECT SUM(load_wh) FROM (SELECT load_wh FROM history_site_energy UNION ALL SELECT load_wh FROM history_site_energy_partial)`).Scan(&legacyLoad); err != nil { + t.Fatal(err) + } + near(t, "unchanged prefix", legacyLoad, 1000.0/120) +} + +func TestSourceLoadV2XDischargeIsNotNegativeEVCharge(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for i := int64(0); i <= 6; i++ { + sourceTick(t, s, base+i*10000, HistorySources{GridW: -1000, V2XW: -2000}) + } + c, err := s.integrateHistoryRange(context.Background(), base, base+60000, []priceSlot{{StartMs: base, EndMs: base + 60000, TotalOre: 100}}, ExportPricing{}) + if err != nil { + t.Fatal(err) + } + near(t, "V2X house Wh", c.LoadWh, 1000.0/60) + near(t, "EV charging Wh", c.EVWh, 0) +} diff --git a/go/internal/state/history_sqlite.go b/go/internal/state/history_sqlite.go index b0a8e3a6c..9fd37d286 100644 --- a/go/internal/state/history_sqlite.go +++ b/go/internal/state/history_sqlite.go @@ -26,7 +26,7 @@ var historyTables = []string{ } var sqliteHistoryBaseTables = append(append([]string{}, historyTables...), "ts_series_hour", "ts_archive_days", "ts_buckets", "ts_bucket_days", "ts_legacy_bucket_days", "ts_aggregate_hours", "ts_latest", "history_dashboard", "history_site_energy", "history_site_cursor") var historyQualityTables = []string{"history_quality", "history_daily_quality", "history_dashboard_partial", "history_site_energy_partial", "history_partial_cursor", "energy_daily_partial"} -var sqliteHistoryTables = append(append(append([]string{}, sqliteHistoryBaseTables...), historyQualityTables...), "history_load_estimates", "history_load_estimate_pending") +var sqliteHistoryTables = append(append(append([]string{}, sqliteHistoryBaseTables...), historyQualityTables...), "history_load_estimates", "history_load_estimate_pending", sourceEnergyTable, "history_source_cursor", "history_source_periods") func ensureHistorySchema(exec func(string) error) error { for _, stmt := range historySchema { @@ -75,6 +75,13 @@ func (s *Store) openHistory() error { s.history = nil } }() + var sourceTables int + if err := db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN ('history_source_energy','history_source_cursor','history_source_periods')`).Scan(&sourceTables); err != nil { + return err + } + if sourceTables != 0 && sourceTables != 3 { + return errors.New("incomplete history source tables; restore a verified full backup") + } if err := ensureHistorySchema(func(stmt string) error { _, err := db.Exec(stmt); return err }); err != nil { return fmt.Errorf("history schema: %w", err) } @@ -149,6 +156,7 @@ func (s *Store) openHistory() error { // A process stop has no continuous sampling proof, even when it was brief. // Keep completed energy and quality, but never bridge the stopped process. if _, err := s.history.Exec(`DELETE FROM history_load_estimate_pending; + DELETE FROM history_source_cursor WHERE key='last'; UPDATE history_partial_cursor SET value=0 WHERE key IN ('last','residual_eligible'); UPDATE history_site_cursor SET value=0 WHERE key='last'`); err != nil { return err @@ -373,8 +381,10 @@ func historyOrder(table string) string { return " ORDER BY path" case "ts_latest": return " ORDER BY driver_id, metric_id" - case "history_load_estimates", "history_dashboard", "history_site_energy", "history_dashboard_partial", "history_site_energy_partial": + case "history_load_estimates", "history_dashboard", "history_site_energy", "history_dashboard_partial", "history_site_energy_partial", sourceEnergyTable: return " ORDER BY start_ms, resolution_ms" + case "history_source_periods": + return " ORDER BY start_ms" case "history_load_estimate_pending": return " ORDER BY from_ms" case "ts_buckets": @@ -503,7 +513,14 @@ func (s *Store) exportHistoryToSQLite(ctx context.Context, path string) error { return errors.New("incomplete history quality tables; keep the source and restore a verified full backup") } // Estimates are additive and optional in published older backups. - for _, table := range []string{"history_load_estimates", "history_load_estimate_pending"} { + var sourceTables int + if err := src.QueryRowContext(ctx, `SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN ('history_source_energy','history_source_cursor','history_source_periods')`).Scan(&sourceTables); err != nil { + return err + } + if sourceTables != 0 && sourceTables != 3 { + return errors.New("incomplete history source tables; keep the source and restore a verified full backup") + } + for _, table := range []string{"history_load_estimates", "history_load_estimate_pending", sourceEnergyTable, "history_source_cursor", "history_source_periods"} { var exists int if err := src.QueryRowContext(ctx, `SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?`, table).Scan(&exists); err != nil { return err @@ -582,6 +599,13 @@ func validateHistorySamples(samples []Sample) error { } func normalizeHistoryPoint(p HistoryPoint) (HistoryPoint, error) { + if p.Sources != nil { + for _, v := range []float64{p.Sources.GridW, p.Sources.PVW, p.Sources.BatteryW, p.Sources.EVW, p.Sources.V2XW} { + if !finiteHistoryValue(v) { + return p, errors.New("non-finite history source") + } + } + } for _, v := range []*float64{&p.GridW, &p.PVW, &p.BatW, &p.LoadW, &p.BatSoC} { if math.IsNaN(*v) || math.IsInf(*v, 0) { return p, errors.New("non-finite history point") diff --git a/go/internal/state/ledger_flows.go b/go/internal/state/ledger_flows.go index 09eb21e12..740e82e77 100644 --- a/go/internal/state/ledger_flows.go +++ b/go/internal/state/ledger_flows.go @@ -20,6 +20,16 @@ func (s *Store) LedgerFlowBuckets(ctx context.Context, since, until int64) ([]Le if s == nil || s.history == nil || until <= since { return nil, nil } + cut, sourceMode, err := s.sourceEnergyFrom(ctx) + if err != nil { + return nil, err + } + if sourceMode && until > cut { + // The savings endpoint then uses site-energy intervals. Mixing the + // counter ledger with receipt-integrated Load would hide different + // coverage and smear a signed balance across tariff boundaries. + return nil, nil + } rows, err := s.history.QueryContext(ctx, ` SELECT bucket_start_ms, bucket_len_ms, flow, SUM(energy_wh) FROM energy_ledger_entries diff --git a/go/internal/state/store.go b/go/internal/state/store.go index e753ff638..18d0e5bd4 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -1393,6 +1393,7 @@ func (s *Store) batteryModelKey(driverName string) string { // HistoryPoint is one row of the history table. type HistoryPoint struct { + Sources *HistorySources `json:",omitempty"` Missing HistoryMissing N int64 `json:"n,omitempty"` ResolutionMS int64 `json:"resolution_ms,omitempty"` @@ -1497,6 +1498,31 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, if s.history == nil || untilMs < sinceMs { return nil, nil } + // Source energy is retained by minute. Do not paint a minute's mean as + // separate ten-second Load observations in either history client. + cut, sourceMode, err := s.sourceEnergyFrom(ctx) + if err != nil { + return nil, err + } + sourceMode = sourceMode && untilMs >= cut + queryFrom := sinceMs + bucketMs := int64(0) + if maxPoints > 0 { + bucketMs = (untilMs - sinceMs + int64(maxPoints)) / int64(maxPoints) + bucketMs = max(1, bucketMs) + } + if sourceMode { + if maxPoints > 0 { + bucketMs = (untilMs - sinceMs + int64(maxPoints) - 1) / int64(maxPoints) + } + var sourceWidth int64 + if err := s.history.QueryRowContext(ctx, `SELECT COALESCE(MAX(resolution_ms),?) FROM history_source_energy WHERE start_ms>=? AND start_ms?`, ArchiveResolutionMS, sinceMs-maxSourceEnergyWidth, untilMs, sinceMs).Scan(&sourceWidth); err != nil { + return nil, err + } + bucketMs = max(ArchiveResolutionMS, ((bucketMs+ArchiveResolutionMS-1)/ArchiveResolutionMS)*ArchiveResolutionMS) + bucketMs = max(sourceWidth, ((bucketMs+sourceWidth-1)/sourceWidth)*sourceWidth) + queryFrom = bucketStart(sinceMs, ArchiveResolutionMS) + } // Union across all three tiers. Dedupe on ts_ms preferring hot over warm over cold. // COALESCE to 0 so NULL columns (from partial aggregations) scan cleanly. const tierUnion = ` @@ -1516,17 +1542,18 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, SELECT * FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY ts_ms ORDER BY tier) AS rn FROM all_rows) WHERE rn=1 ) ` + rangeUnion := tierUnion + if sourceMode { + rangeUnion = strings.ReplaceAll(tierUnion, "ts_ms BETWEEN ? AND ?", "ts_ms >= ? AND ts_ms < ?") + rangeUnion = strings.ReplaceAll(rangeUnion, "last_ms BETWEEN ? AND ?", "last_ms >= ? AND last_ms < ?") + } var ( - rows *sql.Rows - err error + rows *sql.Rows + queryErr error ) - if maxPoints > 0 && untilMs >= sinceMs { + if bucketMs > 0 && untilMs >= sinceMs { // Ceil the bucket width; SQLite MAX selects JSON from its newest row. - bucketMs := (untilMs - sinceMs + int64(maxPoints)) / int64(maxPoints) - if bucketMs < 1 { - bucketMs = 1 - } - rows, err = s.history.QueryContext(ctx, tierUnion+`, bucketed AS ( + rows, queryErr = s.history.QueryContext(ctx, rangeUnion+`, bucketed AS ( SELECT MAX(ts_ms) AS ts_ms, SUM(COALESCE(grid_w, 0)*n)/SUM(n) AS grid_w, SUM(COALESCE(pv_w, 0)*n)/SUM(n) AS pv_w, SUM(COALESCE(bat_w, 0)*n)/SUM(n) AS bat_w, SUM(COALESCE(load_w, 0)*n)/SUM(n) AS load_w, @@ -1538,9 +1565,9 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, (SELECT json FROM history_hot WHERE ts_ms=b.ts_ms), (SELECT json FROM history_warm WHERE ts_ms=b.ts_ms), (SELECT json FROM history_cold WHERE ts_ms=b.ts_ms),'{}'),n,resolution_ms,first_ms,missing - FROM bucketed b ORDER BY ts_ms`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, bucketMs) + FROM bucketed b ORDER BY ts_ms`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, queryFrom, bucketMs) } else { - rows, err = s.history.QueryContext(ctx, tierUnion+` + rows, queryErr = s.history.QueryContext(ctx, rangeUnion+` SELECT ts_ms,COALESCE(grid_w,0),COALESCE(pv_w,0),COALESCE(bat_w,0),COALESCE(load_w,0),COALESCE(bat_soc,0), COALESCE((SELECT json FROM history_dashboard_partial WHERE last_ms=b.ts_ms),(SELECT json FROM history_dashboard WHERE last_ms=b.ts_ms), (SELECT json FROM history_hot WHERE ts_ms=b.ts_ms), @@ -1548,8 +1575,8 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, (SELECT json FROM history_cold WHERE ts_ms=b.ts_ms),'{}'),n,resolution_ms,first_ms,missing FROM deduped b ORDER BY ts_ms LIMIT ?`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, maxRawSeriesPoints+1) } - if err != nil { - return nil, err + if queryErr != nil { + return nil, queryErr } defer rows.Close() @@ -1560,6 +1587,9 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, if err := rows.Scan(&p.TsMs, &p.GridW, &p.PVW, &p.BatW, &p.LoadW, &p.BatSoC, &p.JSON, &p.N, &p.ResolutionMS, &p.FirstMS, &p.Missing); err != nil { return all, err } + if sourceMode { + p.ResolutionMS = max(p.ResolutionMS, bucketMs) + } if p.ResolutionMS > 1 { var detail map[string]any if json.Unmarshal([]byte(p.JSON), &detail) == nil && detail != nil { @@ -1576,7 +1606,16 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, } all = append(all, p) } - return all, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + rows.Close() + if sourceMode { + if err := s.applySourceLoadHistory(ctx, all, queryFrom, untilMs, bucketMs); err != nil { + return nil, err + } + } + return all, nil } // DayEnergy is the set of Wh totals over a time range, in site convention: diff --git a/go/internal/telemetry/forecast.go b/go/internal/telemetry/forecast.go index fc98f3dd6..2ac8457af 100644 --- a/go/internal/telemetry/forecast.go +++ b/go/internal/telemetry/forecast.go @@ -27,6 +27,9 @@ type ForecastFlow struct { FlowID string } type ForecastOptions struct { + // HistoryRevision breaks receipt integration at a saved settings change. + // It does not alter forecast qualification or live power. + HistoryRevision int64 `json:",omitempty"` // HouseholdInvalidReason qualifies unresolved site topology without invalidating independent PV observations. HouseholdInvalidReason string PVInvalidReason string From 01cc8e3d02b5b39c331d016a8db5125579ad38c6 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sun, 18 Oct 2026 20:57:27 +0200 Subject: [PATCH 2/2] fix(history): guard durable day cache against the first source commit --- go/internal/api/api.go | 2 +- go/internal/api/api_source_history_test.go | 8 +++++++ go/internal/state/store.go | 25 ++++++++++++++++++++++ 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/go/internal/api/api.go b/go/internal/api/api.go index 3d7b29341..1909e8ac4 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -2569,7 +2569,7 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { // Best-effort: a write failure is logged but not // surfaced to the operator since the in-memory // cache still serves this request. - if err := s.deps.State.SaveDailyEnergy(dayKey, de); err != nil { + if err := s.deps.State.SaveDailyEnergyForSourceMode(dayKey, de, sourceMode); err != nil { slog.Warn("handleEnergyDaily: persist daily aggregate failed", "err", err, "day", dayKey) } } diff --git a/go/internal/api/api_source_history_test.go b/go/internal/api/api_source_history_test.go index b5ce8c3a0..0fad3ac7a 100644 --- a/go/internal/api/api_source_history_test.go +++ b/go/internal/api/api_source_history_test.go @@ -88,6 +88,14 @@ func testSourceHistoryDayCaches(t *testing.T, firstBatch bool) { if err := st.FlushHistory(context.Background()); err != nil { t.Fatal(err) } + if firstBatch { + // An older request finishes saving after the first source commit. + // Its stale total must not refill the persistent cache under a new RAM key. + day := time.UnixMilli(base).In(now.Location()).Format("2006-01-02") + if err := st.SaveDailyEnergyForSourceMode(day, state.DayEnergy{LoadWh: 1, Intervals: 1}, false); err != nil { + t.Fatal(err) + } + } for _, path := range []string{"/api/energy/daily?days=2", "/api/savings/daily?days=2"} { if got := read(path); math.Abs(got-1000.0/60) > 1e-8 { t.Fatal("same API server retained Load from before final commit") diff --git a/go/internal/state/store.go b/go/internal/state/store.go index 18d0e5bd4..0ab26cafb 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -1821,6 +1821,17 @@ func (s *Store) LoadDailyEnergy(day string) (DayEnergy, bool, error) { // persisted via this method (the day is still accumulating); callers // should gate on "is closed day" before saving. func (s *Store) SaveDailyEnergy(day string, de DayEnergy) error { + return s.saveDailyEnergy(day, de, nil) +} + +// SaveDailyEnergyForSourceMode cannot repopulate a legacy cache after the +// first source batch invalidated it. The check and write share the writer +// lock and transaction; an in-flight older API request may simply skip saving. +func (s *Store) SaveDailyEnergyForSourceMode(day string, de DayEnergy, expectedSourceMode bool) error { + return s.saveDailyEnergy(day, de, &expectedSourceMode) +} + +func (s *Store) saveDailyEnergy(day string, de DayEnergy, expectedSourceMode *bool) error { s.historyWriteMu.Lock() defer s.historyWriteMu.Unlock() const q = ` @@ -1841,6 +1852,20 @@ func (s *Store) SaveDailyEnergy(day string, de DayEnergy) error { return err } defer tx.Rollback() + if expectedSourceMode != nil { + // Acquire SQLite's writer before checking the method, including when + // no cursor exists. Never upgrade a deferred read transaction to write. + if _, err := tx.Exec(`UPDATE history_source_cursor SET value=value WHERE key='through'`); err != nil { + return err + } + var current bool + if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM history_source_cursor WHERE key='through')`).Scan(¤t); err != nil { + return err + } + if current != *expectedSourceMode { + return nil + } + } _, err = tx.Exec(q, day, de.ImportWh, de.ExportWh, de.PVWh, de.BatChargedWh, de.BatDischargedWh, de.LoadWh,