From 84778bc23e9e83f3488f56b9b3bf593126954bbf Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 14 Sep 2026 13:40:53 +0200 Subject: [PATCH 1/4] fix: preserve forecast identity across compression changes Signed-off-by: Fredrik Ahlgren --- .changeset/forecast-compression-identity.md | 5 + .../state/forecast_compression_test.go | 116 ++++++++++++++++++ go/internal/state/forecast_issues.go | 33 ++++- 3 files changed, 152 insertions(+), 2 deletions(-) create mode 100644 .changeset/forecast-compression-identity.md create mode 100644 go/internal/state/forecast_compression_test.go diff --git a/.changeset/forecast-compression-identity.md b/.changeset/forecast-compression-identity.md new file mode 100644 index 00000000..014f3c71 --- /dev/null +++ b/.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. diff --git a/go/internal/state/forecast_compression_test.go b/go/internal/state/forecast_compression_test.go new file mode 100644 index 00000000..c3cd1c77 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/go/internal/state/forecast_issues.go b/go/internal/state/forecast_issues.go index 2ca58dc6..551c7325 100644 --- a/go/internal/state/forecast_issues.go +++ b/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") } } From 51fbcf72eb746371e6c06fb0a60bea03d4c046ae Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 14 Sep 2026 13:40:53 +0200 Subject: [PATCH 2/4] fix: resume hourly history through bounded index reads Signed-off-by: Fredrik Ahlgren --- .changeset/resume-hourly-history.md | 5 + go/internal/state/history_series_hour.go | 233 +++++++++++------- .../state/history_series_hour_dataset_test.go | 62 +++++ .../state/history_series_hour_resume_test.go | 136 ++++++++++ go/internal/state/history_sqlite.go | 2 +- 5 files changed, 347 insertions(+), 91 deletions(-) create mode 100644 .changeset/resume-hourly-history.md create mode 100644 go/internal/state/history_series_hour_dataset_test.go create mode 100644 go/internal/state/history_series_hour_resume_test.go diff --git a/.changeset/resume-hourly-history.md b/.changeset/resume-hourly-history.md new file mode 100644 index 00000000..57617e83 --- /dev/null +++ b/.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. diff --git a/go/internal/state/history_series_hour.go b/go/internal/state/history_series_hour.go index 7a333426..41af7f90 100644 --- a/go/internal/state/history_series_hour.go +++ b/go/internal/state/history_series_hour.go @@ -57,164 +57,217 @@ func (s *Store) startSeriesHourBackfill() { return } s.seriesHourMu.Lock() - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) + if s.seriesHourCancel != nil { + s.seriesHourMu.Unlock() + return + } + ctx, cancel := context.WithCancel(context.Background()) s.seriesHourCancel = cancel s.seriesHourWG.Add(1) s.seriesHourMu.Unlock() go func() { defer s.seriesHourWG.Done() defer cancel() - if err := s.ensureSeriesHours(ctx); err != nil { - slog.Error("hourly series rollup paused; long-range charts keep using raw samples", "err", err) - } + s.runSeriesHourBackfill(ctx, 5*time.Second) }() } -// ensureSeriesHours builds ts_series_hour from ts_samples once. Live writes -// keep the table current afterwards. A completed import must run this before -// year-scale /api/series can use the rollup. -func (s *Store) ensureSeriesHours(ctx context.Context) error { - if s.history == nil { - return nil +func (s *Store) runSeriesHourBackfill(ctx context.Context, retryDelay time.Duration) { + for ctx.Err() == nil { + attemptCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + err := s.ensureSeriesHours(attemptCtx) + cancel() + if err == nil { + return + } + if ctx.Err() != nil { + return + } + slog.Warn("hourly series rollup will resume; raw history remains available", "err", err) + timer := time.NewTimer(retryDelay) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } } - if s.seriesHoursReady() { +} + +// The cursor follows the samples' primary key, skipping empty hours and +// avoiding time-index reads scattered across the whole database. Every +// completed batch saves its cursor in the same transaction as its summaries. +// Live/late writes seed a complete hour before adding samples, so an existing +// summary must never be replaced by a background snapshot. +func (s *Store) ensureSeriesHours(ctx context.Context) error { + if s.history == nil || s.seriesHoursReady() { return nil } - var minTs, maxTs sql.NullInt64 - // Separate extrema let SQLite seek each end of the time index instead of - // scanning the complete history for a combined MIN/MAX aggregate. - rangeCtx, cancelRange := context.WithTimeout(ctx, 5*time.Second) - err := s.history.QueryRowContext(rangeCtx, `SELECT - (SELECT MIN(ts_ms) FROM ts_samples), (SELECT MAX(ts_ms) FROM ts_samples)`).Scan(&minTs, &maxTs) - cancelRange() - if err != nil { - return err - } - if minTs.Valid { - start := seriesHourOf(minTs.Int64) - end := maxTs.Int64 + 1 - const chunk = seriesHourMs - for t := start; t < end; t += chunk { - if err := ctx.Err(); err != nil { - return err - } - tEnd := t + chunk - if tEnd > end { - tEnd = end - } - if err := s.backfillSeriesHour(ctx, t, tEnd); err != nil { + for { + if err := ctx.Err(); err != nil { + return err + } + if s.HistoryWriterStatus().Pending >= historyCommitMaxTicks/2 { + if err := pauseMaintenance(ctx); err != nil { return err } + continue + } + done, err := s.backfillSeriesHours(ctx, 64) + if err != nil { + return err + } + if done { + break + } + if err := pauseMaintenance(ctx); err != nil { + return err } } if err := s.ensureParquetHours(ctx); err != nil { return err } - s.historyWriteMu.Lock() - _, err = s.history.ExecContext(ctx, `INSERT INTO history_migrations(name) VALUES (?) ON CONFLICT DO NOTHING`, seriesHoursMigration) - s.historyWriteMu.Unlock() - if err == nil { - slog.Info("hourly series rollup ready") + defer s.historyWriteMu.Unlock() + tx, err := s.history.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + if _, err := tx.ExecContext(ctx, `INSERT INTO history_migrations(name) VALUES (?) ON CONFLICT DO NOTHING`, seriesHoursMigration); err != nil { + return err } - return err + if _, err := tx.ExecContext(ctx, `DELETE FROM history_sqlite_progress WHERE source=?`, seriesHoursMigration); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return err + } + slog.Info("hourly series rollup ready") + return nil } -// A deferred transaction reads without owning SQLite's writer. If a live write -// changes its snapshot, SQLite rejects the write upgrade; retry the read rather -// than publishing stale aggregates. Never use BEGIN IMMEDIATE here. -func (s *Store) backfillSeriesHour(ctx context.Context, from, until int64) error { +func (s *Store) backfillSeriesHours(ctx context.Context, maxHours int) (bool, error) { var err error for attempt := 0; attempt < 3; attempt++ { - var wrote bool - wrote, err = s.backfillSeriesHourSnapshot(ctx, from, until) + var done bool + done, err = s.backfillSeriesHoursSnapshot(ctx, maxHours) if err == nil { - if wrote { - return pauseMaintenance(ctx) - } - return nil + return done, nil } var sqliteErr *sqlite.Error if !errors.As(err, &sqliteErr) || sqliteErr.Code()&0xff != 5 { - return err + return false, err } if err := pauseMaintenance(ctx); err != nil { - return err + return false, err } } - return err + return false, err } -func (s *Store) backfillSeriesHourSnapshot(ctx context.Context, from, until int64) (bool, error) { +func (s *Store) backfillSeriesHoursSnapshot(ctx context.Context, maxHours int) (bool, error) { txCtx, cancelTx := context.WithTimeout(ctx, 10*time.Second) defer cancelTx() + // Deferred: reading never owns SQLite's writer. A concurrent write makes + // the later upgrade fail with BUSY_SNAPSHOT, including the cursor write. tx, err := s.history.BeginTx(txCtx, nil) if err != nil { return false, err } defer tx.Rollback() - readCtx, cancelRead := context.WithTimeout(ctx, 5*time.Second) + readCtx, cancelRead := context.WithTimeout(txCtx, 5*time.Second) defer cancelRead() - rows, err := tx.QueryContext(readCtx, ` - SELECT driver_id, metric_id, (ts_ms / ?) * ?, SUM(value), MIN(value), MAX(value), COUNT(*), MAX(ts_ms) - FROM ts_samples WHERE ts_ms >= ? AND ts_ms < ? GROUP BY 1, 2, 3`, - seriesHourMs, seriesHourMs, from, until) - if err != nil { + var driver, metric, through, rowsDone int64 + err = tx.QueryRowContext(readCtx, `SELECT rows_done,driver_id,metric_id,ts_ms FROM history_sqlite_progress WHERE source=?`, seriesHoursMigration).Scan(&rowsDone, &driver, &metric, &through) + if err != nil && !errors.Is(err, sql.ErrNoRows) { return false, err } type hour struct { key seriesHourKey acc seriesHourAcc } - var hours []hour - for rows.Next() { - if len(hours) >= 4096 { - rows.Close() - return false, errors.New("hourly series backfill exceeds 4096 series per hour") + hours := make([]hour, 0, maxHours) + done := false + started := time.Now() + for len(hours) < maxHours { + var nextDriver, nextMetric, nextTs int64 + err := tx.QueryRowContext(readCtx, `SELECT driver_id,metric_id,ts_ms FROM ts_samples + WHERE (driver_id,metric_id,ts_ms) > (?,?,?) ORDER BY driver_id,metric_id,ts_ms LIMIT 1`, driver, metric, through).Scan(&nextDriver, &nextMetric, &nextTs) + if errors.Is(err, sql.ErrNoRows) { + done = true + break + } + if err != nil { + return false, err + } + h := hour{key: seriesHourKey{nextDriver, nextMetric, seriesHourOf(nextTs)}} + if h.key.hourMs > math.MaxInt64-seriesHourMs { + return false, errors.New("hourly series timestamp exceeds supported range") } - var h hour - if err := rows.Scan(&h.key.driverID, &h.key.metricID, &h.key.hourMs, - &h.acc.sum, &h.acc.min, &h.acc.max, &h.acc.n, &h.acc.last); err != nil { - rows.Close() + err = tx.QueryRowContext(readCtx, `SELECT SUM(value),MIN(value),MAX(value),COUNT(*),MAX(ts_ms) + FROM ts_samples WHERE driver_id=? AND metric_id=? AND ts_ms>=? AND ts_ms= 500*time.Millisecond { + break + } } - err = errors.Join(rows.Err(), rows.Close()) cancelRead() - if err != nil { - return false, err - } if len(hours) == 0 { - return false, tx.Commit() + return done, tx.Commit() } - // Preserve existing summaries. SQLite validates that the source snapshot - // is still current when this transaction first tries to write. - writeCtx, cancelWrite := context.WithTimeout(ctx, historyCommitTimeout) + writeCtx, cancelWrite := context.WithTimeout(txCtx, historyCommitTimeout) defer cancelWrite() s.historyWriteMu.Lock() defer s.historyWriteMu.Unlock() - stmt, err := tx.PrepareContext(writeCtx, ` - INSERT INTO ts_series_hour (driver_id, metric_id, hour_ms, sum_value, min_value, max_value, n, last_ts_ms) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING`) + stmt, err := tx.PrepareContext(writeCtx, `INSERT INTO ts_series_hour + (driver_id,metric_id,hour_ms,sum_value,min_value,max_value,n,last_ts_ms) + VALUES (?,?,?,?,?,?,?,?) ON CONFLICT DO NOTHING`) if err != nil { return false, err } defer stmt.Close() - wrote := false for _, h := range hours { - res, err := stmt.ExecContext(writeCtx, h.key.driverID, h.key.metricID, h.key.hourMs, - h.acc.sum, h.acc.min, h.acc.max, h.acc.n, h.acc.last) - if err != nil { - return false, err - } - n, err := res.RowsAffected() - if err != nil { + if _, err := stmt.ExecContext(writeCtx, h.key.driverID, h.key.metricID, h.key.hourMs, h.acc.sum, h.acc.min, h.acc.max, h.acc.n, h.acc.last); err != nil { return false, err } - wrote = wrote || n > 0 } - return wrote, tx.Commit() + if _, err := tx.ExecContext(writeCtx, `INSERT INTO history_sqlite_progress(source,rows_done,driver_id,metric_id,ts_ms) + VALUES (?,?,?,?,?) ON CONFLICT(source) DO UPDATE SET rows_done=excluded.rows_done, + driver_id=excluded.driver_id,metric_id=excluded.metric_id,ts_ms=excluded.ts_ms`, + seriesHoursMigration, rowsDone, driver, metric, through); err != nil { + return false, err + } + return done, tx.Commit() +} + +// Raw history remains usable until all summaries and Parquet sources are ready. +func (s *Store) SeriesHourBackfillStatus() map[string]any { + if s == nil || s.history == nil { + return map[string]any{"state": "unavailable"} + } + ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond) + defer cancel() + var rows int64 + var ready bool + err := s.history.QueryRowContext(ctx, `SELECT + EXISTS(SELECT 1 FROM history_migrations WHERE name=?), + COALESCE((SELECT rows_done FROM history_sqlite_progress WHERE source=?),0)`, seriesHoursMigration, seriesHoursMigration).Scan(&ready, &rows) + if err != nil { + return map[string]any{"state": "unknown", "ready": false} + } + if ready { + return map[string]any{"state": "complete", "ready": true} + } + return map[string]any{"state": "rebuilding", "ready": false, "rows_done": rows} } type seriesHourAcc struct { diff --git a/go/internal/state/history_series_hour_dataset_test.go b/go/internal/state/history_series_hour_dataset_test.go new file mode 100644 index 00000000..60e6176f --- /dev/null +++ b/go/internal/state/history_series_hour_dataset_test.go @@ -0,0 +1,62 @@ +package state + +import ( + "context" + "math" + "os" + "testing" + "time" +) + +// Run against a disposable COPY of a large history.db, never a live database: +// FTW_BACKFILL_TEST_DB=/path/to/copy.db go test ./internal/state -run TestSeriesHourBackfillDataset -v -timeout 30m +func TestSeriesHourBackfillDataset(t *testing.T) { + path := os.Getenv("FTW_BACKFILL_TEST_DB") + if path == "" { + t.Skip("set FTW_BACKFILL_TEST_DB to a disposable history copy") + } + db, err := openDurableHistory(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + s := &Store{history: db} + for _, q := range []string{`DELETE FROM ts_series_hour`, `DELETE FROM history_migrations WHERE name='ts-series-hour-v1'`, `DELETE FROM history_sqlite_progress WHERE source='ts-series-hour-v1'`} { + if _, err := db.Exec(q); err != nil { + t.Fatal(err) + } + } + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Minute) + defer cancel() + start := time.Now() + for batch := 0; ; batch++ { + done, err := s.backfillSeriesHours(ctx, 64) + if err != nil { + t.Fatal(err) + } + if batch%200 == 0 { + t.Logf("batch=%d elapsed=%s status=%v", batch, time.Since(start).Round(time.Second), s.SeriesHourBackfillStatus()) + } + if done { + break + } + if err := pauseMaintenance(ctx); err != nil { + t.Fatal(err) + } + } + if err := s.ensureSeriesHours(ctx); err != nil { + t.Fatal(err) + } + var rawN, hourN int64 + var rawSum, hourSum, rawMin, hourMin, rawMax, hourMax float64 + if err := db.QueryRowContext(ctx, `SELECT COUNT(*),SUM(value),MIN(value),MAX(value) FROM ts_samples`).Scan(&rawN, &rawSum, &rawMin, &rawMax); err != nil { + t.Fatal(err) + } + if err := db.QueryRowContext(ctx, `SELECT SUM(n),SUM(sum_value),MIN(min_value),MAX(max_value) FROM ts_series_hour`).Scan(&hourN, &hourSum, &hourMin, &hourMax); err != nil { + t.Fatal(err) + } + if rawN != hourN || rawMin != hourMin || rawMax != hourMax || math.Abs(rawSum-hourSum) > 1e-8*math.Max(1, math.Abs(rawSum)) { + t.Fatalf("raw/summary mismatch: n=%d/%d sum=%g/%g min=%g/%g max=%g/%g", rawN, hourN, rawSum, hourSum, rawMin, hourMin, rawMax, hourMax) + } + t.Logf("verified rows=%d elapsed=%s ready=%v", rawN, time.Since(start).Round(time.Second), s.seriesHoursReady()) +} diff --git a/go/internal/state/history_series_hour_resume_test.go b/go/internal/state/history_series_hour_resume_test.go new file mode 100644 index 00000000..c6f32863 --- /dev/null +++ b/go/internal/state/history_series_hour_resume_test.go @@ -0,0 +1,136 @@ +package state + +import ( + "context" + "database/sql/driver" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + "modernc.org/sqlite" +) + +func unbuiltSeriesHours(t *testing.T, s *Store, hours int) { + t.Helper() + var samples []Sample + for i := 0; i < hours; i++ { + samples = append(samples, Sample{Driver: "meter", Metric: "pv_w", TsMs: int64(i) * seriesHourMs, Value: 1}, + Sample{Driver: "meter", Metric: "pv_w", TsMs: int64(i)*seriesHourMs + 1, Value: 3}) + } + if err := s.RecordSamples(samples); err != nil { + t.Fatal(err) + } + for _, q := range []string{`DELETE FROM ts_series_hour`, `DELETE FROM history_migrations WHERE name='ts-series-hour-v1'`} { + if _, err := s.history.Exec(q); err != nil { + t.Fatal(err) + } + } +} + +func TestSeriesHourBackfillResumesAfterReopenAndLateWrite(t *testing.T) { + s := freshStore(t) + unbuiltSeriesHours(t, s, 4) + if done, err := s.backfillSeriesHours(context.Background(), 1); err != nil || done { + t.Fatalf("first batch: done=%v err=%v", done, err) + } + var rows, through int64 + if err := s.history.QueryRow(`SELECT rows_done,ts_ms FROM history_sqlite_progress WHERE source=?`, seriesHoursMigration).Scan(&rows, &through); err != nil || rows != 2 || through != seriesHourMs-1 { + t.Fatalf("cursor: rows=%d through=%d err=%v", rows, through, err) + } + path := s.mainDBPath + if err := s.Close(); err != nil { + t.Fatal(err) + } + var oldReads atomic.Int64 + function := fmt.Sprintf("test_resume_value_%d", time.Now().UnixNano()) + if err := sqlite.RegisterScalarFunction(function, 2, func(_ *sqlite.FunctionContext, args []driver.Value) (driver.Value, error) { + if args[0].(int64) < seriesHourMs { + oldReads.Add(1) + } + return args[1], nil + }); err != nil { + t.Fatal(err) + } + s, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer s.Close() + // A late sample behind the cursor must update its completed summary. + if err := s.RecordSamples([]Sample{{Driver: "meter", Metric: "pv_w", TsMs: 2, Value: 8}}); err != nil { + t.Fatal(err) + } + if _, err := s.history.Exec(`ALTER TABLE ts_samples RENAME TO resume_source`); err != nil { + t.Fatal(err) + } + if _, err := s.history.Exec(`CREATE VIEW ts_samples AS SELECT driver_id,metric_id,ts_ms,` + function + `(ts_ms,value) AS value FROM resume_source`); err != nil { + t.Fatal(err) + } + if err := s.ensureSeriesHours(context.Background()); err != nil { + t.Fatal(err) + } + if oldReads.Load() != 0 { + t.Fatalf("reread %d values behind durable cursor", oldReads.Load()) + } + var n int64 + var sum float64 + if err := s.history.QueryRow(`SELECT SUM(n),SUM(sum_value) FROM ts_series_hour`).Scan(&n, &sum); err != nil || n != 9 || sum != 24 { + t.Fatalf("rollup: n=%d sum=%v err=%v", n, sum, err) + } + if !s.seriesHoursReady() { + t.Fatal("missing completion marker") + } +} + +func TestSeriesHourBackfillCursorAndSummaryCommitTogether(t *testing.T) { + s := freshStore(t) + unbuiltSeriesHours(t, s, 2) + if _, err := s.history.Exec(`CREATE TRIGGER reject_hour_progress BEFORE INSERT ON history_sqlite_progress + WHEN NEW.source='ts-series-hour-v1' BEGIN SELECT RAISE(ABORT,'test interrupted checkpoint'); END`); err != nil { + t.Fatal(err) + } + if _, err := s.backfillSeriesHours(context.Background(), 1); err == nil { + t.Fatal("expected checkpoint failure") + } + for _, table := range []string{"ts_series_hour", "history_sqlite_progress"} { + var n int + if err := s.history.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&n); err != nil || n != 0 { + t.Fatalf("partial commit in %s: %d %v", table, n, err) + } + } + if _, err := s.history.Exec(`DROP TRIGGER reject_hour_progress`); err != nil { + t.Fatal(err) + } + if err := s.ensureSeriesHours(context.Background()); err != nil { + t.Fatal(err) + } +} + +func TestSeriesHourBackfillRetriesWithoutRestart(t *testing.T) { + var attempts atomic.Int64 + function := fmt.Sprintf("test_retry_hour_value_%d", time.Now().UnixNano()) + if err := sqlite.RegisterScalarFunction(function, 1, func(_ *sqlite.FunctionContext, args []driver.Value) (driver.Value, error) { + if attempts.Add(1) == 1 { + return nil, errors.New("temporary read failure") + } + return args[0], nil + }); err != nil { + t.Fatal(err) + } + s := freshStore(t) + unbuiltSeriesHours(t, s, 2) + if _, err := s.history.Exec(`ALTER TABLE ts_samples RENAME TO retry_source`); err != nil { + t.Fatal(err) + } + if _, err := s.history.Exec(`CREATE VIEW ts_samples AS SELECT driver_id,metric_id,ts_ms,` + function + `(value) AS value FROM retry_source`); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + s.runSeriesHourBackfill(ctx, time.Millisecond) + if !s.seriesHoursReady() { + t.Fatal("background retry did not finish without a restart") + } +} diff --git a/go/internal/state/history_sqlite.go b/go/internal/state/history_sqlite.go index b0e3dacd..a42c0514 100644 --- a/go/internal/state/history_sqlite.go +++ b/go/internal/state/history_sqlite.go @@ -321,7 +321,7 @@ func historyFloatBits(value float64) uint64 { } func (s *Store) HistoryBackend() map[string]any { - info := map[string]any{"engine": "sqlite", "archive": "parquet", "file": filepath.Base(s.historyPath), "writer": s.HistoryWriterStatus(), "migration": s.HistoryMigrationStatus()} + info := map[string]any{"engine": "sqlite", "archive": "parquet", "file": filepath.Base(s.historyPath), "writer": s.HistoryWriterStatus(), "migration": s.HistoryMigrationStatus(), "series_hour": s.SeriesHourBackfillStatus()} for key, path := range map[string]string{"file_bytes": s.historyPath, "wal_bytes": s.historyPath + "-wal"} { if stat, err := os.Stat(path); err == nil { info[key] = stat.Size() From 467512279363655cab75ec84c32516933cacc883 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 14 Sep 2026 13:40:53 +0200 Subject: [PATCH 3/4] fix: drain accepted history before process exit Signed-off-by: Fredrik Ahlgren --- .changeset/drain-before-shutdown.md | 5 ++ docker-compose.macos.yml | 1 + docker-compose.yml | 1 + go/cmd/ftw-updater/main.go | 8 +-- go/cmd/ftw-updater/main_test.go | 10 +-- go/cmd/ftw/main.go | 19 +++-- go/internal/state/history_shutdown_test.go | 82 ++++++++++++++++++++++ go/internal/state/history_writer.go | 23 +++++- go/internal/state/store.go | 47 ++++++++----- 9 files changed, 157 insertions(+), 39 deletions(-) create mode 100644 .changeset/drain-before-shutdown.md create mode 100644 go/internal/state/history_shutdown_test.go diff --git a/.changeset/drain-before-shutdown.md b/.changeset/drain-before-shutdown.md new file mode 100644 index 00000000..64687725 --- /dev/null +++ b/.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. diff --git a/docker-compose.macos.yml b/docker-compose.macos.yml index 48139532..ff75041d 100644 --- a/docker-compose.macos.yml +++ b/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 diff --git a/docker-compose.yml b/docker-compose.yml index ce3e1127..a92a1996 100644 --- a/docker-compose.yml +++ b/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 + diff --git a/go/cmd/ftw-updater/main.go b/go/cmd/ftw-updater/main.go index 6cff3093..858c5a7c 100644 --- a/go/cmd/ftw-updater/main.go +++ b/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 { diff --git a/go/cmd/ftw-updater/main_test.go b/go/cmd/ftw-updater/main_test.go index c53d44b0..87f01d65 100644 --- a/go/cmd/ftw-updater/main_test.go +++ b/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) } }) diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index 1c02a7a5..0067037e 100644 --- a/go/cmd/ftw/main.go +++ b/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) } } diff --git a/go/internal/state/history_shutdown_test.go b/go/internal/state/history_shutdown_test.go new file mode 100644 index 00000000..550542c2 --- /dev/null +++ b/go/internal/state/history_shutdown_test.go @@ -0,0 +1,82 @@ +package state + +import ( + "context" + "strings" + "testing" + "time" +) + +func TestStopHistoryCancelsBackfillAndDrainsMultipleSlowBatches(t *testing.T) { + s := freshStore(t) + w := s.historyWriter + w.commitInterval = time.Hour + w.commitMaxTicks = 1 + w.commitTimeout = 100 * time.Millisecond + w.shutdownTimeout = time.Second + backfill, cancel := context.WithCancel(context.Background()) + s.seriesHourCancel = cancel + s.seriesHourWG.Add(1) + go func() { defer s.seriesHourWG.Done(); <-backfill.Done() }() + w.commitFn = func(ctx context.Context, batches []historyBatch, ack int64) (historyBatchCommit, error) { + select { + case <-backfill.Done(): + case <-ctx.Done(): + return historyBatchCommit{}, ctx.Err() + } + timer := time.NewTimer(40 * time.Millisecond) + defer timer.Stop() + select { + case <-timer.C: + case <-ctx.Done(): + return historyBatchCommit{}, ctx.Err() + } + return s.recordHistoryBatches(ctx, batches, ack) + } + for i := 1; i <= 4; i++ { + if err := s.EnqueueTelemetryTick(nil, []Sample{{Driver: "live", Metric: "power", TsMs: int64(i), Value: float64(i)}}, nil); err != nil { + t.Fatal(err) + } + } + if err := s.StopHistory(); err != nil { + t.Fatal(err) + } + st := s.HistoryWriterStatus() + if st.Accepted != 4 || st.Committed != 4 || st.Pending != 0 || !st.Stopping { + t.Fatalf("incomplete shutdown: %+v", st) + } + got, err := s.LoadSeries("live", "power", 0, 5, 0) + if err != nil || len(got) != 4 { + t.Fatalf("accepted samples missing: %v %v", got, err) + } + if err := s.EnqueueTelemetryTick(nil, []Sample{{Driver: "live", Metric: "power", TsMs: 5, Value: 5}}, nil); err == nil { + t.Fatal("accepted data after shutdown") + } + if err := s.StopHistory(); err != nil { + t.Fatalf("second stop: %v", err) + } +} + +func TestStopHistoryReportsUncommittedTicksAfterBudget(t *testing.T) { + s := freshStore(t) + w := s.historyWriter + w.commitInterval = time.Hour + w.commitTimeout = 20 * time.Millisecond + w.shutdownTimeout = 50 * time.Millisecond + w.commitFn = func(ctx context.Context, _ []historyBatch, _ int64) (historyBatchCommit, error) { + <-ctx.Done() + return historyBatchCommit{}, ctx.Err() + } + if err := s.EnqueueTelemetryTick(nil, []Sample{{Driver: "live", Metric: "power", TsMs: 1, Value: 1}}, nil); err != nil { + t.Fatal(err) + } + if err := s.StopHistory(); err == nil || !strings.Contains(err.Error(), "1 ticks uncommitted") { + t.Fatalf("shutdown result: %v", err) + } + if st := s.HistoryWriterStatus(); st.Committed != 0 || st.Pending != 1 { + t.Fatalf("lost failure state: %+v", st) + } + if err := s.Close(); err == nil { + t.Fatal("Close hid the incomplete drain") + } +} diff --git a/go/internal/state/history_writer.go b/go/internal/state/history_writer.go index 279885e9..bfa7c138 100644 --- a/go/internal/state/history_writer.go +++ b/go/internal/state/history_writer.go @@ -20,6 +20,7 @@ const ( historyBatchBytes = 1 << 20 historyMaintenanceTimeout = 2 * time.Minute historyCommitTimeout = 5 * time.Second + historyShutdownTimeout = 20 * time.Second historyCommitInterval = 15 * time.Second historyCommitMaxTicks = 32 ) @@ -74,6 +75,9 @@ type historyWriter struct { maintenanceRetryDelay time.Duration maintenanceRunning atomic.Bool maintenanceMu sync.Mutex + maintenanceCtx context.Context + maintenanceCancel context.CancelFunc + shutdownTimeout time.Duration maintenanceWG sync.WaitGroup commitInterval time.Duration commitTimeout time.Duration @@ -90,6 +94,7 @@ func newHistoryWriter(s *Store) *historyWriter { maintenanceRowsLimit: 64 * historyImportRows, maintenanceDue: time.Now().Add(time.Hour), maintenanceRetryDelay: 30 * time.Second, commitInterval: historyCommitInterval, commitTimeout: historyCommitTimeout, commitMaxTicks: historyCommitMaxTicks, flushCh: make(chan struct{}, 1)} + w.maintenanceCtx, w.maintenanceCancel = context.WithCancel(ctx) go w.run() return w } @@ -383,6 +388,9 @@ func (w *historyWriter) run() { // Hourly rotation runs in the background so a multi-GB reopen cannot stall // live commits past the site watchdog. func (w *historyWriter) scheduleMaintenance(rows int) { + if w.maintenanceCtx.Err() != nil { + return + } w.maintenanceRows += rows now := time.Now() if now.Before(w.maintenanceRetry) || (w.maintenanceRows < w.maintenanceRowsLimit && now.Before(w.maintenanceDue)) { @@ -411,7 +419,7 @@ func (w *historyWriter) maintainHistory(rows int) { func (w *historyWriter) runMaintenance() { w.maintenanceMu.Lock() defer w.maintenanceMu.Unlock() - ctx, cancel := context.WithTimeout(w.ctx, historyMaintenanceTimeout) + ctx, cancel := context.WithTimeout(w.maintenanceCtx, historyMaintenanceTimeout) err := w.store.checkpointLiveHistory(ctx) cancel() if w.forceRotate.Load() { @@ -474,7 +482,7 @@ func (s *Store) FlushHistory(ctx context.Context) error { return nil } -func (w *historyWriter) close() error { +func (w *historyWriter) stopAdmission() { w.mu.Lock() target := w.status.Accepted if !w.status.Stopping { @@ -484,7 +492,16 @@ func (w *historyWriter) close() error { } w.mu.Unlock() w.requestFlush(target) - timer := time.NewTimer(w.timeout()) +} + +func (w *historyWriter) close() error { + w.stopAdmission() + w.maintenanceCancel() + budget := w.shutdownTimeout + if budget <= 0 { + budget = historyShutdownTimeout + } + timer := time.NewTimer(budget) defer timer.Stop() select { case <-w.done: diff --git a/go/internal/state/store.go b/go/internal/state/store.go index 4fa33e3d..38504373 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -298,6 +298,34 @@ func (s *Store) Close() error { if s == nil { return nil } + err := s.StopHistory() + if s.hot != nil && s.hot != s.history { + err = errors.Join(err, s.hot.Close()) + } + if s.history != nil { + err = errors.Join(err, s.history.Close()) + } + if s.cache != nil { + err = errors.Join(err, s.cache.Close()) + } + if s.db != nil { + if e := s.db.Close(); e != nil { + err = errors.Join(err, e) + } + } + return err +} + +// StopHistory stops admission and background reads before draining accepted +// measurements. Databases remain open for final events and deferred cleanup. +func (s *Store) StopHistory() error { + if s == nil { + return nil + } + if s.historyWriter != nil { + s.historyWriter.stopAdmission() + s.historyWriter.maintenanceCancel() + } // Stop the background integrity scan first: db.Close() blocks until every // in-flight query finishes, and the scan's quick_check can run for minutes on // a large DB. Cancelling it (sqlite3_interrupt) lets the close happen promptly @@ -320,25 +348,10 @@ func (s *Store) Close() error { s.seriesHourMu.Unlock() s.seriesHourWG.Wait() - var err error if s.historyWriter != nil { - err = s.historyWriter.close() - } - if s.hot != nil && s.hot != s.history { - err = errors.Join(err, s.hot.Close()) + return s.historyWriter.close() } - if s.history != nil { - err = errors.Join(err, s.history.Close()) - } - if s.cache != nil { - err = errors.Join(err, s.cache.Close()) - } - if s.db != nil { - if e := s.db.Close(); e != nil { - err = errors.Join(err, e) - } - } - return err + return nil } // resolveMainDBPath is where heal.go drops the clean-shutdown marker. From baf5ad90da1c7fde4ca924cb229a05cab5bba288 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 14 Sep 2026 13:53:33 +0200 Subject: [PATCH 4/4] fix: retain the full work budget for Parquet backfill Signed-off-by: Fredrik Ahlgren --- go/internal/state/history_series_hour.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/go/internal/state/history_series_hour.go b/go/internal/state/history_series_hour.go index 41af7f90..9c11b005 100644 --- a/go/internal/state/history_series_hour.go +++ b/go/internal/state/history_series_hour.go @@ -74,7 +74,11 @@ func (s *Store) startSeriesHourBackfill() { func (s *Store) runSeriesHourBackfill(ctx context.Context, retryDelay time.Duration) { for ctx.Err() == nil { - attemptCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + // SQLite work checkpoints each small batch. Parquet keeps a receipt + // per complete file, so retain its original two-hour work budget; + // a short whole-attempt deadline would restart a slow day forever. + // Individual SQLite reads/writes remain bounded to five seconds. + attemptCtx, cancel := context.WithTimeout(ctx, 2*time.Hour) err := s.ensureSeriesHours(attemptCtx) cancel() if err == nil {