Show history import size, speed and activity

.changeset/history-migration-progress.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Show compressed Parquet source bytes, estimated progress, throughput, elapsed time and remaining time during history import. Report checking, importing, waiting for live writes and checkpointing separately; hide estimates when there is too little evidence. Skip the full SQLite row scan when its import already has a completion receipt.

go/internal/state/history_migration.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"math"
 	"sync"
 	"time"
 
@@ -31,22 +32,34 @@ type HistoryMigrationStatus struct {
 	LastError              string `json:"last_error,omitempty"`
 	IncompleteFromMS       *int64 `json:"incomplete_from_ms,omitempty"`
 	IncompleteUntilMS      *int64 `json:"incomplete_until_ms,omitempty"`
+	// Bytes count compressed Parquet source files only. A partial file uses
+	// its verified row fraction; unknown byte totals and ETA are omitted.
+	SourceBytesTotal *int64  `json:"source_bytes_total,omitempty"`
+	SourceBytesDone  *int64  `json:"source_bytes_done,omitempty"`
+	BytesPerSecond   float64 `json:"bytes_per_second,omitempty"`
+	ElapsedMS        int64   `json:"elapsed_ms"`
+	ETASeconds       *int64  `json:"eta_seconds,omitempty"`
+	Activity         string  `json:"activity,omitempty"`
+	BytesEstimated   bool    `json:"bytes_estimated"`
 }
 
 type historyMigration struct {
-	mu       sync.Mutex
-	status   HistoryMigrationStatus
-	progress func(HistoryMigrationStatus)
-	ctx      context.Context
-	cancel   context.CancelFunc
-	done     chan struct{}
+	mu           sync.Mutex
+	status       HistoryMigrationStatus
+	progress     func(HistoryMigrationStatus)
+	ctx          context.Context
+	cancel       context.CancelFunc
+	done         chan struct{}
+	byteStarted  time.Time
+	bytesThisRun int64
 }
 
 func newHistoryMigration(progress func(HistoryMigrationStatus)) *historyMigration {
 	ctx, cancel := context.WithCancel(context.Background())
 	m := &historyMigration{ctx: ctx, cancel: cancel, done: make(chan struct{}), progress: progress}
 	m.update(func(st *HistoryMigrationStatus) {
 		st.State, st.Phase = "starting", "seed"
+		st.Activity = "checking"
 		st.StartedAtMS = time.Now().UnixMilli()
 	})
 	return m
@@ -56,7 +69,7 @@ func (m *historyMigration) update(change func(*HistoryMigrationStatus)) {
 	m.mu.Lock()
 	change(&m.status)
 	m.status.UpdatedAtMS = time.Now().UnixMilli()
-	st := m.status
+	st := m.snapshotLocked(time.Now())
 	m.mu.Unlock()
 	if m.progress != nil {
 		m.progress(st)
@@ -70,19 +83,74 @@ func (s *Store) HistoryMigrationStatus() HistoryMigrationStatus {
 	m := s.historyMigration
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	return m.status
+	return m.snapshotLocked(time.Now())
+}
+
+func (m *historyMigration) snapshotLocked(now time.Time) HistoryMigrationStatus {
+	st := m.status
+	end := now.UnixMilli()
+	if st.State == "complete" || st.State == "failed" {
+		end = st.UpdatedAtMS
+	}
+	st.ElapsedMS = max(0, end-st.StartedAtMS)
+	seconds := now.Sub(m.byteStarted).Seconds()
+	if st.State == "running" && st.SourceBytesTotal != nil && st.SourceBytesDone != nil && seconds >= 5 && m.bytesThisRun > 0 {
+		st.BytesPerSecond = float64(m.bytesThisRun) / seconds
+		if st.Activity == "importing" {
+			eta := int64(math.Ceil(float64(max(0, *st.SourceBytesTotal-*st.SourceBytesDone)) / st.BytesPerSecond))
+			st.ETASeconds = &eta
+		}
+	}
+	return st
+}
+
+func (m *historyMigration) startSourceBytes(total, done *int64) {
+	m.update(func(st *HistoryMigrationStatus) {
+		st.SourceBytesTotal, st.SourceBytesDone = total, done
+		m.byteStarted, m.bytesThisRun = time.Now(), 0
+	})
+}
+
+func (m *historyMigration) addSourceBytes(delta int64, estimated, newWork bool) {
+	m.update(func(st *HistoryMigrationStatus) {
+		if st.SourceBytesDone == nil || st.SourceBytesTotal == nil {
+			return
+		}
+		done := min(*st.SourceBytesTotal, *st.SourceBytesDone+max(0, delta))
+		st.SourceBytesDone, st.BytesEstimated = &done, estimated
+		if newWork {
+			m.bytesThisRun += max(0, delta)
+		}
+	})
+}
+
+func (s *Store) historyActivity(activity string) {
+	if s.historyMigration != nil {
+		s.historyMigration.update(func(st *HistoryMigrationStatus) { st.Activity = activity })
+	}
+}
+
+func (s *Store) checkpointHistoryImport(ctx context.Context) error {
+	previous := s.HistoryMigrationStatus().Activity
+	s.historyActivity("checkpointing")
+	defer s.historyActivity(previous)
+	return s.CheckpointHistory(ctx)
 }
 
 func (s *Store) runHistoryMigration(coldDir string) {
 	m := s.historyMigration
 	defer close(m.done)
-	m.update(func(st *HistoryMigrationStatus) { st.State, st.Phase = "running", "sqlite" })
+	m.update(func(st *HistoryMigrationStatus) {
+		st.State, st.Phase, st.Activity = "running", "sqlite", "checking"
+		st.CurrentSource = "state.db/ts_samples"
+		st.CurrentSourceRowsDone, st.CurrentSourceRowsTotal = 0, 0
+	})
 	err := s.importSQLiteSamples(m.ctx)
 	if err == nil {
 		err = s.ImportLegacyParquet(m.ctx, coldDir)
 	}
 	if err == nil {
-		err = s.CheckpointHistory(m.ctx)
+		err = s.checkpointHistoryImport(m.ctx)
 	}
 	if err == nil {
 		s.historyWriteMu.Lock()
@@ -99,6 +167,7 @@ func (s *Store) runHistoryMigration(coldDir string) {
 	}
 	m.update(func(st *HistoryMigrationStatus) {
 		st.State, st.Phase, st.HistoryComplete = "complete", "complete", true
+		st.Activity = ""
 		st.CurrentSource = ""
 		st.CurrentSourceRowsDone, st.CurrentSourceRowsTotal = 0, 0
 		st.IncompleteFromMS, st.IncompleteUntilMS = nil, nil
@@ -116,7 +185,9 @@ func (s *Store) importSQLiteSamples(ctx context.Context) error {
 	}
 	if complete != 0 {
 		var count int64
-		if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM ts_samples`).Scan(&count); err != nil {
+		// Do not scan the frozen SQLite table on every restart. Older releases
+		// have no row cursor; leave their row count unknown instead of scanning.
+		if err := s.history.QueryRowContext(ctx, `SELECT COALESCE((SELECT rows_done FROM history_sqlite_progress WHERE source='ts_samples'),0)`).Scan(&count); err != nil {
 			return err
 		}
 		s.historyMigration.update(func(st *HistoryMigrationStatus) { st.RowsDone = count })
@@ -137,6 +208,7 @@ func (s *Store) importSQLiteSamples(ctx context.Context) error {
 		st.CurrentSourceRowsDone, st.CurrentSourceRowsTotal = done, count
 	})
 	for {
+		s.historyActivity("importing")
 		if err := s.yieldHistoryImport(ctx); err != nil {
 			return err
 		}
@@ -178,7 +250,7 @@ func (s *Store) importSQLiteSamples(ctx context.Context) error {
 		done = next
 		s.historyMigration.update(func(st *HistoryMigrationStatus) { st.RowsDone, st.CurrentSourceRowsDone = done, done })
 		if done%(64*historyImportRows) == 0 {
-			if err := s.CheckpointHistory(ctx); err != nil {
+			if err := s.checkpointHistoryImport(ctx); err != nil {
 				return err
 			}
 		}
@@ -195,13 +267,25 @@ func (s *Store) importSQLiteSamples(ctx context.Context) error {
 // Let queued live ticks commit before importing another bounded chunk. No
 // database or catalog lock is held while the importer yields.
 func (s *Store) yieldHistoryImport(ctx context.Context) error {
+	waiting := false
+	previous := ""
+	defer func() {
+		if waiting {
+			s.historyActivity(previous)
+		}
+	}()
 	for {
 		if err := ctx.Err(); err != nil {
 			return err
 		}
 		if s.HistoryWriterStatus().Pending == 0 {
 			return nil
 		}
+		if !waiting {
+			previous = s.HistoryMigrationStatus().Activity
+			s.historyActivity("waiting_for_live")
+			waiting = true
+		}
 		timer := time.NewTimer(10 * time.Millisecond)
 		select {
 		case <-ctx.Done():

go/internal/state/history_migration_progress_test.go

@@ -0,0 +1,122 @@
+package state
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestMigrationByteRateExcludesPriorProgressAndWaitETA(t *testing.T) {
+	m := newHistoryMigration(nil)
+	defer m.cancel()
+	s := &Store{historyMigration: m}
+	unknown, _ := json.Marshal(s.HistoryMigrationStatus())
+	if strings.Contains(string(unknown), "source_bytes_total") || strings.Contains(string(unknown), "eta_seconds") {
+		t.Fatalf("invented unknown metrics: %s", unknown)
+	}
+	total, prior := int64(1000), int64(500)
+	m.startSourceBytes(&total, &prior)
+	m.update(func(st *HistoryMigrationStatus) {
+		st.State, st.Phase, st.Activity = "running", "parquet", "importing"
+		st.StartedAtMS = time.Now().Add(-20 * time.Second).UnixMilli()
+		m.byteStarted = time.Now().Add(-10 * time.Second)
+	})
+	if st := s.HistoryMigrationStatus(); st.ETASeconds != nil || st.BytesPerSecond != 0 {
+		t.Fatalf("old bytes became current throughput: %+v", st)
+	}
+	m.addSourceBytes(100, true, true)
+	st := s.HistoryMigrationStatus()
+	if *st.SourceBytesDone != 600 || *st.SourceBytesTotal != 1000 || !st.BytesEstimated || st.BytesPerSecond < 9 || st.BytesPerSecond > 11 || st.ETASeconds == nil || *st.ETASeconds < 40 || *st.ETASeconds > 42 || st.ElapsedMS < 20000 {
+		t.Fatalf("wrong byte progress: %+v", st)
+	}
+	s.historyActivity("waiting_for_live")
+	if st := s.HistoryMigrationStatus(); st.ETASeconds != nil {
+		t.Fatalf("ETA claimed during blocked writer: %+v", st)
+	}
+	if prior != 500 {
+		t.Fatal("mutated a published status snapshot")
+	}
+}
+
+func TestCompletedSQLiteImportDoesNotReadSQLite(t *testing.T) {
+	s := freshStore(t)
+	m := newHistoryMigration(nil)
+	s.historyMigration = m
+	defer func() { s.historyMigration = nil; m.cancel() }()
+	if err := s.db.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.importSQLiteSamples(context.Background()); err != nil {
+		t.Fatalf("completed import touched closed SQLite: %v", err)
+	}
+}
+
+func TestParquetByteProgressCountsCompressedSourcesAndPartialRows(t *testing.T) {
+	path, cold := legacyMigrationFixture(t, 0)
+	day := filepath.Join(cold, "2026", "01")
+	if err := os.MkdirAll(day, 0700); err != nil {
+		t.Fatal(err)
+	}
+	first, second := filepath.Join(day, "01.parquet"), filepath.Join(day, "02.parquet")
+	if err := writeParquetDay(first, []parquetSampleRow{{TsMs: 1, Driver: "meter", Metric: "power", Value: 1}}); err != nil {
+		t.Fatal(err)
+	}
+	s, err := OpenWithLegacyHistory(path, cold)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Close(); err != nil {
+		t.Fatal(err)
+	}
+	points := make([]parquetSampleRow, historyImportRows+13)
+	for i := range points {
+		points[i] = parquetSampleRow{TsMs: int64(10000 + i), Driver: "meter", Metric: "power", Value: float64(i)}
+	}
+	if err := writeParquetDay(second, points); err != nil {
+		t.Fatal(err)
+	}
+	f1, err := os.Stat(first)
+	if err != nil {
+		t.Fatal(err)
+	}
+	f2, err := os.Stat(second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	reached, release := make(chan struct{}), make(chan struct{})
+	var once, unblock sync.Once
+	s, err = OpenWithBackgroundHistory(path, cold, func(st HistoryMigrationStatus) {
+		if st.BytesEstimated && st.CurrentSourceRowsDone == historyImportRows {
+			once.Do(func() { close(reached); <-release })
+		}
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { unblock.Do(func() { close(release) }); s.Close() }()
+	select {
+	case <-reached:
+	case <-time.After(10 * time.Second):
+		t.Fatal("no partial byte progress")
+	}
+	st := s.HistoryMigrationStatus()
+	want := f1.Size() + estimatedSourceBytes(f2.Size(), historyImportRows, int64(len(points)))
+	if st.SourceBytesDone == nil || *st.SourceBytesDone != want || st.SourceBytesTotal == nil || *st.SourceBytesTotal != f1.Size()+f2.Size() || st.Activity != "importing" {
+		t.Fatalf("wrong source bytes: %+v", st)
+	}
+	unblock.Do(func() { close(release) })
+	select {
+	case <-s.historyMigration.done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("import did not finish")
+	}
+	st = s.HistoryMigrationStatus()
+	if !st.HistoryComplete || st.BytesEstimated || *st.SourceBytesDone != *st.SourceBytesTotal {
+		t.Fatalf("wrong final progress: %+v", st)
+	}
+}

go/internal/state/history_migration_test.go

@@ -593,7 +593,9 @@ func TestBackgroundHistoryContinuesBetaOneReceiptsAndSequences(t *testing.T) {
 		t.Fatal("legacy import did not finish")
 	}
 	st := s.HistoryMigrationStatus()
-	if seedRepeated || !st.HistoryComplete || st.FilesDone != 2 || st.RowsDone != 20 {
+	// beta.1 has no saved SQLite row count. Resume must not scan that table
+	// just to fill a status counter; only the three known Parquet rows count.
+	if seedRepeated || !st.HistoryComplete || st.FilesDone != 2 || st.RowsDone != 3 {
 		t.Fatalf("seedRepeated=%v status=%+v", seedRepeated, st)
 	}
 	if got, err := s.historyConfig("history_duckdb_generation"); err != nil || got != generation {