diff --git a/CHANGELOG.md b/CHANGELOG.md index ff0edc8eb..ae66af3ba 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/framework/db/simple_module_db/session.py b/framework/db/simple_module_db/session.py index 735765b7b..cfaa2d0ff 100644 --- a/framework/db/simple_module_db/session.py +++ b/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() diff --git a/framework/db/tests/test_db_sqlite_wal_switch.py b/framework/db/tests/test_db_sqlite_wal_switch.py new file mode 100644 index 000000000..60291fef3 --- /dev/null +++ b/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()