downloader: fix "database is locked" on read-then-write dispatch

The download dispatcher (_dispatch_new_downloads) reads the `new` downloads
then claims each by writing last_download_attempt, all in one ORM
transaction. Since ORM transactions begin *deferred*, that write is a lock
upgrade (SHARED->RESERVED). SQLite skips busy_timeout for lock upgrades to
avoid deadlock, so under concurrency the write failed instantly with
"database is locked" instead of waiting out the 30s busy_timeout. This
surfaced repeatedly on the SQLite build after the Postgres->SQLite cutover.

Add get_immediate_db_session(), a committing session that takes the write
lock up front (BEGIN IMMEDIATE) so busy_timeout absorbs the contention --
mirroring what the raw-cursor get_db_curs path already does. Plain
get_db_session stays deferred so ordinary writes and reads keep WAL
read-concurrency; only read-then-write call sites opt in. Convert the
dispatcher to use it.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

wrolpi/db.py

@@ -25,6 +25,10 @@ from wrolpi.vars import PYTEST
 
 logger = logger.getChild(__name__)
 
+# Toggled only by `get_immediate_db_session()` (below) while such a session is open on this thread,
+# so the engine's `begin` listener knows to emit `BEGIN IMMEDIATE` instead of a deferred `BEGIN`.
+_immediate_txn = threading.local()
+
 
 def _adapt_datetime(value):
     """Store datetimes from raw SQL exactly like SQLAlchemy does: naive UTC with microseconds."""
@@ -82,7 +86,12 @@ def create_wrolpi_engine(target: Union[str, pathlib.Path]) -> sqlalchemy.engine.
 
     @event.listens_for(engine, 'begin')
     def _sqlite_do_begin(conn):
-        conn.execute('BEGIN')
+        # `get_immediate_db_session()` sessions take the write lock up front; everything else stays
+        # deferred so WAL readers never block (see `_immediate_txn`).
+        if getattr(_immediate_txn, 'active', False):
+            conn.execute('BEGIN IMMEDIATE')
+        else:
+            conn.execute('BEGIN')
 
     return engine
 
@@ -150,6 +159,28 @@ def get_db_session(commit: bool = False) -> Generator[Session, Any, None]:
             session.rollback()
 
 
+@contextmanager
+def get_immediate_db_session() -> Generator[Session, Any, None]:
+    """A committing session that takes the SQLite write lock up front (`BEGIN IMMEDIATE`).
+
+    Use this for read-then-write transactions (read some rows, then UPDATE/INSERT them).  A plain
+    `get_db_session(commit=True)` begins *deferred* and only upgrades to a writer at its first
+    write; if another connection holds the write lock at that moment, SQLite returns "database is
+    locked" *immediately* — `busy_timeout` is skipped for lock upgrades to avoid deadlock.  Taking
+    the write lock at BEGIN instead lets `busy_timeout` (30s) absorb the contention.
+
+    The `_immediate_txn` flag is read by the engine's `begin` listener when the transaction opens
+    (on the caller's first statement), so keep queries inside this `with` block.
+    """
+    previous = getattr(_immediate_txn, 'active', False)
+    _immediate_txn.active = True
+    try:
+        with get_db_session(commit=True) as session:
+            yield session
+    finally:
+        _immediate_txn.active = previous
+
+
 @contextmanager
 def get_db_curs(commit: bool = False) -> Generator[sqlite3.Cursor, Any, None]:
     """

wrolpi/downloader.py

@@ -35,7 +35,7 @@ from wrolpi.common import Base, ModelHelper, logger, wrol_mode_check, zig_zag, C
     wrol_mode_enabled, background_task, get_absolute_media_path, timer, aiohttp_get, \
     get_download_info, trim_file_name, get_wrolpi_config, TRACE_LEVEL, normalize_domain
 from wrolpi.dates import TZDateTime, now, Seconds
-from wrolpi.db import get_db_session, get_db_curs
+from wrolpi.db import get_db_session, get_db_curs, get_immediate_db_session
 from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \
     ValidationError, DownloadError
 from wrolpi.events import Events
@@ -1077,8 +1077,12 @@ class DownloadManager:
     async def _dispatch_new_downloads(self):
         """Claim and dispatch the next batch of `new` downloads.  Split out from `dispatch_downloads`
         so a transient "database is locked" can be handled there without retrying the signal
-        dispatches inside this transaction."""
-        with get_db_session(commit=True) as session:
+        dispatches inside this transaction.
+
+        Uses `get_immediate_db_session` because this is a read-then-write transaction (it reads the
+        `new` downloads, then claims each by setting `last_download_attempt`); a deferred lock
+        upgrade here fails instantly with "database is locked" under concurrency."""
+        with get_immediate_db_session() as session:
             new_downloads = list(session.query(Download).filter(
                 Download.status == 'new',
                 Download.domain not in self.processing_domains,

wrolpi/test/test_db.py

@@ -1 +1,72 @@
-# Placeholder test file - optional_session tests removed since decorator is deprecated.
+import sqlite3
+
+from wrolpi.conftest import production_like_sessions
+from wrolpi.db import get_db_session, get_immediate_db_session
+
+
+def _probe_write_lock_is_held(db_file: str) -> bool:
+    """Return True if a competing connection cannot immediately take the write lock.
+
+    Uses a zero busy-timeout so the probe fails instantly if another connection holds a
+    RESERVED (write) lock, rather than waiting.  In WAL mode a plain reader holds no such
+    lock, so this only returns True when a writer transaction is actually in progress.
+    """
+    probe = sqlite3.connect(db_file, timeout=0)
+    try:
+        probe.execute('PRAGMA busy_timeout=0')
+        try:
+            probe.execute('BEGIN IMMEDIATE')
+        except sqlite3.OperationalError as e:
+            if 'database is locked' in str(e):
+                return True
+            raise
+        probe.execute('ROLLBACK')
+        return False
+    finally:
+        probe.close()
+
+
+def test_immediate_session_takes_write_lock_up_front(test_session):
+    """`get_immediate_db_session` must acquire the SQLite write lock at BEGIN.
+
+    Regression test for `database is locked` on `UPDATE download SET last_download_attempt`:
+    a deferred (read-then-write) transaction that tries to upgrade its lock while another
+    connection holds the write lock gets SQLITE_BUSY *immediately* — busy_timeout is ignored
+    for lock upgrades to avoid deadlock.  Taking the write lock up front (BEGIN IMMEDIATE)
+    lets busy_timeout absorb the contention instead of erroring instantly.
+    """
+    db_file = test_session.get_bind().url.database
+
+    with production_like_sessions(test_session):
+        with get_immediate_db_session() as session:
+            # Trigger the transaction's BEGIN with a trivial read (as the download dispatcher
+            # does before writing).  This session already holds the write lock.
+            session.execute('SELECT 1')
+            assert _probe_write_lock_is_held(db_file), \
+                'immediate session did not hold the write lock after BEGIN'
+
+
+def test_plain_write_session_stays_deferred(test_session):
+    """A plain `get_db_session(commit=True)` stays deferred (no write lock until it actually writes).
+
+    Only `get_immediate_db_session` opts into the up-front write lock, so ordinary write sessions
+    keep WAL read-concurrency and don't serialize behind each other at BEGIN.
+    """
+    db_file = test_session.get_bind().url.database
+
+    with production_like_sessions(test_session):
+        with get_db_session(commit=True) as session:
+            session.execute('SELECT 1')
+            assert not _probe_write_lock_is_held(db_file), \
+                'plain write session unexpectedly holds the write lock before writing'
+
+
+def test_read_session_does_not_take_write_lock(test_session):
+    """A read-only `get_db_session()` must stay deferred so WAL read-concurrency is preserved."""
+    db_file = test_session.get_bind().url.database
+
+    with production_like_sessions(test_session):
+        with get_db_session() as session:
+            session.execute('SELECT 1')
+            assert not _probe_write_lock_is_held(db_file), \
+                'read-only session unexpectedly holds the write lock'