diff --git a/.changeset/clock-jump-learning-recovery.md b/.changeset/clock-jump-learning-recovery.md new file mode 100644 index 000000000..16622b802 --- /dev/null +++ b/.changeset/clock-jump-learning-recovery.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Recover saved household load learning when the raw interval archive proves the old clock-jump collection bug. Keep solar learning, charging goals and the raw record. A permanently rejected archived observation no longer blocks later valid training; disk, worker and model failures still retry. diff --git a/go/cmd/ftw/forecast_learning.go b/go/cmd/ftw/forecast_learning.go index d3c78ab44..550674b6d 100644 --- a/go/cmd/ftw/forecast_learning.go +++ b/go/cmd/ftw/forecast_learning.go @@ -47,6 +47,9 @@ func (f *forecastTracker) restoreLearning(ctx context.Context) error { } f.learningMu.Lock() defer f.learningMu.Unlock() + if err := f.recoverClockJumpLearningLocked(ctx); err != nil { + slog.Warn("forecast load learning recovery pending", "err", err) + } for _, signal := range []string{"pv", "load"} { if err := f.applyLearningLocked(ctx, signal); err != nil { slog.Warn("forecast learning restart pending", "signal", signal, "err", err) @@ -113,6 +116,9 @@ func (f *forecastTracker) reconcileLearning(ctx context.Context) { } f.learningMu.Lock() defer f.learningMu.Unlock() + if err := f.recoverClockJumpLearningLocked(ctx); err != nil { + slog.Warn("forecast load learning recovery pending", "err", err) + } site := f.learningSiteLocked(f.site()) if site.IdentityPending { return @@ -176,6 +182,19 @@ func (f *forecastTracker) RestartLearning(ctx context.Context, signal string) er if f.learningErrors[signal] != nil && previous > 0 { cutoff = previous } + return f.restartLearningLocked(ctx, site, signal, cutoff) +} + +// Caller holds configMu and learningMu. Manual resets and archive-based recovery +// share the durable intent and idempotent, signal-scoped apply. +func (f *forecastTracker) restartLearningLocked(ctx context.Context, site forecastSite, signal string, cutoff int64) error { + if err := ctx.Err(); err != nil { + return err + } + next := f.learningPeriods + if next.ConfigRevision != rustConfigRevision(site) { + next = forecastLearningPeriods{ConfigRevision: rustConfigRevision(site)} + } if signal == "pv" { next.PVMS = cutoff } else { diff --git a/go/cmd/ftw/forecast_learning_recovery.go b/go/cmd/ftw/forecast_learning_recovery.go new file mode 100644 index 000000000..6212276a3 --- /dev/null +++ b/go/cmd/ftw/forecast_learning_recovery.go @@ -0,0 +1,114 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log/slog" + "math" + "time" + + "github.com/srcfl/ftw/go/internal/forecasting" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +// The former wall/monotonic-clock bug could emit several old quarters from one +// reading. A complete_balance_v1 collector with a two-minute maximum gap cannot +// emit adjacent quarters at the same late capture time. Disk retries cannot +// create this signature: AvailableAtMS freezes before the observation is queued. +// Power is deliberately absent from this test; genuine household consumption +// may exceed the grid fuse when solar or the battery supplies the difference. +func clockJumpObservationCohorts(rows []forecasting.Observation, cutoff, now int64) map[string]int64 { + type capture struct { + cohort string + start int64 + at int64 + } + seen := make(map[capture]bool) + bad := make(map[string]int64) + const quarterMS = int64(15 * time.Minute / time.Millisecond) + const maxGapMS = int64(2 * time.Minute / time.Millisecond) + for _, o := range rows { + if !o.LoadKnown || o.Quality != telemetry.ForecastIntervalQuality || o.StartMS < cutoff || + o.AvailableAtMS > now || o.EndMS-o.StartMS != quarterMS || o.StartMS%quarterMS != 0 || + o.AvailableAtMS-o.EndMS <= maxGapMS || o.Validate() != nil { + continue + } + previous := capture{o.ConfigVersion, o.StartMS - quarterMS, o.AvailableAtMS} + next := capture{o.ConfigVersion, o.EndMS, o.AvailableAtMS} + if seen[previous] || seen[next] { + bad[o.ConfigVersion] = max(bad[o.ConfigVersion], o.AvailableAtMS) + } + seen[capture{o.ConfigVersion, o.StartMS, o.AvailableAtMS}] = true + } + return bad +} + +// Caller holds configMu and learningMu, before either pipeline can be captured. +// Check once per physical identity in this process. The existing durable load +// cutoff makes the recovery idempotent across crashes and future restarts. +func (f *forecastTracker) recoverClockJumpLearningLocked(ctx context.Context) error { + site := f.learningSiteLocked(f.site()) + identity := site.SiteID + "/" + rustConfigRevision(site) + if site.IdentityPending || f.store == nil || f.load == nil { + return nil + } + now := f.now().UnixMilli() + if now <= 0 { + return errors.New("clock unavailable for forecast learning recovery") + } + if f.learningRecoveryChecked == identity && (f.learningRecoveryAfterMS == 0 || now < f.learningRecoveryAfterMS) { + return nil + } + // Finish a saved reset before considering a new one. Never move its cutoff + // forward while a worker or disk failure still leaves that intent pending. + if site.LoadLearningStartedMS > 0 && f.learningErrors["load"] != nil { + return nil + } + workCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + // An RPi may boot with a wall clock older than its archive. Load future + // rows too, without treating them as available evidence. Cache the next + // capture time so clock sync can reopen this check without repeated reads. + rows, err := f.store.LoadForecastObservations(workCtx, 0, math.MaxInt64) + if err != nil { + return fmt.Errorf("read clock-jump evidence: %w", err) + } + var after int64 + for _, o := range rows { + if o.AvailableAtMS > now && (after == 0 || o.AvailableAtMS < after) { + after = o.AvailableAtMS + } + } + bad := clockJumpObservationCohorts(rows, site.LoadLearningStartedMS, now) + matched := bad[site.Revision] > 0 + if !matched && len(bad) > 0 { + // Evaluation cohorts change with the worker build or planning policy, + // while learning survives. An archived site binding proves that an old + // cohort belongs to this same physical model; other sites stay untouched. + for cohort := range bad { + // The identity binding is timeless metadata, not a forecast input. + // It may have been archived after this pre-sync boot clock. + binding, err := f.store.LoadForecastCohortSite(workCtx, cohort, math.MaxInt64) + if err != nil { + return fmt.Errorf("read clock-jump site binding: %w", err) + } + if binding != nil && binding.SiteID == site.SiteID && binding.LearningRevision == rustConfigRevision(site) { + matched = true + break + } + } + } + if !matched { + f.learningRecoveryChecked = identity + f.learningRecoveryAfterMS = after + return nil + } + if err = f.restartLearningLocked(workCtx, site, "load", now); err != nil { + return err + } + f.learningRecoveryChecked = identity + f.learningRecoveryAfterMS = after + slog.Warn("forecast load learning restarted after invalid clock-jump intervals", "learning_started_ms", now) + return nil +} diff --git a/go/cmd/ftw/forecast_learning_recovery_test.go b/go/cmd/ftw/forecast_learning_recovery_test.go new file mode 100644 index 000000000..6b73cf91f --- /dev/null +++ b/go/cmd/ftw/forecast_learning_recovery_test.go @@ -0,0 +1,375 @@ +package main + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "errors" + "io" + "math" + "os" + "path/filepath" + "reflect" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/energyforecast" + "github.com/srcfl/ftw/go/internal/forecasting" + "github.com/srcfl/ftw/go/internal/loadmodel" + "github.com/srcfl/ftw/go/internal/modelstate" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +// Synthetic complete intervals with one impossible shared capture time. No +// customer identifiers, measurements, model state or timestamps belong here. +func clockJumpObservations(site forecastSite, start time.Time) []forecasting.Observation { + rows := make([]forecasting.Observation, 2) + for i := range rows { + rows[i] = learningObservation(site, start.Add(time.Duration(i)*15*time.Minute)) + rows[i].Quality = telemetry.ForecastIntervalQuality + rows[i].AvailableAtMS = start.Add(time.Hour).UnixMilli() + rows[i].LoadW = float64(20+i*15) * 1000 + } + return rows +} + +func TestForecastClockJumpRecoverySignature(t *testing.T) { + start := time.Now().Add(-4 * time.Hour).Truncate(15 * time.Minute) + base := clockJumpObservations(hostForecastSite(), start) + now := start.Add(3 * time.Hour).UnixMilli() + tests := []struct { + name string + change func([]forecasting.Observation) + cutoff int64 + want bool + }{ + {"shared capture", nil, 0, true}, + {"low power still has impossible time", func(o []forecasting.Observation) { o[0].LoadW, o[1].LoadW = 700, 800 }, 0, true}, + {"ordinary high household power", func(o []forecasting.Observation) { + for i := range o { + o[i].AvailableAtMS = o[i].EndMS + 1000 + o[i].LoadW = 107000 + } + }, 0, false}, + {"exactly maximum valid delay", func(o []forecasting.Observation) { + for i := range o { + o[i].AvailableAtMS = o[i].EndMS + 120000 + } + }, 0, false}, + {"only one late quarter", func(o []forecasting.Observation) { o[0].AvailableAtMS = o[0].EndMS }, 0, false}, + {"different captures", func(o []forecasting.Observation) { o[0].AvailableAtMS++ }, 0, false}, + {"different cohorts", func(o []forecasting.Observation) { o[0].ConfigVersion = "other" }, 0, false}, + {"missing quarter", func(o []forecasting.Observation) { o[0].StartMS -= 900000; o[0].EndMS -= 900000 }, 0, false}, + {"same quarter twice", func(o []forecasting.Observation) { o[0] = o[1] }, 0, false}, + {"PV only", func(o []forecasting.Observation) { o[0].LoadKnown = false }, 0, false}, + {"other quality", func(o []forecasting.Observation) { o[0].Quality = "import" }, 0, false}, + {"partial interval", func(o []forecasting.Observation) { o[0].StartMS++ }, 0, false}, + {"NaN", func(o []forecasting.Observation) { o[0].LoadW = math.NaN() }, 0, false}, + {"infinite", func(o []forecasting.Observation) { o[0].LoadW = math.Inf(1) }, 0, false}, + {"future capture", func(o []forecasting.Observation) { + for i := range o { + o[i].AvailableAtMS = now + 1 + } + }, 0, false}, + {"already reset", nil, start.Add(2 * time.Hour).UnixMilli(), false}, + {"cutoff straddles evidence", nil, start.Add(time.Minute).UnixMilli(), false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + rows := append([]forecasting.Observation(nil), base...) + if tc.change != nil { + tc.change(rows) + } + got := clockJumpObservationCohorts(rows, tc.cutoff, now) + if (len(got) > 0) != tc.want { + t.Fatalf("cohorts=%v want corruption=%v", got, tc.want) + } + rows[0], rows[1] = rows[1], rows[0] + if !reflect.DeepEqual(got, clockJumpObservationCohorts(rows, tc.cutoff, now)) { + t.Fatal("row order changed diagnosis") + } + }) + } +} + +func saveClockJumpEvidence(t *testing.T, st *state.Store, rows []forecasting.Observation) { + t.Helper() + for _, row := range rows { + if err := st.SaveForecastObservation(context.Background(), row); err != nil { + t.Fatal(err) + } + } +} + +func seedLegacyRecoveryModel(t *testing.T, st *state.Store, site forecastSite, at time.Time) { + t.Helper() + for _, profile := range loadmodel.Profiles() { + m := loadmodel.NewModel(4000) + m.ConfigRevision, m.Timezone = rustConfigRevision(site), site.Timezone + m.Update(at, 35000, 5) + data, err := modelstate.Wrap(loadmodel.FeatureHash(), m) + if err != nil { + t.Fatal(err) + } + if err := st.SaveConfig("loadmodel/state_utc:"+string(profile), data); err != nil { + t.Fatal(err) + } + } +} + +func TestForecastClockJumpRecoveryKeepsHealthyLearning(t *testing.T) { + for _, kind := range []string{"healthy", "different learning identity", "different site", "unbound older cohort"} { + t.Run(kind, func(t *testing.T) { + st := hostForecastDB(t) + site := hostForecastSite() + start := time.Now().Add(-4 * time.Hour).Truncate(time.Hour) + seedLegacyRecoveryModel(t, st, site, start) + rows := clockJumpObservations(site, start) + if kind == "healthy" { + for i := range rows { + rows[i].AvailableAtMS = rows[i].EndMS + 1000 + rows[i].LoadW = 70000 + } + } else { + for i := range rows { + rows[i].ConfigVersion = "old-worker-cohort" + } + if kind != "unbound older cohort" { + issue := hostForecastIssue(start.Add(2*time.Hour), start.Add(90*time.Minute), 1) + issue.ConfigVersion = rows[0].ConfigVersion + issue.Site = &forecasting.SiteContext{SiteID: site.SiteID, LearningRevision: rustConfigRevision(site), Timezone: site.Timezone} + if kind == "different site" { + issue.Site.SiteID = "different-site" + } else { + issue.Site.LearningRevision = "different-learning" + } + if err := st.SaveForecastIssue(context.Background(), issue); err != nil { + t.Fatal(err) + } + } + } + saveClockJumpEvidence(t, st, rows) + f := learningTracker(st, &site, start.Add(3*time.Hour), nil) + before := f.load.Snapshot() + if err := f.restoreLearning(context.Background()); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(before, f.load.Snapshot()) { + t.Fatal("unproven corruption changed healthy model") + } + if _, ok := st.LoadConfig(forecastLearningKey); ok { + t.Fatal("unproven corruption persisted a reset") + } + }) + } +} + +func TestForecastClockJumpRecoveryFromBackupRetainsEvidenceAndGoals(t *testing.T) { + st := hostForecastDB(t) + site := hostForecastSite() + start := time.Now().Add(-4 * time.Hour).Truncate(time.Hour) + rows := clockJumpObservations(site, start) + seedLegacyRecoveryModel(t, st, site, start) + saveClockJumpEvidence(t, st, rows) + goal := `{"soc":0.8,"time_of_day_min_utc":300,"recurring":true}` + if err := st.SaveConfig("loadpoint_schedule:garage", goal); err != nil { + t.Fatal(err) + } + backup := filepath.Join(t.TempDir(), "backup.gz") + if err := st.BackupToCompressed(backup); err != nil { + t.Fatal(err) + } + in, err := os.Open(backup) + if err != nil { + t.Fatal(err) + } + defer in.Close() + z, err := gzip.NewReader(in) + if err != nil { + t.Fatal(err) + } + defer z.Close() + data, err := io.ReadAll(z) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "restored.db") + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + restored, err := state.Open(path) + if err != nil { + t.Fatal(err) + } + defer restored.Close() + now := start.Add(3 * time.Hour) + f := learningTracker(restored, &site, now, nil) + beforePV := f.pv.Model() + if err := f.restoreLearning(context.Background()); err != nil { + t.Fatal(err) + } + for _, profile := range f.load.Snapshot().Profiles { + if profile.Samples != 0 || profile.LearningStartedMS != now.UnixMilli() { + t.Fatalf("poisoned restored profile survived: %+v", profile) + } + } + if !reflect.DeepEqual(beforePV, f.pv.Model()) { + t.Fatal("load recovery changed PV") + } + got, err := restored.LoadForecastObservations(context.Background(), start.UnixMilli(), now.UnixMilli()) + if err != nil || !reflect.DeepEqual(got, rows) { + t.Fatalf("raw archive changed: %+v %v", got, err) + } + if got, _ := restored.LoadConfig("loadpoint_schedule:garage"); got != goal { + t.Fatal("EV goal changed") + } + // A second process sees the durable cutoff, not a fresh recovery epoch. + f = learningTracker(restored, &site, now.Add(time.Hour), nil) + if err := f.restoreLearning(context.Background()); err != nil { + t.Fatal(err) + } + if f.load.LearningStartedMS() != now.UnixMilli() { + t.Fatal("restart repeated reset") + } +} + +func TestForecastClockJumpRecoveryWaitsForBootClockSync(t *testing.T) { + st := hostForecastDB(t) + site := hostForecastSite() + start := time.Now().Add(-4 * time.Hour).Truncate(time.Hour) + saveClockJumpEvidence(t, st, clockJumpObservations(site, start)) + seedLegacyRecoveryModel(t, st, site, start) + now := start.Add(-time.Hour) + f := learningTracker(st, &site, now, nil) + f.clock = func() time.Time { return now } + if err := f.restoreLearning(context.Background()); err != nil { + t.Fatal(err) + } + if f.load.LearningStartedMS() != 0 { + t.Fatal("future archive reset load before clock sync") + } + if f.learningRecoveryAfterMS != start.Add(time.Hour).UnixMilli() { + t.Fatal("pre-sync startup permanently skipped future evidence") + } + now = start.Add(3 * time.Hour) + f.reconcileLearning(context.Background()) + if f.load.LearningStartedMS() != now.UnixMilli() { + t.Fatal("same-identity clock sync did not reopen recovery") + } + cutoff := f.load.LearningStartedMS() + now = now.Add(time.Hour) + f.reconcileLearning(context.Background()) + if f.load.LearningStartedMS() != cutoff { + t.Fatal("clock sync caused repeated resets") + } +} + +func TestForecastClockJumpRecoveryWaitsForIdentityAndArchive(t *testing.T) { + st := hostForecastDB(t) + site := hostForecastSite() + start := time.Now().Add(-4 * time.Hour).Truncate(time.Hour) + saveClockJumpEvidence(t, st, clockJumpObservations(site, start)) + seedLegacyRecoveryModel(t, st, site, start) + site.IdentityPending = true + f := learningTracker(st, &site, start.Add(3*time.Hour), nil) + if err := f.restoreLearning(context.Background()); err != nil { + t.Fatal(err) + } + if f.load.LearningStartedMS() != 0 { + t.Fatal("unconfirmed hardware identity reset learning") + } + site.IdentityPending = false + ctx, cancel := context.WithCancel(context.Background()) + cancel() + f.reconcileLearning(ctx) + if f.load.LearningStartedMS() != 0 || f.learningRecoveryChecked != "" { + t.Fatal("unavailable archive caused a reset or consumed recovery check") + } + f.reconcileLearning(context.Background()) + if f.load.LearningStartedMS() != start.Add(3*time.Hour).UnixMilli() { + t.Fatal("valid identity/archive did not recover") + } +} + +func TestForecastClockJumpRecoveryNativeCrashAndWorkerUpdate(t *testing.T) { + st := hostForecastDB(t) + site := hostForecastSite() + site.LearningRevision = "physical-site-stable" + start := time.Now().Add(-4 * time.Hour).Truncate(time.Hour) + rows := clockJumpObservations(site, start) + saveClockJumpEvidence(t, st, rows) + r := learningNative(t, st) + for _, row := range rows { + if err := r.Update(context.Background(), site, row, nil, false); err != nil { + t.Fatal(err) + } + } + beforePV := learningModel(t, r, "pv") + beforeLoad := learningModel(t, r, "load") + // Save the old cohort's physical identity before a new worker changes the + // evaluation revision. The private model remains opaque to Core. + issue := hostForecastIssue(start.Add(2*time.Hour), start.Add(90*time.Minute), 1) + issue.Site = &forecasting.SiteContext{SiteID: site.SiteID, LearningRevision: site.LearningRevision, Timezone: site.Timezone} + if err := st.SaveForecastIssue(context.Background(), issue); err != nil { + t.Fatal(err) + } + site.Revision = "new-worker-cohort" + now := start.Add(3 * time.Hour) + f := learningTracker(st, &site, now, r) + transport := r.transport.(energyforecast.RoundTripper) + r.client = energyforecast.NewClient(hostForecastExchange(func(ctx context.Context, p []byte) ([]byte, error) { + var header struct { + Action string `json:"action"` + } + _ = json.Unmarshal(p, &header) + if header.Action == "reset" { + return nil, errors.New("injected worker outage after durable intent") + } + return transport.RoundTrip(ctx, p) + })) + if err := f.restoreLearning(context.Background()); err != nil { + t.Fatal(err) + } + if f.learningPeriods.LoadMS != now.UnixMilli() || f.learningPeriods.PVMS != 0 { + t.Fatal("recovery intent absent or crossed signals") + } + if !bytes.Equal(beforeLoad, learningModel(t, r, "load")) { + t.Fatal("failed worker reset published changed state") + } + // Simulate a crash between durable reset intent and the worker state save. + r = learningNative(t, st) + f = learningTracker(st, &site, now.Add(time.Hour), r) + if err := f.restoreLearning(context.Background()); err != nil { + t.Fatal(err) + } + if f.learningPeriods.LoadMS != now.UnixMilli() || f.load.LearningStartedMS() != now.UnixMilli() { + t.Fatal("crash retry moved reset cutoff") + } + if !bytes.Equal(beforePV, learningModel(t, r, "pv")) { + t.Fatal("recovery changed PV state") + } + if bytes.Equal(beforeLoad, learningModel(t, r, "load")) { + t.Fatal("recovery left poisoned load state") + } + if got := f.LearningStatus("load"); got.Status != "cold_start" || got.LatestTrainingMS != 0 { + t.Fatalf("recovery forged training: %+v", got) + } + eligible := learningObservation(site, now.Add(15*time.Minute)) + if err := f.updateObservation(context.Background(), forecastObservationJob{site: site, observation: eligible}); err != nil { + t.Fatal(err) + } + learned := r.Snapshot() + r = learningNative(t, st) + f = learningTracker(st, &site, now.Add(2*time.Hour), r) + if err := f.restoreLearning(context.Background()); err != nil { + t.Fatal(err) + } + if !bytes.Equal(learned, r.Snapshot()) { + t.Fatal("second restart erased healthy post-recovery learning") + } + if got := f.LearningStatus("load"); got.LatestTrainingMS != eligible.EndMS { + t.Fatalf("healthy learning lost: %+v", got) + } +} diff --git a/go/cmd/ftw/forecast_observation_queue.go b/go/cmd/ftw/forecast_observation_queue.go index 11d1430b1..4352c5cfd 100644 --- a/go/cmd/ftw/forecast_observation_queue.go +++ b/go/cmd/ftw/forecast_observation_queue.go @@ -6,6 +6,7 @@ import ( "log/slog" "time" + "github.com/srcfl/ftw/go/internal/energyforecast" "github.com/srcfl/ftw/go/internal/forecasting" "github.com/srcfl/ftw/go/internal/state" ) @@ -106,10 +107,14 @@ func (f *forecastTracker) flushObservationsWith(ctx context.Context, f.observationError = "" } f.mu.Unlock() - if err != nil { + rejected := job.archived && energyforecast.IsObservationRejected(err) && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) + if err != nil && !rejected { slog.Warn("forecast archive: observation retained for retry", "phase", phase, "start_ms", job.observation.StartMS, "err", err) return } + if rejected { + slog.Warn("forecast archive: archived observation rejected by model", "start_ms", job.observation.StartMS, "err", err) + } f.pendingObservations[0] = forecastObservationJob{} f.pendingObservations = f.pendingObservations[1:] f.requestScoring() diff --git a/go/cmd/ftw/forecast_observation_queue_test.go b/go/cmd/ftw/forecast_observation_queue_test.go index e22551de4..1a070f73a 100644 --- a/go/cmd/ftw/forecast_observation_queue_test.go +++ b/go/cmd/ftw/forecast_observation_queue_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/srcfl/ftw/go/internal/energyforecast" "github.com/srcfl/ftw/go/internal/forecasting" "github.com/srcfl/ftw/go/internal/state" ) @@ -239,3 +240,85 @@ func TestForecastObservationQueueOverflowHealthRecovers(t *testing.T) { t.Fatal("current health did not recover or historical loss was erased") } } + +func TestForecastObservationPermanentRejectionDoesNotStarveNextInterval(t *testing.T) { + for _, kind := range []string{"observation", "transport", "state", "deadline", "archive"} { + t.Run(kind, func(t *testing.T) { + f := trackerFixture(time.Now()) + bad := observationJobFixture() + good := bad + good.observation.StartMS += 900000 + good.observation.EndMS += 900000 + good.observation.AvailableAtMS += 900000 + f.enqueueObservation(bad) + f.enqueueObservation(good) + var saved []forecasting.Observation + var trained []forecasting.Observation + failed := false + save := func(_ context.Context, o forecasting.Observation) error { + if kind == "archive" && !failed { + failed = true + return &energyforecast.ObservationRejectedError{Err: errors.New("no raw archive acknowledgement")} + } + saved = append(saved, o) + return nil + } + update := func(_ context.Context, job forecastObservationJob) error { + if job.observation == bad.observation && !failed { + failed = true + switch kind { + case "observation": + return &energyforecast.ObservationRejectedError{Err: errors.New("household load exceeds configured limit")} + case "transport", "state": + return errors.New("retry model pipeline failure") + case "deadline": + return errors.Join(&energyforecast.ObservationRejectedError{Err: errors.New("rejection")}, context.DeadlineExceeded) + } + } + trained = append(trained, job.observation) + return nil + } + f.flushObservationsWith(context.Background(), save, update) + if kind == "observation" { + if len(f.pendingObservations) != 0 || !reflect.DeepEqual(trained, []forecasting.Observation{good.observation}) { + t.Fatal("permanent bad head starved valid training") + } + } else { + if len(f.pendingObservations) != 2 || f.observationError == "" { + t.Fatal("transient/pipeline/unarchived failure discarded evidence") + } + f.flushObservationsWith(context.Background(), save, update) + if len(f.pendingObservations) != 0 || len(trained) != 2 { + t.Fatal("retry did not train retained evidence") + } + } + if !reflect.DeepEqual(saved, []forecasting.Observation{bad.observation, good.observation}) { + t.Fatalf("archive changed or repeated: %+v", saved) + } + }) + } +} + +func TestForecastObservationNativeLimitRejectionKeepsRawAndTrainsNext(t *testing.T) { + st := hostForecastDB(t) + r := learningNative(t, st) + site := hostForecastSite() + start := time.Now().Add(-time.Hour).Truncate(time.Hour) + bad := learningObservation(site, start) + bad.LoadW = 107000 // Worker acceptance limit, not a site's grid fuse. + good := learningObservation(site, start.Add(15*time.Minute)) + f := learningTracker(st, &site, start.Add(time.Hour), r) + f.enqueueObservation(forecastObservationJob{site: site, observation: bad}) + f.enqueueObservation(forecastObservationJob{site: site, observation: good}) + f.flushObservations(context.Background()) + if len(f.pendingObservations) != 0 { + t.Fatal("permanently rejected load blocked next interval") + } + if got := r.LearningStatus(site, "load"); got.LatestTrainingMS != good.EndMS { + t.Fatalf("next good interval did not train: %+v", got) + } + rows, err := st.LoadForecastObservations(context.Background(), start.UnixMilli(), start.Add(time.Hour).UnixMilli()) + if err != nil || !reflect.DeepEqual(rows, []forecasting.Observation{bad, good}) { + t.Fatalf("rejection rewrote raw evidence: %+v %v", rows, err) + } +} diff --git a/go/cmd/ftw/forecast_rust.go b/go/cmd/ftw/forecast_rust.go index 635fad021..20cab2c68 100644 --- a/go/cmd/ftw/forecast_rust.go +++ b/go/cmd/ftw/forecast_rust.go @@ -131,7 +131,7 @@ func (r *rustForecast) Update(ctx context.Context, site forecastSite, o forecast } origin := max(o.AvailableAtMS, saved.LatestAvailableMS) if weather != nil && weather.FetchedAtMs > o.AvailableAtMS { - return errors.New("observation weather arrived after update origin") + return &energyforecast.ObservationRejectedError{Err: errors.New("observation weather arrived after update origin")} } input := energyforecast.Observation{Interval: energyforecast.Interval{ValidStartMs: o.StartMS, ValidEndMs: o.EndMS}, Features: rustFeatures(time.UnixMilli(o.StartMS), site, !away, weather), AvailableAtMs: o.AvailableAtMS, diff --git a/go/cmd/ftw/forecast_tracking.go b/go/cmd/ftw/forecast_tracking.go index 0c5a8d35f..5309dd28f 100644 --- a/go/cmd/ftw/forecast_tracking.go +++ b/go/cmd/ftw/forecast_tracking.go @@ -89,18 +89,21 @@ type forecastTracker struct { observationLoadValid bool issueArchiveError bool sourceChoice forecastSourceChoice // guarded by mu + + learningRecoveryChecked string // guarded by learningMu; physical learning identity + learningRecoveryAfterMS int64 // a pre-sync clock retries when archived evidence becomes available } func (f *forecastTracker) Start(ctx context.Context) error { - if err := f.restoreLearning(ctx); err != nil { - return err - } initCtx, cancel := context.WithTimeout(ctx, 3*time.Second) err := f.store.InitForecastArchive(initCtx) cancel() if err != nil { return err } + if err := f.restoreLearning(ctx); err != nil { + return err + } f.queue = make(chan forecastJob, 8) f.scoreWake = make(chan struct{}, 1) workerCtx, stop := context.WithCancel(ctx) diff --git a/go/internal/energyforecast/client.go b/go/internal/energyforecast/client.go index bb6900578..6d97b0d2f 100644 --- a/go/internal/energyforecast/client.go +++ b/go/internal/energyforecast/client.go @@ -25,7 +25,23 @@ func NewClient(transport RoundTripper) *Client { return &Client{transport: transport, Timeout: DefaultTimeout} } -type workerError struct{ detail string } +type workerError struct { + detail string + code string + message string +} + +// ObservationRejectedError marks input that cannot become trainable on retry. +// The caller must retain its raw archive before it advances the training queue. +type ObservationRejectedError struct{ Err error } + +func (e *ObservationRejectedError) Error() string { return e.Err.Error() } +func (e *ObservationRejectedError) Unwrap() error { return e.Err } + +func IsObservationRejected(err error) bool { + var rejected *ObservationRejectedError + return errors.As(err, &rejected) +} func (e workerError) Error() string { return "forecast worker rejected request: " + e.detail } @@ -34,64 +50,79 @@ func (c *Client) Update(ctx context.Context, request UpdateRequest) (UpdateReply if err := validateContext(request.RequestContext); err != nil { return reply, err } - if len(request.Observations) > MaxObservations { - return reply, errors.New("forecast update accepts at most 4096 observations") + if err := validateUpdateObservations(request); err != nil { + return reply, &ObservationRejectedError{Err: err} } if request.Observations == nil { request.Observations = []Observation{} } + payload := struct { + Op string `json:"op"` + Version int `json:"version"` + Action string `json:"action"` + UpdateRequest + }{"forecast", ProtocolVersion, "update", request} + if err := c.call(ctx, payload, request.RequestContext, "update", &reply); err != nil { + var rejected workerError + // A bad state/configuration affects the pipeline, not this measured + // interval. Only explicit observation rejections may advance its queue. + if errors.As(err, &rejected) && rejected.code == "invalid_request" && + (rejected.message == "household load exceeds configured limit" || + rejected.message == "PV observation exceeds interval or power limit") { + return UpdateReply{}, &ObservationRejectedError{Err: err} + } + return UpdateReply{}, err + } + if err := validateState(reply.State, true); err != nil { + return UpdateReply{}, err + } + if err := validateCounts(reply.Updates.PV, request.Config.PV != nil, len(request.Observations)); err != nil { + return UpdateReply{}, err + } + if err := validateCounts(reply.Updates.Load, request.Config.Load != nil, len(request.Observations)); err != nil { + return UpdateReply{}, err + } + return reply, nil +} + +func validateUpdateObservations(request UpdateRequest) error { + if len(request.Observations) > MaxObservations { + return errors.New("forecast update accepts at most 4096 observations") + } var previousEnd int64 for i, o := range request.Observations { if err := validateInterval(o.Interval); err != nil { - return reply, fmt.Errorf("observation %d: %w", i, err) + return fmt.Errorf("observation %d: %w", i, err) } if o.ValidEndMs-o.ValidStartMs != 900000 || o.ValidStartMs%900000 != 0 { - return reply, errors.New("forecast observation must cover a complete quarter-hour") + return errors.New("forecast observation must cover a complete quarter-hour") } if i > 0 && o.ValidStartMs < previousEnd { - return reply, errors.New("forecast observations overlap or are out of order") + return errors.New("forecast observations overlap or are out of order") } previousEnd = o.ValidEndMs if o.AvailableAtMs < o.ValidEndMs || o.AvailableAtMs > request.OriginMs { - return reply, errors.New("forecast observation was not available at origin") + return errors.New("forecast observation was not available at origin") } if err := validateFeatures(o.Features, request.OriginMs); err != nil { - return reply, fmt.Errorf("observation %d: %w", i, err) + return fmt.Errorf("observation %d: %w", i, err) } for _, signal := range []struct { value *float64 quality ObservationQuality }{{o.PVAvailableW, o.PVQuality}, {o.HouseholdLoadW, o.LoadQuality}} { if !validObservationQuality(signal.quality) { - return reply, errors.New("forecast observation needs explicit quality") + return errors.New("forecast observation needs explicit quality") } if signal.value != nil && (!finite(*signal.value) || *signal.value < 0) { - return reply, errors.New("forecast observation power must be finite and nonnegative") + return errors.New("forecast observation power must be finite and nonnegative") } if signal.value == nil && (signal.quality == QualityGood || signal.quality == QualityClipped) { - return reply, errors.New("good forecast observation has no measured power") + return errors.New("good forecast observation has no measured power") } } } - payload := struct { - Op string `json:"op"` - Version int `json:"version"` - Action string `json:"action"` - UpdateRequest - }{"forecast", ProtocolVersion, "update", request} - if err := c.call(ctx, payload, request.RequestContext, "update", &reply); err != nil { - return UpdateReply{}, err - } - if err := validateState(reply.State, true); err != nil { - return UpdateReply{}, err - } - if err := validateCounts(reply.Updates.PV, request.Config.PV != nil, len(request.Observations)); err != nil { - return UpdateReply{}, err - } - if err := validateCounts(reply.Updates.Load, request.Config.Load != nil, len(request.Observations)); err != nil { - return UpdateReply{}, err - } - return reply, nil + return nil } func (c *Client) Predict(ctx context.Context, request PredictRequest) (PredictReply, error) { @@ -211,7 +242,12 @@ func (c *Client) call(ctx context.Context, request any, expected RequestContext, if detail == "" || detail == "null" { detail = "unspecified error" } - return workerError{detail} + var reason struct { + Code string `json:"code"` + Message string `json:"message"` + } + _ = json.Unmarshal(echo.Error, &reason) + return workerError{detail: detail, code: reason.Code, message: reason.Message} } if raw, ok := fields["model_revision"]; !ok || bytes.Equal(raw, []byte("null")) { return errors.New("forecast reply lacks a model revision") diff --git a/go/internal/energyforecast/client_test.go b/go/internal/energyforecast/client_test.go index 40e5bdaaf..2fbd5c133 100644 --- a/go/internal/energyforecast/client_test.go +++ b/go/internal/energyforecast/client_test.go @@ -83,6 +83,46 @@ func TestClientPreservesUnknownAndProvisional(t *testing.T) { } } +func TestClientObservationRejectionLeavesModelFailuresRetryable(t *testing.T) { + for _, tc := range []struct { + code, message string + permanent bool + }{ + {"invalid_request", "household load exceeds configured limit", true}, + {"invalid_request", "PV observation exceeds interval or power limit", true}, + {"invalid_request", "invalid household-load configuration", false}, + {"invalid_state", "household load exceeds configured limit", false}, + {"invalid_state", "saved model is malformed", false}, + {"numerical_failure", "update failed", false}, + {"future_data", "clock precedes saved state", false}, + } { + t.Run(tc.code+"/"+tc.message, func(t *testing.T) { + c := replying(func(r map[string]any) { + r["ok"] = false + r["error"] = map[string]string{"code": tc.code, "message": tc.message} + }) + _, err := c.Update(context.Background(), updateRequest()) + if err == nil || IsObservationRejected(err) != tc.permanent { + t.Fatalf("classification=%v err=%v", IsObservationRejected(err), err) + } + }) + } + for _, value := range []float64{math.NaN(), math.Inf(1), -1} { + r := updateRequest() + r.Observations[0].HouseholdLoadW = ptr(value) + _, err := replying(nil).Update(context.Background(), r) + if !IsObservationRejected(err) { + t.Fatalf("invalid observation was not classified: %v", err) + } + } + r := updateRequest() + r.State = json.RawMessage(`{"bad-state":`) + _, err := replying(nil).Update(context.Background(), r) + if err == nil || IsObservationRejected(err) { + t.Fatalf("invalid state discarded valid observation: %v", err) + } +} + func TestClientRejectsWrongReplyIdentity(t *testing.T) { for _, key := range []string{"op", "version", "action", "request_id", "site_id", "config_revision", "origin_ms", "ok"} { t.Run(key, func(t *testing.T) { diff --git a/go/internal/state/forecast_issues.go b/go/internal/state/forecast_issues.go index d64d93c15..863a5ec62 100644 --- a/go/internal/state/forecast_issues.go +++ b/go/internal/state/forecast_issues.go @@ -428,6 +428,29 @@ func (s *Store) LoadForecastIssues(ctx context.Context, since, until int64, limi return out, nil } +// LoadForecastCohortSite reads one bounded issue to bind an older evaluation +// cohort to its physical learning identity. Cohorts change on worker updates; +// loading every issue and model snapshot would delay recovery on small boxes. +func (s *Store) LoadForecastCohortSite(ctx context.Context, cohort string, until int64) (*forecasting.SiteContext, error) { + var data []byte + err := s.db.QueryRowContext(ctx, `SELECT payload FROM forecast_issues + WHERE config_version=? AND issued_at_ms<=? ORDER BY issued_at_ms DESC,id DESC LIMIT 1`, cohort, until).Scan(&data) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + issue, _, err := decodeForecastIssue(data) + if err != nil { + return nil, err + } + if issue.ConfigVersion != cohort { + return nil, errors.New("forecast cohort payload does not match archive key") + } + return issue.Site, nil +} + func (s *Store) LoadForecastObservations(ctx context.Context, since, until int64) (out []forecasting.Observation, err error) { timing := newForecastReadTiming(ctx, "observations") defer timing.finish(&err)