fix(ledger): survive concurrent ledger opens on a fresh database

internal/dbmig/dbmig.go

@@ -3,6 +3,12 @@
 // Apply works with database/sql so callers own SQLite connection setup. The cr
 // ledger startup path should still enforce its own connection contract,
 // including single-writer behavior or db.SetMaxOpenConns(1).
+//
+// Concurrent callers must open SQLite with immediate transactions. Apply
+// re-reads schema_version inside each migration transaction, and that read is
+// only race-free when BEGIN takes the write lock up front; with a deferred
+// transaction the following DDL upgrades a read lock and SQLite skips the busy
+// handler for that upgrade.
 package dbmig
 
 import (
@@ -75,11 +81,27 @@ func Apply(ctx context.Context, db *sql.DB, migrations []Migration) (Result, err
 		if migration.Version <= current {
 			continue
 		}
-		if err := applyMigration(ctx, db, migration); err != nil {
+		observed, applied, err := applyMigration(ctx, db, migration)
+		if err != nil {
 			return result, err
 		}
-		current = migration.Version
+		// Trust the version read under the write lock, not the planned one: a
+		// concurrent newer binary may have moved the schema past this target.
+		current = observed
 		result.CurrentVersion = current
+		if current > target {
+			return result, fmt.Errorf("%w: database version %d, code version %d", ErrDowngrade, current, target)
+		}
+		// A migration that ran must leave meta exactly where it put it. Any
+		// other value means something moved the schema underneath this run,
+		// including into the band below target where the check above cannot
+		// see it, and later migrations would be skipped without running.
+		if applied && current != migration.Version {
+			return result, fmt.Errorf("%w: schema_version %d after migration %d", ErrDowngrade, current, migration.Version)
+		}
+		if !applied {
+			continue
+		}
 		result.Applied = append(result.Applied, AppliedMigration{
 			Version: migration.Version,
 			Name:    migration.Name,
@@ -239,33 +261,63 @@ func requireMetaColumn(columns map[string]metaColumn, name, columnType string) e
 	return nil
 }
 
-func applyMigration(ctx context.Context, db *sql.DB, migration Migration) error {
+func applyMigration(ctx context.Context, db *sql.DB, migration Migration) (int, bool, error) {
 	tx, err := db.BeginTx(ctx, nil)
 	if err != nil {
-		return fmt.Errorf("dbmig: begin migration %d %q: %w", migration.Version, migration.Name, err)
+		return 0, false, fmt.Errorf("dbmig: begin migration %d %q: %w", migration.Version, migration.Name, err)
 	}
 	defer func() {
 		_ = tx.Rollback()
 	}()
 
+	// Re-read the version inside the transaction. A concurrent process may have
+	// applied this migration between ensureMeta and now; skip it rather than
+	// failing on DDL that already exists. This is only race-free because
+	// callers open SQLite with immediate transactions (see the ledger DSN), so
+	// the read happens under the write lock.
+	var current int
+	if err := tx.QueryRowContext(ctx, "SELECT schema_version FROM meta").Scan(&current); err != nil {
+		return 0, false, fmt.Errorf("%w: reading schema_version before migration %d: %w", ErrInvalidMeta, migration.Version, err)
+	}
+	if current >= migration.Version {
+		return current, false, nil
+	}
+	if current != migration.Version-1 {
+		return current, false, fmt.Errorf("%w: schema_version %d before migration %d", ErrInvalidMeta, current, migration.Version)
+	}
+
 	if err := migration.Up(ctx, tx); err != nil {
-		return fmt.Errorf("dbmig: apply migration %d %q: %w", migration.Version, migration.Name, err)
+		return current, false, fmt.Errorf("dbmig: apply migration %d %q: %w", migration.Version, migration.Name, err)
 	}
 
 	result, err := tx.ExecContext(ctx, "UPDATE meta SET schema_version = ? WHERE rowid = (SELECT MIN(rowid) FROM meta)", migration.Version)
 	if err != nil {
-		return fmt.Errorf("%w: updating schema_version for migration %d: %w", ErrInvalidMeta, migration.Version, err)
+		return current, false, fmt.Errorf("%w: updating schema_version for migration %d: %w", ErrInvalidMeta, migration.Version, err)
 	}
 	rowsAffected, err := result.RowsAffected()
 	if err != nil {
-		return fmt.Errorf("dbmig: checking schema_version update for migration %d: %w", migration.Version, err)
+		return current, false, fmt.Errorf("dbmig: checking schema_version update for migration %d: %w", migration.Version, err)
 	}
 	if rowsAffected != 1 {
-		return fmt.Errorf("%w: schema_version update affected %d rows", ErrInvalidMeta, rowsAffected)
+		return current, false, fmt.Errorf("%w: schema_version update affected %d rows", ErrInvalidMeta, rowsAffected)
+	}
+
+	// Read the stored value back rather than trusting the planned version: a
+	// trigger or a concurrent writer can leave meta somewhere other than where
+	// this migration put it, in either direction.
+	var stored int
+	if err := tx.QueryRowContext(ctx, "SELECT schema_version FROM meta").Scan(&stored); err != nil {
+		return current, false, fmt.Errorf("%w: reading schema_version after migration %d: %w", ErrInvalidMeta, migration.Version, err)
+	}
+	// A value below what this migration wrote means meta was tampered with
+	// inside this transaction; a value above it means a newer binary moved the
+	// schema forward, which Apply reports as a downgrade once it sees it.
+	if stored < migration.Version {
+		return current, false, fmt.Errorf("%w: schema_version %d after migration %d", ErrInvalidMeta, stored, migration.Version)
 	}
 
 	if err := tx.Commit(); err != nil {
-		return fmt.Errorf("dbmig: commit migration %d %q: %w", migration.Version, migration.Name, err)
+		return current, false, fmt.Errorf("dbmig: commit migration %d %q: %w", migration.Version, migration.Name, err)
 	}
-	return nil
+	return stored, true, nil
 }

internal/dbmig/dbmig_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"fmt"
 	"path/filepath"
 	"reflect"
 	"testing"
@@ -494,3 +495,95 @@ func execSQL(t *testing.T, db *sql.DB, statement string, args ...any) {
 		t.Fatalf("exec %q: %v", statement, err)
 	}
 }
+
+func TestApplyMigrationReportsVersionObservedUnderWriteLock(t *testing.T) {
+	ctx := context.Background()
+	db := openTestDB(t)
+	applied := 0
+	second := countedMigration(2, "create reviews", &applied, "CREATE TABLE reviews (id INTEGER PRIMARY KEY)")
+	migrations := []Migration{
+		countedMigration(1, "create widgets", &applied, "CREATE TABLE widgets (id INTEGER PRIMARY KEY)"),
+		second,
+	}
+	if _, err := Apply(ctx, db, migrations); err != nil {
+		t.Fatalf("seed Apply: %v", err)
+	}
+	if _, err := db.ExecContext(ctx, "UPDATE meta SET schema_version = 7"); err != nil {
+		t.Fatalf("advance schema_version: %v", err)
+	}
+
+	// A concurrent newer binary moved the schema past this migration. The
+	// version observed under the write lock must be reported so Apply can
+	// refuse the downgrade instead of continuing on a schema it does not know.
+	observed, didApply, err := applyMigration(ctx, db, second)
+	if err != nil {
+		t.Fatalf("applyMigration: %v", err)
+	}
+	if didApply {
+		t.Fatal("applyMigration reapplied a migration the database already has")
+	}
+	if observed != 7 {
+		t.Fatalf("observed version = %d, want 7", observed)
+	}
+}
+
+func TestApplyRefusesDowngradeDiscoveredMidRun(t *testing.T) {
+	// Stand in for a newer binary that wins the race and advances the schema
+	// past this run's target. The trigger fires on the schema_version write
+	// that ends its migration, so the version moves underneath this run. It
+	// must be refused whether that happens on an early migration or on the
+	// last one in the plan.
+	advanceOn := func(version int, name, ddl string, to int) Migration {
+		return Migration{
+			Version: version,
+			Name:    name,
+			Up: func(ctx context.Context, tx *sql.Tx) error {
+				if _, err := tx.ExecContext(ctx, ddl); err != nil {
+					return err
+				}
+				_, err := tx.ExecContext(ctx, fmt.Sprintf("CREATE TRIGGER advance AFTER UPDATE ON meta BEGIN UPDATE meta SET schema_version = %d; END", to))
+				return err
+			},
+		}
+	}
+
+	for _, tt := range []struct {
+		name    string
+		advance int
+		to      int
+		wantErr error
+	}{
+		{name: "first migration advances", advance: 1, to: 9, wantErr: ErrDowngrade},
+		{name: "last migration advances", advance: 2, to: 9, wantErr: ErrDowngrade},
+		{name: "first migration lowers", advance: 1, to: 0, wantErr: ErrInvalidMeta},
+		{name: "last migration lowers", advance: 2, to: 0, wantErr: ErrInvalidMeta},
+		{name: "advances only to the target", advance: 1, to: 2, wantErr: ErrDowngrade},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+			db := openTestDB(t)
+			applied := 0
+
+			migrations := make([]Migration, 0, 2)
+			for _, m := range []struct {
+				version int
+				name    string
+				ddl     string
+			}{
+				{1, "create widgets", "CREATE TABLE widgets (id INTEGER PRIMARY KEY)"},
+				{2, "create reviews", "CREATE TABLE reviews (id INTEGER PRIMARY KEY)"},
+			} {
+				if m.version == tt.advance {
+					migrations = append(migrations, advanceOn(m.version, m.name, m.ddl, tt.to))
+					continue
+				}
+				migrations = append(migrations, countedMigration(m.version, m.name, &applied, m.ddl))
+			}
+
+			_, err := Apply(ctx, db, migrations)
+			if !errors.Is(err, tt.wantErr) {
+				t.Fatalf("Apply error = %v, want %v", err, tt.wantErr)
+			}
+		})
+	}
+}

internal/ledger/ledger.go

@@ -7,8 +7,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net/url"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -29,6 +31,7 @@ const (
 	// DefaultBusyTimeout is the SQLite busy timeout configured at open.
 	DefaultBusyTimeout = 5 * time.Second
 	writeQueueSize     = 64
+	walRetryInterval   = 10 * time.Millisecond
 )
 
 var (
@@ -413,7 +416,11 @@ func Open(ctx context.Context, path string) (*Store, error) {
 		return nil, fmt.Errorf("ledger: create db parent: %w", err)
 	}
 
-	db, err := sql.Open("sqlite", path)
+	dsn, err := sqliteDataSourceName(path)
+	if err != nil {
+		return nil, err
+	}
+	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
 		return nil, fmt.Errorf("ledger: open sqlite: %w", err)
 	}
@@ -496,19 +503,109 @@ func (s *Store) checkOpen() error {
 	return nil
 }
 
+// sqliteDataSourceName builds the DSN used to open path.
+//
+// busy_timeout and foreign_keys are per-connection settings, so they are carried
+// in the DSN and applied by the driver to every connection it dials. Applying
+// them once through configureSQLite is not enough: database/sql discards a
+// connection that returns driver.ErrBadConn and replaces it with one that never
+// sees configureSQLite.
+//
+// The DSN is a SQLite "file:" URI rather than a bare path because the driver
+// splits the query string at the first '?', and a filesystem path may itself
+// contain '?', '#', or '%'. The absolute path is percent-encoded as a URI path.
+// The localhost authority keeps the URI well formed: SQLite accepts only an
+// empty or "localhost" authority, and without one url.URL would render a
+// Windows drive path such as "C:/data" as an authority that SQLite rejects.
+func sqliteDataSourceName(path string) (string, error) {
+	absolute, err := filepath.Abs(path)
+	if err != nil {
+		return "", fmt.Errorf("ledger: resolve db path: %w", err)
+	}
+
+	query := url.Values{}
+	query.Add("_pragma", "busy_timeout="+strconv.FormatInt(DefaultBusyTimeout.Milliseconds(), 10))
+	query.Add("_pragma", "foreign_keys=ON")
+	// Immediate transactions take the write lock before reading, so concurrent
+	// writers wait on busy_timeout instead of failing instantly with
+	// SQLITE_BUSY_SNAPSHOT. Startup migrations rely on that to serialize
+	// against another process opening the same fresh ledger.
+	// Every BeginTx on this handle is therefore a write transaction: do not add
+	// a read-only BeginTx here, it would serialize behind the writer. Reads run
+	// as autocommit queries instead. dbmig.Apply also depends on this — its
+	// in-transaction schema_version re-read is only race-free under the write
+	// lock — so this setting is a cross-package contract, pinned by
+	// TestSQLiteDataSourceName.
+	query.Set("_txlock", "immediate")
+
+	uri := url.URL{
+		Scheme:   "file",
+		Host:     "localhost",
+		Path:     filepath.ToSlash(absolute),
+		RawQuery: query.Encode(),
+	}
+	return uri.String(), nil
+}
+
+// configureSQLite applies the database-wide pragmas that only need setting once.
+// Per-connection pragmas live in the DSN instead (see sqliteDataSourceName).
 func configureSQLite(ctx context.Context, db *sql.DB) error {
-	if _, err := db.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil {
-		return fmt.Errorf("ledger: enable foreign keys: %w", err)
-	}
-	if _, err := db.ExecContext(ctx, "PRAGMA journal_mode = WAL"); err != nil {
+	// journal_mode is persisted in the database header, not per connection, so it
+	// is set once here. Converting a fresh database to WAL upgrades a read
+	// transaction to a write transaction, and SQLite skips the busy handler for
+	// that upgrade, so a concurrent holder returns SQLITE_BUSY immediately; retry
+	// instead of relying on busy_timeout.
+	mode, err := retryWhileBusy(ctx, walRetryInterval, DefaultBusyTimeout, func(ctx context.Context) (string, error) {
+		return setWALJournalMode(ctx, db)
+	})
+	if err != nil {
 		return fmt.Errorf("ledger: enable WAL: %w", err)
 	}
-	if _, err := db.ExecContext(ctx, fmt.Sprintf("PRAGMA busy_timeout = %d", DefaultBusyTimeout.Milliseconds())); err != nil {
-		return fmt.Errorf("ledger: set busy timeout: %w", err)
+	if mode != "wal" {
+		return fmt.Errorf("ledger: enable WAL: journal_mode = %q, want wal", mode)
 	}
 	return nil
 }
 
+// retryWhileBusy repeats attempt until it succeeds, returns a non-busy error,
+// or budget elapses.
+func retryWhileBusy(ctx context.Context, interval, budget time.Duration, attempt func(context.Context) (string, error)) (string, error) {
+	deadline := time.Now().Add(budget)
+	for {
+		value, err := attempt(ctx)
+		if err == nil {
+			return value, nil
+		}
+		if !isSQLiteBusyError(err) || !time.Now().Before(deadline) {
+			return "", err
+		}
+		select {
+		case <-ctx.Done():
+			return "", ctx.Err()
+		case <-time.After(interval):
+		}
+	}
+}
+
+// setWALJournalMode asks SQLite for WAL mode and returns the resulting mode.
+func setWALJournalMode(ctx context.Context, db *sql.DB) (string, error) {
+	var mode string
+	if err := db.QueryRowContext(ctx, "PRAGMA journal_mode = WAL").Scan(&mode); err != nil {
+		return "", err
+	}
+	return strings.ToLower(mode), nil
+}
+
+// isSQLiteBusyError reports whether err is lock contention reported by SQLite.
+func isSQLiteBusyError(err error) bool {
+	var sqliteErr *sqlite.Error
+	if !errors.As(err, &sqliteErr) {
+		return false
+	}
+	// The driver enables extended result codes, so mask to the primary code.
+	return sqliteErr.Code()&0xff == sqlite3.SQLITE_BUSY
+}
+
 func migrations() []dbmig.Migration {
 	return []dbmig.Migration{
 		{