From a2337619bf022a1d84212255a219f615bea4e476 Mon Sep 17 00:00:00 2001 From: jarl9804 Date: Wed, 15 Jul 2026 11:29:35 +0200 Subject: [PATCH 1/5] fix(store): set busy_timeout before WAL switch, retry it, and persist WAL files Cold-start ordering bug: journal_mode = WAL ran before busy_timeout was set, so concurrent engram processes racing the rollback-journal-to-WAL conversion failed immediately with SQLITE_BUSY (#559). busy_timeout now comes first and the WAL switch is wrapped in a bounded growing-backoff retry (mirroring withSQLiteWriteRetry) for the lock paths busy_timeout does not cover. Enable SQLITE_FCNTL_PERSIST_WAL on the single pooled connection via modernc.org/sqlite's FileControl so closing the last connection no longer unlinks the -wal/-shm files. The last-closer unlink, combined with divergent SHM views across process generations, allowed a process holding an orphaned SHM to delete the live WAL on clean shutdown and corrupt the database (#477, #571). Both New and newWithoutRepair now share one newStore path so the pragma setup can never diverge again. --- internal/store/store.go | 122 ++++++++++++++++++++++++++++------------ 1 file changed, 85 insertions(+), 37 deletions(-) diff --git a/internal/store/store.go b/internal/store/store.go index 24037c4a..fc5f7d7e 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -6,6 +6,7 @@ package store import ( + "context" "crypto/rand" "crypto/sha256" "database/sql" @@ -610,6 +611,16 @@ func (s *Store) commitHook(tx *sql.Tx) error { } func New(cfg Config) (*Store, error) { + return newStore(cfg, true) +} + +// newWithoutRepair is the same as New but skips repairEnrolledProjectSyncMutations. +// It exists solely to support tests that need to seed data and call repair manually. +func newWithoutRepair(cfg Config) (*Store, error) { + return newStore(cfg, false) +} + +func newStore(cfg Config, withRepair bool) (*Store, error) { if !filepath.IsAbs(cfg.DataDir) { return nil, fmt.Errorf("engram: data directory must be an absolute path, got %q — set ENGRAM_DATA_DIR or ensure your home directory is resolvable", cfg.DataDir) } @@ -624,64 +635,101 @@ func New(cfg Config) (*Store, error) { } db.SetMaxOpenConns(1) - // SQLite performance pragmas - pragmas := []string{ - "PRAGMA journal_mode = WAL", - "PRAGMA busy_timeout = 5000", - "PRAGMA synchronous = NORMAL", - "PRAGMA foreign_keys = ON", - } - for _, p := range pragmas { - if _, err := db.Exec(p); err != nil { - return nil, fmt.Errorf("engram: pragma %q: %w", p, err) - } + if err := configureSQLiteConnection(db); err != nil { + _ = db.Close() + return nil, err } s := &Store{db: db, cfg: cfg, hooks: defaultStoreHooks()} if err := s.migrate(); err != nil { + _ = db.Close() return nil, fmt.Errorf("engram: migration: %w", err) } - if err := s.repairEnrolledProjectSyncMutations(); err != nil { - return nil, fmt.Errorf("engram: repair enrolled sync journal: %w", err) + if withRepair { + if err := s.repairEnrolledProjectSyncMutations(); err != nil { + _ = db.Close() + return nil, fmt.Errorf("engram: repair enrolled sync journal: %w", err) + } } - return s, nil } -// newWithoutRepair is the same as New but skips repairEnrolledProjectSyncMutations. -// It exists solely to support tests that need to seed data and call repair manually. -func newWithoutRepair(cfg Config) (*Store, error) { - if !filepath.IsAbs(cfg.DataDir) { - return nil, fmt.Errorf("engram: data directory must be an absolute path, got %q — set ENGRAM_DATA_DIR or ensure your home directory is resolvable", cfg.DataDir) +// walSwitchRetryBackoffs paces retries of the cold-start `journal_mode = WAL` +// switch. Kept separate from sqliteWriteRetryBackoffs (used for row writes) +// because the WAL switch races entire process cold-starts, which need a +// longer tail to drain. +var walSwitchRetryBackoffs = []time.Duration{ + 10 * time.Millisecond, + 25 * time.Millisecond, + 50 * time.Millisecond, + 100 * time.Millisecond, + 200 * time.Millisecond, +} + +// configureSQLiteConnection applies the session pragmas and enables +// persistent WAL on the single pooled SQLite connection. +// +// Order matters: busy_timeout MUST be set before switching journal_mode to +// WAL. On a cold start several engram processes race to perform the +// rollback-journal→WAL conversion, and without a busy timeout the losers +// fail immediately with SQLITE_BUSY (#559). The switch is additionally +// wrapped in a bounded retry because busy_timeout does not cover every lock +// path of a journal-mode change. +func configureSQLiteConnection(db *sql.DB) error { + ctx := context.Background() + + // Pin the pool's single connection so every statement below — and the + // persist-WAL file control — lands on the same physical connection. + conn, err := db.Conn(ctx) + if err != nil { + return fmt.Errorf("engram: acquire connection: %w", err) } - if err := os.MkdirAll(cfg.DataDir, 0755); err != nil { - return nil, fmt.Errorf("engram: create data dir: %w", err) + defer conn.Close() + + const busyTimeoutPragma = "PRAGMA busy_timeout = 5000" + if _, err := conn.ExecContext(ctx, busyTimeoutPragma); err != nil { + return fmt.Errorf("engram: pragma %q: %w", busyTimeoutPragma, err) } - dbPath := filepath.Join(cfg.DataDir, "engram.db") - db, err := openDB("sqlite", dbPath) - if err != nil { - return nil, fmt.Errorf("engram: open database: %w", err) + const walPragma = "PRAGMA journal_mode = WAL" + var lastErr error + for attempt := 0; attempt <= len(walSwitchRetryBackoffs); attempt++ { + _, lastErr = conn.ExecContext(ctx, walPragma) + if lastErr == nil { + break + } + if !isRetryableSQLiteLockError(lastErr) || attempt == len(walSwitchRetryBackoffs) { + return fmt.Errorf("engram: pragma %q: %w", walPragma, lastErr) + } + time.Sleep(walSwitchRetryBackoffs[attempt]) } - db.SetMaxOpenConns(1) - pragmas := []string{ - "PRAGMA journal_mode = WAL", - "PRAGMA busy_timeout = 5000", + for _, p := range []string{ "PRAGMA synchronous = NORMAL", "PRAGMA foreign_keys = ON", - } - for _, p := range pragmas { - if _, err := db.Exec(p); err != nil { - return nil, fmt.Errorf("engram: pragma %q: %w", p, err) + } { + if _, err := conn.ExecContext(ctx, p); err != nil { + return fmt.Errorf("engram: pragma %q: %w", p, err) } } - s := &Store{db: db, cfg: cfg, hooks: defaultStoreHooks()} - if err := s.migrate(); err != nil { - return nil, fmt.Errorf("engram: migration: %w", err) + // Persistent WAL (SQLITE_FCNTL_PERSIST_WAL): keep the -wal/-shm files on + // disk when the last connection closes. Without it, SQLite's last-closer + // unlink — combined with divergent SHM views across process generations — + // can delete a live WAL out from under other processes and corrupt the + // database (#477, #571). + if err := conn.Raw(func(driverConn any) error { + fc, ok := driverConn.(sqlite.FileControl) + if !ok { + return fmt.Errorf("driver connection %T does not implement sqlite.FileControl", driverConn) + } + _, fcErr := fc.FileControlPersistWAL("main", 1) + return fcErr + }); err != nil { + return fmt.Errorf("engram: enable persistent WAL: %w", err) } - return s, nil + + return nil } func (s *Store) Close() error { From 4cfa25e09789a1cb0e2712df44eea1ad9670d0bd Mon Sep 17 00:00:00 2001 From: jarl9804 Date: Wed, 15 Jul 2026 11:30:13 +0200 Subject: [PATCH 2/5] fix(store): gate migrations behind user_version with an inter-process lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every engram subcommand ran the full migrate() suite plus the enrolled- project sync repair unconditionally and without inter-process exclusion. Three migrations are destructive check-then-act rebuilds (FTS topic_key, sync_chunks, legacy observations): two processes racing a cold start could both pass the check and double-run DROP/recreate, corrupting the database. Startup now reads PRAGMA user_version and compares it to a new schemaVersion constant (1 = the fully-migrated v1.19.0 schema): - equal: skip migrate() and the startup repair entirely (fast read-only startup for every warm process). - older: take an exclusive advisory file lock on /.migrate.lock, re-read user_version under the lock (another process may have finished first), run migrate() plus the repair, stamp user_version, unlock. - newer: warn and skip — never run old migrations against a schema written by a newer engram. A file lock (flock on unix, LockFileEx on Windows) was chosen over one giant BEGIN IMMEDIATE transaction because migrate() executes dozens of statements directly on s.db and its rebuild helpers open their own transactions internally; wrapping the suite in a single SQL transaction would nest BEGINs and require rewriting every helper. The per-step transactions inside the destructive rebuilds keep crash atomicity; the file lock adds the missing cross-process mutual exclusion. The user_version stamp is written only after migrate() and the repair both succeed, so a failed run is retried on the next startup. Promotes golang.org/x/sys to a direct dependency for LockFileEx. --- go.mod | 2 +- internal/store/migration_lock_unix.go | 33 +++++++ internal/store/migration_lock_windows.go | 35 +++++++ internal/store/store.go | 114 +++++++++++++++++++++-- 4 files changed, 175 insertions(+), 9 deletions(-) create mode 100644 internal/store/migration_lock_unix.go create mode 100644 internal/store/migration_lock_windows.go diff --git a/go.mod b/go.mod index a3c8dd9c..056d29e9 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/jackc/pgx/v5 v5.7.6 github.com/mark3labs/mcp-go v0.44.0 golang.org/x/net v0.52.0 + golang.org/x/sys v0.43.0 modernc.org/sqlite v1.45.0 ) @@ -59,7 +60,6 @@ require ( golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect golang.org/x/mod v0.34.0 // indirect golang.org/x/sync v0.20.0 // indirect - golang.org/x/sys v0.43.0 // indirect golang.org/x/text v0.36.0 // indirect golang.org/x/tools v0.43.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/internal/store/migration_lock_unix.go b/internal/store/migration_lock_unix.go new file mode 100644 index 00000000..4cba4923 --- /dev/null +++ b/internal/store/migration_lock_unix.go @@ -0,0 +1,33 @@ +//go:build unix + +package store + +import ( + "fmt" + "os" + "syscall" +) + +// acquireMigrationLock takes an exclusive advisory flock(2) on path, blocking +// until it is granted, and returns a function that releases the lock. It +// serializes whole processes around the migration suite so that the +// destructive check-then-act rebuilds inside migrate() can never run twice +// concurrently against the same database. +// +// The lock file is deliberately left in place after unlock: unlinking it +// would open a race where a third process re-creates the path and flocks a +// different inode, defeating the exclusion. +func acquireMigrationLock(path string) (func(), error) { + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + return nil, fmt.Errorf("open migration lock file %q: %w", path, err) + } + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil { + _ = f.Close() + return nil, fmt.Errorf("flock migration lock file %q: %w", path, err) + } + return func() { + _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) + _ = f.Close() + }, nil +} diff --git a/internal/store/migration_lock_windows.go b/internal/store/migration_lock_windows.go new file mode 100644 index 00000000..90b37508 --- /dev/null +++ b/internal/store/migration_lock_windows.go @@ -0,0 +1,35 @@ +//go:build windows + +package store + +import ( + "fmt" + "os" + + "golang.org/x/sys/windows" +) + +// acquireMigrationLock takes an exclusive LockFileEx lock on path, blocking +// until it is granted, and returns a function that releases the lock. It +// serializes whole processes around the migration suite so that the +// destructive check-then-act rebuilds inside migrate() can never run twice +// concurrently against the same database. +// +// The lock file is deliberately left in place after unlock: deleting it +// would open a race where a third process re-creates the path and locks a +// different file object, defeating the exclusion. +func acquireMigrationLock(path string) (func(), error) { + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + return nil, fmt.Errorf("open migration lock file %q: %w", path, err) + } + ol := new(windows.Overlapped) + if err := windows.LockFileEx(windows.Handle(f.Fd()), windows.LOCKFILE_EXCLUSIVE_LOCK, 0, 1, 0, ol); err != nil { + _ = f.Close() + return nil, fmt.Errorf("lock migration lock file %q: %w", path, err) + } + return func() { + _ = windows.UnlockFileEx(windows.Handle(f.Fd()), 0, 1, 0, new(windows.Overlapped)) + _ = f.Close() + }, nil +} diff --git a/internal/store/store.go b/internal/store/store.go index fc5f7d7e..cfc4ef1f 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -20,6 +20,7 @@ import ( "regexp" "strconv" "strings" + "sync/atomic" "time" "github.com/Gentleman-Programming/engram/internal/timeutil" @@ -641,15 +642,9 @@ func newStore(cfg Config, withRepair bool) (*Store, error) { } s := &Store{db: db, cfg: cfg, hooks: defaultStoreHooks()} - if err := s.migrate(); err != nil { + if err := s.runStartupMigrations(withRepair); err != nil { _ = db.Close() - return nil, fmt.Errorf("engram: migration: %w", err) - } - if withRepair { - if err := s.repairEnrolledProjectSyncMutations(); err != nil { - _ = db.Close() - return nil, fmt.Errorf("engram: repair enrolled sync journal: %w", err) - } + return nil, err } return s, nil } @@ -738,6 +733,109 @@ func (s *Store) Close() error { // ─── Migrations ────────────────────────────────────────────────────────────── +// schemaVersion is the store schema generation recorded in SQLite's +// PRAGMA user_version. Version 1 corresponds to the fully-migrated v1.19.0 +// schema (the current migrate() suite). Bump this constant whenever migrate() +// gains a new step so that databases stamped with an older version run the +// suite again on next startup. +// +// Gate semantics (runStartupMigrations): +// - user_version == schemaVersion → schema is current; skip migrate() and +// the startup repair entirely (fast, read-only startup). +// - user_version < schemaVersion → run migrate() under an exclusive +// inter-process lock, then stamp user_version. +// - user_version > schemaVersion → database was migrated by a newer engram; +// warn and skip (never run old migrations against a newer schema). +const schemaVersion = 1 + +// migrateRunCount counts executions of the gated migration block across the +// process. It exists for test observability only — asserting that the +// user_version gate skips migrations on already-current databases. +var migrateRunCount atomic.Int64 + +// runStartupMigrations runs migrate() (and, when withRepair is set, the +// enrolled-project sync repair) exactly once per schema generation. +// +// Exclusivity is provided by an advisory file lock (acquireMigrationLock) +// on /.migrate.lock rather than by one giant BEGIN IMMEDIATE +// transaction: migrate() executes dozens of statements through s.db and its +// helper migrations (migrateSyncChunksTable, migrateLegacyObservationsTable) +// open their own transactions internally, so wrapping the whole suite in a +// single SQL transaction would require rewriting every migration helper and +// would still deadlock on the nested BEGINs. The file lock serializes whole +// processes around the suite, and the destructive rebuild steps keep their +// own per-step transactions for atomicity against crashes. +func (s *Store) runStartupMigrations(withRepair bool) error { + current, err := s.readUserVersion() + if err != nil { + return fmt.Errorf("engram: read user_version: %w", err) + } + if shouldSkipMigrations(current) { + return nil + } + + unlock, err := acquireMigrationLock(filepath.Join(s.cfg.DataDir, ".migrate.lock")) + if err != nil { + return fmt.Errorf("engram: acquire migration lock: %w", err) + } + defer unlock() + + // Double-check under the lock: another process may have completed the + // migration while this one was waiting on the lock. + current, err = s.readUserVersion() + if err != nil { + return fmt.Errorf("engram: re-read user_version: %w", err) + } + if shouldSkipMigrations(current) { + return nil + } + + migrateRunCount.Add(1) + if err := s.migrate(); err != nil { + return fmt.Errorf("engram: migration: %w", err) + } + if withRepair { + if err := s.repairEnrolledProjectSyncMutations(); err != nil { + return fmt.Errorf("engram: repair enrolled sync journal: %w", err) + } + } + // Stamp only after the full block succeeded so a failed migration or + // repair is retried on the next startup (migrate() is idempotent). + if err := s.setUserVersion(schemaVersion); err != nil { + return fmt.Errorf("engram: set user_version: %w", err) + } + return nil +} + +// shouldSkipMigrations reports whether the migration suite must be skipped +// for the given user_version. A database stamped by a NEWER engram is left +// untouched: running this binary's older migrations against a newer schema +// is exactly the kind of blind rebuild that corrupts databases. +func shouldSkipMigrations(current int) bool { + if current == schemaVersion { + return true + } + if current > schemaVersion { + log.Printf("[store] database schema version %d is newer than this binary's %d — skipping migrations (upgrade engram to manage this database)", current, schemaVersion) + return true + } + return false +} + +func (s *Store) readUserVersion() (int, error) { + var v int + if err := s.db.QueryRow("PRAGMA user_version").Scan(&v); err != nil { + return 0, err + } + return v, nil +} + +func (s *Store) setUserVersion(v int) error { + // PRAGMA does not support bound parameters; v is a trusted constant. + _, err := s.db.Exec(fmt.Sprintf("PRAGMA user_version = %d", v)) + return err +} + func (s *Store) migrate() error { schema := ` CREATE TABLE IF NOT EXISTS sessions ( From 3af305164ff4366edd7c6fbaef90a2f96d4ff5f6 Mon Sep 17 00:00:00 2001 From: jarl9804 Date: Wed, 15 Jul 2026 11:33:41 +0200 Subject: [PATCH 3/5] test(store): cover persistent WAL, cold-start races, and migration gate - persistent WAL: the -wal file must survive Close (a last-closer unlink here is the #477/#571 corruption mechanism). - concurrent cold start: 8 goroutines and 3 child processes open the same fresh database simultaneously; all must succeed, the migration suite must run exactly once, and the result must pass integrity_check (#559). - user_version gate: a second open skips the migration suite (asserted via migrateRunCount), and a database stamped by a newer engram is opened without running this binary's migrations and without rewriting its user_version. - acquireMigrationLock: a second acquire blocks until the first holder releases. --- internal/store/startup_gate_test.go | 351 ++++++++++++++++++++++++++++ 1 file changed, 351 insertions(+) create mode 100644 internal/store/startup_gate_test.go diff --git a/internal/store/startup_gate_test.go b/internal/store/startup_gate_test.go new file mode 100644 index 00000000..32ec5bc6 --- /dev/null +++ b/internal/store/startup_gate_test.go @@ -0,0 +1,351 @@ +package store + +// Tests for the SQLite-corruption fixes: +// +// 1. persistent WAL — closing the store must NOT unlink the -wal file +// (mechanism behind upstream #477/#571), +// 2. cold-start concurrency — many stores opening the same fresh database +// simultaneously, in-process and across child processes (#559), and +// 3. the user_version migration gate — the migration suite runs exactly +// once per schema generation and is never run against a database +// stamped by a newer engram. + +import ( + "database/sql" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +func TestPersistentWALSurvivesClose(t *testing.T) { + cfg := mustDefaultConfig(t) + cfg.DataDir = t.TempDir() + + s, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := s.CreateSession("wal-session", "wal-project", cfg.DataDir); err != nil { + t.Fatalf("CreateSession: %v", err) + } + + walPath := filepath.Join(cfg.DataDir, "engram.db-wal") + if _, err := os.Stat(walPath); err != nil { + t.Fatalf("-wal file missing while store is open: %v", err) + } + + if err := s.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + if _, err := os.Stat(walPath); err != nil { + t.Fatalf("-wal file was unlinked on close — persistent WAL is not active: %v", err) + } +} + +func TestUserVersionGateSkipsSecondOpen(t *testing.T) { + cfg := mustDefaultConfig(t) + cfg.DataDir = t.TempDir() + + before := migrateRunCount.Load() + + s1, err := New(cfg) + if err != nil { + t.Fatalf("first New: %v", err) + } + var v int + if err := s1.db.QueryRow("PRAGMA user_version").Scan(&v); err != nil { + t.Fatalf("read user_version: %v", err) + } + if v != schemaVersion { + t.Fatalf("user_version after first open = %d, want %d", v, schemaVersion) + } + if err := s1.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + afterFirst := migrateRunCount.Load() + if got := afterFirst - before; got != 1 { + t.Fatalf("migration suite ran %d times on a fresh database, want exactly 1", got) + } + + s2, err := New(cfg) + if err != nil { + t.Fatalf("second New: %v", err) + } + defer s2.Close() + + if got := migrateRunCount.Load() - afterFirst; got != 0 { + t.Fatalf("migration suite ran %d times on an already-migrated database, want 0", got) + } + + // The gated (skipped-migration) store must still be fully usable. + if err := s2.CreateSession("gate-session", "gate-project", cfg.DataDir); err != nil { + t.Fatalf("CreateSession on gated store: %v", err) + } +} + +func TestNewerSchemaVersionIsLeftUntouched(t *testing.T) { + cfg := mustDefaultConfig(t) + cfg.DataDir = t.TempDir() + + s1, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := s1.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + // Simulate a database already migrated by a NEWER engram. + future := schemaVersion + 7 + raw, err := sql.Open("sqlite", filepath.Join(cfg.DataDir, "engram.db")) + if err != nil { + t.Fatalf("raw open: %v", err) + } + if _, err := raw.Exec(fmt.Sprintf("PRAGMA user_version = %d", future)); err != nil { + t.Fatalf("stamp future user_version: %v", err) + } + if err := raw.Close(); err != nil { + t.Fatalf("raw close: %v", err) + } + + before := migrateRunCount.Load() + + s2, err := New(cfg) + if err != nil { + t.Fatalf("New on newer-schema database: %v", err) + } + defer s2.Close() + + if got := migrateRunCount.Load() - before; got != 0 { + t.Fatalf("migration suite ran %d times against a newer schema, want 0", got) + } + + var v int + if err := s2.db.QueryRow("PRAGMA user_version").Scan(&v); err != nil { + t.Fatalf("read user_version: %v", err) + } + if v != future { + t.Fatalf("user_version was rewritten to %d, want it left at %d", v, future) + } +} + +func TestConcurrentColdStartGoroutines(t *testing.T) { + cfg := mustDefaultConfig(t) + cfg.DataDir = t.TempDir() + + before := migrateRunCount.Load() + + const n = 8 + start := make(chan struct{}) + errs := make(chan error, n) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + s, err := New(cfg) + if err != nil { + errs <- fmt.Errorf("goroutine %d: New: %w", i, err) + return + } + defer s.Close() + if err := s.CreateSession(fmt.Sprintf("cold-%d", i), "coldstart", cfg.DataDir); err != nil { + errs <- fmt.Errorf("goroutine %d: CreateSession: %w", i, err) + return + } + errs <- nil + }(i) + } + close(start) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Error(err) + } + } + if t.Failed() { + return + } + + if got := migrateRunCount.Load() - before; got != 1 { + t.Errorf("migration suite ran %d times across %d concurrent cold starts, want exactly 1", got, n) + } + + // Verify the resulting database is complete and healthy. + s, err := New(cfg) + if err != nil { + t.Fatalf("verification New: %v", err) + } + defer s.Close() + + var sessions int + if err := s.db.QueryRow("SELECT COUNT(*) FROM sessions").Scan(&sessions); err != nil { + t.Fatalf("count sessions: %v", err) + } + if sessions != n { + t.Errorf("sessions = %d, want %d", sessions, n) + } + var v int + if err := s.db.QueryRow("PRAGMA user_version").Scan(&v); err != nil { + t.Fatalf("read user_version: %v", err) + } + if v != schemaVersion { + t.Errorf("user_version = %d, want %d", v, schemaVersion) + } + var integrity string + if err := s.db.QueryRow("PRAGMA integrity_check").Scan(&integrity); err != nil { + t.Fatalf("integrity_check: %v", err) + } + if integrity != "ok" { + t.Errorf("integrity_check = %q, want ok", integrity) + } +} + +// coldStartChildEnv points a re-executed child copy of the test binary at the +// shared data directory used by TestConcurrentColdStartProcesses. +const coldStartChildEnv = "ENGRAM_TEST_COLDSTART_DIR" + +// TestColdStartChildProcess is not a standalone test: it is re-executed as a +// child process by TestConcurrentColdStartProcesses and skips otherwise. +func TestColdStartChildProcess(t *testing.T) { + dir := os.Getenv(coldStartChildEnv) + if dir == "" { + t.Skip("runs only as a child of TestConcurrentColdStartProcesses") + } + + cfg, err := DefaultConfig() + if err != nil { + t.Fatalf("DefaultConfig: %v", err) + } + cfg.DataDir = dir + cfg.DedupeWindow = time.Hour + + s, err := New(cfg) + if err != nil { + t.Fatalf("child cold-start New: %v", err) + } + defer s.Close() + + if err := s.CreateSession(fmt.Sprintf("proc-%d", os.Getpid()), "coldstart-proc", dir); err != nil { + t.Fatalf("child CreateSession: %v", err) + } +} + +func TestConcurrentColdStartProcesses(t *testing.T) { + if os.Getenv(coldStartChildEnv) != "" { + t.Skip("child process mode") + } + + exe, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable: %v", err) + } + + dir := t.TempDir() + const procs = 3 + + cmds := make([]*exec.Cmd, procs) + outputs := make([]*strings.Builder, procs) + for i := range cmds { + outputs[i] = &strings.Builder{} + cmd := exec.Command(exe, "-test.run", "^TestColdStartChildProcess$", "-test.v", "-test.timeout", "60s") + cmd.Env = append(os.Environ(), coldStartChildEnv+"="+dir) + cmd.Stdout = outputs[i] + cmd.Stderr = outputs[i] + if err := cmd.Start(); err != nil { + t.Fatalf("start child %d: %v", i, err) + } + cmds[i] = cmd + } + for i, cmd := range cmds { + if err := cmd.Wait(); err != nil { + t.Errorf("child process %d failed: %v\noutput:\n%s", i, err, outputs[i].String()) + } + } + if t.Failed() { + return + } + + // Every child cold-started against the same fresh database and wrote one + // session. Open from the parent and verify the result is consistent. + cfg, err := DefaultConfig() + if err != nil { + t.Fatalf("DefaultConfig: %v", err) + } + cfg.DataDir = dir + cfg.DedupeWindow = time.Hour + + s, err := New(cfg) + if err != nil { + t.Fatalf("parent verification New: %v", err) + } + defer s.Close() + + var sessions int + if err := s.db.QueryRow("SELECT COUNT(*) FROM sessions").Scan(&sessions); err != nil { + t.Fatalf("count sessions: %v", err) + } + if sessions != procs { + t.Errorf("sessions = %d, want %d", sessions, procs) + } + var v int + if err := s.db.QueryRow("PRAGMA user_version").Scan(&v); err != nil { + t.Fatalf("read user_version: %v", err) + } + if v != schemaVersion { + t.Errorf("user_version = %d, want %d", v, schemaVersion) + } + var integrity string + if err := s.db.QueryRow("PRAGMA integrity_check").Scan(&integrity); err != nil { + t.Fatalf("integrity_check: %v", err) + } + if integrity != "ok" { + t.Errorf("integrity_check = %q, want ok", integrity) + } +} + +func TestAcquireMigrationLockExcludes(t *testing.T) { + path := filepath.Join(t.TempDir(), ".migrate.lock") + + unlock1, err := acquireMigrationLock(path) + if err != nil { + t.Fatalf("first acquire: %v", err) + } + + acquired := make(chan struct{}) + go func() { + unlock2, err := acquireMigrationLock(path) + if err != nil { + t.Errorf("second acquire: %v", err) + close(acquired) + return + } + close(acquired) + unlock2() + }() + + select { + case <-acquired: + t.Fatal("second acquire succeeded while the first lock was still held") + case <-time.After(150 * time.Millisecond): + // Still blocked — expected. + } + + unlock1() + + select { + case <-acquired: + // Granted after release — expected. + case <-time.After(5 * time.Second): + t.Fatal("second acquire did not proceed after the first lock was released") + } +} From 7df66c4af18ccd4b60c6816fc411a13258be5fbd Mon Sep 17 00:00:00 2001 From: jarl9804 Date: Wed, 15 Jul 2026 12:02:16 +0200 Subject: [PATCH 4/5] fix(store): survive connection replacement, repair every open, bound lock wait MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fixes from reliability review: Connection replacement (critical). The session pragmas and persist-WAL were applied once, to the first pooled connection only. database/sql silently discards a modernc connection after a context-cancelled query interrupts it (IsValid/ResetSession fail once sqlite3_is_interrupted) and opens an unconfigured replacement — busy_timeout back to 0 (#559 returns) and persist-WAL unset (#477 returns) for the rest of the process life. Production paths pass cancellable request contexts into the pool (server and MCP doctor handlers), so this was reachable. Now: - the four pragmas travel in the DSN as _pragma query parameters, which the driver applies to every physical connection it opens (and it orders busy_timeout ahead of the WAL switch itself); - persist-WAL is enabled by a driver-global RegisterConnectionHook (registered once via sync.Once, best-effort so exotic VFS opens never break) and strictly verified on the store's own database at startup; - the cold-start SQLITE_BUSY retry now wraps the first connection open, because DSN pragmas run inside driver.Open and a lost WAL-switch race fails the open as a whole. Replacement opens re-run journal_mode(WAL) as a cheap no-op read once the database is in WAL mode. Repair frequency. The user_version gate had reduced the enrolled-project sync repair from every-open to once per schema generation, regressing its self-healing contract. The repair runs on every open again, but inside the same exclusive migration lock so N concurrent startups cannot storm identical backfill writes. Exclusive (not shared) locking is correct because the repair writes when a backfill is needed; the healthy-path cost is one brief lock hold plus the repair's read-only fast path. Lock robustness. The migration lock was a blocking flock: a hung holder would silently stall every engram process on the machine forever. Both platform paths now acquire non-blocking (LOCK_NB / LOCKFILE_FAIL_IMMEDIATELY) inside a bounded retry loop (poll capped at 100ms, 60s total budget) and fail with an error naming the lock file and pointing at a stuck engram process. Shared loop lives in migration_lock.go; platform files provide only try/unlock primitives. --- internal/store/migration_lock.go | 68 ++++++++ internal/store/migration_lock_unix.go | 39 ++--- internal/store/migration_lock_windows.go | 43 +++-- internal/store/store.go | 207 +++++++++++++++-------- 4 files changed, 239 insertions(+), 118 deletions(-) create mode 100644 internal/store/migration_lock.go diff --git a/internal/store/migration_lock.go b/internal/store/migration_lock.go new file mode 100644 index 00000000..2a256a85 --- /dev/null +++ b/internal/store/migration_lock.go @@ -0,0 +1,68 @@ +package store + +import ( + "fmt" + "os" + "time" +) + +// migrationLockTimeout bounds how long a process waits for the migration +// lock. A hung holder must produce a loud, actionable error instead of +// silently blocking every engram process on the machine forever. It is a +// variable (not a constant) so tests can shorten the timeout path. +var migrationLockTimeout = 60 * time.Second + +// acquireMigrationLock takes an exclusive advisory lock on path and returns +// a function that releases it. It serializes whole processes around the +// migration suite and the startup repair so that the destructive +// check-then-act rebuilds inside migrate() can never run twice concurrently +// against the same database. +// +// Acquisition is non-blocking with a bounded growing backoff (up to +// migrationLockTimeout total) rather than a blocking lock: a stuck holder +// then surfaces as a clear error naming the lock file instead of a silent +// machine-wide hang. +// +// The lock file is deliberately left in place after unlock: unlinking it +// would open a race where a third process re-creates the path and locks a +// different inode/file object, defeating the exclusion. +func acquireMigrationLock(path string) (func(), error) { + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + return nil, fmt.Errorf("open migration lock file %q: %w", path, err) + } + + deadline := time.Now().Add(migrationLockTimeout) + backoff := 10 * time.Millisecond + for { + acquired, err := tryLockMigrationFile(f) + if err != nil { + _ = f.Close() + return nil, fmt.Errorf("lock migration lock file %q: %w", path, err) + } + if acquired { + return func() { + _ = unlockMigrationFile(f) + _ = f.Close() + }, nil + } + if time.Now().After(deadline) { + _ = f.Close() + return nil, fmt.Errorf( + "timed out after %s waiting for migration lock %q — another engram process appears to be holding it; check for a stuck engram process (and terminate it) before retrying", + migrationLockTimeout, path, + ) + } + time.Sleep(backoff) + // Grow the poll interval but cap it low: healthy holders release the + // lock within milliseconds (the startup repair fast path is read-only), + // and every engram subcommand acquires this lock, so an aggressive cap + // keeps contended cold starts snappy. + if backoff < 100*time.Millisecond { + backoff *= 2 + if backoff > 100*time.Millisecond { + backoff = 100 * time.Millisecond + } + } + } +} diff --git a/internal/store/migration_lock_unix.go b/internal/store/migration_lock_unix.go index 4cba4923..1f01067e 100644 --- a/internal/store/migration_lock_unix.go +++ b/internal/store/migration_lock_unix.go @@ -3,31 +3,28 @@ package store import ( - "fmt" + "errors" "os" "syscall" ) -// acquireMigrationLock takes an exclusive advisory flock(2) on path, blocking -// until it is granted, and returns a function that releases the lock. It -// serializes whole processes around the migration suite so that the -// destructive check-then-act rebuilds inside migrate() can never run twice -// concurrently against the same database. -// -// The lock file is deliberately left in place after unlock: unlinking it -// would open a race where a third process re-creates the path and flocks a -// different inode, defeating the exclusion. -func acquireMigrationLock(path string) (func(), error) { - f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) - if err != nil { - return nil, fmt.Errorf("open migration lock file %q: %w", path, err) +// tryLockMigrationFile attempts a non-blocking exclusive flock(2) on f. +// It reports (false, nil) when another process (or file description) holds +// the lock, so the caller can retry with backoff. +func tryLockMigrationFile(f *os.File) (bool, error) { + err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + if err == nil { + return true, nil } - if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil { - _ = f.Close() - return nil, fmt.Errorf("flock migration lock file %q: %w", path, err) + // EWOULDBLOCK/EAGAIN: lock is held elsewhere. EINTR: interrupted by a + // signal. Both are retryable, not failures. + if errors.Is(err, syscall.EWOULDBLOCK) || errors.Is(err, syscall.EAGAIN) || errors.Is(err, syscall.EINTR) { + return false, nil } - return func() { - _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) - _ = f.Close() - }, nil + return false, err +} + +// unlockMigrationFile releases the flock taken by tryLockMigrationFile. +func unlockMigrationFile(f *os.File) error { + return syscall.Flock(int(f.Fd()), syscall.LOCK_UN) } diff --git a/internal/store/migration_lock_windows.go b/internal/store/migration_lock_windows.go index 90b37508..92372024 100644 --- a/internal/store/migration_lock_windows.go +++ b/internal/store/migration_lock_windows.go @@ -3,33 +3,32 @@ package store import ( - "fmt" + "errors" "os" "golang.org/x/sys/windows" ) -// acquireMigrationLock takes an exclusive LockFileEx lock on path, blocking -// until it is granted, and returns a function that releases the lock. It -// serializes whole processes around the migration suite so that the -// destructive check-then-act rebuilds inside migrate() can never run twice -// concurrently against the same database. -// -// The lock file is deliberately left in place after unlock: deleting it -// would open a race where a third process re-creates the path and locks a -// different file object, defeating the exclusion. -func acquireMigrationLock(path string) (func(), error) { - f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) - if err != nil { - return nil, fmt.Errorf("open migration lock file %q: %w", path, err) - } +// tryLockMigrationFile attempts a non-blocking exclusive LockFileEx on f. +// It reports (false, nil) when another process holds the lock, so the +// caller can retry with backoff. +func tryLockMigrationFile(f *os.File) (bool, error) { ol := new(windows.Overlapped) - if err := windows.LockFileEx(windows.Handle(f.Fd()), windows.LOCKFILE_EXCLUSIVE_LOCK, 0, 1, 0, ol); err != nil { - _ = f.Close() - return nil, fmt.Errorf("lock migration lock file %q: %w", path, err) + err := windows.LockFileEx( + windows.Handle(f.Fd()), + windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY, + 0, 1, 0, ol, + ) + if err == nil { + return true, nil + } + if errors.Is(err, windows.ERROR_LOCK_VIOLATION) { + return false, nil } - return func() { - _ = windows.UnlockFileEx(windows.Handle(f.Fd()), 0, 1, 0, new(windows.Overlapped)) - _ = f.Close() - }, nil + return false, err +} + +// unlockMigrationFile releases the lock taken by tryLockMigrationFile. +func unlockMigrationFile(f *os.File) error { + return windows.UnlockFileEx(windows.Handle(f.Fd()), 0, 1, 0, new(windows.Overlapped)) } diff --git a/internal/store/store.go b/internal/store/store.go index cfc4ef1f..918f6ef7 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -15,11 +15,13 @@ import ( "errors" "fmt" "log" + "net/url" "os" "path/filepath" "regexp" "strconv" "strings" + "sync" "sync/atomic" "time" @@ -630,13 +632,14 @@ func newStore(cfg Config, withRepair bool) (*Store, error) { } dbPath := filepath.Join(cfg.DataDir, "engram.db") - db, err := openDB("sqlite", dbPath) + registerPersistWALHook() + db, err := openDB("sqlite", storeDSN(dbPath)) if err != nil { return nil, fmt.Errorf("engram: open database: %w", err) } db.SetMaxOpenConns(1) - if err := configureSQLiteConnection(db); err != nil { + if err := primeConnection(db); err != nil { _ = db.Close() return nil, err } @@ -649,10 +652,69 @@ func newStore(cfg Config, withRepair bool) (*Store, error) { return s, nil } -// walSwitchRetryBackoffs paces retries of the cold-start `journal_mode = WAL` -// switch. Kept separate from sqliteWriteRetryBackoffs (used for row writes) -// because the WAL switch races entire process cold-starts, which need a -// longer tail to drain. +// storeDSN builds the modernc.org/sqlite DSN for dbPath with the session +// pragmas encoded as _pragma query parameters. The driver applies these to +// EVERY physical connection it opens — including replacements the +// database/sql pool creates after a context-cancelled query poisons a +// connection (an interrupted modernc conn fails IsValid/ResetSession and is +// silently discarded). Configuring only the first connection would leave +// such replacements with busy_timeout 0 (#559 returns) and no persistent +// WAL (#477 returns). +// +// Order also matters: busy_timeout must be active before the WAL switch; +// the driver itself sorts busy_timeout ahead of the other _pragma params +// (see modernc.org/sqlite applyQueryParams), so every fresh connection gets +// the 5s busy handler before running journal_mode(WAL). +// +// The plain-path DSN form is used deliberately: modernc strips everything +// after '?' from a non-"file:" DSN and passes the path through verbatim, so +// no URL escaping of the path is needed (a literal '?' inside DataDir is the +// only unsupported case, as in the driver generally). +func storeDSN(dbPath string) string { + q := url.Values{} + for _, p := range []string{ + "busy_timeout(5000)", + "journal_mode(WAL)", + "synchronous(NORMAL)", + "foreign_keys(1)", + } { + q.Add("_pragma", p) + } + return dbPath + "?" + q.Encode() +} + +// persistWALHookOnce guards the process-global driver hook registration. +var persistWALHookOnce sync.Once + +// registerPersistWALHook registers a modernc.org/sqlite connection hook that +// enables SQLITE_FCNTL_PERSIST_WAL on every new physical connection, so a +// pool-created replacement connection can never reintroduce the last-closer +// WAL unlink (#477, #571). +// +// The hook is driver-global (it fires for every sqlite connection this +// process opens, not just the store's), so it is registered exactly once and +// kept best-effort: engram only opens its own databases, persist-WAL is +// harmless-to-beneficial on ancillary opens (backups, diagnostics, tests), +// and a VFS that does not support the file control must not break those +// opens. The store's own database is strictly verified in primeConnection. +func registerPersistWALHook() { + persistWALHookOnce.Do(func() { + sqlite.RegisterConnectionHook(func(conn sqlite.ExecQuerierContext, _ string) error { + if fc, ok := conn.(sqlite.FileControl); ok { + _, _ = fc.FileControlPersistWAL("main", 1) + } + return nil + }) + }) +} + +// walSwitchRetryBackoffs paces retries of the first connection open during a +// cold start. The DSN pragmas run inside driver.Open, so when several +// processes race the rollback-journal→WAL conversion a losing open fails +// with SQLITE_BUSY as a whole; the retry therefore wraps connection +// acquisition rather than a single statement. Kept separate from +// sqliteWriteRetryBackoffs (used for row writes) because the WAL switch +// races entire process cold-starts, which need a longer tail to drain. var walSwitchRetryBackoffs = []time.Duration{ 10 * time.Millisecond, 25 * time.Millisecond, @@ -661,70 +723,44 @@ var walSwitchRetryBackoffs = []time.Duration{ 200 * time.Millisecond, } -// configureSQLiteConnection applies the session pragmas and enables -// persistent WAL on the single pooled SQLite connection. -// -// Order matters: busy_timeout MUST be set before switching journal_mode to -// WAL. On a cold start several engram processes race to perform the -// rollback-journal→WAL conversion, and without a busy timeout the losers -// fail immediately with SQLITE_BUSY (#559). The switch is additionally -// wrapped in a bounded retry because busy_timeout does not cover every lock -// path of a journal-mode change. -func configureSQLiteConnection(db *sql.DB) error { +// primeConnection forces the pool to open its first physical connection — +// running the DSN pragmas — with a bounded cold-start retry, and then +// verifies persistent WAL is active on the store's database. Replacement +// connections repeat the same configuration automatically via the DSN +// pragmas and the driver connection hook; once the database is in WAL mode +// the journal_mode(WAL) pragma on those opens is a cheap no-op read, so the +// cold-start SQLITE_BUSY window exists only here. +func primeConnection(db *sql.DB) error { ctx := context.Background() - // Pin the pool's single connection so every statement below — and the - // persist-WAL file control — lands on the same physical connection. - conn, err := db.Conn(ctx) - if err != nil { - return fmt.Errorf("engram: acquire connection: %w", err) - } - defer conn.Close() - - const busyTimeoutPragma = "PRAGMA busy_timeout = 5000" - if _, err := conn.ExecContext(ctx, busyTimeoutPragma); err != nil { - return fmt.Errorf("engram: pragma %q: %w", busyTimeoutPragma, err) - } - - const walPragma = "PRAGMA journal_mode = WAL" + var conn *sql.Conn var lastErr error for attempt := 0; attempt <= len(walSwitchRetryBackoffs); attempt++ { - _, lastErr = conn.ExecContext(ctx, walPragma) + conn, lastErr = db.Conn(ctx) if lastErr == nil { break } if !isRetryableSQLiteLockError(lastErr) || attempt == len(walSwitchRetryBackoffs) { - return fmt.Errorf("engram: pragma %q: %w", walPragma, lastErr) + return fmt.Errorf("engram: open initial connection: %w", lastErr) } time.Sleep(walSwitchRetryBackoffs[attempt]) } + defer conn.Close() - for _, p := range []string{ - "PRAGMA synchronous = NORMAL", - "PRAGMA foreign_keys = ON", - } { - if _, err := conn.ExecContext(ctx, p); err != nil { - return fmt.Errorf("engram: pragma %q: %w", p, err) - } - } - - // Persistent WAL (SQLITE_FCNTL_PERSIST_WAL): keep the -wal/-shm files on - // disk when the last connection closes. Without it, SQLite's last-closer - // unlink — combined with divergent SHM views across process generations — - // can delete a live WAL out from under other processes and corrupt the - // database (#477, #571). - if err := conn.Raw(func(driverConn any) error { + return conn.Raw(func(driverConn any) error { fc, ok := driverConn.(sqlite.FileControl) if !ok { - return fmt.Errorf("driver connection %T does not implement sqlite.FileControl", driverConn) + return fmt.Errorf("engram: driver connection %T does not implement sqlite.FileControl", driverConn) } - _, fcErr := fc.FileControlPersistWAL("main", 1) - return fcErr - }); err != nil { - return fmt.Errorf("engram: enable persistent WAL: %w", err) - } - - return nil + mode, err := fc.FileControlPersistWAL("main", 1) + if err != nil { + return fmt.Errorf("engram: enable persistent WAL: %w", err) + } + if mode != 1 { + return fmt.Errorf("engram: persistent WAL not active (file control returned mode %d)", mode) + } + return nil + }) } func (s *Store) Close() error { @@ -740,12 +776,16 @@ func (s *Store) Close() error { // suite again on next startup. // // Gate semantics (runStartupMigrations): -// - user_version == schemaVersion → schema is current; skip migrate() and -// the startup repair entirely (fast, read-only startup). +// - user_version == schemaVersion → schema is current; skip migrate(). // - user_version < schemaVersion → run migrate() under an exclusive // inter-process lock, then stamp user_version. // - user_version > schemaVersion → database was migrated by a newer engram; // warn and skip (never run old migrations against a newer schema). +// +// The enrolled-project sync repair is independent of the gate: it runs on +// every open (self-healing, as before the gate existed) but inside the same +// inter-process lock so concurrent startups cannot storm identical backfill +// writes. const schemaVersion = 1 // migrateRunCount counts executions of the gated migration block across the @@ -753,8 +793,10 @@ const schemaVersion = 1 // user_version gate skips migrations on already-current databases. var migrateRunCount atomic.Int64 -// runStartupMigrations runs migrate() (and, when withRepair is set, the -// enrolled-project sync repair) exactly once per schema generation. +// runStartupMigrations runs migrate() at most once per schema generation +// (gated on PRAGMA user_version) and, when withRepair is set, the +// enrolled-project sync repair on every open — both inside one exclusive +// inter-process lock. // // Exclusivity is provided by an advisory file lock (acquireMigrationLock) // on /.migrate.lock rather than by one giant BEGIN IMMEDIATE @@ -765,13 +807,22 @@ var migrateRunCount atomic.Int64 // would still deadlock on the nested BEGINs. The file lock serializes whole // processes around the suite, and the destructive rebuild steps keep their // own per-step transactions for atomicity against crashes. +// +// The repair deliberately runs on every open — its self-healing behavior +// predates the gate and must not regress — but it lives inside the same +// exclusive lock so N concurrent startups cannot storm identical backfill +// writes; the exclusive (not shared) lock is correct because the repair +// writes when a backfill is needed. The per-open cost is small: the repair's +// own fast path (projectNeedsBackfill) is read-only, so a healthy database +// pays one brief lock acquisition plus a few reads. func (s *Store) runStartupMigrations(withRepair bool) error { current, err := s.readUserVersion() if err != nil { return fmt.Errorf("engram: read user_version: %w", err) } - if shouldSkipMigrations(current) { - return nil + skipMigrate := shouldSkipMigrations(current) + if skipMigrate && !withRepair { + return nil // schema current and no repair requested: nothing to do } unlock, err := acquireMigrationLock(filepath.Join(s.cfg.DataDir, ".migrate.lock")) @@ -780,29 +831,35 @@ func (s *Store) runStartupMigrations(withRepair bool) error { } defer unlock() - // Double-check under the lock: another process may have completed the - // migration while this one was waiting on the lock. - current, err = s.readUserVersion() - if err != nil { - return fmt.Errorf("engram: re-read user_version: %w", err) - } - if shouldSkipMigrations(current) { - return nil + ranMigrate := false + if !skipMigrate { + // Double-check under the lock: another process may have completed + // the migration while this one was waiting on the lock. + current, err = s.readUserVersion() + if err != nil { + return fmt.Errorf("engram: re-read user_version: %w", err) + } + if !shouldSkipMigrations(current) { + migrateRunCount.Add(1) + if err := s.migrate(); err != nil { + return fmt.Errorf("engram: migration: %w", err) + } + ranMigrate = true + } } - migrateRunCount.Add(1) - if err := s.migrate(); err != nil { - return fmt.Errorf("engram: migration: %w", err) - } if withRepair { if err := s.repairEnrolledProjectSyncMutations(); err != nil { return fmt.Errorf("engram: repair enrolled sync journal: %w", err) } } + // Stamp only after the full block succeeded so a failed migration or // repair is retried on the next startup (migrate() is idempotent). - if err := s.setUserVersion(schemaVersion); err != nil { - return fmt.Errorf("engram: set user_version: %w", err) + if ranMigrate { + if err := s.setUserVersion(schemaVersion); err != nil { + return fmt.Errorf("engram: set user_version: %w", err) + } } return nil } From a5cd6d843a0dd06df11c875345ac9183f318d2a8 Mon Sep 17 00:00:00 2001 From: jarl9804 Date: Wed, 15 Jul 2026 12:02:27 +0200 Subject: [PATCH 5/5] test(store): cover connection replacement and lock timeout diagnostics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - TestConnectionReplacementKeepsConfiguration: plants a session-scoped cache_size marker, interrupts a long query via context timeout so the pool discards the poisoned connection, proves the physical connection was replaced (marker gone), and asserts the replacement is fully configured: busy_timeout 5000, journal_mode wal, foreign_keys on, and persist-WAL mode 1 via the file-control query — then writes, closes, and checks the -wal file still exists. - TestAcquireMigrationLockTimesOutWithDiagnostic: with a shortened timeout, a second acquire against a held lock fails after the budget with an error naming the lock file and a stuck engram process. --- internal/store/startup_gate_test.go | 146 ++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) diff --git a/internal/store/startup_gate_test.go b/internal/store/startup_gate_test.go index 32ec5bc6..aa65d8ee 100644 --- a/internal/store/startup_gate_test.go +++ b/internal/store/startup_gate_test.go @@ -11,6 +11,7 @@ package store // stamped by a newer engram. import ( + "context" "database/sql" "fmt" "os" @@ -313,6 +314,151 @@ func TestConcurrentColdStartProcesses(t *testing.T) { } } +// TestConnectionReplacementKeepsConfiguration guards the pool-replacement +// regression: database/sql silently discards a modernc connection after a +// context-cancelled query interrupts it (IsValid/ResetSession fail once +// sqlite3_is_interrupted) and opens a fresh one. Because the pragmas travel +// in the DSN and persist-WAL is applied by a driver connection hook, the +// replacement connection must come up fully configured — busy_timeout 5000 +// (#559) and persistent WAL (#477) intact. +func TestConnectionReplacementKeepsConfiguration(t *testing.T) { + cfg := mustDefaultConfig(t) + cfg.DataDir = t.TempDir() + + s, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + defer s.Close() + + // Plant a session-scoped marker on the current physical connection. + // PRAGMA cache_size is per-connection and not part of the DSN, so its + // disappearance later proves the pool swapped in a new connection. + if _, err := s.db.Exec("PRAGMA cache_size = -12345"); err != nil { + t.Fatalf("set cache_size marker: %v", err) + } + var marker int + if err := s.db.QueryRow("PRAGMA cache_size").Scan(&marker); err != nil { + t.Fatalf("read cache_size marker: %v", err) + } + if marker != -12345 { + t.Fatalf("cache_size marker = %d, want -12345", marker) + } + + // Interrupt a long-running query via context timeout. modernc's + // interruptOnDone calls sqlite3_interrupt, poisoning the connection so + // the pool discards it on release. + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + var n int64 + if err := s.db.QueryRowContext(ctx, + `WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c) SELECT count(*) FROM c`, + ).Scan(&n); err == nil { + t.Fatal("expected the interrupted query to fail, it succeeded") + } + + // The pool must have replaced the physical connection... + var cacheSize int + if err := s.db.QueryRow("PRAGMA cache_size").Scan(&cacheSize); err != nil { + t.Fatalf("query after interruption: %v", err) + } + if cacheSize == -12345 { + t.Fatal("cache_size marker survived — connection was not replaced; test cannot exercise the replacement path") + } + + // ...and the replacement must be fully configured. + var busy int + if err := s.db.QueryRow("PRAGMA busy_timeout").Scan(&busy); err != nil { + t.Fatalf("read busy_timeout: %v", err) + } + if busy != 5000 { + t.Errorf("busy_timeout on replacement connection = %d, want 5000 (#559 regression)", busy) + } + var journalMode string + if err := s.db.QueryRow("PRAGMA journal_mode").Scan(&journalMode); err != nil { + t.Fatalf("read journal_mode: %v", err) + } + if !strings.EqualFold(journalMode, "wal") { + t.Errorf("journal_mode on replacement connection = %q, want wal", journalMode) + } + var foreignKeys int + if err := s.db.QueryRow("PRAGMA foreign_keys").Scan(&foreignKeys); err != nil { + t.Fatalf("read foreign_keys: %v", err) + } + if foreignKeys != 1 { + t.Errorf("foreign_keys on replacement connection = %d, want 1", foreignKeys) + } + + // Persist-WAL must be held on the replacement connection (query mode -1). + conn, err := s.db.Conn(context.Background()) + if err != nil { + t.Fatalf("pin replacement connection: %v", err) + } + if err := conn.Raw(func(driverConn any) error { + fc, ok := driverConn.(interface { + FileControlPersistWAL(string, int) (int, error) + }) + if !ok { + return fmt.Errorf("driver connection %T has no FileControlPersistWAL", driverConn) + } + mode, err := fc.FileControlPersistWAL("main", -1) + if err != nil { + return err + } + if mode != 1 { + return fmt.Errorf("persist-WAL mode on replacement connection = %d, want 1 (#477 regression)", mode) + } + return nil + }); err != nil { + t.Error(err) + } + if err := conn.Close(); err != nil { + t.Fatalf("release pinned connection: %v", err) + } + + // End to end: write through the replacement connection, close, and the + // -wal file must survive. + if err := s.CreateSession("replacement-session", "replacement-project", cfg.DataDir); err != nil { + t.Fatalf("CreateSession on replacement connection: %v", err) + } + if err := s.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + walPath := filepath.Join(cfg.DataDir, "engram.db-wal") + if _, err := os.Stat(walPath); err != nil { + t.Fatalf("-wal file was unlinked on close after connection replacement — persist-WAL was lost: %v", err) + } +} + +func TestAcquireMigrationLockTimesOutWithDiagnostic(t *testing.T) { + origTimeout := migrationLockTimeout + migrationLockTimeout = 300 * time.Millisecond + t.Cleanup(func() { migrationLockTimeout = origTimeout }) + + path := filepath.Join(t.TempDir(), ".migrate.lock") + + unlock, err := acquireMigrationLock(path) + if err != nil { + t.Fatalf("first acquire: %v", err) + } + defer unlock() + + start := time.Now() + _, err = acquireMigrationLock(path) + if err == nil { + t.Fatal("second acquire succeeded while the first lock was held; want timeout error") + } + if elapsed := time.Since(start); elapsed < 300*time.Millisecond { + t.Errorf("timed out after %s, want at least the %s budget", elapsed, 300*time.Millisecond) + } + if !strings.Contains(err.Error(), path) { + t.Errorf("timeout error does not name the lock file: %v", err) + } + if !strings.Contains(err.Error(), "stuck engram process") { + t.Errorf("timeout error does not point at a stuck process: %v", err) + } +} + func TestAcquireMigrationLockExcludes(t *testing.T) { path := filepath.Join(t.TempDir(), ".migrate.lock")