fix(store): run SQLite startup migrations inside one transaction (BACKLOG #1586)

messagefoundry/store/store.py

@@ -2591,8 +2591,17 @@ async def open(
             await db.execute("PRAGMA foreign_keys=ON")
             await db.execute("PRAGMA busy_timeout=5000")
             await db.executescript(_SCHEMA)
-            await cls._migrate(db)
-            await db.commit()
+            # BACKLOG #1586: the migrations run in ONE transaction, so an interrupted run leaves no
+            # trace. Outside one, each ALTER ... ADD COLUMN commits on its own, and a failure before
+            # its paired backfill leaves the column present -- the next open's column-missing guard
+            # then skips that backfill for good. The transaction opens exactly here and no earlier:
+            # before the PRAGMAs, foreign_keys=ON is a silent no-op and journal_mode=WAL raises;
+            # before executescript, its implicit COMMIT ends the transaction before _migrate runs.
+            # _writer_txn rolls back on BaseException; the lock is a fresh one because nothing else
+            # can reach this connection yet.
+            async with _writer_txn(db, asyncio.Lock()):
+                await cls._migrate(db)
+                await db.commit()
             # Tighten permissions now that the file (and its WAL siblings) exist — they hold PHI.
             # Off the loop (BACKLOG #1634): three files, each an icacls subprocess on Windows. Open
             # completes before anything is serving, so this one is consistency rather than a fix.
@@ -3771,16 +3780,14 @@ async def _migrate(db: aiosqlite.Connection) -> None:
         # seq/rowid but KEPT the names ix_queue_fifo_in/out with CREATE IF NOT EXISTS — so an upgraded DB
         # silently keeps its old created_at-trailing index and never adopts the seq-only claim's index.
         # Drop the old-named indexes and build the seq-trailing ones under a NEW name (so name-existence is
-        # a correct discriminator). This is NOT a transactional swap on SQLite — Python's sqlite3 auto-
-        # commits DDL — but it does not need to be: the FIFO index is CORRECTNESS-NEUTRAL (the claim orders
-        # by rowid and names no index, ADR 0059), so a crash in the DROP→CREATE gap leaves a lane
-        # transiently unindexed (claims stay correct, just slower) and the next open's idempotent re-run
-        # (DROP IF EXISTS / CREATE IF NOT EXISTS) converges to the seq-trailing pair. This runs at open,
-        # before serving, so the transient gap is never observed by a live claim. DROP-old before
-        # CREATE-new so the on-disk FIFO index count never doubles; a fresh DB no-ops the drops and a
-        # re-opened migrated DB no-ops everything. (The server backends run the same swap inside a real
-        # schema transaction, so they additionally get atomicity — see ADR 0060 / sqlserver.py /
-        # postgres.py.)
+        # a correct discriminator). The swap is atomic now that `open` runs this whole method in one
+        # transaction (BACKLOG #1586), as it already was on the server backends (ADR 0060 /
+        # sqlserver.py / postgres.py). It never depended on that: the FIFO index is CORRECTNESS-NEUTRAL
+        # (the claim orders by rowid and names no index, ADR 0059), so any partial index state still
+        # claims correctly, just slower, and the next open's idempotent re-run (DROP IF EXISTS / CREATE
+        # IF NOT EXISTS) converges to the seq-trailing pair. DROP-old before CREATE-new so the on-disk
+        # FIFO index count never doubles; a fresh DB no-ops the drops and a re-opened migrated DB
+        # no-ops everything.
         await db.execute("DROP INDEX IF EXISTS ix_queue_fifo_in")
         await db.execute("DROP INDEX IF EXISTS ix_queue_fifo_out")
         await db.execute(

tests/test_backlog1586_sqlite_migrate_txn.py

@@ -0,0 +1,266 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+# Copyright (C) 2026 MessageFoundry Foundation, LLC and contributors
+"""BACKLOG #1586: SQLite runs its startup migrations inside ONE transaction.
+
+Outside a transaction, every ``ALTER TABLE ... ADD COLUMN`` in ``MessageStore._migrate`` commits on
+its own. Two of them are paired with a one-time backfill (``users.password_claimed_at``, #1245, and
+``users.notify_email``, #1139), and each backfill sits inside a column-missing guard. So a startup
+that stops between the ADD and its backfill leaves the column present, and every later open skips
+the backfill for good: the claim stamp and the notification address stay NULL on every existing row.
+
+The fix is one ``BEGIN`` in ``MessageStore.open``, and its PLACEMENT is the whole fix, so these
+tests pin the placement rather than the happy path. Each wrong spot fails at least one of them:
+
+- no ``BEGIN`` at all: the ADD survives the failed open, so both interrupt tests fail;
+- ``BEGIN`` before the PRAGMAs: ``journal_mode=WAL`` raises inside a transaction, so every open fails;
+- ``BEGIN`` between ``synchronous`` and ``foreign_keys``: ``foreign_keys=ON`` is a silent no-op
+  there, which the PRAGMA test catches;
+- ``BEGIN`` before ``executescript(_SCHEMA)``: ``executescript`` COMMITs first, so the migration runs
+  outside the transaction again and the interrupt tests fail.
+
+The legacy database is built by hand, not by dropping columns from a fresh store: a users table as
+it stood before any column ``_migrate`` adds, with ``_SCHEMA`` then run over it by the stdlib driver.
+That file is exactly what ``_migrate`` sees on a real upgrade, primary key and all, so a failed open
+must leave its schema and rows exactly as they were.
+"""
+
+from __future__ import annotations
+
+import shutil
+import sqlite3
+import threading
+from contextlib import closing
+from pathlib import Path
+from typing import Any
+
+import aiosqlite
+import pytest
+
+from messagefoundry.store.store import _SCHEMA, MessageStore
+from tests.test_store import _watch_aiosqlite_connects
+
+#: ``users`` before any column ``_migrate`` adds. A frozen historical shape on purpose: the test in
+#: this file that reaches the current shape checks the two converge, so drift reds rather than hides.
+_LEGACY_USERS = """
+CREATE TABLE users (
+    id                   TEXT PRIMARY KEY,
+    username             TEXT NOT NULL UNIQUE,
+    auth_provider        TEXT NOT NULL,
+    display_name         TEXT,
+    email                TEXT,
+    disabled             INTEGER NOT NULL DEFAULT 0,
+    created_at           REAL NOT NULL,
+    updated_at           REAL NOT NULL,
+    last_login_at        REAL,
+    password_hash        TEXT,
+    password_changed_at  REAL,
+    must_change_password INTEGER NOT NULL DEFAULT 0,
+    failed_attempts      INTEGER NOT NULL DEFAULT 0,
+    locked_until         REAL
+);
+"""
+
+#: One row per backfill outcome: alice gets both, bob neither, carol only the address.
+_LEGACY_ROWS = [
+    ("u1", "alice", "local", "a@example.org", "h1", 1000.0, 0),
+    ("u2", "bob", "local", None, "h2", 2000.0, 1),
+    ("u3", "carol", "ad", "c@corp.example", None, None, 0),
+]
+
+#: What the two backfills must produce on those rows, by id: (password_claimed_at, notify_email).
+_EXPECTED = {
+    "u1": (1000.0, "a@example.org"),
+    "u2": (None, None),
+    "u3": (None, "c@corp.example"),
+}
+
+_BACKFILLED = ("password_claimed_at", "notify_email")
+
+
+class _Interrupted(Exception):  # noqa: N818 -- stands in for a stop, it is not an error condition
+    """Whatever stops a startup between a column's ADD and the statement after it."""
+
+
+def _normal(sql: str) -> str:
+    return " ".join(sql.split())
+
+
+def _add_column(column: str) -> str:
+    return f"ALTER TABLE users ADD COLUMN {column}"
+
+
+def _build_legacy(path: Path) -> None:
+    with closing(sqlite3.connect(path)) as con:
+        con.executescript(_LEGACY_USERS)
+        con.executemany(
+            "INSERT INTO users (id, username, auth_provider, email, password_hash,"
+            " password_changed_at, must_change_password, created_at, updated_at)"
+            " VALUES (?,?,?,?,?,?,?,1.0,1.0)",
+            _LEGACY_ROWS,
+        )
+        con.commit()
+        con.executescript(_SCHEMA)
+        cols = {row[1] for row in con.execute("PRAGMA table_info(users)")}
+        # Positive control: _SCHEMA's CREATE TABLE IF NOT EXISTS must not have widened the table.
+        assert not cols & set(_BACKFILLED), cols
+
+
+def _dump(path: Path) -> tuple[list[tuple[Any, ...]], list[tuple[Any, ...]]]:
+    """The file's whole schema and its users rows, read by the stdlib driver."""
+    with closing(sqlite3.connect(path)) as con:
+        schema = con.execute(
+            "SELECT type, name, tbl_name, sql FROM sqlite_master ORDER BY type, name"
+        ).fetchall()
+        return schema, con.execute("SELECT * FROM users ORDER BY id").fetchall()
+
+
+def _users(path: Path) -> list[dict[str, Any]]:
+    with closing(sqlite3.connect(path)) as con:
+        con.row_factory = sqlite3.Row
+        return [dict(row) for row in con.execute("SELECT * FROM users ORDER BY id")]
+
+
+def _users_columns(path: Path) -> list[str]:
+    with closing(sqlite3.connect(path)) as con:
+        return [row[1] for row in con.execute("PRAGMA table_info(users)")]
+
+
+def _backfilled(rows: list[dict[str, Any]]) -> dict[str, tuple[Any, ...]]:
+    return {row["id"]: tuple(row[c] for c in _BACKFILLED) for row in rows}
+
+
+def _interrupt_after(m: pytest.MonkeyPatch, prefix: str) -> list[str]:
+    """Raise :class:`_Interrupted` right after the first statement starting with ``prefix`` runs.
+
+    The statement itself completes first, so what is under test is exactly the gap the row names:
+    the ADD has happened and the statement after it has not. Every other call is passed through
+    untouched, so ``async with db.execute(...)`` still works on the ones that do not match.
+    """
+    real = aiosqlite.Connection.execute
+    fired: list[str] = []
+
+    def execute(self: aiosqlite.Connection, sql: str, parameters: Any = None) -> Any:
+        result = real(self, sql, parameters)
+        if fired or not _normal(sql).startswith(prefix):
+            return result
+
+        async def interrupted() -> None:
+            await result
+            fired.append(sql)
+            raise _Interrupted(sql)
+
+        return interrupted()
+
+    m.setattr(aiosqlite.Connection, "execute", execute)
+    return fired
+
+
+async def _open_and_close(path: Path) -> None:
+    store = await MessageStore.open(path)
+    await store.close()
+
+
+async def _clean_migration(tmp_path: Path, legacy: Path) -> list[dict[str, Any]]:
+    clean = tmp_path / "clean.db"
+    shutil.copyfile(legacy, clean)
+    await _open_and_close(clean)
+    return _users(clean)
+
+
+@pytest.fixture
+def legacy(tmp_path: Path) -> Path:
+    path = tmp_path / "legacy.db"
+    _build_legacy(path)
+    return path
+
+
+@pytest.mark.parametrize("column", _BACKFILLED)
+async def test_an_interrupted_backfill_heals_to_the_clean_migration(
+    tmp_path: Path, legacy: Path, column: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The row's acceptance: interrupt after the ADD, reopen twice, match a clean migration."""
+    clean = await _clean_migration(tmp_path, legacy)
+    # Controls on the reference itself. A clean run that skipped a backfill would make the equality
+    # below pass on two identically wrong files, and a legacy shape that no longer converges on the
+    # current one would make this test about a database nobody can have.
+    assert _backfilled(clean) == _EXPECTED
+    fresh = tmp_path / "fresh.db"
+    await _open_and_close(fresh)
+    assert set(clean[0]) == set(_users_columns(fresh))
+
+    case = tmp_path / f"interrupted-{column}.db"
+    shutil.copyfile(legacy, case)
+    with monkeypatch.context() as m:
+        watched: set[threading.Thread] = set()
+        opened = _watch_aiosqlite_connects(m, watched)
+        fired = _interrupt_after(m, _add_column(column))
+        with pytest.raises(_Interrupted):
+            await MessageStore.open(case)
+    assert fired, f"the interrupt never fired; no statement began {_add_column(column)!r}"
+    # The #1670 cleanup still runs on this path: the writer is the only connection a failed
+    # migration has opened, and its worker thread is gone by the time open() has raised.
+    assert len(opened) == 1, opened
+    assert not [t.name for t in watched if t.is_alive()], "a failed open left a live worker"
+    # The rollback is what makes the reopen below a real migration rather than a skip.
+    assert _dump(case) == _dump(legacy), "the interrupted migration left changes on disk"
+
+    for reopen in (1, 2):
+        await _open_and_close(case)
+        assert _users(case) == clean, f"reopen {reopen} does not match a clean migration"
+
+
+async def test_an_interrupt_after_any_column_addition_leaves_the_file_as_it_was(
+    tmp_path: Path, legacy: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Every ADD the migration issues on this file, not only the two with a backfill.
+
+    The additions are DISCOVERED from a clean run rather than listed, so a column added to
+    ``_migrate`` later is covered without anyone editing this test.
+    """
+    seen: list[str] = []
+    real = aiosqlite.Connection.execute
+
+    def record(self: aiosqlite.Connection, sql: str, parameters: Any = None) -> Any:
+        text = _normal(sql)
+        if text.startswith("ALTER TABLE") and " ADD COLUMN " in text:
+            seen.append(text)
+        return real(self, sql, parameters)
+
+    with monkeypatch.context() as m:
+        m.setattr(aiosqlite.Connection, "execute", record)
+        await _clean_migration(tmp_path, legacy)
+    # Liveness: a recorder that saw nothing would make the loop below pass by never running.
+    for column in _BACKFILLED:
+        assert any(s.startswith(_add_column(column)) for s in seen), (column, seen)
+    assert len(seen) > len(_BACKFILLED), seen
+
+    before = _dump(legacy)
+    for index, statement in enumerate(seen):
+        case = tmp_path / f"add-{index}.db"
+        shutil.copyfile(legacy, case)
+        with monkeypatch.context() as m:
+            fired = _interrupt_after(m, statement)
+            with pytest.raises(_Interrupted):
+                await MessageStore.open(case)
+        assert [_normal(sql) for sql in fired] == [statement]
+        assert _dump(case) == before, f"interrupted after {statement!r}, the file changed"
+
+
+async def test_the_connection_pragmas_hold_after_the_migration_transaction(
+    tmp_path: Path,
+) -> None:
+    """``foreign_keys=ON`` is a silent no-op inside a transaction; nothing else would notice."""
+    store = await MessageStore.open(tmp_path / "pragmas.db")
+    try:
+        assert await _pragma(store, "foreign_keys") == 1
+        assert await _pragma(store, "journal_mode") == "wal"
+        assert not store._db.in_transaction, "open() returned with its transaction still open"
+    finally:
+        await store.close()
+
+
+async def _pragma(store: MessageStore, name: str) -> Any:
+    cur = await store._db.execute(f"PRAGMA {name}")
+    row = await cur.fetchone()
+    assert row is not None
+    return row[0]

tests/test_fifo_index_migration.py

@@ -341,8 +341,9 @@ async def test_idempotent_reopen_is_stable(store: Any) -> None:
 
 
 async def test_partial_migration_converges_and_stays_correct(tmp_path: Path) -> None:
-    """SQLite auto-commits DDL (legacy isolation), so a crash mid-migration can leave a PARTIAL index
-    state (some old dropped, some new created). B10 does NOT rely on atomicity here — the FIFO index is
+    """A PARTIAL index state (some old dropped, some new created) must stay safe. The swap has run
+    inside open's migration transaction since BACKLOG #1586, but a store written before that, or edited
+    by hand, can still hold one. B10 does NOT rely on atomicity here — the FIFO index is
     correctness-neutral and the migration is idempotent — so ANY partial state (a) still claims in strict
     seq order and (b) converges to the seq-trailing pair on the next open. Construct a representative
     mid-crash partial state (old-in dropped + new-in created, but old-out still present + new-out missing)