fix: complete forecast, history rebuild and shutdown recovery

.changeset/drain-before-shutdown.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Stop history admission and background work before draining the accepted queue. Give the whole queue a separate shutdown budget, finish deferred cleanup on restart, and report an incomplete drain through a failed process exit. Allow 60 seconds for container shutdown during updates and restarts.

.changeset/forecast-compression-identity.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Keep forecast and model IDs valid when a software update changes gzip encoding. Compare bounded, verified record contents and retain the original archive bytes. Changed or corrupt records still fail validation.

.changeset/resume-hourly-history.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Build hourly history in small batches that follow the primary index. Save progress with each committed batch, resume after timeouts and restarts, and expose unfinished work separately from raw-history migration. Keep live writes and late samples ahead of background work.

docker-compose.macos.yml

@@ -52,6 +52,7 @@ services:
     image: ghcr.io/srcfl/ftw:${FTW_IMAGE_TAG:-latest}
     container_name: ftw
     restart: unless-stopped
+    stop_grace_period: 60s
 
     environment:
       # In-app self-update feature (version banner + Update/Restart

docker-compose.yml

@@ -41,6 +41,7 @@ services:
     image: ghcr.io/srcfl/ftw:${FTW_IMAGE_TAG:-latest}
     container_name: ftw
     restart: unless-stopped
+    stop_grace_period: 60s
 
     environment:
       # Turns on the in-app self-update feature (version-check banner +

go/cmd/ftw-updater/main.go

@@ -472,7 +472,7 @@ func (s *server) restartExisting(spec componentSpec, startedAt time.Time) {
 	s.writeState(st)
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
 	err := s.runWithStateHeartbeat(st, func() error {
-		return s.runner(ctx, nil, s.composeArgs("restart", "--no-deps", spec.service)...)
+		return s.runner(ctx, nil, s.composeArgs("restart", "--no-deps", "--timeout", "60", spec.service)...)
 	})
 	cancel()
 	if err == nil && s.healthCheck != nil {
@@ -608,7 +608,7 @@ func (s *server) runComponentJob(action, target, component string, startedAt tim
 	upCtx, upCancel := context.WithTimeout(context.Background(), 10*time.Minute)
 	defer upCancel()
 
-	upArgs := s.composeArgs("up", "-d", spec.service)
+	upArgs := s.composeArgs("up", "-d", "--timeout", "60", spec.service)
 	if err := s.runWithStateHeartbeat(restartState, func() error {
 		return s.runner(upCtx, env, upArgs...)
 	}); err != nil {
@@ -940,7 +940,7 @@ func (s *server) runRollback(snapshotID string, files []string, safetySnapshotID
 
 	// 1. Stop the main service so SQLite isn't holding a file handle
 	// while we swap state.db under it.
-	if err := s.runner(ctx, nil, "stop", "--time", "30", containerID); err != nil {
+	if err := s.runner(ctx, nil, "stop", "--time", "60", containerID); err != nil {
 		s.writeState(State{State: "failed", Action: base.Action, Snapshot: base.Snapshot, StartedAt: now, UpdatedAt: time.Now(), Message: "container stop failed: " + err.Error()})
 		return
 	}
@@ -1200,7 +1200,7 @@ func decompressGzipFile(src, dst string) error {
 
 func (s *server) recoverRollbackSafety(ctx context.Context, base State, safetySnapshotID string, safetyFiles []string, containerID, imageRef, cause string) {
 	s.writeState(State{State: "restoring", Action: base.Action, Snapshot: base.Snapshot, StartedAt: base.StartedAt, UpdatedAt: time.Now(), Message: "rollback failed; restoring pre-rollback safety backup"})
-	_ = s.runner(ctx, nil, "stop", "--time", "30", containerID)
+	_ = s.runner(ctx, nil, "stop", "--time", "60", containerID)
 	restoreErr := s.restoreSnapshotFiles(ctx, safetySnapshotID, safetyFiles, containerID, imageRef)
 	var startErr error
 	if restoreErr == nil {

go/cmd/ftw-updater/main_test.go

@@ -246,7 +246,7 @@ func TestHandleUpdate_RestartDoesNotRecreate(t *testing.T) {
 	}
 	waitForState(t, s, "done")
 	calls := runner.snapshot()
-	if len(calls) != 1 || strings.Join(calls[0], " ") != strings.Join(s.composeArgs("restart", "--no-deps", s.mainServiceName), " ") {
+	if len(calls) != 1 || strings.Join(calls[0], " ") != strings.Join(s.composeArgs("restart", "--no-deps", "--timeout", "60", s.mainServiceName), " ") {
 		t.Fatalf("restart must only restart the existing container: %v", calls)
 	}
 }
@@ -670,7 +670,7 @@ func TestHandleUpdate_RestartKeepsEveryRunningImage(t *testing.T) {
 				t.Fatalf("restart state = %+v", st)
 			}
 			calls := runner.snapshot()
-			if len(calls) != 1 || strings.Join(calls[0], " ") != strings.Join(s.composeArgs("restart", "--no-deps", canonicalMainServiceName), " ") {
+			if len(calls) != 1 || strings.Join(calls[0], " ") != strings.Join(s.composeArgs("restart", "--no-deps", "--timeout", "60", canonicalMainServiceName), " ") {
 				t.Fatalf("restart selected or pulled a replacement image: %v", calls)
 			}
 			for _, env := range runner.envSnapshot() {
@@ -760,7 +760,7 @@ func TestHandleUpdate_RollbackRestoresFiles(t *testing.T) {
 	if len(calls) != 5 {
 		t.Fatalf("want 5 docker calls, got %d: %v", len(calls), calls)
 	}
-	if got := strings.Join(calls[0], " "); got != "stop --time 30 ftw-container" {
+	if got := strings.Join(calls[0], " "); got != "stop --time 60 ftw-container" {
 		t.Errorf("first call must stop the exact running container: %v", calls[0])
 	}
 	for i, f := range []string{"state.db", "config.yaml"} {
@@ -1243,7 +1243,7 @@ func TestRecoverCrashedRollbackRestoresSafetyBackup(t *testing.T) {
 		t.Fatalf("crashed rollback recovery = %+v", state)
 	}
 	calls := runner.snapshot()
-	if len(calls) != 5 || strings.Join(calls[0], " ") != "stop --time 30 ftw-container" || strings.Join(calls[4], " ") != "start ftw-container" {
+	if len(calls) != 5 || strings.Join(calls[0], " ") != "stop --time 60 ftw-container" || strings.Join(calls[4], " ") != "start ftw-container" {
 		t.Fatalf("crashed rollback recovery calls = %v", calls)
 	}
 }
@@ -1289,7 +1289,7 @@ func TestUpdateReadinessFailureNeverRevertsImage(t *testing.T) {
 				t.Fatalf("state=%+v", st)
 			}
 			calls := runner.snapshot()
-			if len(calls) != 2 || !strings.Contains(strings.Join(calls[0], " "), "pull ftw") || !strings.Contains(strings.Join(calls[1], " "), "up -d ftw") {
+			if len(calls) != 2 || !strings.Contains(strings.Join(calls[0], " "), "pull ftw") || !strings.Contains(strings.Join(calls[1], " "), "up -d --timeout 60 ftw") {
 				t.Fatalf("readiness failure changed the running image: %v", calls)
 			}
 		})

go/cmd/ftw/main.go

@@ -298,6 +298,12 @@ func adoptGatewayIdentityWith(
 }
 
 func main() {
+	exitCode := 0
+	defer func() {
+		if exitCode != 0 {
+			os.Exit(exitCode)
+		}
+	}()
 	imageTag := os.Getenv("FTW_IMAGE_TAG")
 	builtVersion := Version
 	resolvedVersion, imageTagApplied := runtimeVersionFromImageTag(builtVersion, CandidateTag, imageTag)
@@ -465,6 +471,7 @@ func main() {
 	defer func() {
 		if err := st.Close(); err != nil {
 			slog.Error("state shutdown failed", "err", err)
+			exitCode = 1
 		}
 	}()
 	if *retiredShadowSocket != "" {
@@ -614,18 +621,12 @@ func main() {
 	// Closing restartCh from /api/restart drops the main control loop out
 	// of its select, which returns from main() so every defer (HA Stop,
 	// state.Close, http.Shutdown, …) runs in normal LIFO order. The
-	// bottom-of-stack `os.Exit` defer below then translates exitCode 1
+	// first registered `os.Exit` defer then translates exitCode 1
 	// into a non-zero process exit so docker (`unless-stopped`) and
 	// systemd (`Restart=on-failure`) bring the binary back up. SIGTERM /
 	// SIGINT take the same return path with exitCode 0.
 	restartCh := make(chan struct{})
 	var restartOnce sync.Once
-	exitCode := 0
-	defer func() {
-		if exitCode != 0 {
-			os.Exit(exitCode)
-		}
-	}()
 
 	// ---- Driver registry ----
 	ctx, cancel := context.WithCancel(context.Background())
@@ -3382,9 +3383,7 @@ func doRolloff(ctx context.Context, st *state.Store, coldDir string) {
 }
 
 func flushHistoryOnStop(st *state.Store) {
-	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
-	defer cancel()
-	if err := st.FlushHistory(ctx); err != nil {
+	if err := st.StopHistory(); err != nil {
 		slog.Warn("history flush on shutdown", "err", err)
 	}
 }

go/internal/state/forecast_compression_test.go

@@ -0,0 +1,116 @@
+package state
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/forecasting"
+)
+
+func differentForecastGzip(t *testing.T, raw []byte) []byte {
+	t.Helper()
+	var b bytes.Buffer
+	z, err := gzip.NewWriterLevel(&b, gzip.BestSpeed)
+	if err != nil {
+		t.Fatal(err)
+	}
+	z.Name = "older-encoder"
+	if _, err := z.Write(raw); err != nil {
+		t.Fatal(err)
+	}
+	if err := z.Close(); err != nil {
+		t.Fatal(err)
+	}
+	return b.Bytes()
+}
+
+func TestForecastArchiveKeepsIDsAcrossCompressionChanges(t *testing.T) {
+	s := openForecastArchive(t)
+	ctx := context.Background()
+	now := time.Now().Add(-time.Hour).UnixMilli()
+	issue := archiveIssue("before-upgrade", now, now)
+	issue.Models = []forecasting.ModelState{{Name: "pv", Version: "v1", Quality: forecasting.ModelQualityWarm,
+		UpdatedAtMS: now, State: json.RawMessage(`{"weights":[1,2,3],"samples":400}`)}}
+	if err := s.SaveForecastIssue(ctx, issue); err != nil {
+		t.Fatal(err)
+	}
+	prepared, err := prepareForecastIssue(issue)
+	if err != nil {
+		t.Fatal(err)
+	}
+	state := prepared.modelStates[0]
+	oldModel := differentForecastGzip(t, issue.Models[0].State)
+	rawIssue, err := json.Marshal(prepared.issue)
+	if err != nil {
+		t.Fatal(err)
+	}
+	oldIssue := differentForecastGzip(t, rawIssue)
+	if bytes.Equal(oldModel, state.payload) || bytes.Equal(oldIssue, prepared.payload) {
+		t.Fatal("fixture must use different compression")
+	}
+	if _, err := s.db.Exec(`UPDATE forecast_model_states SET payload=? WHERE id=?`, oldModel, state.id); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.db.Exec(`UPDATE forecast_issues SET payload=? WHERE id=?`, oldIssue, issue.ID); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.SaveForecastIssue(ctx, issue); err != nil {
+		t.Fatalf("retry after encoder change: %v", err)
+	}
+	issue.ID = "after-upgrade"
+	if err := s.SaveForecastIssue(ctx, issue); err != nil {
+		t.Fatalf("new forecast referencing existing model: %v", err)
+	}
+	loaded, err := s.LoadForecastModelState(ctx, state.id)
+	if err != nil || !bytes.Equal(loaded, issue.Models[0].State) {
+		t.Fatalf("model changed: %s %v", loaded, err)
+	}
+	var count int
+	var payload []byte
+	if err := s.db.QueryRow(`SELECT COUNT(*) FROM forecast_model_states`).Scan(&count); err != nil || count != 1 {
+		t.Fatalf("model count %d: %v", count, err)
+	}
+	if err := s.db.QueryRow(`SELECT payload FROM forecast_model_states WHERE id=?`, state.id).Scan(&payload); err != nil || !bytes.Equal(payload, oldModel) {
+		t.Fatal("rewrote the existing model")
+	}
+	if err := s.db.QueryRow(`SELECT payload FROM forecast_issues WHERE id='before-upgrade'`).Scan(&payload); err != nil || !bytes.Equal(payload, oldIssue) {
+		t.Fatal("rewrote the existing forecast")
+	}
+}
+
+func TestForecastArchiveStillRejectsChangedOrCorruptModel(t *testing.T) {
+	for _, kind := range []string{"changed", "truncated", "too-large"} {
+		t.Run(kind, func(t *testing.T) {
+			s := openForecastArchive(t)
+			now := time.Now().Add(-time.Hour).UnixMilli()
+			issue := archiveIssue("one", now, now)
+			issue.Models = []forecasting.ModelState{{Name: "pv", Version: "v1", Quality: forecasting.ModelQualityWarm,
+				UpdatedAtMS: now, State: json.RawMessage(`{"weights":[1,2,3]}`)}}
+			if err := s.SaveForecastIssue(context.Background(), issue); err != nil {
+				t.Fatal(err)
+			}
+			payload := differentForecastGzip(t, []byte(`{"weights":[4,5,6]}`))
+			if kind == "truncated" {
+				payload = payload[:len(payload)-4]
+			}
+			if kind == "too-large" {
+				payload = differentForecastGzip(t, bytes.Repeat([]byte("x"), forecasting.MaxModelStateBytes+1))
+			}
+			if _, err := s.db.Exec(`UPDATE forecast_model_states SET payload=?`, payload); err != nil {
+				t.Fatal(err)
+			}
+			issue.ID = "two"
+			if err := s.SaveForecastIssue(context.Background(), issue); err == nil {
+				t.Fatal("accepted invalid existing model")
+			}
+			var n int
+			if err := s.db.QueryRow(`SELECT COUNT(*) FROM forecast_issues`).Scan(&n); err != nil || n != 1 {
+				t.Fatalf("failed save was not atomic: %d %v", n, err)
+			}
+		})
+	}
+}

go/internal/state/forecast_issues.go

@@ -106,6 +106,35 @@ func gzipForecastModelState(state json.RawMessage) ([]byte, error) {
 	return buf.Bytes(), nil
 }
 
+// Identity belongs to the uncompressed record. Gzip headers and encoder
+// versions may change without changing a model or an issued forecast.
+func sameForecastPayload(a, b []byte, compressedLimit, expandedLimit int) bool {
+	if len(a) > compressedLimit || len(b) > compressedLimit {
+		return false
+	}
+	if bytes.Equal(a, b) {
+		return true
+	}
+	expand := func(data []byte) ([]byte, error) {
+		z, err := gzip.NewReader(bytes.NewReader(data))
+		if err != nil {
+			return nil, err
+		}
+		plain, readErr := io.ReadAll(io.LimitReader(z, int64(expandedLimit)+1))
+		closeErr := z.Close()
+		if len(plain) > expandedLimit {
+			return nil, errors.New("oversized forecast payload")
+		}
+		return plain, errors.Join(readErr, closeErr)
+	}
+	plainA, err := expand(a)
+	if err != nil {
+		return false
+	}
+	plainB, err := expand(b)
+	return err == nil && bytes.Equal(plainA, plainB)
+}
+
 type preparedForecastModelState struct {
 	id            string
 	expandedBytes int
@@ -185,7 +214,7 @@ func storePreparedForecastModelStates(ctx context.Context, tx *sql.Tx, prepared
 			if err = tx.QueryRowContext(ctx, "SELECT expanded_bytes,payload FROM forecast_model_states WHERE id=?", state.id).Scan(&expanded, &old); err != nil {
 				return err
 			}
-			if expanded != state.expandedBytes || !bytes.Equal(old, state.payload) {
+			if expanded != state.expandedBytes || !sameForecastPayload(old, state.payload, maxForecastModelCompressed, forecasting.MaxModelStateBytes) {
 				return errors.New("forecast model state ID is immutable")
 			}
 		}
@@ -261,7 +290,7 @@ func (s *Store) SaveForecastIssue(ctx context.Context, issue forecasting.Issue)
 		if err = tx.QueryRowContext(ctx, "SELECT payload FROM forecast_issues WHERE id=?", prepared.issue.ID).Scan(&old); err != nil {
 			return fmt.Errorf("read existing forecast issue: %w", err)
 		}
-		if !bytes.Equal(old, prepared.payload) {
+		if !sameForecastPayload(old, prepared.payload, maxForecastCompressedBytes, forecasting.MaxPayloadBytes) {
 			return errors.New("forecast issue ID is immutable")
 		}
 	}