From e48af5dc9cd2031f75dcdb1ac352464aa84f61d0 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 18:27:40 +0000 Subject: [PATCH] store: apply a new database's migrations in one transaction A new database applied each of its 37 migrations in a transaction of its own, and every commit flushes the rollback journal and the database to disk: 156 fsyncs before store.Open returned. On a busy Windows CI runner that took over thirty seconds, so a freshly started daemon could miss the 30s the CLI e2e harness gives it. CI failed with "daemon never became ready" and empty output; a goroutine dump of the daemon at the deadline shows it still in store.Open -> migrate -> tx.Commit -> FlushFileBuffers, and its log says it started 1.5s after the test gave up. Starts of 5 to 20 seconds, all spent before the address was published, were common on the same runners. Pending migrations now go in one transaction: one commit, 8 fsyncs for a new database (counted with strace), none for one already up to date. A failed step now leaves the schema as it was instead of half upgraded. TestOpeningANewDatabaseCommitsOnce counts commits with the driver's commit hook: 38 before this change, 1 after. Co-Authored-By: Claude Opus 5.5 --- internal/store/db.go | 122 +++++++++++++++++++------------ internal/store/migrate_test.go | 130 +++++++++++++++++++++++++++++++++ 2 files changed, 206 insertions(+), 46 deletions(-) create mode 100644 internal/store/migrate_test.go diff --git a/internal/store/db.go b/internal/store/db.go index 9e69aeac..bcdbfd2f 100644 --- a/internal/store/db.go +++ b/internal/store/db.go @@ -29,22 +29,10 @@ type Store struct { // Open creates (if needed) and opens the SQLite database at path, applying // any migrations that haven't run yet. func Open(path string) (*Store, error) { - // _pragma params ensure foreign keys are enforced (SQLite defaults them - // off per-connection) and busy_timeout avoids spurious SQLITE_BUSY - // errors from the watcher/api/p2p goroutines all touching the DB. - dsn := fmt.Sprintf("file:%s?_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)", path) - db, err := sqlx.Open("sqlite", dsn) + db, err := openDB(path) if err != nil { - return nil, fmt.Errorf("open sqlite database: %w", err) + return nil, err } - db.SetMaxOpenConns(1) // modernc.org/sqlite + a single file: avoid concurrent-writer lock contention - - // Tolerate columns the struct doesn't know about. Queries here use - // SELECT *, so without this a database written by a NEWER build hard-fails - // an older one ("missing destination name ") and the app won't start - // at all — a downgrade, or just testing a beta and going back, bricks it. - // Ignoring unmapped columns makes the schema forward-compatible instead. - db = db.Unsafe() s := &Store{db: db} if err := s.migrate(); err != nil { @@ -62,6 +50,26 @@ func Open(path string) (*Store, error) { return s, nil } +// openDB opens the SQLite connection Open builds a Store on. +func openDB(path string) (*sqlx.DB, error) { + // _pragma params ensure foreign keys are enforced (SQLite defaults them + // off per-connection) and busy_timeout avoids spurious SQLITE_BUSY + // errors from the watcher/api/p2p goroutines all touching the DB. + dsn := fmt.Sprintf("file:%s?_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)", path) + db, err := sqlx.Open("sqlite", dsn) + if err != nil { + return nil, fmt.Errorf("open sqlite database: %w", err) + } + db.SetMaxOpenConns(1) // modernc.org/sqlite + a single file: avoid concurrent-writer lock contention + + // Tolerate columns the struct doesn't know about. Queries here use + // SELECT *, so without this a database written by a NEWER build hard-fails + // an older one ("missing destination name ") and the app won't start + // at all — a downgrade, or just testing a beta and going back, bricks it. + // Ignoring unmapped columns makes the schema forward-compatible instead. + return db.Unsafe(), nil +} + // restrictToOwner sets a file to 0600 where mode bits are the access control. // Missing files are fine (the WAL and SHM files exist only while in use). func restrictToOwner(path string) { @@ -78,52 +86,74 @@ func (s *Store) Close() error { return s.db.Close() } -func (s *Store) migrate() error { - if _, err := s.db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (name TEXT PRIMARY KEY)`); err != nil { - return fmt.Errorf("create schema_migrations table: %w", err) - } +// migration is one embedded schema step, applied in name order. +type migration struct { + name string + sql string +} +func (s *Store) migrate() error { entries, err := fs.ReadDir(migrationsFS, "migrations") if err != nil { return fmt.Errorf("read embedded migrations: %w", err) } - var names []string + var migrations []migration for _, e := range entries { - if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") { - names = append(names, e.Name()) - } - } - sort.Strings(names) - - for _, name := range names { - var applied int - if err := s.db.Get(&applied, `SELECT COUNT(*) FROM schema_migrations WHERE name = ?`, name); err != nil { - return fmt.Errorf("check migration %s: %w", name, err) - } - if applied > 0 { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") { continue } - - sqlBytes, err := migrationsFS.ReadFile("migrations/" + name) + sqlBytes, err := migrationsFS.ReadFile("migrations/" + e.Name()) if err != nil { - return fmt.Errorf("read migration %s: %w", name, err) + return fmt.Errorf("read migration %s: %w", e.Name(), err) } + migrations = append(migrations, migration{name: e.Name(), sql: string(sqlBytes)}) + } + sort.Slice(migrations, func(i, j int) bool { return migrations[i].name < migrations[j].name }) + return s.applyMigrations(migrations) +} - tx, err := s.db.Beginx() - if err != nil { - return fmt.Errorf("begin migration tx %s: %w", name, err) - } - if _, err := tx.Exec(string(sqlBytes)); err != nil { - tx.Rollback() - return fmt.Errorf("apply migration %s: %w", name, err) +// applyMigrations applies the migrations not yet recorded, all in one +// transaction. +// +// One, not one per migration: every commit flushes the journal and the +// database to disk, and a new database has every migration to apply. Committed +// one by one that was some 150 flushes before the daemon could start, which a +// busy Windows disk took over thirty seconds to do — past the time the CLI and +// the tests give a daemon to come up. In one transaction it is a single +// commit, and a failed step leaves the schema as it was rather than half +// upgraded. +func (s *Store) applyMigrations(migrations []migration) error { + tx, err := s.db.Beginx() + if err != nil { + return fmt.Errorf("begin migrations: %w", err) + } + defer tx.Rollback() // a no-op once committed + + if _, err := tx.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (name TEXT PRIMARY KEY)`); err != nil { + return fmt.Errorf("create schema_migrations table: %w", err) + } + var names []string + if err := tx.Select(&names, `SELECT name FROM schema_migrations`); err != nil { + return fmt.Errorf("read applied migrations: %w", err) + } + applied := make(map[string]bool, len(names)) + for _, name := range names { + applied[name] = true + } + + for _, m := range migrations { + if applied[m.name] { + continue } - if _, err := tx.Exec(`INSERT INTO schema_migrations (name) VALUES (?)`, name); err != nil { - tx.Rollback() - return fmt.Errorf("record migration %s: %w", name, err) + if _, err := tx.Exec(m.sql); err != nil { + return fmt.Errorf("apply migration %s: %w", m.name, err) } - if err := tx.Commit(); err != nil { - return fmt.Errorf("commit migration %s: %w", name, err) + if _, err := tx.Exec(`INSERT INTO schema_migrations (name) VALUES (?)`, m.name); err != nil { + return fmt.Errorf("record migration %s: %w", m.name, err) } } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit migrations: %w", err) + } return nil } diff --git a/internal/store/migrate_test.go b/internal/store/migrate_test.go new file mode 100644 index 00000000..9fdb93d6 --- /dev/null +++ b/internal/store/migrate_test.go @@ -0,0 +1,130 @@ +package store + +import ( + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + + "modernc.org/sqlite" +) + +// commitCounts counts the commits made on each database file, by the path in +// its connection string. The hook is the driver's and so global: registered +// once, it watches every store any test in this package opens. +var commitCounts = struct { + once sync.Once + mu sync.Mutex + n map[string]*atomic.Int64 +}{n: map[string]*atomic.Int64{}} + +func countCommits(t *testing.T, dbPath string) *atomic.Int64 { + t.Helper() + commitCounts.once.Do(func() { + sqlite.RegisterConnectionHook(func(conn sqlite.ExecQuerierContext, dsn string) error { + hooks, ok := conn.(sqlite.HookRegisterer) + if !ok { + return nil + } + commitCounts.mu.Lock() + defer commitCounts.mu.Unlock() + for path, n := range commitCounts.n { + if strings.Contains(dsn, path) { + hooks.RegisterCommitHook(func() int32 { n.Add(1); return 0 }) + } + } + return nil + }) + }) + n := new(atomic.Int64) + commitCounts.mu.Lock() + commitCounts.n[dbPath] = n + commitCounts.mu.Unlock() + return n +} + +// A new database is set up in one commit, not one per migration. +// +// Every commit flushes the journal and the database to disk, and a new +// database used to take one per migration: 37 of them, some 150 flushes +// before the daemon could start. On a busy Windows disk that took over thirty +// seconds, so a freshly installed daemon could miss the 30 s its clients +// give it to come up — CI saw `daemon never became ready` with the daemon +// still inside store.Open, flushing a migration's commit. +func TestOpeningANewDatabaseCommitsOnce(t *testing.T) { + path := filepath.Join(t.TempDir(), "opensave.db") + commits := countCommits(t, path) + + s, err := Open(path) + if err != nil { + t.Fatalf("Open() error = %v", err) + } + defer s.Close() + + if got := commits.Load(); got != 1 { + t.Errorf("setting up a new database took %d commits, want 1", got) + } + + // Every migration is recorded as applied, so the next open does nothing. + var applied int + if err := s.db.Get(&applied, `SELECT COUNT(*) FROM schema_migrations`); err != nil { + t.Fatal(err) + } + entries, err := migrationsFS.ReadDir("migrations") + if err != nil { + t.Fatal(err) + } + if applied != len(entries) { + t.Errorf("%d migrations recorded, want all %d", applied, len(entries)) + } +} + +// Opening a database that is already up to date writes nothing. +func TestOpeningAnUpToDateDatabaseDoesNotCommit(t *testing.T) { + path := filepath.Join(t.TempDir(), "opensave.db") + s, err := Open(path) + if err != nil { + t.Fatal(err) + } + s.Close() + + commits := countCommits(t, path) + s, err = Open(path) + if err != nil { + t.Fatalf("reopening: %v", err) + } + defer s.Close() + if got := commits.Load(); got != 0 { + t.Errorf("reopening an up-to-date database made %d commits, want 0", got) + } +} + +// A migration that fails leaves the database as it was: none of the ones +// before it in the same run are recorded either, so the next start retries +// the whole step instead of resuming from a half-applied schema. +func TestAFailedMigrationAppliesNothing(t *testing.T) { + path := filepath.Join(t.TempDir(), "opensave.db") + s := &Store{} + db, err := openDB(path) + if err != nil { + t.Fatal(err) + } + s.db = db + defer s.Close() + + err = s.applyMigrations([]migration{ + {name: "0001_ok.sql", sql: `CREATE TABLE fine (id INTEGER)`}, + {name: "0002_broken.sql", sql: `CREATE TABLE broken (`}, + }) + if err == nil || !strings.Contains(err.Error(), "0002_broken.sql") { + t.Fatalf("applyMigrations error = %v, want one naming the broken migration", err) + } + var tables int + if err := s.db.Get(&tables, `SELECT COUNT(*) FROM sqlite_master WHERE name IN ('fine', 'schema_migrations')`); err != nil { + t.Fatal(err) + } + if tables != 0 { + t.Errorf("%d tables left behind by a failed migration run, want none", tables) + } +}