diff --git a/.changeset/fresh-history-cutoff.md b/.changeset/fresh-history-cutoff.md new file mode 100644 index 00000000..0b21ac7c --- /dev/null +++ b/.changeset/fresh-history-cutoff.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Keep fresh measurements that arrive during a control tick in history and the household energy ledger. Check their age when taking the snapshot while preserving the tick timestamp. diff --git a/go/cmd/ftw/energy_history_test.go b/go/cmd/ftw/energy_history_test.go index e480c5b1..63a01d0f 100644 --- a/go/cmd/ftw/energy_history_test.go +++ b/go/cmd/ftw/energy_history_test.go @@ -144,6 +144,60 @@ func TestBuildHistoryPointRequiresFreshEVAndV2X(t *testing.T) { } +func TestPersistTelemetryTickUsesPersistenceFreshness(t *testing.T) { + for _, tc := range []struct { + name string + sampleAge time.Duration + wantSaved bool + }{ + {"poll after tick start", 0, true}, + {"stale reading", -2 * time.Minute, false}, + {"future reading", time.Minute, false}, + } { + t.Run(tc.name, func(t *testing.T) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + tickMS := time.Now().Add(-5 * time.Second).UnixMilli() + tel := telemetry.NewStore() + tel.EnsureDriverHealth("meter") + tel.Update("meter", telemetry.DerMeter, 1200, nil, nil) + tel.RecordDriverSuccess("meter") + sampleAt := time.Now().Add(tc.sampleAge) + tel.Get("meter", telemetry.DerMeter).UpdatedAt = sampleAt + ctrl := &control.State{SiteMeterDriver: "meter"} + if _, err := persistTelemetryTick(st, tel, ctrl, tickMS, time.Minute); err != nil { + t.Fatal(err) + } + history, err := st.LoadHistory(tickMS-1, tickMS+1, 0) + if err != nil { + t.Fatal(err) + } + 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]) + } + assets, err := st.EnergyAssets() + if err != nil { + t.Fatal(err) + } + consumerSaved := false + for _, asset := range assets { + if asset.AssetID == observedConsumerAssetID { + consumerSaved = asset.LastSeenMS == sampleAt.UnixMilli() + } + } + if consumerSaved != tc.wantSaved { + t.Fatalf("consumer ledger assets = %+v, want saved=%v", assets, tc.wantSaved) + } + }) + } +} + func TestStaleMeterTickKeepsSamplesAndIndependentLedgerWithoutDispatch(t *testing.T) { st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) if err != nil { diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index 885afe39..667b14f7 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -4088,11 +4088,13 @@ func buildHistoryPoint(tel *telemetry.Store, ctrl *control.State, nowMs int64, h opts = options[0] opts.MaxAge = historyMaxAge } - now := time.UnixMilli(nowMs).Add(time.Millisecond - time.Nanosecond) - balance := tel.ForecastMeasurement(now, ctrl.SiteMeterDriver, opts) + // Polls can update telemetry while dispatch runs. Keep the tick's history + // timestamp, but judge reading freshness when this snapshot is captured. + balance := tel.ForecastMeasurementNow(ctrl.SiteMeterDriver, opts) if !balance.Valid { return unavailable, false } + now := balance.At gridW, pvW, batW := balance.GridW, balance.PVW, balance.BatteryW evW, v2xW, loadW := balance.EVW, balance.V2XW, balance.HouseholdW readingUsable := func(driver string, updatedAt time.Time) bool { diff --git a/go/internal/telemetry/forecast.go b/go/internal/telemetry/forecast.go index 8ed11ddd..d2344cc3 100644 --- a/go/internal/telemetry/forecast.go +++ b/go/internal/telemetry/forecast.go @@ -49,6 +49,25 @@ type ForecastReading struct { // does not invalidate a fresh meter. Missing configured flows invalidate the // balance, including devices that have never emitted. No fuse limits house W. func (s *Store) ForecastMeasurement(now time.Time, siteMeter string, opts ForecastOptions) ForecastReading { + if s != nil { + s.mu.RLock() + defer s.mu.RUnlock() + } + return s.forecastMeasurementLocked(now, siteMeter, opts) +} + +// ForecastMeasurementNow captures the time after acquiring the telemetry lock. +// A poll already in progress can finish before the snapshot without appearing +// to come from the future. Explicit forecast origins still use ForecastMeasurement. +func (s *Store) ForecastMeasurementNow(siteMeter string, opts ForecastOptions) ForecastReading { + if s != nil { + s.mu.RLock() + defer s.mu.RUnlock() + } + return s.forecastMeasurementLocked(time.Now(), siteMeter, opts) +} + +func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts ForecastOptions) ForecastReading { out := ForecastReading{At: now, Valid: true, PVValid: true} if s == nil { out.Valid = false @@ -63,8 +82,6 @@ func (s *Store) ForecastMeasurement(now time.Time, siteMeter string, opts Foreca if opts.MaxSkew <= 0 { opts.MaxSkew = 30 * time.Second } - s.mu.RLock() - defer s.mu.RUnlock() flows := map[string]ForecastFlow{} key := func(f ForecastFlow) string { return f.Driver + ":" + f.DerType.String() } fail := func(reason string, pv bool) { diff --git a/go/internal/telemetry/forecast_test.go b/go/internal/telemetry/forecast_test.go index 01763fc8..c2b1001e 100644 --- a/go/internal/telemetry/forecast_test.go +++ b/go/internal/telemetry/forecast_test.go @@ -3,6 +3,8 @@ package telemetry import ( "encoding/json" "math" + "runtime" + "strings" "testing" "time" ) @@ -20,6 +22,73 @@ func forecastStore(now time.Time, grid, pv, bat float64) *Store { } return s } + +func TestForecastMeasurementNowCapturesCutoffAfterConcurrentWriter(t *testing.T) { + s := forecastStore(time.Now().Add(-time.Second), 4000, 0, 3000) + s.mu.Lock() + locked := true + defer func() { + if locked { + s.mu.Unlock() + } + }() + result := make(chan ForecastReading, 1) + go func() { result <- s.ForecastMeasurementNow("site", ForecastOptions{}) }() + + // A start channel alone would not prove the reader reached RLock before + // this writer publishes. Observe that blocked stack so an implementation + // that captures time before RLock deterministically retains an old cutoff. + stack := make([]byte, 128<<10) + deadline := time.Now().Add(5 * time.Second) + for { + blocked := false + for _, goroutine := range strings.Split(string(stack[:runtime.Stack(stack, true)]), "\n\n") { + if strings.Contains(goroutine, "(*Store).ForecastMeasurementNow(") && strings.Contains(goroutine, "sync.(*RWMutex).RLock(") { + blocked = true + break + } + } + if blocked { + break + } + if time.Now().After(deadline) { + t.Fatal("current measurement did not wait on the active writer") + } + runtime.Gosched() + } + published := time.Now() + for _, reading := range s.readings { + reading.UpdatedAt = published + } + s.readings["site:meter"].RawW = 4200 + s.mu.Unlock() + locked = false + + select { + case reading := <-result: + if !reading.Valid || !reading.PVValid || reading.HouseholdW != 1200 { + t.Fatalf("fresh writer update rejected or missed: %+v", reading) + } + if reading.At.Before(published) || !reading.Latest.Equal(published) { + t.Fatalf("cutoff %s precedes published reading %s (latest %s)", reading.At, published, reading.Latest) + } + case <-time.After(5 * time.Second): + t.Fatal("current measurement did not finish after writer released lock") + } +} + +func TestForecastExplicitOriginStillRejectsLaterPublishedMeasurement(t *testing.T) { + origin := time.Now().Add(-time.Second) + s := forecastStore(origin, 4000, 0, 3000) + s.mu.Lock() + s.readings["site:meter"].UpdatedAt = origin.Add(time.Millisecond) + s.mu.Unlock() + reading := s.ForecastMeasurement(origin, "site", ForecastOptions{}) + if reading.Valid || reading.Reason != "future:site:meter" || !reading.At.Equal(origin) { + t.Fatalf("explicit causal origin was replaced with current time: %+v", reading) + } +} + func TestForecastCompleteBalance(t *testing.T) { now := time.Now() s := forecastStore(now, 4000, 0, 3000)