From 9e66b539a7e538482e4ad09bc8fcec356ff07a46 Mon Sep 17 00:00:00 2001
From: Aaron Wong <6979793+zzwong@users.noreply.github.com>
Date: Sat, 12 Sep 2026 18:11:44 -0500
Subject: [PATCH 1/6] fix(ledger): survive concurrent ledger opens on a fresh
 database

Starting several reviews at once failed during runtime construction with
"ledger: enable WAL: database is locked". Two independent causes had to be
fixed before concurrent opens survive.

Per-connection pragmas moved into the DSN. busy_timeout and foreign_keys were
applied once after opening, but database/sql discards a connection returning
driver.ErrBadConn and dials a replacement that never sees that setup, so the
settings were not guaranteed on the connection actually in use. The DSN is a
file: URI so that a path containing a query or fragment character cannot
corrupt the parameters.

WAL conversion now retries. Reordering the pragmas is not sufficient: the
conversion begins a read transaction and upgrades it to a write transaction,
and SQLite consults the busy handler only while no transaction is open, so a
concurrent holder returns SQLITE_BUSY immediately no matter how large
busy_timeout is. The conversion is retried within the same budget instead.

Migrations re-check the schema version inside their own transaction. Apply read
the version before opening each migration transaction, so two processes racing
a fresh ledger could both plan migration 1 and the loser failed on DDL that
already existed. The read now happens under the write lock, which immediate
transactions guarantee, and an already-applied migration is skipped.
---
 internal/dbmig/dbmig.go        | 39 ++++++++++----
 internal/ledger/ledger.go      | 99 +++++++++++++++++++++++++++++++---
 internal/ledger/ledger_test.go | 83 ++++++++++++++++++++++++++++
 3 files changed, 204 insertions(+), 17 deletions(-)

diff --git a/internal/dbmig/dbmig.go b/internal/dbmig/dbmig.go
index a989c0ad..18ca3773 100644
--- a/internal/dbmig/dbmig.go
+++ b/internal/dbmig/dbmig.go
@@ -75,11 +75,16 @@ 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 {
+		applied, err := applyMigration(ctx, db, migration)
+		if err != nil {
 			return result, err
 		}
 		current = migration.Version
 		result.CurrentVersion = current
+		if !applied {
+			// Another process applied this migration after Apply planned the run.
+			continue
+		}
 		result.Applied = append(result.Applied, AppliedMigration{
 			Version: migration.Version,
 			Name:    migration.Name,
@@ -239,33 +244,49 @@ 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) (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 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 false, fmt.Errorf("%w: reading schema_version before migration %d: %w", ErrInvalidMeta, migration.Version, err)
+	}
+	if current >= migration.Version {
+		return false, nil
+	}
+	if current != migration.Version-1 {
+		return 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 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 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 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 false, fmt.Errorf("%w: schema_version update affected %d rows", ErrInvalidMeta, rowsAffected)
 	}
 
 	if err := tx.Commit(); err != nil {
-		return fmt.Errorf("dbmig: commit migration %d %q: %w", migration.Version, migration.Name, err)
+		return false, fmt.Errorf("dbmig: commit migration %d %q: %w", migration.Version, migration.Name, err)
 	}
-	return nil
+	return true, nil
 }
diff --git a/internal/ledger/ledger.go b/internal/ledger/ledger.go
index 2916cc75..f742a1bd 100644
--- a/internal/ledger/ledger.go
+++ b/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,17 +503,93 @@ 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.
+	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)
+	// 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.
+	deadline := time.Now().Add(DefaultBusyTimeout)
+	for {
+		mode, err := setWALJournalMode(ctx, db)
+		if err == nil {
+			if mode != "wal" {
+				return fmt.Errorf("ledger: enable WAL: journal_mode = %q, want wal", mode)
+			}
+			return nil
+		}
+		if !isSQLiteBusyError(err) || !time.Now().Before(deadline) {
+			return fmt.Errorf("ledger: enable WAL: %w", err)
+		}
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("ledger: enable WAL: %w", ctx.Err())
+		case <-time.After(walRetryInterval):
+		}
 	}
-	if _, err := db.ExecContext(ctx, "PRAGMA journal_mode = WAL"); err != nil {
-		return fmt.Errorf("ledger: enable WAL: %w", err)
+}
+
+// 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
 	}
-	if _, err := db.ExecContext(ctx, fmt.Sprintf("PRAGMA busy_timeout = %d", DefaultBusyTimeout.Milliseconds())); err != nil {
-		return fmt.Errorf("ledger: set busy timeout: %w", err)
+	switch sqliteErr.Code() & 0xff {
+	case sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED:
+		return true
+	default:
+		return false
 	}
-	return nil
 }
 
 func migrations() []dbmig.Migration {
diff --git a/internal/ledger/ledger_test.go b/internal/ledger/ledger_test.go
index 18243ea2..974bfe52 100644
--- a/internal/ledger/ledger_test.go
+++ b/internal/ledger/ledger_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"fmt"
 	"path/filepath"
 	"reflect"
 	"slices"
@@ -57,6 +58,88 @@ func TestOpenMigratesFreshDatabaseAndAppliesStartupContract(t *testing.T) {
 	}
 }
 
+func TestOpenAppliesPerConnectionPragmasFromDSN(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "ledger.db")
+	dsn, err := sqliteDataSourceName(path)
+	if err != nil {
+		t.Fatalf("sqliteDataSourceName: %v", err)
+	}
+
+	// A connection dialled from the DSN alone must already carry the
+	// per-connection pragmas. This is what protects connections that
+	// database/sql dials after discarding one, which never run
+	// configureSQLite.
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		t.Fatalf("sql.Open(%q): %v", dsn, err)
+	}
+	t.Cleanup(func() {
+		if err := db.Close(); err != nil {
+			t.Fatalf("close DSN connection: %v", err)
+		}
+	})
+	assertSQLitePragmas(t, db)
+
+	// Open must expose the same values on the connection it uses.
+	store := openStoreAt(t, path)
+	assertSQLitePragmas(t, store.db)
+}
+
+func TestOpenConcurrentOnSamePathSucceeds(t *testing.T) {
+	const openers = 8
+
+	path := filepath.Join(t.TempDir(), "ledger.db")
+	start := make(chan struct{})
+	errs := make([]error, openers)
+	stores := make([]*Store, openers)
+
+	var wg sync.WaitGroup
+	for i := range openers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			stores[i], errs[i] = Open(context.Background(), path)
+		}()
+	}
+
+	// Release every opener at once so they genuinely race for the WAL lock.
+	close(start)
+	wg.Wait()
+
+	t.Cleanup(func() {
+		for _, store := range stores {
+			if store == nil {
+				continue
+			}
+			if err := store.Close(); err != nil {
+				t.Errorf("Close: %v", err)
+			}
+		}
+	})
+
+	failures := make([]error, 0, openers)
+	for i, err := range errs {
+		if err != nil {
+			failures = append(failures, fmt.Errorf("opener %d: %w", i, err))
+		}
+	}
+	if len(failures) > 0 {
+		t.Fatalf("%d/%d concurrent Open calls failed:\n%v", len(failures), openers, errors.Join(failures...))
+	}
+}
+
+func assertSQLitePragmas(t *testing.T, db *sql.DB) {
+	t.Helper()
+
+	if got := queryInt(t, db, "PRAGMA busy_timeout"); int64(got) != DefaultBusyTimeout.Milliseconds() {
+		t.Fatalf("PRAGMA busy_timeout = %d, want %d", got, DefaultBusyTimeout.Milliseconds())
+	}
+	if got := queryInt(t, db, "PRAGMA foreign_keys"); got != 1 {
+		t.Fatalf("PRAGMA foreign_keys = %d, want 1", got)
+	}
+}
+
 func TestReviewerCohortReplaceAndSessionUpdateAreAtomic(t *testing.T) {
 	store := openStore(t)
 	ctx := context.Background()

From 4b040a9a28f16b4bec83af6cfa9891da61c6c0d5 Mon Sep 17 00:00:00 2001
From: Aaron Wong <6979793+zzwong@users.noreply.github.com>
Date: Sat, 12 Sep 2026 18:39:59 -0500
Subject: [PATCH 2/6] fix(dbmig): refuse a downgrade discovered after the
 migration run starts

Apply checked for a newer schema once, before the loop, using the version read
outside any transaction. The in-transaction re-check then discarded the version
it observed and advanced CurrentVersion from the plan instead, so an older
binary that lost the race to a newer one skipped every migration, returned no
error, and went on to operate on a schema it does not know. That is the state
ErrDowngrade exists to refuse, and before the re-check existed it failed loudly
on duplicate DDL.

applyMigration now reports the version it read under the write lock, Apply
trusts that instead of the planned version, and a mid-run downgrade is refused.
The busy-error check drops its SQLITE_LOCKED arm, which the driver already
handles internally and which a retry cannot resolve, and the dbmig package doc
records that concurrent callers must open with immediate transactions.

The concurrency test now asserts every racing open lands on a fully migrated
database in WAL mode rather than merely avoiding an error.
---
 internal/dbmig/dbmig.go        | 38 ++++++++++++--------
 internal/dbmig/dbmig_test.go   | 65 ++++++++++++++++++++++++++++++++++
 internal/ledger/ledger.go      |  8 ++---
 internal/ledger/ledger_test.go | 21 +++++++++++
 4 files changed, 112 insertions(+), 20 deletions(-)

diff --git a/internal/dbmig/dbmig.go b/internal/dbmig/dbmig.go
index 18ca3773..0ae68b64 100644
--- a/internal/dbmig/dbmig.go
+++ b/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,14 +81,18 @@ func Apply(ctx context.Context, db *sql.DB, migrations []Migration) (Result, err
 		if migration.Version <= current {
 			continue
 		}
-		applied, err := applyMigration(ctx, db, migration)
+		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)
+		}
 		if !applied {
-			// Another process applied this migration after Apply planned the run.
 			continue
 		}
 		result.Applied = append(result.Applied, AppliedMigration{
@@ -244,10 +254,10 @@ func requireMetaColumn(columns map[string]metaColumn, name, columnType string) e
 	return nil
 }
 
-func applyMigration(ctx context.Context, db *sql.DB, migration Migration) (bool, error) {
+func applyMigration(ctx context.Context, db *sql.DB, migration Migration) (int, bool, error) {
 	tx, err := db.BeginTx(ctx, nil)
 	if err != nil {
-		return false, 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()
@@ -260,33 +270,33 @@ func applyMigration(ctx context.Context, db *sql.DB, migration Migration) (bool,
 	// 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 false, fmt.Errorf("%w: reading schema_version before migration %d: %w", ErrInvalidMeta, migration.Version, err)
+		return 0, false, fmt.Errorf("%w: reading schema_version before migration %d: %w", ErrInvalidMeta, migration.Version, err)
 	}
 	if current >= migration.Version {
-		return false, nil
+		return current, false, nil
 	}
 	if current != migration.Version-1 {
-		return false, fmt.Errorf("%w: schema_version %d before migration %d", ErrInvalidMeta, current, migration.Version)
+		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 false, 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 false, 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 false, 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 false, 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)
 	}
 
 	if err := tx.Commit(); err != nil {
-		return false, 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 true, nil
+	return migration.Version, true, nil
 }
diff --git a/internal/dbmig/dbmig_test.go b/internal/dbmig/dbmig_test.go
index 8440b3b6..30f5c90b 100644
--- a/internal/dbmig/dbmig_test.go
+++ b/internal/dbmig/dbmig_test.go
@@ -494,3 +494,68 @@ 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) {
+	ctx := context.Background()
+	db := openTestDB(t)
+	applied := 0
+
+	// Stand in for a newer binary that wins the race and advances the schema
+	// past this run's target while this run is between migrations. The trigger
+	// fires on the schema_version write that ends migration 1, so migration 2
+	// re-reads a version this code does not know.
+	advance := Migration{
+		Version: 1,
+		Name:    "create widgets",
+		Up: func(ctx context.Context, tx *sql.Tx) error {
+			if _, err := tx.ExecContext(ctx, "CREATE TABLE widgets (id INTEGER PRIMARY KEY)"); err != nil {
+				return err
+			}
+			_, err := tx.ExecContext(ctx, "CREATE TRIGGER advance AFTER UPDATE ON meta BEGIN UPDATE meta SET schema_version = 9; END")
+			return err
+		},
+	}
+	migrations := []Migration{
+		advance,
+		countedMigration(2, "create reviews", &applied, "CREATE TABLE reviews (id INTEGER PRIMARY KEY)"),
+	}
+
+	_, err := Apply(ctx, db, migrations)
+	if !errors.Is(err, ErrDowngrade) {
+		t.Fatalf("Apply error = %v, want ErrDowngrade", err)
+	}
+	if applied != 0 {
+		t.Fatalf("migration 2 ran %d times against a newer schema, want 0", applied)
+	}
+}
diff --git a/internal/ledger/ledger.go b/internal/ledger/ledger.go
index f742a1bd..d84404e5 100644
--- a/internal/ledger/ledger.go
+++ b/internal/ledger/ledger.go
@@ -584,12 +584,8 @@ func isSQLiteBusyError(err error) bool {
 	if !errors.As(err, &sqliteErr) {
 		return false
 	}
-	switch sqliteErr.Code() & 0xff {
-	case sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED:
-		return true
-	default:
-		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 {
diff --git a/internal/ledger/ledger_test.go b/internal/ledger/ledger_test.go
index 974bfe52..4410b706 100644
--- a/internal/ledger/ledger_test.go
+++ b/internal/ledger/ledger_test.go
@@ -9,6 +9,7 @@ import (
 	"reflect"
 	"slices"
 	"strconv"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -127,6 +128,26 @@ func TestOpenConcurrentOnSamePathSucceeds(t *testing.T) {
 	if len(failures) > 0 {
 		t.Fatalf("%d/%d concurrent Open calls failed:\n%v", len(failures), openers, errors.Join(failures...))
 	}
+
+	// A racing open must not merely avoid erroring; it must land on a fully
+	// migrated database in WAL mode.
+	wantVersion := len(migrations())
+	for i, store := range stores {
+		var version int
+		if err := store.db.QueryRowContext(context.Background(), "SELECT schema_version FROM meta").Scan(&version); err != nil {
+			t.Fatalf("opener %d: read schema_version: %v", i, err)
+		}
+		if version != wantVersion {
+			t.Fatalf("opener %d: schema_version = %d, want %d", i, version, wantVersion)
+		}
+		var mode string
+		if err := store.db.QueryRowContext(context.Background(), "PRAGMA journal_mode").Scan(&mode); err != nil {
+			t.Fatalf("opener %d: read journal_mode: %v", i, err)
+		}
+		if strings.ToLower(mode) != "wal" {
+			t.Fatalf("opener %d: journal_mode = %q, want wal", i, mode)
+		}
+	}
 }
 
 func assertSQLitePragmas(t *testing.T, db *sql.DB) {

From bd9dbea167167bad13180ab71f9758f3f26d1271 Mon Sep 17 00:00:00 2001
From: Aaron Wong <6979793+zzwong@users.noreply.github.com>
Date: Sat, 12 Sep 2026 18:57:58 -0500
Subject: [PATCH 3/6] style(ledger): use the US spelling of dialed in a test
 comment

---
 internal/ledger/ledger_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/internal/ledger/ledger_test.go b/internal/ledger/ledger_test.go
index 4410b706..346974e3 100644
--- a/internal/ledger/ledger_test.go
+++ b/internal/ledger/ledger_test.go
@@ -66,7 +66,7 @@ func TestOpenAppliesPerConnectionPragmasFromDSN(t *testing.T) {
 		t.Fatalf("sqliteDataSourceName: %v", err)
 	}
 
-	// A connection dialled from the DSN alone must already carry the
+	// A connection dialed from the DSN alone must already carry the
 	// per-connection pragmas. This is what protects connections that
 	// database/sql dials after discarding one, which never run
 	// configureSQLite.

From a15d3c73f7d9db500872863acc3282c8b47aa712 Mon Sep 17 00:00:00 2001
From: Aaron Wong <6979793+zzwong@users.noreply.github.com>
Date: Sat, 12 Sep 2026 19:15:09 -0500
Subject: [PATCH 4/6] fix(dbmig): report the stored schema version and pin the
 DSN contract

The downgrade guard only saw versions that moved before a migration's DDL,
because applyMigration returned the planned version on success rather than the
value the transaction actually left behind. A version that moved during the
last migration in a plan was therefore invisible: the loop ended, the guard
never fired, and Apply returned no error on a newer schema. It now reads the
stored value back inside the same transaction, and the downgrade test covers an
advance on the last migration as well as an early one.

The DSN's encoding contract was untested. Every input was a generated temp
path, so replacing the URI construction with plain concatenation passed the
whole suite while breaking any data root containing a query or fragment
character. A table test now opens real databases through paths containing a
space, ?, #, % and &=, and asserts the pragmas survive the encoding.

_txlock=immediate is a cross-package precondition for dbmig's in-transaction
re-read, guarded until now only by a scheduling-dependent test. It is asserted
directly from the DSN, and the invariant that every BeginTx on this handle is a
write transaction is recorded next to the setting.
---
 internal/dbmig/dbmig.go        | 10 +++-
 internal/dbmig/dbmig_test.go   | 74 ++++++++++++++++++----------
 internal/ledger/ledger.go      |  6 +++
 internal/ledger/ledger_test.go | 89 ++++++++++++++++++++++++++++++++++
 4 files changed, 152 insertions(+), 27 deletions(-)

diff --git a/internal/dbmig/dbmig.go b/internal/dbmig/dbmig.go
index 0ae68b64..f6cdbd62 100644
--- a/internal/dbmig/dbmig.go
+++ b/internal/dbmig/dbmig.go
@@ -295,8 +295,16 @@ func applyMigration(ctx context.Context, db *sql.DB, migration Migration) (int,
 		return current, false, fmt.Errorf("%w: schema_version update affected %d rows", ErrInvalidMeta, rowsAffected)
 	}
 
+	// Read the stored value back rather than returning the planned version: a
+	// trigger or a concurrent writer can leave meta somewhere other than where
+	// this migration put it, and Apply's downgrade check needs the truth.
+	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)
+	}
+
 	if err := tx.Commit(); err != nil {
 		return current, false, fmt.Errorf("dbmig: commit migration %d %q: %w", migration.Version, migration.Name, err)
 	}
-	return migration.Version, true, nil
+	return stored, true, nil
 }
diff --git a/internal/dbmig/dbmig_test.go b/internal/dbmig/dbmig_test.go
index 30f5c90b..2544edfb 100644
--- a/internal/dbmig/dbmig_test.go
+++ b/internal/dbmig/dbmig_test.go
@@ -527,35 +527,57 @@ func TestApplyMigrationReportsVersionObservedUnderWriteLock(t *testing.T) {
 }
 
 func TestApplyRefusesDowngradeDiscoveredMidRun(t *testing.T) {
-	ctx := context.Background()
-	db := openTestDB(t)
-	applied := 0
-
 	// Stand in for a newer binary that wins the race and advances the schema
-	// past this run's target while this run is between migrations. The trigger
-	// fires on the schema_version write that ends migration 1, so migration 2
-	// re-reads a version this code does not know.
-	advance := Migration{
-		Version: 1,
-		Name:    "create widgets",
-		Up: func(ctx context.Context, tx *sql.Tx) error {
-			if _, err := tx.ExecContext(ctx, "CREATE TABLE widgets (id INTEGER PRIMARY KEY)"); err != nil {
+	// 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) 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, "CREATE TRIGGER advance AFTER UPDATE ON meta BEGIN UPDATE meta SET schema_version = 9; END")
 				return err
-			}
-			_, err := tx.ExecContext(ctx, "CREATE TRIGGER advance AFTER UPDATE ON meta BEGIN UPDATE meta SET schema_version = 9; END")
-			return err
-		},
-	}
-	migrations := []Migration{
-		advance,
-		countedMigration(2, "create reviews", &applied, "CREATE TABLE reviews (id INTEGER PRIMARY KEY)"),
+			},
+		}
 	}
 
-	_, err := Apply(ctx, db, migrations)
-	if !errors.Is(err, ErrDowngrade) {
-		t.Fatalf("Apply error = %v, want ErrDowngrade", err)
-	}
-	if applied != 0 {
-		t.Fatalf("migration 2 ran %d times against a newer schema, want 0", applied)
+	for _, tt := range []struct {
+		name    string
+		advance int
+	}{
+		{name: "first migration", advance: 1},
+		{name: "last migration", advance: 2},
+	} {
+		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))
+					continue
+				}
+				migrations = append(migrations, countedMigration(m.version, m.name, &applied, m.ddl))
+			}
+
+			_, err := Apply(ctx, db, migrations)
+			if !errors.Is(err, ErrDowngrade) {
+				t.Fatalf("Apply error = %v, want ErrDowngrade", err)
+			}
+		})
 	}
 }
diff --git a/internal/ledger/ledger.go b/internal/ledger/ledger.go
index d84404e5..4c159162 100644
--- a/internal/ledger/ledger.go
+++ b/internal/ledger/ledger.go
@@ -530,6 +530,12 @@ func sqliteDataSourceName(path string) (string, error) {
 	// 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{
diff --git a/internal/ledger/ledger_test.go b/internal/ledger/ledger_test.go
index 346974e3..e545bddf 100644
--- a/internal/ledger/ledger_test.go
+++ b/internal/ledger/ledger_test.go
@@ -5,6 +5,8 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"net/url"
+	"os"
 	"path/filepath"
 	"reflect"
 	"slices"
@@ -1996,3 +1998,90 @@ func indexColumns(t *testing.T, db *sql.DB, name string) []string {
 func strPtr(value string) *string {
 	return &value
 }
+
+func TestSQLiteDataSourceName(t *testing.T) {
+	// The helper builds a file: URI rather than concatenating the path, because
+	// the driver splits the query string at the first '?'. A data root
+	// containing any of these characters would otherwise produce a DSN whose
+	// parameters are truncated or whose path is wrong.
+	for _, name := range []string{
+		"ledger.db",
+		"with space.db",
+		"question?.db",
+		"hash#.db",
+		"percent%41.db",
+		"amp&eq=.db",
+	} {
+		t.Run(name, func(t *testing.T) {
+			path := filepath.Join(t.TempDir(), name)
+			dsn, err := sqliteDataSourceName(path)
+			if err != nil {
+				t.Fatalf("sqliteDataSourceName(%q): %v", path, err)
+			}
+			if !strings.HasPrefix(dsn, "file://") {
+				t.Fatalf("dsn = %q, want a file: URI", dsn)
+			}
+
+			db, err := sql.Open("sqlite", dsn)
+			if err != nil {
+				t.Fatalf("sql.Open(%q): %v", dsn, err)
+			}
+			t.Cleanup(func() {
+				if err := db.Close(); err != nil {
+					t.Fatalf("close: %v", err)
+				}
+			})
+			// The pragmas surviving the encoding is the proof the query string
+			// was not truncated by a character in the path.
+			assertSQLitePragmas(t, db)
+
+			var opened string
+			if err := db.QueryRowContext(context.Background(), "SELECT file FROM pragma_database_list WHERE name = 'main'").Scan(&opened); err != nil {
+				t.Fatalf("read opened path: %v", err)
+			}
+			// Resolve both sides: macOS reports the temp dir through
+			// /private/var while t.TempDir hands back /var.
+			wantPath, err := filepath.EvalSymlinks(path)
+			if err != nil {
+				t.Fatalf("resolve %q: %v", path, err)
+			}
+			gotPath, err := filepath.EvalSymlinks(opened)
+			if err != nil {
+				t.Fatalf("resolve %q: %v", opened, err)
+			}
+			if gotPath != wantPath {
+				t.Fatalf("opened %q, want %q", gotPath, wantPath)
+			}
+		})
+	}
+
+	t.Run("resolves a relative path", func(t *testing.T) {
+		dsn, err := sqliteDataSourceName("ledger.db")
+		if err != nil {
+			t.Fatalf("sqliteDataSourceName: %v", err)
+		}
+		cwd, err := os.Getwd()
+		if err != nil {
+			t.Fatalf("Getwd: %v", err)
+		}
+		if !strings.Contains(dsn, filepath.ToSlash(cwd)) {
+			t.Fatalf("dsn = %q, want it to resolve against %q", dsn, cwd)
+		}
+	})
+
+	t.Run("requires immediate transactions", func(t *testing.T) {
+		dsn, err := sqliteDataSourceName(filepath.Join(t.TempDir(), "ledger.db"))
+		if err != nil {
+			t.Fatalf("sqliteDataSourceName: %v", err)
+		}
+		parsed, err := url.Parse(dsn)
+		if err != nil {
+			t.Fatalf("parse dsn %q: %v", dsn, err)
+		}
+		// dbmig.Apply's in-transaction re-read is only race-free under the
+		// write lock, so this is a cross-package contract, not a tuning knob.
+		if got := parsed.Query().Get("_txlock"); got != "immediate" {
+			t.Fatalf("_txlock = %q, want immediate", got)
+		}
+	})
+}

From 2e801a76db6466b67748f269c423b746cb608fd6 Mon Sep 17 00:00:00 2001
From: Aaron Wong <6979793+zzwong@users.noreply.github.com>
Date: Sat, 12 Sep 2026 19:47:52 -0500
Subject: [PATCH 5/6] fix(dbmig): reject a lowered schema version and cover the
 retry policy directly

The version read back after a migration was compared only against the run's
target, so a value that landed below what the migration wrote passed unchecked:
Apply returned no error, recorded the migration as applied, and reported a
version behind the schema it had installed. A stored value lower than the
migration's own version is now an invalid-meta error, while a higher one still
reaches Apply's downgrade check, and the table covers both directions on an
early and a final migration.

The WAL retry policy is extracted so its branches can be exercised without
relying on scheduling. Its test drives a genuine SQLITE_BUSY from the driver,
provoked by a contended write, because the driver exposes no constructor for
its error type and the retry check matches only that concrete type.

Three assertions were unsound. The relative-path check compared against a raw
path, so any checkout containing a character the URI encodes would fail it. The
URI form was guarded by a file:// prefix that also accepts the host-less
rendering, which SQLite rejects for Windows drive paths, so the authority is
asserted directly. The concurrency test derived the expected version from the
migration count rather than the declared SchemaVersion, and the startup test
kept its own copy of assertions now owned by a helper.
---
 internal/dbmig/dbmig.go        |  10 ++-
 internal/dbmig/dbmig_test.go   |  19 +++--
 internal/ledger/ledger.go      |  30 ++++---
 internal/ledger/ledger_test.go | 141 ++++++++++++++++++++++++++++++---
 4 files changed, 171 insertions(+), 29 deletions(-)

diff --git a/internal/dbmig/dbmig.go b/internal/dbmig/dbmig.go
index f6cdbd62..ea09d14d 100644
--- a/internal/dbmig/dbmig.go
+++ b/internal/dbmig/dbmig.go
@@ -295,13 +295,19 @@ func applyMigration(ctx context.Context, db *sql.DB, migration Migration) (int,
 		return current, false, fmt.Errorf("%w: schema_version update affected %d rows", ErrInvalidMeta, rowsAffected)
 	}
 
-	// Read the stored value back rather than returning the planned version: a
+	// 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, and Apply's downgrade check needs the truth.
+	// 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 current, false, fmt.Errorf("dbmig: commit migration %d %q: %w", migration.Version, migration.Name, err)
diff --git a/internal/dbmig/dbmig_test.go b/internal/dbmig/dbmig_test.go
index 2544edfb..d3785f07 100644
--- a/internal/dbmig/dbmig_test.go
+++ b/internal/dbmig/dbmig_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"fmt"
 	"path/filepath"
 	"reflect"
 	"testing"
@@ -532,7 +533,7 @@ func TestApplyRefusesDowngradeDiscoveredMidRun(t *testing.T) {
 	// 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) Migration {
+	advanceOn := func(version int, name, ddl string, to int) Migration {
 		return Migration{
 			Version: version,
 			Name:    name,
@@ -540,7 +541,7 @@ func TestApplyRefusesDowngradeDiscoveredMidRun(t *testing.T) {
 				if _, err := tx.ExecContext(ctx, ddl); err != nil {
 					return err
 				}
-				_, err := tx.ExecContext(ctx, "CREATE TRIGGER advance AFTER UPDATE ON meta BEGIN UPDATE meta SET schema_version = 9; END")
+				_, err := tx.ExecContext(ctx, fmt.Sprintf("CREATE TRIGGER advance AFTER UPDATE ON meta BEGIN UPDATE meta SET schema_version = %d; END", to))
 				return err
 			},
 		}
@@ -549,9 +550,13 @@ func TestApplyRefusesDowngradeDiscoveredMidRun(t *testing.T) {
 	for _, tt := range []struct {
 		name    string
 		advance int
+		to      int
+		wantErr error
 	}{
-		{name: "first migration", advance: 1},
-		{name: "last migration", advance: 2},
+		{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},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			ctx := context.Background()
@@ -568,15 +573,15 @@ func TestApplyRefusesDowngradeDiscoveredMidRun(t *testing.T) {
 				{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))
+					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, ErrDowngrade) {
-				t.Fatalf("Apply error = %v, want ErrDowngrade", err)
+			if !errors.Is(err, tt.wantErr) {
+				t.Fatalf("Apply error = %v, want %v", err, tt.wantErr)
 			}
 		})
 	}
diff --git a/internal/ledger/ledger.go b/internal/ledger/ledger.go
index 4c159162..c8ddac0c 100644
--- a/internal/ledger/ledger.go
+++ b/internal/ledger/ledger.go
@@ -555,22 +555,34 @@ func configureSQLite(ctx context.Context, db *sql.DB) error {
 	// 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.
-	deadline := time.Now().Add(DefaultBusyTimeout)
+	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 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 {
-		mode, err := setWALJournalMode(ctx, db)
+		value, err := attempt(ctx)
 		if err == nil {
-			if mode != "wal" {
-				return fmt.Errorf("ledger: enable WAL: journal_mode = %q, want wal", mode)
-			}
-			return nil
+			return value, nil
 		}
 		if !isSQLiteBusyError(err) || !time.Now().Before(deadline) {
-			return fmt.Errorf("ledger: enable WAL: %w", err)
+			return "", err
 		}
 		select {
 		case <-ctx.Done():
-			return fmt.Errorf("ledger: enable WAL: %w", ctx.Err())
-		case <-time.After(walRetryInterval):
+			return "", ctx.Err()
+		case <-time.After(interval):
 		}
 	}
 }
diff --git a/internal/ledger/ledger_test.go b/internal/ledger/ledger_test.go
index e545bddf..e73c2e80 100644
--- a/internal/ledger/ledger_test.go
+++ b/internal/ledger/ledger_test.go
@@ -27,15 +27,10 @@ func TestOpenMigratesFreshDatabaseAndAppliesStartupContract(t *testing.T) {
 	if version := queryInt(t, store.db, "SELECT schema_version FROM meta"); version != SchemaVersion {
 		t.Fatalf("schema_version = %d, want %d", version, SchemaVersion)
 	}
-	if got := queryInt(t, store.db, "PRAGMA foreign_keys"); got != 1 {
-		t.Fatalf("PRAGMA foreign_keys = %d, want 1", got)
-	}
+	assertSQLitePragmas(t, store.db)
 	if got := queryString(t, store.db, "PRAGMA journal_mode"); got != "wal" {
 		t.Fatalf("PRAGMA journal_mode = %q, want wal", got)
 	}
-	if got := queryInt(t, store.db, "PRAGMA busy_timeout"); int64(got) != DefaultBusyTimeout.Milliseconds() {
-		t.Fatalf("PRAGMA busy_timeout = %d, want %d", got, DefaultBusyTimeout.Milliseconds())
-	}
 
 	for _, table := range []string{"prs", "runs", "sessions", "findings", "planned_actions", "named_sessions", "reviewer_cohorts", "reviewer_cohort_members"} {
 		if !tableExists(t, store.db, table) {
@@ -133,7 +128,7 @@ func TestOpenConcurrentOnSamePathSucceeds(t *testing.T) {
 
 	// A racing open must not merely avoid erroring; it must land on a fully
 	// migrated database in WAL mode.
-	wantVersion := len(migrations())
+	wantVersion := SchemaVersion
 	for i, store := range stores {
 		var version int
 		if err := store.db.QueryRowContext(context.Background(), "SELECT schema_version FROM meta").Scan(&version); err != nil {
@@ -2018,8 +2013,14 @@ func TestSQLiteDataSourceName(t *testing.T) {
 			if err != nil {
 				t.Fatalf("sqliteDataSourceName(%q): %v", path, err)
 			}
-			if !strings.HasPrefix(dsn, "file://") {
-				t.Fatalf("dsn = %q, want a file: URI", dsn)
+			parsedDSN, err := url.Parse(dsn)
+			if err != nil {
+				t.Fatalf("parse dsn %q: %v", dsn, err)
+			}
+			// SQLite requires an empty or "localhost" authority; without one a
+			// Windows drive path renders as an invalid authority.
+			if parsedDSN.Scheme != "file" || parsedDSN.Host != "localhost" {
+				t.Fatalf("dsn = %q, want a file://localhost URI", dsn)
 			}
 
 			db, err := sql.Open("sqlite", dsn)
@@ -2064,8 +2065,12 @@ func TestSQLiteDataSourceName(t *testing.T) {
 		if err != nil {
 			t.Fatalf("Getwd: %v", err)
 		}
-		if !strings.Contains(dsn, filepath.ToSlash(cwd)) {
-			t.Fatalf("dsn = %q, want it to resolve against %q", dsn, cwd)
+		parsed, err := url.Parse(dsn)
+		if err != nil {
+			t.Fatalf("parse dsn %q: %v", dsn, err)
+		}
+		if want := filepath.ToSlash(filepath.Join(cwd, "ledger.db")); parsed.Path != want {
+			t.Fatalf("dsn path = %q, want %q", parsed.Path, want)
 		}
 	})
 
@@ -2085,3 +2090,117 @@ func TestSQLiteDataSourceName(t *testing.T) {
 		}
 	})
 }
+
+// busyError returns a genuine SQLITE_BUSY from the driver. The driver exposes
+// no constructor for sqlite.Error, and isSQLiteBusyError matches only that
+// concrete type, so the error is provoked by holding a write lock and writing
+// from a second connection with no busy timeout.
+func busyError(t *testing.T) error {
+	t.Helper()
+
+	path := filepath.Join(t.TempDir(), "busy.db")
+	dsn := "file://localhost" + filepath.ToSlash(path) + "?_pragma=busy_timeout(0)&_txlock=immediate"
+	holder := openRawSQLite(t, dsn)
+	contender := openRawSQLite(t, dsn)
+
+	ctx := context.Background()
+	if _, err := holder.ExecContext(ctx, "CREATE TABLE lock_probe (id INTEGER PRIMARY KEY)"); err != nil {
+		t.Fatalf("create lock probe: %v", err)
+	}
+	tx, err := holder.BeginTx(ctx, nil)
+	if err != nil {
+		t.Fatalf("begin holder transaction: %v", err)
+	}
+	t.Cleanup(func() {
+		_ = tx.Rollback()
+	})
+	if _, err := tx.ExecContext(ctx, "INSERT INTO lock_probe DEFAULT VALUES"); err != nil {
+		t.Fatalf("take write lock: %v", err)
+	}
+
+	_, err = contender.ExecContext(ctx, "INSERT INTO lock_probe DEFAULT VALUES")
+	if !isSQLiteBusyError(err) {
+		t.Fatalf("contended write error = %v, want a SQLITE_BUSY the retry loop recognizes", err)
+	}
+	return err
+}
+
+func openRawSQLite(t *testing.T, dsn string) *sql.DB {
+	t.Helper()
+
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		t.Fatalf("sql.Open(%q): %v", dsn, err)
+	}
+	t.Cleanup(func() {
+		if err := db.Close(); err != nil {
+			t.Errorf("close %q: %v", dsn, err)
+		}
+	})
+	return db
+}
+
+func TestRetryWhileBusy(t *testing.T) {
+	t.Run("retries until success", func(t *testing.T) {
+		busy := busyError(t)
+		calls := 0
+		value, err := retryWhileBusy(context.Background(), time.Millisecond, time.Second, func(context.Context) (string, error) {
+			calls++
+			if calls < 3 {
+				return "", busy
+			}
+			return "wal", nil
+		})
+		if err != nil {
+			t.Fatalf("retryWhileBusy: %v", err)
+		}
+		if value != "wal" {
+			t.Fatalf("value = %q, want wal", value)
+		}
+		if calls != 3 {
+			t.Fatalf("attempts = %d, want 3", calls)
+		}
+	})
+
+	t.Run("gives up at the budget", func(t *testing.T) {
+		busy := busyError(t)
+		calls := 0
+		_, err := retryWhileBusy(context.Background(), time.Millisecond, 20*time.Millisecond, func(context.Context) (string, error) {
+			calls++
+			return "", busy
+		})
+		if !isSQLiteBusyError(err) {
+			t.Fatalf("err = %v, want the busy error surfaced", err)
+		}
+		if calls < 2 {
+			t.Fatalf("attempts = %d, want more than one", calls)
+		}
+	})
+
+	t.Run("returns a non-busy error immediately", func(t *testing.T) {
+		calls := 0
+		wantErr := errors.New("boom")
+		_, err := retryWhileBusy(context.Background(), time.Millisecond, time.Second, func(context.Context) (string, error) {
+			calls++
+			return "", wantErr
+		})
+		if !errors.Is(err, wantErr) {
+			t.Fatalf("err = %v, want %v", err, wantErr)
+		}
+		if calls != 1 {
+			t.Fatalf("attempts = %d, want 1", calls)
+		}
+	})
+
+	t.Run("stops on a canceled context", func(t *testing.T) {
+		busy := busyError(t)
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		_, err := retryWhileBusy(ctx, time.Millisecond, time.Second, func(context.Context) (string, error) {
+			return "", busy
+		})
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("err = %v, want context.Canceled", err)
+		}
+	})
+}

From 27842bbeebd7be2af8facfbdb2db887355c768a8 Mon Sep 17 00:00:00 2001
From: Aaron Wong <6979793+zzwong@users.noreply.github.com>
Date: Sat, 12 Sep 2026 20:06:12 -0500
Subject: [PATCH 6/6] fix(dbmig): reject any schema version a migration did not
 itself write

The read-back guards were asymmetric: applyMigration rejected a value below the
migration it had just run, and Apply rejected one above the run's target, so a
value in between passed both. With two migrations and a target of two, a writer
that moved schema_version to two during the first migration left the second one
skipped without running, and Apply returned no error.

A migration that ran must leave meta exactly where it put it, so any other
value is now refused.
---
 internal/dbmig/dbmig.go      | 7 +++++++
 internal/dbmig/dbmig_test.go | 1 +
 2 files changed, 8 insertions(+)

diff --git a/internal/dbmig/dbmig.go b/internal/dbmig/dbmig.go
index ea09d14d..898760a4 100644
--- a/internal/dbmig/dbmig.go
+++ b/internal/dbmig/dbmig.go
@@ -92,6 +92,13 @@ func Apply(ctx context.Context, db *sql.DB, migrations []Migration) (Result, err
 		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
 		}
diff --git a/internal/dbmig/dbmig_test.go b/internal/dbmig/dbmig_test.go
index d3785f07..27cf5ef5 100644
--- a/internal/dbmig/dbmig_test.go
+++ b/internal/dbmig/dbmig_test.go
@@ -557,6 +557,7 @@ func TestApplyRefusesDowngradeDiscoveredMidRun(t *testing.T) {
 		{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()