fix(history): keep fresh polls received during dispatch

.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.

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 {

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 {

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) {

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)