From 7a343be36a3d239e39a129cee8136f9b8648aea0 Mon Sep 17 00:00:00 2001 From: Roland Date: Wed, 22 Jul 2026 12:07:26 -0600 Subject: [PATCH] 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) --- wrolpi/db.py | 33 ++++++++++++++++++- wrolpi/downloader.py | 10 ++++-- wrolpi/test/test_db.py | 73 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 111 insertions(+), 5 deletions(-) diff --git a/wrolpi/db.py b/wrolpi/db.py index a45b3a40..33c078ce 100644 --- a/wrolpi/db.py +++ b/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]: """ diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index e44569c4..7327b82a 100644 --- a/wrolpi/downloader.py +++ b/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, diff --git a/wrolpi/test/test_db.py b/wrolpi/test/test_db.py index 1031e1ab..aa353c2e 100644 --- a/wrolpi/test/test_db.py +++ b/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'