fix(db): tolerate a concurrent WAL switch on a fresh SQLite file

CHANGELOG.md

@@ -56,6 +56,11 @@ All notable changes to this project are documented in this file. The format is b
   most 32 characters now fail at registration (#707).
 
 ### Fixed
+- Two connections opening a fresh SQLite file at the same time no longer fail
+  with "database is locked": switching the file to WAL needs an exclusive lock
+  that SQLite refuses at once instead of waiting out `busy_timeout`. A
+  connection now skips the switch when the file is already in WAL, and retries a
+  lost switch until `busy_timeout` runs out.
 - `file_storage_stored_file.size_bytes` is now `BIGINT` (migration `f5b1c3e8a207`,
   which also gives `file_storage` its own branch label): on PostgreSQL it was a
   32-bit `INTEGER`, so a file of 2 GiB or more failed on insert (#708).

framework/db/simple_module_db/session.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import sqlite3
+import time
 from collections.abc import Callable
 from dataclasses import dataclass, field
 
@@ -153,6 +155,34 @@ def _is_file_database(database_url: str) -> bool:
     return bool(path) and path != ":memory:"
 
 
+def _ensure_wal(cursor, timeout_ms: int = 5000) -> None:
+    """Switch the file to WAL unless it already is.
+
+    Switching needs an exclusive lock, and SQLite fails it with "database is
+    locked" at once, without waiting out ``busy_timeout``, when another
+    connection holds a lock. That happens when two connections open a fresh
+    file together. So read the mode first (WAL persists in the file, so later
+    connections skip the switch), and retry a lost switch until the winner has
+    finished or ``timeout_ms`` has passed. On aiosqlite the sleep blocks the
+    event loop, but only while a fresh file is being switched.
+    """
+    deadline = time.monotonic() + timeout_ms / 1000
+    while _journal_mode(cursor) != "wal":
+        try:
+            cursor.execute("PRAGMA journal_mode=WAL")
+            return
+        except sqlite3.OperationalError:
+            if time.monotonic() >= deadline:
+                raise
+            time.sleep(0.02)
+
+
+def _journal_mode(cursor) -> str:
+    # The aiosqlite adapter's ``execute`` returns None, so fetch from the cursor.
+    cursor.execute("PRAGMA journal_mode")
+    return cursor.fetchone()[0].lower()
+
+
 def _configure_sqlite(
     engine: AsyncEngine,
     database_url: str,
@@ -166,7 +196,7 @@ def _configure_sqlite(
     Registered on the *sync* engine's ``connect`` event — the async engine is a
     facade, and ``connect`` fires on the DBAPI connection underneath it. All
     three PRAGMAs must be re-issued per connection except ``journal_mode``,
-    which is a property of the file; re-issuing it is cheap and idempotent.
+    which is a property of the file (see :func:`_ensure_wal`).
 
     Also registers a ``savepoint`` listener that opens the transaction before an
     outermost SAVEPOINT (pysqlite would otherwise let its RELEASE commit; GH #350).
@@ -182,7 +212,7 @@ def _set_sqlite_pragmas(dbapi_connection, _connection_record) -> None:
             if busy_timeout_ms is not None:
                 cursor.execute(f"PRAGMA busy_timeout={int(busy_timeout_ms)}")
             if apply_wal:
-                cursor.execute("PRAGMA journal_mode=WAL")
+                _ensure_wal(cursor, busy_timeout_ms or 0)
         finally:
             cursor.close()
 

framework/db/tests/test_db_sqlite_wal_switch.py

@@ -0,0 +1,87 @@
+"""The per-connection WAL switch must tolerate a concurrent opener (fresh file)."""
+
+from __future__ import annotations
+
+import sqlite3
+import threading
+
+import pytest
+from simple_module_db.session import _ensure_wal
+
+
+class _Cursor:
+    """A cursor whose switch loses the race; the winner leaves ``after``."""
+
+    def __init__(self, before: str, after: str) -> None:
+        self.mode = before
+        self.after = after
+        self.switches = 0
+
+    def execute(self, sql: str):
+        if sql == "PRAGMA journal_mode=WAL":
+            self.switches += 1
+            self.mode = self.after
+            raise sqlite3.OperationalError("database is locked")
+        return self
+
+    def fetchone(self):
+        return (self.mode,)
+
+
+def test_already_wal_skips_the_switch():
+    cursor = _Cursor("wal", "wal")
+    _ensure_wal(cursor)
+    assert cursor.switches == 0
+
+
+def test_losing_the_switch_to_another_connection_is_fine():
+    cursor = _Cursor("delete", "wal")
+    _ensure_wal(cursor)
+    assert cursor.switches == 1
+
+
+def test_a_switch_still_in_progress_is_retried():
+    cursor = _Cursor("delete", "delete")
+    original = cursor.execute
+
+    def execute(sql):
+        if sql == "PRAGMA journal_mode=WAL" and cursor.switches == 2:
+            cursor.mode = "wal"
+            return cursor
+        return original(sql)
+
+    cursor.execute = execute
+    _ensure_wal(cursor)
+    assert (cursor.switches, cursor.mode) == (2, "wal")
+
+
+def test_a_switch_that_leaves_the_file_unjournaled_still_raises():
+    with pytest.raises(sqlite3.OperationalError, match="locked"):
+        _ensure_wal(_Cursor("delete", "delete"), timeout_ms=50)
+
+
+def test_fresh_file_waits_for_a_concurrent_switch(tmp_path):
+    """The old one-shot switch failed here at once: the other opener held a lock."""
+    path = tmp_path / "app.db"
+    other = sqlite3.connect(path, check_same_thread=False)
+    other.execute("CREATE TABLE t (x)")
+    other.commit()
+    other.execute("BEGIN IMMEDIATE")
+    other.execute("INSERT INTO t VALUES (1)")
+
+    def finish_and_switch():
+        other.commit()
+        other.execute("PRAGMA journal_mode=WAL")
+
+    timer = threading.Timer(0.2, finish_and_switch)
+    opener = sqlite3.connect(path, timeout=0)
+    try:
+        with pytest.raises(sqlite3.OperationalError, match="locked"):
+            opener.execute("PRAGMA journal_mode=WAL")
+        timer.start()
+        _ensure_wal(opener.cursor(), timeout_ms=5000)
+        assert opener.execute("PRAGMA journal_mode").fetchone()[0] == "wal"
+    finally:
+        timer.join()
+        opener.close()
+        other.close()