store: apply a new database's migrations in one transaction

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 <col>") 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 <col>") 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
 }

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)
+	}
+}