Harden SQLite connection defaults with WAL and busy timeout

backend/app/db.py

@@ -106,6 +106,13 @@ _SCHEMA_STATEMENTS: list[str] = [
 # ---------------------------------------------------------------------------
 
 
+async def _configure_connection(db: aiosqlite.Connection) -> None:
+    """Apply hardening pragmas to a newly-opened SQLite connection."""
+    await db.execute("PRAGMA journal_mode=WAL;")
+    await db.execute("PRAGMA foreign_keys=ON;")
+    await db.execute("PRAGMA busy_timeout=5000;")
+
+
 async def init_db(db: aiosqlite.Connection) -> None:
     """Create all BanGUI application tables if they do not already exist.
 
@@ -117,10 +124,7 @@ async def init_db(db: aiosqlite.Connection) -> None:
         db: An open :class:`aiosqlite.Connection` to the application database.
     """
     log.info("initialising_database_schema")
-    async with db.execute("PRAGMA journal_mode=WAL;"):
-        pass
-    async with db.execute("PRAGMA foreign_keys=ON;"):
-        pass
+    await _configure_connection(db)
     for statement in _SCHEMA_STATEMENTS:
         await db.executescript(statement)
     await db.commit()
@@ -138,5 +142,5 @@ async def open_db(database_path: str) -> aiosqlite.Connection:
     """
     db = await aiosqlite.connect(database_path)
     db.row_factory = aiosqlite.Row
-    await db.execute("PRAGMA foreign_keys=ON;")
+    await _configure_connection(db)
     return db

backend/tests/test_db.py

@@ -0,0 +1,58 @@
+import asyncio
+from pathlib import Path
+
+import aiosqlite
+
+from app.db import open_db
+
+
+async def test_open_db_applies_hardening_pragmas(tmp_path: Path) -> None:
+    database_path = str(tmp_path / "bangui_test.db")
+    db = await open_db(database_path)
+    try:
+        async with db.execute("PRAGMA journal_mode;") as cursor:
+            row = await cursor.fetchone()
+        assert row is not None and row[0].lower() == "wal"
+
+        async with db.execute("PRAGMA foreign_keys;") as cursor:
+            row = await cursor.fetchone()
+        assert row is not None and row[0] == 1
+
+        async with db.execute("PRAGMA busy_timeout;") as cursor:
+            row = await cursor.fetchone()
+        assert row is not None and row[0] == 5000
+    finally:
+        await db.close()
+
+
+async def test_open_db_respects_busy_timeout_for_concurrent_writes(tmp_path: Path) -> None:
+    database_path = str(tmp_path / "bangui_lock.db")
+    connection_a = await open_db(database_path)
+    try:
+        await connection_a.execute(
+            "CREATE TABLE IF NOT EXISTS test_lock (id INTEGER PRIMARY KEY, value TEXT);"
+        )
+        await connection_a.commit()
+
+        await connection_a.execute("BEGIN EXCLUSIVE;")
+
+        async def write_after_lock() -> None:
+            connection_b = await open_db(database_path)
+            try:
+                await connection_b.execute(
+                    "INSERT INTO test_lock (value) VALUES ('locked');"
+                )
+                await connection_b.commit()
+            finally:
+                await connection_b.close()
+
+        task = asyncio.create_task(write_after_lock())
+        await asyncio.sleep(0.1)
+        await connection_a.commit()
+        await task
+
+        async with connection_a.execute("SELECT value FROM test_lock;") as cursor:
+            row = await cursor.fetchone()
+        assert row is not None and row[0] == "locked"
+    finally:
+        await connection_a.close()