fix(state): keep maintenance bounded and retain forecast observations

.changeset/storage-maintenance-recovery.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Keep live storage responsive during archive and history maintenance. Bound write transactions, retain completed forecast observations for retry, and expose current storage and forecast health with failure history for diagnostics.

go/cmd/ftw/forecast_learning.go

@@ -201,7 +201,7 @@ func (f *forecastTracker) RestartLearning(ctx context.Context, signal string) er
 }
 
 func (f *forecastTracker) LearningStatus(signal string) forecasting.LearningStatus {
-	status := forecasting.LearningStatus{Engine: "legacy", Status: "unavailable"}
+	status := forecasting.LearningStatus{Engine: "legacy", Status: "unavailable", Health: "unknown"}
 	if f == nil {
 		return status
 	}
@@ -249,9 +249,65 @@ func (f *forecastTracker) LearningStatus(signal string) forecasting.LearningStat
 	if site.IdentityPending || (f.learningPeriods.ConfigRevision == rustConfigRevision(site) && f.learningErrors[signal] != nil) {
 		status.Status = "unavailable"
 	}
+	status.Health, status.HealthReason = f.learningHealth(signal, status)
 	return status
 }
 
+// Health is current pipeline health, separate from the model's learning stage.
+// A ready model alone cannot establish working collection or persistence.
+func (f *forecastTracker) learningHealth(signal string, status forecasting.LearningStatus) (string, string) {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+	if f.stopped {
+		return "unknown", "stopped"
+	}
+	if f.observationOverflow {
+		return "degraded", "observation_queue_full"
+	}
+	if f.observationError != "" {
+		return "degraded", f.observationError
+	}
+	if f.issueArchiveError {
+		return "degraded", "forecast_archive_pending"
+	}
+	if f.store != nil {
+		writer := f.store.HistoryWriterStatus()
+		if writer.LastError != "" {
+			return "degraded", "history_write_pending"
+		}
+		if writer.MaintenanceError != "" || f.store.HistoryMaintenanceStatus().LastError != "" {
+			return "degraded", "history_maintenance_failed"
+		}
+	}
+	if status.Status == "unavailable" {
+		return "unknown", "model_unavailable"
+	}
+	if f.observationCheckedMS == 0 {
+		return "unknown", "not_checked"
+	}
+	age := f.now().UnixMilli() - f.observationCheckedMS
+	valid := f.observationLoadValid
+	trainingAgeLimit := 2 * time.Hour
+	if signal == "pv" {
+		valid = f.observationPVValid
+		trainingAgeLimit = 36 * time.Hour
+	}
+	if age < 0 || age > (30*time.Second).Milliseconds() || !valid {
+		return "waiting_for_data", "measurements_unavailable"
+	}
+	if status.Status != "ready" {
+		return "unknown", "learning"
+	}
+	trainingAge := f.now().UnixMilli() - status.LatestTrainingMS
+	if status.LatestTrainingMS <= 0 || trainingAge < 0 || trainingAge > trainingAgeLimit.Milliseconds() {
+		return "waiting_for_data", "training_outdated"
+	}
+	if status.Engine != "energyplan" {
+		return "unknown", "model_persistence_unchecked"
+	}
+	return "healthy", ""
+}
+
 // Calibration is per signal: a PV reset also drops joint net errors, while
 // retaining load evidence. Neither the archive nor its measured truth is erased.
 func learningEvidence(history []forecasting.ErrorSample, observations []forecasting.Observation, pvMS, loadMS int64) ([]forecasting.ErrorSample, []forecasting.Observation) {

go/cmd/ftw/forecast_observation_queue.go

@@ -0,0 +1,143 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/forecasting"
+	"github.com/srcfl/ftw/go/internal/state"
+)
+
+// Keep a bounded backlog of completed intervals. The observer owns this queue;
+// issuing forecasts and scoring them must not postpone the next measurement.
+const maxPendingForecastObservations = 64
+
+type forecastObservationJob struct {
+	observation forecasting.Observation
+	site        forecastSite
+	weather     *state.ForecastPoint
+	away        bool
+	archived    bool
+}
+
+func (f *forecastTracker) runObservations(ctx context.Context) {
+	defer func() {
+		drain, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
+		defer cancel()
+		f.flushObservations(drain)
+		if len(f.pendingObservations) > 0 {
+			slog.Error("forecast archive: shutdown observations still pending", "intervals", len(f.pendingObservations))
+		}
+	}()
+	tick := time.NewTicker(10 * time.Second)
+	defer tick.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+			f.observe(ctx)
+		}
+	}
+}
+
+// Capture new intervals before disk work, but let a recovered backlog free
+// capacity before rejecting the captured evidence. The same deadline bounds
+// both drains; recovery never postpones the next observation indefinitely.
+func (f *forecastTracker) acceptObservations(ctx context.Context, jobs []forecastObservationJob, flush func(context.Context)) {
+	if len(f.pendingObservations)+len(jobs) > maxPendingForecastObservations {
+		flush(ctx)
+	}
+	for _, job := range jobs {
+		f.enqueueObservation(job)
+	}
+	flush(ctx)
+}
+
+func (f *forecastTracker) enqueueObservation(job forecastObservationJob) {
+	if len(f.pendingObservations) == maxPendingForecastObservations {
+		f.mu.Lock()
+		f.observationOverflow = true
+		f.observationDrops++
+		f.mu.Unlock()
+		slog.Error("forecast archive: observation queue full", "start_ms", job.observation.StartMS)
+		return
+	}
+	f.pendingObservations = append(f.pendingObservations, job)
+}
+
+func (f *forecastTracker) flushObservations(ctx context.Context) {
+	f.flushObservationsWith(ctx, f.store.SaveForecastObservation, f.updateObservation)
+}
+
+// Save and update retries keep the original evidence and captured features.
+// A successful save followed by a failed model update skips saving on retry.
+func (f *forecastTracker) flushObservationsWith(ctx context.Context,
+	save func(context.Context, forecasting.Observation) error,
+	update func(context.Context, forecastObservationJob) error) {
+	for len(f.pendingObservations) > 0 {
+		if ctx.Err() != nil {
+			return
+		}
+		job := &f.pendingObservations[0]
+		writeCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+		phase := "observation_save_pending"
+		var err error
+		if !job.archived {
+			err = save(writeCtx, job.observation)
+			if err == nil {
+				job.archived = true
+			}
+		}
+		if err == nil {
+			phase = "model_save_pending"
+			err = update(writeCtx, *job)
+		}
+		if err != nil && writeCtx.Err() != nil {
+			err = errors.Join(err, writeCtx.Err())
+		}
+		cancel()
+		f.mu.Lock()
+		if err != nil {
+			f.observationError = phase
+		} else {
+			f.observationError = ""
+		}
+		f.mu.Unlock()
+		if err != nil {
+			slog.Warn("forecast archive: observation retained for retry", "phase", phase, "start_ms", job.observation.StartMS, "err", err)
+			return
+		}
+		f.pendingObservations[0] = forecastObservationJob{}
+		f.pendingObservations = f.pendingObservations[1:]
+		f.requestScoring()
+	}
+	f.mu.Lock()
+	if f.observationOverflow {
+		slog.Info("forecast archive: observation queue recovered", "dropped_intervals", f.observationDrops)
+	}
+	f.observationOverflow = false
+	f.mu.Unlock()
+}
+
+func (f *forecastTracker) updateObservation(ctx context.Context, job forecastObservationJob) error {
+	if f.candidate == nil {
+		return nil
+	}
+	if f.configMu != nil {
+		f.configMu.RLock()
+		defer f.configMu.RUnlock()
+	}
+	f.learningMu.RLock()
+	defer f.learningMu.RUnlock()
+	current := f.site()
+	// Old configuration evidence remains in the archive, but must not switch
+	// the current model back to a previous identity when the queue recovers.
+	if current.IdentityPending || current.Revision != job.site.Revision {
+		return nil
+	}
+	site := f.learningSiteLocked(job.site)
+	return f.candidate.Update(ctx, site, job.observation, job.weather, job.away)
+}