From e3076a480f3b939238594e2757a7584a43035f36 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 8 Sep 2026 04:52:43 +0200 Subject: [PATCH] fix(forecast): capture live observation time with telemetry --- .changeset/current-forecast-observation.md | 5 + go/cmd/ftw/forecast_observation_time_test.go | 115 +++++++++++++++++++ go/cmd/ftw/forecast_occupancy_test.go | 2 +- go/cmd/ftw/forecast_tracking.go | 11 +- 4 files changed, 128 insertions(+), 5 deletions(-) create mode 100644 .changeset/current-forecast-observation.md create mode 100644 go/cmd/ftw/forecast_observation_time_test.go diff --git a/.changeset/current-forecast-observation.md b/.changeset/current-forecast-observation.md new file mode 100644 index 000000000..f15a5f5ca --- /dev/null +++ b/.changeset/current-forecast-observation.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Use the telemetry snapshot time for live forecast observations so delayed archive writes do not mistake fresh readings for future data. diff --git a/go/cmd/ftw/forecast_observation_time_test.go b/go/cmd/ftw/forecast_observation_time_test.go new file mode 100644 index 000000000..36809f76d --- /dev/null +++ b/go/cmd/ftw/forecast_observation_time_test.go @@ -0,0 +1,115 @@ +package main + +import ( + "context" + "math" + "path/filepath" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/forecasting" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +func TestForecastObservationUsesCurrentTelemetryCutoff(t *testing.T) { + ctx := context.Background() + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err = st.InitForecastArchive(ctx); err != nil { + t.Fatal(err) + } + + tel := telemetry.NewStore() + site := trackerSite() + f := trackerFixture(time.Now()) + f.store = st + f.tele = tel + f.site = func() forecastSite { return site } + var observedAt time.Time + f.curtailed = func(at time.Time) bool { + observedAt = at + return false + } + reading := func(at time.Time) telemetry.ForecastReading { + return telemetry.ForecastReading{ + At: at, Earliest: at, Latest: at, PVEarliest: at, PVLatest: at, + Valid: true, PVValid: true, HouseholdW: 1200, PVW: -800, + } + } + feed := func(at time.Time) []forecasting.Observation { + return f.observationIntervals(reading(at), site, at) + } + + // Prime the real accumulator from the start of the current quarter through + // the scheduled tick. A synchronous archive retry can delay observe until a + // newer telemetry sample has arrived. + scheduledAt := time.Now().UTC() + quarterStart := scheduledAt.Truncate(15 * time.Minute) + if out := feed(quarterStart); len(out) != 0 { + t.Fatalf("first sample completed an interval: %+v", out) + } + for at := quarterStart.Add(time.Minute); at.Before(scheduledAt); at = at.Add(time.Minute) { + if out := feed(at); len(out) != 0 { + t.Fatalf("partial quarter completed early: %+v", out) + } + } + if scheduledAt.After(quarterStart) { + if out := feed(scheduledAt); len(out) != 0 { + t.Fatalf("scheduled sample completed early: %+v", out) + } + } + + tel.Update("site", telemetry.DerMeter, 400, nil, nil) + tel.Update("pv", telemetry.DerPV, -800, nil, nil) + tel.RecordDriverSuccess("site") + tel.RecordDriverSuccess("pv") + f.observe(ctx) + if observedAt.IsZero() { + t.Fatal("observation did not apply the curtailment cutoff") + } + + var completed []forecasting.Observation + quarterEnd := quarterStart.Add(15 * time.Minute) + if observedAt.Before(quarterEnd) { + for at := observedAt.Add(time.Minute); at.Before(quarterEnd); at = at.Add(time.Minute) { + completed = append(completed, feed(at)...) + } + completed = append(completed, feed(quarterEnd)...) + } + persisted, err := st.LoadForecastObservations(ctx, quarterStart.UnixMilli(), quarterEnd.UnixMilli()) + if err != nil { + t.Fatal(err) + } + completed = append(completed, persisted...) + if len(completed) != 1 || !completed[0].LoadKnown || completed[0].Quality != telemetry.ForecastIntervalQuality || + !completed[0].PVKnown || math.Abs(completed[0].LoadW-1200) > 1e-6 || math.Abs(completed[0].PVW-800) > 1e-6 || + completed[0].StartMS != quarterStart.UnixMilli() || completed[0].EndMS != quarterEnd.UnixMilli() { + t.Fatalf("delayed observation lost or changed the complete quarter: %+v", completed) + } +} + +func TestForecastObservationStillRejectsLongGap(t *testing.T) { + f := &forecastTracker{} + site := trackerSite() + start := time.Date(2026, 1, 5, 12, 0, 0, 0, time.UTC) + reading := func(at time.Time) telemetry.ForecastReading { + return telemetry.ForecastReading{At: at, Earliest: at, Latest: at, Valid: true, HouseholdW: 1200} + } + if out := f.observationIntervals(reading(start), site, start); len(out) != 0 { + t.Fatalf("first sample completed an interval: %+v", out) + } + if out := f.observationIntervals(reading(start.Add(3*time.Minute)), site, start.Add(3*time.Minute)); len(out) != 0 { + t.Fatalf("long gap produced an interval: %+v", out) + } + var completed []forecasting.Observation + for at := start.Add(4 * time.Minute); !at.After(start.Add(15 * time.Minute)); at = at.Add(time.Minute) { + completed = append(completed, f.observationIntervals(reading(at), site, at)...) + } + if len(completed) != 0 { + t.Fatalf("quarter with a long gap was accepted: %+v", completed) + } +} diff --git a/go/cmd/ftw/forecast_occupancy_test.go b/go/cmd/ftw/forecast_occupancy_test.go index af9e41d56..d02eac075 100644 --- a/go/cmd/ftw/forecast_occupancy_test.go +++ b/go/cmd/ftw/forecast_occupancy_test.go @@ -167,7 +167,7 @@ func TestForecastPendingIdentityCannotUseSavedModels(t *testing.T) { // Pending startup must not touch live telemetry or the archive either. f.tele = nil f.store = nil - f.observe(context.Background(), at) + f.observe(context.Background()) if refreshes != 2 { t.Fatalf("identity refreshes=%d want2", refreshes) } diff --git a/go/cmd/ftw/forecast_tracking.go b/go/cmd/ftw/forecast_tracking.go index 28e05311e..1d95cbc88 100644 --- a/go/cmd/ftw/forecast_tracking.go +++ b/go/cmd/ftw/forecast_tracking.go @@ -166,8 +166,8 @@ func (f *forecastTracker) run(ctx context.Context) { continue } pending = nil - case now := <-tick.C: - f.observe(ctx, now) + case <-tick.C: + f.observe(ctx) } } } @@ -220,7 +220,7 @@ func forecastArchiveRetryable(err error) bool { return false } -func (f *forecastTracker) observe(ctx context.Context, now time.Time) { +func (f *forecastTracker) observe(ctx context.Context) { if f.refreshIdentity != nil { f.refreshIdentity() } @@ -239,7 +239,10 @@ func (f *forecastTracker) observe(ctx context.Context, now time.Time) { return } - r := f.tele.ForecastMeasurement(now, site.Meter, site.Options) + // A delayed tick only wakes the observer. Capture time under the telemetry + // lock so fresh polls are not compared with an old scheduled timestamp. + r := f.tele.ForecastMeasurementNow(site.Meter, site.Options) + now := r.At.UTC() if f.configMu != nil { f.configMu.RUnlock() }