[PATCH RFC v2 07/25] review: serialize schema migrations against a concurrent opener

_migrate_db_if_needed() reads the schema version and then issues DDL in
autocommit, so two processes opening the same database both decide to
migrate.  The loser dies on `duplicate column name: message_count`, or,
having read the version before the winner's DROP COLUMN landed, on
`no such column: message_count`.  Reproduced 6/6 with two threads
opening one v10 database.

busy_timeout, which _configure_conn sets for exactly this pair of
writers, does not help: neither side ever asks for a lock.

The TUI and a `b4 review cron` sweep are that pair.  The race has been
latent since the v8 migrations because nothing has needed migrating
since; the next schema bump makes it fire, once, on the first launch
after an upgrade, for every maintainer with the timer installed.  It does
not corrupt anything, since the loser's work rolls back and the winner
completes, but it surfaces as a raw sqlite traceback out of get_db(), and
update_revision_message_counts() catches only FileNotFoundError.

Take the write lock before re-reading the version, so the second process
finds the work already done.  sqlite's DDL is transactional, so this also
makes the migration atomic: the version bump can no longer commit
separately from the schema it describes, and an interrupted migration
rolls back whole rather than leaving a half-migrated database stamped
with the old version.

The version is still read once without the lock first.  The answer is "no
migration pending" on every open but the one after an upgrade, and that
path must not serialize every connection behind a write lock.

Signed-off-by: Christian Brauner (Amutable) <[email protected]>
---
 src/b4/review/tracking.py | 44 ++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 42 insertions(+), 2 deletions(-)

diff --git a/src/b4/review/tracking.py b/src/b4/review/tracking.py
index 2722e6c3..cecfdb9c 100644
--- a/src/b4/review/tracking.py
+++ b/src/b4/review/tracking.py
@@ -135,10 +135,45 @@ def init_db(identifier: str) -> sqlite3.Connection:
 
 
 def _migrate_db_if_needed(conn: sqlite3.Connection) -> None:
-    """Apply any pending schema migrations in-place."""
+    """Apply any pending schema migrations in-place.
+
+    Serialized against other processes, because the TUI and a
+    ``b4 review cron`` sweep open the same database and a pending
+    migration is exactly what both of them find on the first run after an
+    upgrade.  Reading the version and then issuing DDL in autocommit let
+    both decide to migrate: the loser died on `duplicate column name`, or
+    -- having read the version before the winner's DROP landed -- on
+    `no such column`.  busy_timeout cannot help, since neither side ever
+    asked for a lock.
+
+    BEGIN IMMEDIATE takes the write lock before the version is re-read, so
+    the second process finds the work already done.  It also makes the
+    whole migration one transaction -- sqlite's DDL is transactional -- so
+    the version bump can no longer commit separately from the schema it
+    describes, and an interrupted migration rolls back whole.
+
+    The version is read once without the lock first: the answer is "no" on
+    every open but the one after an upgrade, and that path must not
+    serialize every connection behind a write lock.
+    """
+    row = conn.execute('SELECT version FROM schema_version').fetchone()
+    if row is not None and row[0] >= SCHEMA_VERSION:
+        return
+    conn.execute('BEGIN IMMEDIATE')
+    try:
+        _run_migrations(conn)
+    except Exception:
+        conn.rollback()
+        raise
+
+
+def _run_migrations(conn: sqlite3.Connection) -> None:
+    """The migration ladder, under the write lock :func:`_migrate_db_if_needed` took."""
     row = conn.execute('SELECT version FROM schema_version').fetchone()
     version = row[0] if row else 0
     if version >= SCHEMA_VERSION:
+        # Another process migrated while we waited for the lock.
+        conn.rollback()
         return
     if version < 2:
         conn.execute('ALTER TABLE series ADD COLUMN branch_sha TEXT')
@@ -198,7 +233,12 @@ def _migrate_db_if_needed(conn: sqlite3.Connection) -> None:
             conn.execute(
                 'ALTER TABLE revisions ADD COLUMN is_rethreaded INTEGER DEFAULT 0'
             )
-    conn.execute('UPDATE schema_version SET version = ?', (SCHEMA_VERSION,))
+    # Not an UPDATE: `version` is the primary key, so an UPDATE writes
+    # nothing at all against an empty table -- and the read above maps "no
+    # row" to version 0, so such a database would re-run the whole ladder
+    # on every open and never record that it had finished.
+    conn.execute('DELETE FROM schema_version')
+    conn.execute('INSERT INTO schema_version (version) VALUES (?)', (SCHEMA_VERSION,))
     conn.commit()
 
 

-- 
2.53.0