From cee71eaf825d26b4281fece52c243cb32fd614c2 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 14 Sep 2026 11:14:03 +0200 Subject: [PATCH] fix: let offline backups copy without live-work pacing OpenBackupSource no longer inherits the live 100ms yield that made a two-hour export deadline unreachable for large converted histories. Live Core backups still pause between copy batches. Verification extracts beside the archive, and create refuses to start without room for the raw export, compressed file and extract. Addresses #1259 Signed-off-by: Fredrik Ahlgren --- .changeset/offline-backup-pacing.md | 5 + docs/backup-and-restore.md | 7 +- go/internal/backup/archive.go | 17 ++- go/internal/backup/archive_test.go | 69 +++++++++- go/internal/state/backup_pace_test.go | 156 ++++++++++++++++++++++ go/internal/state/backup_state.go | 88 +++++++++++- go/internal/state/history_primary_test.go | 2 +- go/internal/state/history_sqlite.go | 6 +- go/internal/state/maintenance_io.go | 12 +- go/internal/state/store.go | 20 ++- 10 files changed, 358 insertions(+), 24 deletions(-) create mode 100644 .changeset/offline-backup-pacing.md create mode 100644 go/internal/state/backup_pace_test.go diff --git a/.changeset/offline-backup-pacing.md b/.changeset/offline-backup-pacing.md new file mode 100644 index 00000000..9cede322 --- /dev/null +++ b/.changeset/offline-backup-pacing.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Let the offline backup helper copy large histories without the live 100 ms pause that made a two-hour export deadline unreachable. Keep live backup yielding so charging-goal saves stay within their latency limit, and refuse to start when the destination cannot hold the raw export, compressed archive and verification extract. diff --git a/docs/backup-and-restore.md b/docs/backup-and-restore.md index 87a52984..68a92a61 100644 --- a/docs/backup-and-restore.md +++ b/docs/backup-and-restore.md @@ -80,8 +80,11 @@ ftw-backup revert -data /var/lib/ftw -safety /var/lib/.ftw-pre-restore-... -yes Stop the native FTW service before `restore` or `revert`. `create` opens the existing database read-only and does not migrate or repair its schema. -Pass `-config` to `create` when the seed has a name other than -`/config.yaml`. The config seed must be inside the data directory. +It copies without the live 100 ms yield used while Core is collecting, and +refuses to start when the destination cannot hold the raw export, compressed +archive and verification extract. Pass `-config` to `create` when the seed has +a name other than `/config.yaml`. The config seed must be inside the +data directory. ## Svenska – kortversion att skicka till en användare diff --git a/go/internal/backup/archive.go b/go/internal/backup/archive.go index 739f195d..1589bda5 100644 --- a/go/internal/backup/archive.go +++ b/go/internal/backup/archive.go @@ -146,6 +146,9 @@ func Create(ctx context.Context, opts CreateOptions) (Info, error) { if err := os.MkdirAll(outputDir, 0o700); err != nil { return Info{}, fmt.Errorf("backup: create output dir: %w", err) } + if err := state.EnsureDiskSpace(outputDir, state.BackupArchiveScratch(opts.State.BackupSourceBytes(), 0)); err != nil { + return Info{}, err + } if opts.Maintenance != nil { opts.Maintenance.Lock() defer opts.Maintenance.Unlock() @@ -245,7 +248,7 @@ func Create(ctx context.Context, opts CreateOptions) (Info, error) { } tmpPath := filepath.Join(outputDir, "."+id+".tmp") - if err := writeArchive(ctx, tmpPath, manifest, sources); err != nil { + if err := writeArchive(ctx, tmpPath, manifest, sources, !opts.State.OfflineBackup()); err != nil { _ = os.Remove(tmpPath) return Info{}, err } @@ -381,7 +384,7 @@ func describeSource(ctx context.Context, dataDir string, source sourceEntry) (Fi return entry, err } -func writeArchive(ctx context.Context, dst string, manifest Manifest, sources []sourceEntry) error { +func writeArchive(ctx context.Context, dst string, manifest Manifest, sources []sourceEntry, live bool) error { f, err := os.OpenFile(dst, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) if err != nil { return err @@ -393,7 +396,7 @@ func writeArchive(ctx context.Context, dst string, manifest Manifest, sources [] _ = os.Remove(dst) } }() - zw, err := gzip.NewWriterLevel(state.NewMaintenanceWriter(ctx, f), gzip.BestSpeed) + zw, err := gzip.NewWriterLevel(state.NewMaintenanceWriterPaced(ctx, f, live), gzip.BestSpeed) if err != nil { return err } @@ -491,7 +494,7 @@ func Verify(archivePath string) (Manifest, error) { want[entry.Path] = entry } seen := make(map[string]bool, len(want)) - tmpDir, err := os.MkdirTemp("", ".ftw-backup-verify-") + tmpDir, err := os.MkdirTemp(filepath.Dir(archivePath), ".ftw-backup-verify-") if err != nil { return Manifest{}, err } @@ -569,7 +572,7 @@ func Verify(archivePath string) (Manifest, error) { if len(seen) != len(want) { return Manifest{}, errors.New("backup: archive is missing one or more manifest files") } - if err := verifyCompressedDatabase(dbGzip); err != nil { + if err := verifyCompressedDatabase(dbGzip, tmpDir); err != nil { return Manifest{}, err } return manifest, nil @@ -1055,8 +1058,8 @@ func extractArchive(archivePath, staging string) error { return verifyDatabase(filepath.Join(staging, filepath.FromSlash(manifest.DatabaseFile))) } -func verifyCompressedDatabase(src string) error { - tmp, err := os.CreateTemp("", ".ftw-backup-db-*.sqlite") +func verifyCompressedDatabase(src, workDir string) error { + tmp, err := os.CreateTemp(workDir, ".ftw-backup-db-*.sqlite") if err != nil { return err } diff --git a/go/internal/backup/archive_test.go b/go/internal/backup/archive_test.go index 3a95da7e..96c6f8de 100644 --- a/go/internal/backup/archive_test.go +++ b/go/internal/backup/archive_test.go @@ -71,7 +71,7 @@ func TestBackupRejectsUnreadableParquetWithMatchingHash(t *testing.T) { manifest.Files = append(manifest.Files, entry) } archive := filepath.Join(t.TempDir(), "bad.ftwbak") - if err := writeArchive(context.Background(), archive, manifest, sources); err != nil { + if err := writeArchive(context.Background(), archive, manifest, sources, true); err != nil { t.Fatal(err) } if _, err := Verify(archive); err == nil { @@ -81,6 +81,73 @@ func TestBackupRejectsUnreadableParquetWithMatchingHash(t *testing.T) { } } +func TestOfflineBackupCreateVerifyRestore(t *testing.T) { + root := t.TempDir() + dataDir := filepath.Join(root, "data") + if err := os.MkdirAll(dataDir, 0o700); err != nil { + t.Fatal(err) + } + statePath := filepath.Join(dataDir, "state.db") + st, err := state.Open(statePath) + if err != nil { + t.Fatal(err) + } + if err := st.SaveConfig("ev_goal", "80% by 07:00"); err != nil { + t.Fatal(err) + } + points := make([]state.HistoryPoint, 2500) + for i := range points { + points[i] = state.HistoryPoint{TsMs: int64(i + 1), GridW: float64(i)} + } + if err := st.BulkRecordHistory(points); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + src, err := state.OpenBackupSource(statePath) + if err != nil { + t.Fatal(err) + } + defer src.Close() + if !src.OfflineBackup() { + t.Fatal("offline helper was treated as a live Core store") + } + info, err := Create(context.Background(), CreateOptions{ + State: src, StatePath: statePath, DataDir: dataDir, + OutputDir: filepath.Join(root, "backups"), + }) + if err != nil { + t.Fatal(err) + } + blocked := filepath.Join(t.TempDir(), "not-a-dir") + if err := os.WriteFile(blocked, []byte("file"), 0o600); err != nil { + t.Fatal(err) + } + t.Setenv("TMPDIR", blocked) + t.Setenv("TMP", blocked) + t.Setenv("TEMP", blocked) + if _, err := Verify(info.Path); err != nil { + t.Fatal(err) + } + restoredDir := filepath.Join(root, "restored") + if _, err := Restore(info.Path, restoredDir, time.Time{}); err != nil { + t.Fatal(err) + } + restored, err := state.Open(filepath.Join(restoredDir, "state.db")) + if err != nil { + t.Fatal(err) + } + defer restored.Close() + if goal, ok := restored.LoadConfig("ev_goal"); !ok || goal != "80% by 07:00" { + t.Fatalf("restored goal = %q ok=%v", goal, ok) + } + h, err := restored.LoadHistory(0, 3000, 0) + if err != nil || len(h) != 2500 || h[0].GridW != 0 || h[2499].GridW != 2499 { + t.Fatalf("restored history = %d %v %+v", len(h), err, h) + } +} + func TestCreateVerifyAndRestoreCompleteBackup(t *testing.T) { root := t.TempDir() dataDir := filepath.Join(root, "data") diff --git a/go/internal/state/backup_pace_test.go b/go/internal/state/backup_pace_test.go new file mode 100644 index 00000000..226a23ca --- /dev/null +++ b/go/internal/state/backup_pace_test.go @@ -0,0 +1,156 @@ +package state + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "testing" + "time" +) + +func TestBackupWorkContextOmitsDeadlineWhenOffline(t *testing.T) { + offline := &Store{offlineBackup: true} + ctx, cancel := offline.backupWorkContext() + defer cancel() + if _, ok := ctx.Deadline(); ok { + t.Fatal("offline backup must not inherit the live export deadline") + } + + live := &Store{} + ctx, cancel = live.backupWorkContext() + defer cancel() + deadline, ok := ctx.Deadline() + if !ok { + t.Fatal("live backup must keep an export deadline") + } + until := time.Until(deadline) + if until > liveBackupTimeout || until < liveBackupTimeout-time.Minute { + t.Fatalf("live backup deadline = %v", until) + } +} + +func TestLiveBackupCopyYieldsBetweenBatches(t *testing.T) { + s := freshStore(t) + pauses := 0 + s.backupPause = func(context.Context) error { pauses++; return nil } + if err := s.BulkRecordHistory(pacedBackupPoints(2500)); err != nil { + t.Fatal(err) + } + dst := filepath.Join(t.TempDir(), "export.db") + if err := s.copyStateForBackup(context.Background(), dst); err != nil { + t.Fatal(err) + } + if err := s.exportHistoryToSQLite(context.Background(), dst); err != nil { + t.Fatal(err) + } + if pauses == 0 { + t.Fatal("live backup must yield between copy batches") + } +} + +func TestOfflineBackupCopyDoesNotYieldBetweenBatches(t *testing.T) { + s := freshStore(t) + if err := s.BulkRecordHistory(pacedBackupPoints(2500)); err != nil { + t.Fatal(err) + } + path := s.mainDBPath + if err := s.Close(); err != nil { + t.Fatal(err) + } + src, err := OpenBackupSource(path) + if err != nil { + t.Fatal(err) + } + defer src.Close() + if !src.OfflineBackup() { + t.Fatal("OpenBackupSource must mark the helper offline") + } + pauses := 0 + src.backupPause = func(context.Context) error { + pauses++ + return nil + } + dst := filepath.Join(t.TempDir(), "export.db") + if err := src.copyStateForBackup(context.Background(), dst); err != nil { + t.Fatal(err) + } + if err := src.exportHistoryToSQLite(context.Background(), dst); err != nil { + t.Fatal(err) + } + if pauses != 0 { + t.Fatalf("offline backup paused %d times; live 100ms yield must not apply", pauses) + } + db, err := sql.Open("sqlite", dst) + if err != nil { + t.Fatal(err) + } + defer db.Close() + var n int + if err := db.QueryRow(`SELECT COUNT(*) FROM history_hot`).Scan(&n); err != nil || n != 2500 { + t.Fatalf("offline export lost rows: %d %v", n, err) + } +} + +func TestBackupArchiveScratchReservesRawGzipAndVerify(t *testing.T) { + const source, extra int64 = 100, 7 + got := BackupArchiveScratch(source, extra) + if got < 3*source+extra+backupScratchHeadroom { + t.Fatalf("scratch %d omits a coexistence phase", got) + } + // Issue #1259: 174,629,397 sample rows with a 100 ms pause every 1,024 + // rows spend 4h44m paused, so the live 2h export deadline must fire. + const rows int64 = 174629397 + livePause := time.Duration(rows/1024) * maintenancePause + if livePause <= liveBackupTimeout { + t.Fatalf("live pause floor %v no longer exceeds the %v export deadline", livePause, liveBackupTimeout) + } + if backupCopyScratch(source) < 2*source { + t.Fatal("copy scratch must hold raw export and gzip together") + } +} + +func TestEnsureDiskSpaceRejectsShortFilesystem(t *testing.T) { + orig := backupDiskAvail + t.Cleanup(func() { backupDiskAvail = orig }) + needed := BackupArchiveScratch(100, 0) + backupDiskAvail = func(string) (int64, error) { return needed - 1, nil } + if err := EnsureDiskSpace(t.TempDir(), needed); err == nil { + t.Fatal("accepted a filesystem smaller than the export scratch") + } + backupDiskAvail = func(string) (int64, error) { return needed, nil } + if err := EnsureDiskSpace(t.TempDir(), needed); err != nil { + t.Fatal(err) + } +} + +func TestEnsureDiskSpaceSkipsUnknownProbe(t *testing.T) { + orig := backupDiskAvail + t.Cleanup(func() { backupDiskAvail = orig }) + backupDiskAvail = func(string) (int64, error) { return 0, errors.New("unsupported") } + if err := EnsureDiskSpace(t.TempDir(), 1<<40); err != nil { + t.Fatal(err) + } +} + +func TestBackupToCompressedPreflightUsesSourceSize(t *testing.T) { + s := freshStore(t) + if err := s.SaveConfig("goal", "80%"); err != nil { + t.Fatal(err) + } + orig := backupDiskAvail + t.Cleanup(func() { backupDiskAvail = orig }) + backupDiskAvail = func(string) (int64, error) { return 1, nil } + err := s.BackupToCompressed(filepath.Join(t.TempDir(), "full.gz")) + if err == nil { + t.Fatal("published a backup without room for scratch files") + } +} + +func pacedBackupPoints(n int) []HistoryPoint { + points := make([]HistoryPoint, n) + for i := range points { + points[i] = HistoryPoint{TsMs: int64(i + 1), GridW: float64(i)} + } + return points +} diff --git a/go/internal/state/backup_state.go b/go/internal/state/backup_state.go index 7ef5ad0a..85a514b1 100644 --- a/go/internal/state/backup_state.go +++ b/go/internal/state/backup_state.go @@ -5,17 +5,97 @@ import ( "database/sql" "errors" "fmt" + "os" "sort" "strings" "time" ) +const liveBackupTimeout = 2 * time.Hour +const backupScratchHeadroom = 64 << 20 + +var backupDiskAvail = diskAvail + +// OfflineBackup reports whether this store is the read-only helper used by +// ftw-backup. Live Core backups keep their 100 ms yield; the helper does not. +func (s *Store) OfflineBackup() bool { + return s != nil && s.offlineBackup +} + +func (s *Store) backupWorkContext() (context.Context, context.CancelFunc) { + if s.OfflineBackup() { + return context.WithCancel(context.Background()) + } + return context.WithTimeout(context.Background(), liveBackupTimeout) +} + +func (s *Store) backupCopyYield(ctx context.Context) func() error { + if s.OfflineBackup() { + return nil + } + pause := s.backupPause + if pause == nil { + pause = pauseMaintenance + } + return func() error { return pause(ctx) } +} + +// BackupSourceBytes is the on-disk size of state and history files, including +// WAL and SHM. Used to preflight scratch space before a portable export. +func (s *Store) BackupSourceBytes() int64 { + if s == nil { + return 0 + } + var n int64 + for _, p := range []string{s.mainDBPath, s.historyPath} { + n += fileSizeOrZero(p) + fileSizeOrZero(p+"-wal") + fileSizeOrZero(p+"-shm") + } + return n +} + +func fileSizeOrZero(path string) int64 { + if path == "" { + return 0 + } + st, err := os.Stat(path) + if err != nil { + return 0 + } + return st.Size() +} + +// backupCopyScratch is raw export plus gzip of that file, which coexist. +func backupCopyScratch(sourceBytes int64) int64 { + return 2*sourceBytes + backupScratchHeadroom +} + +// BackupArchiveScratch is raw export, gzip, and a later uncompressed +// verification extract on the same filesystem as the published archive. +func BackupArchiveScratch(sourceBytes, extraBytes int64) int64 { + return 3*sourceBytes + extraBytes + backupScratchHeadroom +} + +// EnsureDiskSpace refuses to start a backup when dir cannot hold needed bytes. +// A probe error (including Windows) does not block; the copy still fails if +// the filesystem fills. +func EnsureDiskSpace(dir string, needed int64) error { + if needed <= 0 { + return nil + } + avail, err := backupDiskAvail(dir) + if err != nil { + return nil + } + if avail < needed { + return fmt.Errorf("backup: need %d bytes free in %s for the raw export, compressed archive and verification extract; have %d", needed, dir, avail) + } + return nil +} + // Copy one coherent state snapshot without recopying frozen legacy history. // The selected history database is exported separately. The destination is // temporary until the complete backup has passed verification and fsync. -func (s *Store) copyStateForBackup(path string) error { - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) - defer cancel() +func (s *Store) copyStateForBackup(ctx context.Context, path string) error { src, err := s.db.BeginTx(ctx, nil) if err != nil { return err @@ -59,7 +139,7 @@ func (s *Store) copyStateForBackup(path string) error { if _, err := dst.ExecContext(ctx, item.sqlText); err != nil { return fmt.Errorf("backup create %s: %w", item.name, err) } - if err := copyVerifiedTable(ctx, src, dst, item.name, scanSQLiteBackupTable, func() error { return pauseMaintenance(ctx) }); err != nil { + if err := copyVerifiedTable(ctx, src, dst, item.name, scanSQLiteBackupTable, s.backupCopyYield(ctx)); err != nil { return err } } diff --git a/go/internal/state/history_primary_test.go b/go/internal/state/history_primary_test.go index a86f4ad7..4d95139a 100644 --- a/go/internal/state/history_primary_test.go +++ b/go/internal/state/history_primary_test.go @@ -238,7 +238,7 @@ func TestOfflineBackupIncludesSQLiteCorrectionsAndRestoresBesideOldPrimary(t *te if _, err := backup.db.Exec(`VACUUM INTO '` + dst + `'`); err != nil { t.Fatal(err) } - if err := backup.exportHistoryToSQLite(dst); err != nil { + if err := backup.exportHistoryToSQLite(context.Background(), dst); err != nil { t.Fatal(err) } backup.Close() diff --git a/go/internal/state/history_sqlite.go b/go/internal/state/history_sqlite.go index b0e3dacd..874d6556 100644 --- a/go/internal/state/history_sqlite.go +++ b/go/internal/state/history_sqlite.go @@ -423,12 +423,10 @@ func hashHistoryRows(rows *sql.Rows) (string, int64, error) { return fmt.Sprintf("%x", h.Sum(nil)), n, rows.Err() } -func (s *Store) exportHistoryToSQLite(path string) error { +func (s *Store) exportHistoryToSQLite(ctx context.Context, path string) error { if s.history == nil { return nil } - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) - defer cancel() if err := s.FlushHistory(ctx); err != nil { return err } @@ -460,7 +458,7 @@ func (s *Store) exportHistoryToSQLite(path string) error { if _, err := dest.ExecContext(ctx, `DELETE FROM `+table); err != nil { return err } - if err := copyHistoryTablePaced(ctx, src, dest, table, func() error { return pauseMaintenance(ctx) }); err != nil { + if err := copyHistoryTablePaced(ctx, src, dest, table, s.backupCopyYield(ctx)); err != nil { return err } } diff --git a/go/internal/state/maintenance_io.go b/go/internal/state/maintenance_io.go index 49780907..5c06478f 100644 --- a/go/internal/state/maintenance_io.go +++ b/go/internal/state/maintenance_io.go @@ -10,7 +10,17 @@ import ( // The caller still owns the file and must sync the final tail before publishing. // It must not wrap goal/session writes: those must never wait for this pacing. func NewMaintenanceWriter(ctx context.Context, f *os.File) io.Writer { - return &maintenanceWriter{ctx: ctx, file: f, limit: 1 << 20, pause: pauseMaintenance} + return NewMaintenanceWriterPaced(ctx, f, true) +} + +// NewMaintenanceWriterPaced still syncs each bounded dirty batch. Live Core +// backups then pause so goal writes can catch up; the offline helper does not. +func NewMaintenanceWriterPaced(ctx context.Context, f *os.File, live bool) io.Writer { + pause := func(context.Context) error { return nil } + if live { + pause = pauseMaintenance + } + return &maintenanceWriter{ctx: ctx, file: f, limit: 1 << 20, pause: pause} } type maintenanceWriteSyncer interface { diff --git a/go/internal/state/store.go b/go/internal/state/store.go index 4fa33e3d..9d24b81f 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -82,6 +82,12 @@ type Store struct { seriesHourWG sync.WaitGroup seriesHourMu sync.Mutex seriesHourCancel context.CancelFunc + + // offlineBackup is set by OpenBackupSource. Live Core backups yield + // between copy batches so goal and control writes stay within latency + // limits; the offline helper must not inherit that 100 ms live pause. + offlineBackup bool + backupPause func(context.Context) error } // Open initializes (or creates) the precious state.db at path plus the @@ -247,7 +253,7 @@ func OpenBackupSource(path string) (*Store, error) { db.Close() return nil, err } - s := &Store{db: db, mainDBPath: abs, historyPath: historyDatabasePath(abs)} + s := &Store{db: db, mainDBPath: abs, historyPath: historyDatabasePath(abs), offlineBackup: true} // Offline helpers must export the primary database, never frozen legacy rows. var configTable int if err := db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE name='config'`).Scan(&configTable); err != nil { @@ -610,15 +616,21 @@ func (s *Store) backupToCompressed(dstPath string, report func(BackupProgress), return fmt.Errorf("backup: stat dst %s: %w", dstPath, err) } + ctx, cancel := s.backupWorkContext() + defer cancel() + if err := EnsureDiskSpace(filepath.Dir(dstPath), backupCopyScratch(s.BackupSourceBytes())); err != nil { + return err + } + rawPath := dstPath + ".raw.tmp" _ = os.Remove(rawPath) defer os.Remove(rawPath) reportBackupProgress(report, BackupProgress{Phase: BackupPhaseCopying}) - if err := s.copyStateForBackup(rawPath); err != nil { + if err := s.copyStateForBackup(ctx, rawPath); err != nil { return fmt.Errorf("backup state: %w", err) } - if err := s.exportHistoryToSQLite(rawPath); err != nil { + if err := s.exportHistoryToSQLite(ctx, rawPath); err != nil { return fmt.Errorf("backup history: %w", err) } @@ -653,7 +665,7 @@ func (s *Store) backupToCompressed(dstPath string, report func(BackupProgress), } }() - zw, err := gzip.NewWriterLevel(NewMaintenanceWriter(context.Background(), out), gzip.BestSpeed) + zw, err := gzip.NewWriterLevel(NewMaintenanceWriterPaced(ctx, out, !s.offlineBackup), gzip.BestSpeed) if err != nil { return fmt.Errorf("create gzip writer: %w", err) }