From 81e4ed689bdc94814821cbade6d930b7b7729345 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 8 Sep 2026 13:31:44 +0200 Subject: [PATCH 1/3] feat(web): show migration size speed and time estimates --- web/history-migration.js | 49 ++++++++++++++++++++++++++++++---- web/history-migration.test.mjs | 39 +++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 5 deletions(-) diff --git a/web/history-migration.js b/web/history-migration.js index d5bb8050e..651181974 100644 --- a/web/history-migration.js +++ b/web/history-migration.js @@ -1,6 +1,15 @@ // Shared by the startup page, update dialog and live dashboard. const count = value => Math.max(0, Number.isFinite(Number(value)) ? Number(value) : 0); const number = value => count(value).toLocaleString("en-US"); +const decimal = value => value.toLocaleString("en-US", { maximumFractionDigits: 1 }); +const megabytes = value => `${decimal(count(value) / 1e6)} MB`; +function duration(seconds) { + seconds = Math.max(0, Math.round(seconds)); + if (seconds < 60) return `${seconds} s`; + const minutes = Math.floor(seconds / 60); + if (minutes < 60) return `${minutes} min ${seconds % 60} s`; + return `${Math.floor(minutes / 60)} h ${minutes % 60} min`; +} const escape = value => String(value ?? "").replace(/[&<>"']/g, c => ({ "&": "&", "<": "<", ">": ">", '"': """, "'": "'", })[c]); @@ -12,7 +21,10 @@ export function migrationView(migration, { boot = false, connected = true, now = const seed = migration.phase === "seed"; const total = count(archive || seed ? migration.current_source_rows_total : migration.rows_total); const done = count(archive || seed ? migration.current_source_rows_done : migration.rows_done); - const progress = total > 0 ? { value: Math.min(done, total), max: total, label: archive ? "Current archive file" : seed ? "Current preparation step" : "History import progress" } : null; + const byteTotal = archive ? count(migration.source_bytes_total) : 0; + const byteDone = Math.min(count(migration.source_bytes_done), byteTotal); + const progress = byteTotal > 0 ? { value: byteDone, max: byteTotal, label: "Older archive data processed" } + : total > 0 ? { value: Math.min(done, total), max: total, label: archive ? "Current archive file" : seed ? "Current preparation step" : "History import progress" } : null; const details = []; if (seed && done > 0) details.push(`${number(done)}${total > 0 ? ` of ${number(total)}` : ""} saved items copied in this step`); else if (total > 0) details.push(`${archive ? "Current archive: " : ""}${number(done)} of ${number(total)} readings imported`); @@ -28,13 +40,38 @@ export function migrationView(migration, { boot = false, connected = true, now = const unit = age < 60 ? "second" : "minute"; const activity = age === null ? "Waiting for the first import report." : `Last import report ${elapsed} ${unit}${elapsed === 1 ? "" : "s"} ago.`; + const steps = { + checking: archive ? "Checking saved archive files" : "Checking saved history", + importing: archive ? "Importing archive readings" : "Importing saved readings", + waiting_for_live: "Waiting for new measurements to be saved", + checkpointing: "Saving database progress", + }; + const step = !connected ? "" : steps[migration.activity] || ""; + const metrics = []; + if (byteTotal > 0) metrics.push(`${migration.bytes_estimated ? "About " : ""}${megabytes(byteDone)} of ${megabytes(byteTotal)} processed · ${megabytes(byteTotal - byteDone)} remaining`); + const rate = count(migration.bytes_per_second); + const moving = connected && !failed && (age === null || age <= 30) && migration.activity === "importing"; + if (byteTotal > 0 && moving && rate > 0) metrics.push(`Speed: ${rate >= 1e6 ? megabytes(rate) : `${decimal(rate / 1e3)} KB`}/s of archive data`); + const started = count(migration.started_at_ms); + const elapsedMS = migration.elapsed_ms == null + ? started > 0 ? Math.max(0, (connected && !failed ? now : count(updated) || started) - started) : 0 + : count(migration.elapsed_ms); + if (started > 0 || migration.elapsed_ms != null) metrics.push(`Elapsed: ${duration(elapsedMS / 1000)}`); + const eta = Number(migration.eta_seconds); + const estimate = !connected || failed ? "Time remaining unavailable" + : byteTotal > 0 && moving && rate > 0 && migration.eta_seconds != null && Number.isFinite(eta) && eta >= 0 + ? `Estimated time remaining: about ${duration(Math.max(1, eta))}` + : "Time remaining: estimating…"; + metrics.push(estimate); return { title: !connected ? "History import status unavailable" : failed ? "History import paused" : boot ? "Preparing FTW" : "Importing older history", description: !connected ? "The box is not responding. Showing its last import report." : failed ? "The import needs attention. Your original history files are still kept." : boot ? "FTW is preparing the data it needs to start. Control has not started yet." : "Core is running while FTW imports older readings in the background.", - details: details.join(" · "), coverage, activity, progress, + details: details.join(" · "), coverage, activity, progress, step, + metrics: metrics.join(" · "), + sizeNote: byteTotal > 0 ? `Sizes refer to compressed archive files${migration.bytes_estimated ? "; the current file is estimated" : ""}.` : "", error: failed ? migration.last_error || "The box could not finish importing history." : "", guidance: failed || !connected ? "Keep the original data and backup. Reload this page to check the current status." : "Keep the box powered. You can close this page and return later. Import progress is saved so it can resume after a restart.", @@ -51,7 +88,9 @@ export function migrationHTML(migration, options) { const heading = options?.boot ? "h1" : "h3"; return `
<${heading} role="status">${escape(view.title)}

${escape(view.description)}

- ${progress}

${escape(view.details)}

${escape(view.coverage)}

+ ${view.step ? `

${escape(view.step)}

` : ""} + ${progress}

${escape(view.metrics)}

${view.sizeNote ? `

${escape(view.sizeNote)}

` : ""} +

${escape(view.details)}

${escape(view.coverage)}

${escape(view.activity)}

${view.error ? `

${escape(view.error)}

` : ""}

${escape(view.guidance)}

`; @@ -74,10 +113,10 @@ export function updateMigrationBanner(health) { banner.className = "storage-banner"; banner.setAttribute("aria-label", "History import"); banner.style.cssText = "display:block;text-align:left;padding:16px 24px"; - banner.innerHTML = '

'; + banner.innerHTML = '

'; main.parentNode.insertBefore(banner, main); } - for (const key of ["title", "description", "details", "coverage", "activity", "error", "guidance"]) { + for (const key of ["title", "description", "step", "metrics", "sizeNote", "details", "coverage", "activity", "error", "guidance"]) { const element = banner.querySelector(`[data-field="${key}"]`); if (element.textContent !== view[key]) element.textContent = view[key]; element.hidden = !view[key]; diff --git a/web/history-migration.test.mjs b/web/history-migration.test.mjs index 5227d7c88..35e5a2206 100644 --- a/web/history-migration.test.mjs +++ b/web/history-migration.test.mjs @@ -34,3 +34,42 @@ test("failure keeps incomplete coverage and escapes diagnostic text", () => { assert.match(html, /<script>/); assert.match(html, /not complete yet/); }); +const measured = {...running, activity:"importing", source_bytes_total:500e6, source_bytes_done:200e6, bytes_per_second:250000, bytes_estimated:true, started_at_ms:1000, elapsed_ms:800000, eta_seconds:1200}; +test("archive progress shows source MB, rate, elapsed time and a qualified ETA", () => { + const view = migrationView(measured, {now:110000}); + assert.deepEqual(view.progress, {value:200e6, max:500e6, label:"Older archive data processed"}); + assert.match(view.metrics, /About 200 MB of 500 MB processed · 300 MB remaining/); + assert.match(view.metrics, /250 KB\/s of archive data/); + assert.match(view.metrics, /Elapsed: 13 min 20 s/); + assert.match(view.metrics, /Estimated time remaining: about 20 min 0 s/); + assert.match(view.sizeNote, /compressed archive files.*current file is estimated/); +}); +test("checking, paused and stale progress do not keep advertising a live rate or ETA", () => { + for (const [data, options] of [ + [{...measured, activity:"checking"}, {now:110000}], + [{...measured, activity:"waiting_for_live"}, {now:110000}], + [{...measured, state:"failed"}, {now:110000}], + [measured, {now:150000}], + [measured, {now:110000, connected:false}], + ]) { + const view = migrationView(data, options); + assert.doesNotMatch(view.metrics, /Speed:|Estimated time remaining: about/); + } + assert.equal(migrationView({...measured, activity:"checking"}).step, "Checking saved archive files"); +}); +test("unknown byte totals and estimates never become zero remaining", () => { + const view = migrationView({...running, phase:"sqlite", activity:"checking", started_at_ms:1000}, {now:131000}); + assert.match(view.metrics, /Elapsed: 2 min 10 s/); + assert.match(view.metrics, /Time remaining: estimating/); + assert.doesNotMatch(view.metrics, / MB|Speed:/); + assert.equal(view.sizeNote, ""); + assert.equal(view.progress, null); +}); +test("all bytes processed still means verification is pending until Core confirms completion", () => { + const view = migrationView({...measured, source_bytes_done:600e6, activity:"checkpointing"}, {now:110000}); + assert.equal(view.progress.value, view.progress.max); + assert.match(view.metrics, /0 MB remaining/); + assert.match(view.coverage, /not complete yet/); + assert.equal(view.step, "Saving database progress"); + assert.doesNotMatch(view.metrics, /Speed:|Estimated time remaining: about/); +}); From bff52e6bc44dd800f1f4666ad018ee2b0ff350fc Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 8 Sep 2026 13:37:01 +0200 Subject: [PATCH 2/3] fix(web): use archive bytes as the overall progress count --- web/history-migration.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/web/history-migration.js b/web/history-migration.js index 651181974..576c2a867 100644 --- a/web/history-migration.js +++ b/web/history-migration.js @@ -28,7 +28,7 @@ export function migrationView(migration, { boot = false, connected = true, now = const details = []; if (seed && done > 0) details.push(`${number(done)}${total > 0 ? ` of ${number(total)}` : ""} saved items copied in this step`); else if (total > 0) details.push(`${archive ? "Current archive: " : ""}${number(done)} of ${number(total)} readings imported`); - if ((archive || !total) && count(migration.rows_done) > 0) details.push(`${number(migration.rows_done)} readings imported in total`); + if (!byteTotal && (archive || !total) && count(migration.rows_done) > 0) details.push(`${number(migration.rows_done)} readings imported in total`); if (count(migration.files_total) > 0) details.push(`${number(migration.files_done)} of ${number(migration.files_total)} archive files complete`); const from = Number(migration.incomplete_from_ms), until = Number(migration.incomplete_until_ms); const coverage = Number.isFinite(from) && Number.isFinite(until) && from > 0 && until >= from From 891dfb23183cae78b7f639325f6db3c4a0a7e7da Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 8 Sep 2026 13:39:20 +0200 Subject: [PATCH 3/3] feat(state): report source import progress without rescanning SQLite --- .changeset/history-migration-progress.md | 5 + go/internal/state/history_migration.go | 108 ++++++++++++++-- .../state/history_migration_progress_test.go | 122 ++++++++++++++++++ go/internal/state/history_migration_test.go | 4 +- go/internal/state/history_parquet_import.go | 52 +++++++- 5 files changed, 274 insertions(+), 17 deletions(-) create mode 100644 .changeset/history-migration-progress.md create mode 100644 go/internal/state/history_migration_progress_test.go diff --git a/.changeset/history-migration-progress.md b/.changeset/history-migration-progress.md new file mode 100644 index 000000000..fb7b7a9e8 --- /dev/null +++ b/.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. diff --git a/go/internal/state/history_migration.go b/go/internal/state/history_migration.go index 6e6bd83e0..ec965aa14 100644 --- a/go/internal/state/history_migration.go +++ b/go/internal/state/history_migration.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "log/slog" + "math" "sync" "time" @@ -31,15 +32,26 @@ 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 { @@ -47,6 +59,7 @@ func newHistoryMigration(progress func(HistoryMigrationStatus)) *historyMigratio 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,6 +267,13 @@ 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 @@ -202,6 +281,11 @@ func (s *Store) yieldHistoryImport(ctx context.Context) error { 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(): diff --git a/go/internal/state/history_migration_progress_test.go b/go/internal/state/history_migration_progress_test.go new file mode 100644 index 000000000..703e8e2a5 --- /dev/null +++ b/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) + } +} diff --git a/go/internal/state/history_migration_test.go b/go/internal/state/history_migration_test.go index 3acfe5ce0..fc61e6fb4 100644 --- a/go/internal/state/history_migration_test.go +++ b/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 { diff --git a/go/internal/state/history_parquet_import.go b/go/internal/state/history_parquet_import.go index ac6568c3b..f67e97f1b 100644 --- a/go/internal/state/history_parquet_import.go +++ b/go/internal/state/history_parquet_import.go @@ -41,21 +41,34 @@ func (s *Store) ImportLegacyParquet(ctx context.Context, coldDir string) error { if err := s.bindLegacyParquetSources(coldDir); err != nil { return err } - rows, err := s.history.QueryContext(ctx, `SELECT m.path,COALESCE(s.rows,0) FROM history_parquet_manifest m LEFT JOIN history_parquet_sources s ON s.path=m.path ORDER BY m.path`) + rows, err := s.history.QueryContext(ctx, `SELECT m.path,COALESCE(s.rows,0),s.path IS NOT NULL FROM history_parquet_manifest m LEFT JOIN history_parquet_sources s ON s.path=m.path ORDER BY m.path`) if err != nil { return err } paths := []string{} var completedRows int64 + var totalBytes, completedBytes int64 + bytesKnown := true for rows.Next() { var path string var count int64 - if err := rows.Scan(&path, &count); err != nil { + var complete bool + if err := rows.Scan(&path, &count, &complete); err != nil { rows.Close() return err } paths = append(paths, path) completedRows += count + if info, err := os.Stat(path); err == nil { + totalBytes += info.Size() + if complete { + completedBytes += info.Size() + } + } else { + // A verified source may already have been removed. Its old compressed + // size is unknown; do not present a partial inventory as the total. + bytesKnown = false + } } if err := errors.Join(rows.Err(), rows.Close()); err != nil { return err @@ -63,11 +76,15 @@ func (s *Store) ImportLegacyParquet(ctx context.Context, coldDir string) error { if s.historyMigration != nil { s.historyMigration.update(func(st *HistoryMigrationStatus) { st.Phase = "parquet" + st.Activity = "checking" st.FilesTotal = len(paths) st.CurrentSource = "" st.RowsDone += completedRows st.RowsTotal = 0 }) + if bytesKnown { + s.historyMigration.startSourceBytes(&totalBytes, &completedBytes) + } } for _, path := range paths { if err := s.yieldHistoryImport(ctx); err != nil { @@ -122,6 +139,14 @@ func (s *Store) bindLegacyParquetSources(coldDir string) error { } func (s *Store) importHistoryFile(ctx context.Context, path, coldDir string) error { + if s.historyMigration != nil { + name, _ := filepath.Rel(coldDir, path) + s.historyMigration.update(func(st *HistoryMigrationStatus) { + st.CurrentSource = filepath.ToSlash(name) + st.Activity = "checking" + st.CurrentSourceRowsDone, st.CurrentSourceRowsTotal = 0, 0 + }) + } digest, err := historyFileHash(path) if err != nil { // A verified file may have been removed after a complete backup. Its @@ -157,7 +182,12 @@ func (s *Store) importHistoryFile(ctx context.Context, path, coldDir string) err st.RowsTotal = 0 }) } - if err := s.rotateHistory(ctx); err != nil { + info, err := os.Stat(path) + if err != nil { + return err + } + sourceBytes := info.Size() + if err := s.checkpointHistoryImport(ctx); err != nil { return err } // This instance owns all file-sized buffers and may spill to disk. Closing @@ -225,8 +255,10 @@ func (s *Store) importHistoryFile(ctx context.Context, path, coldDir string) err st.CurrentSourceRowsTotal = count st.RowsDone += offset }) + s.historyMigration.addSourceBytes(estimatedSourceBytes(sourceBytes, offset, count), offset > 0, false) } for offset < count { + s.historyActivity("importing") if err := s.yieldHistoryImport(ctx); err != nil { return err } @@ -259,14 +291,16 @@ func (s *Store) importHistoryFile(ctx context.Context, path, coldDir string) err } if s.historyMigration != nil { s.historyMigration.update(func(st *HistoryMigrationStatus) { st.CurrentSourceRowsDone = end; st.RowsDone += end - offset }) + s.historyMigration.addSourceBytes(estimatedSourceBytes(sourceBytes, end, count)-estimatedSourceBytes(sourceBytes, offset, count), true, true) } offset = end if offset%(64*historyImportRows) == 0 { - if err := s.CheckpointHistory(ctx); err != nil { + if err := s.checkpointHistoryImport(ctx); err != nil { return err } } } + s.historyActivity("checking") if after, err := historyFileHash(path); err != nil || after != digest { return errors.Join(err, errors.New("Parquet changed during import; restore the original source")) } @@ -289,10 +323,20 @@ func (s *Store) importHistoryFile(ctx context.Context, path, coldDir string) err if err := tx.Commit(); err != nil { return err } + if s.historyMigration != nil { + s.historyMigration.addSourceBytes(sourceBytes-estimatedSourceBytes(sourceBytes, offset, count), false, true) + } slog.Info("history: verified Parquet import", "file", filepath.Base(path), "rows", count, "sha256", digest) return nil } +func estimatedSourceBytes(size, rows, totalRows int64) int64 { + if totalRows <= 0 { + return 0 + } + return int64(float64(size) * float64(min(rows, totalRows)) / float64(totalRows)) +} + func stageHistoryParquet(ctx context.Context, conn *sql.Conn, path string, progress ...func(int64)) (int64, error) { f, err := os.Open(path) if err != nil {