From daf2ba39d4c6246c742e54ee6dd86d4593f9f2d2 Mon Sep 17 00:00:00 2001 From: Roland Date: Fri, 7 Aug 2026 19:40:41 -0600 Subject: [PATCH] db: write intent is the opt-in, not a second session helper Three "database is locked" failures on the docker WROLPi, one cause each: * `INSERT INTO download` from POST /api/download -- a user creating a video download while a refresh ran (2026-08-06 14:06:46); * six `INSERT INTO file_group` from `upsert_file`, uploads landing in the same three-minute window; * `UPDATE download SET status` from `renew_recurring_downloads`. All three are the deferred read-then-write trap: a transaction that begins as a reader and upgrades at its first write is refused *instantly* when another connection wrote in between, because busy_timeout is skipped for lock upgrades. `get_immediate_db_session` existed to avoid it, but an opt-in helper only works if every author remembers, and read-then-write is the ordinary shape of a write. So it is gone: `get_db_session(commit=True)` -- the existing declaration of write intent -- now begins IMMEDIATE. Rails 7.1 and Django 5.1 made the same call, defaulting SQLite transactions to IMMEDIATE rather than asking application code to opt in. That default is only safe if write transactions stay short, and one did not. `video_cleanup` ran its whole Channel-claiming join -- `UPDATE video ... FROM channel ... LEFT JOIN file_group`, O(channels x file_groups) -- inside one write transaction: 110 to 130 seconds each of the eight times it appears in the log, on a library with ~1500 channels. While it ran, *every* writer on the box failed, including ones that began IMMEDIATE, because the hold far exceeds the 30s busy_timeout. The 14:04-14:06 failures above sit inside its 130-second window. The join is now a read -- WAL readers never block writers -- and only the resulting UPDATEs take the lock, 500 rows per transaction. `claim_videos_for_channels` had the same join and the same fix; channel import calls it with every channel in channels.yaml. Also here: * `renew_recurring_downloads` was a read session with an inner commit; it is a write session now. Its computed `next_download` values now persist even when nothing was renewed -- that is the point of computing them. * POST /api/download no longer writes through `request.ctx.session`. That session is committed by the response middleware, so a writer would hold the lock across the whole request; a write session commits before the handler returns, ahead of the background dispatch it fires. * `_immediate_txn` is a ContextVar, not a threading.local. The writers that set it include coroutines, and one asyncio thread runs many of them interleaved. * `write_lock_held_briefly` moved from the move tests into conftest, since two suites now need it. Both regression tests were verified by mutation: reverting either site alone fails its own test, and the API one fails with the production error verbatim. Sources: fractaledmind.com/2023/12/11/sqlite-on-rails-improving-concurrency, tenthousandmeters.com/blog/sqlite-concurrent-writes-and-database-is-locked-errors, berthub.eu/articles/posts/a-brief-post-on-sqlite3-database-locked-despite-timeout Co-Authored-By: Claude Opus 5 (1M context) --- modules/videos/__init__.py | 116 +++++++++++++----- .../test/test_channel_directory_scenarios.py | 71 +++++++++++ wrolpi/api_utils.py | 9 +- wrolpi/conftest.py | 31 +++++ wrolpi/db.py | 55 ++++----- wrolpi/downloader.py | 15 ++- wrolpi/files/test/test_worker_move.py | 35 +----- wrolpi/files/worker.py | 6 +- wrolpi/root_api.py | 18 ++- wrolpi/test/test_db.py | 32 ++--- wrolpi/test/test_downloader.py | 2 +- wrolpi/test/test_root_api.py | 32 +++++ 12 files changed, 293 insertions(+), 129 deletions(-) diff --git a/modules/videos/__init__.py b/modules/videos/__init__.py index fd71e210..42f778d2 100644 --- a/modules/videos/__init__.py +++ b/modules/videos/__init__.py @@ -76,38 +76,98 @@ async def video_modeler(progress_callback: Callable[[int], None] = None): await asyncio.sleep(0) +# Rows written per transaction when claiming Videos for their Channels. Small enough that the +# write lock is held for milliseconds at a time; large enough that a big library does not pay a +# transaction per row. +CLAIM_CHUNK_SIZE = 500 + + +def _chunks(items: list, size: int = CLAIM_CHUNK_SIZE): + for start in range(0, len(items), size): + yield items[start:start + size] + + +def _first_claim_per_video(rows) -> List[Tuple[int, int]]: + """One `(video_id, channel_id)` per Video. + + A Video nested under two Channel directories matches both; the single `UPDATE ... FROM` this + replaced also picked one arbitrarily, so keep the first row and drop the rest rather than + writing the same Video twice.""" + claims = dict() + for video_id, channel_id in rows: + claims.setdefault(video_id, channel_id) + return list(claims.items()) + + +def _claim_videos(claims: List[Tuple[int, int]]): + """Assign `(video_id, channel_id)` pairs, a chunk per write transaction. + + The pairs are computed by a *read* — WAL readers never block writers — and only these short + UPDATEs take the write lock. Doing the whole thing as one `UPDATE ... FROM channel` held the + lock for the entire join: 110-130 seconds per refresh on a library with ~1500 channels, which + is well past the 30s `busy_timeout`, so every other writer on the box failed with "database is + locked" for two solid minutes. + + `channel_id IS NULL` is re-checked in the UPDATE because the read ran in an earlier + transaction: a download may have claimed the Video since, and it should win. + """ + if not claims: + return + claimed = 0 + for chunk in _chunks(claims): + with get_db_curs(commit=True) as curs: + curs.executemany('UPDATE video SET channel_id = ? WHERE id = ? AND channel_id IS NULL', + [(channel_id, video_id) for video_id, channel_id in chunk]) + claimed += len(chunk) + logger.debug(f'Claimed {claimed} Videos for their Channels') + + +# Videos in a Channel's directory (or below it) that no Channel has claimed yet. +_UNCLAIMED_VIDEOS_SQL = ''' + SELECT v.id, c.id + FROM video v + INNER JOIN file_group fg ON fg.id = v.file_group_id + INNER JOIN collection col ON fg.directory = col.directory + OR fg.directory LIKE col.directory || '/%' + INNER JOIN channel c ON c.collection_id = col.id + WHERE v.channel_id IS NULL + ''' + + @register_refresh_cleanup @limit_concurrent(1) def video_cleanup(): logger.info('Claiming Videos for their Channels') - with get_db_curs(commit=True) as curs: - # Delete all Videos if the FileModel no longer contains a video or audio file. + # Read the FileGroups that are no longer video/audio, then unmodel them in short transactions. + with get_db_curs() as curs: curs.execute(''' - UPDATE file_group - SET model = NULL + SELECT id + FROM file_group WHERE model = 'video' AND mimetype NOT LIKE 'video/%' AND mimetype NOT LIKE 'audio/%' - RETURNING id ''') - deleted_ids = [i['id'] for i in curs.fetchall()] - if deleted_ids: + stale_ids = [i['id'] for i in curs.fetchall()] + for chunk in _chunks(stale_ids): + ids = json.dumps(chunk) + with get_db_curs(commit=True) as curs: + curs.execute(''' + UPDATE file_group + SET model = NULL + WHERE id IN (SELECT value FROM json_each(:ids)) + ''', dict(ids=ids)) curs.execute(''' DELETE FROM video WHERE file_group_id IN (SELECT value FROM json_each(:ids)) - ''', dict(ids=json.dumps(deleted_ids))) - # Claim all Videos in a Channel's directory for that Channel. But, only if they have not yet been claimed. - curs.execute(''' - UPDATE video AS v - SET channel_id = c.id - FROM channel c - INNER JOIN collection col ON col.id = c.collection_id - LEFT JOIN file_group fg ON fg.directory = col.directory - OR fg.directory LIKE col.directory || '/%' - WHERE v.channel_id IS NULL - AND fg.id = v.file_group_id - ''') + ''', dict(ids=ids)) + + # Claim all Videos in a Channel's directory for that Channel. But, only if they have not yet + # been claimed. The join is the slow part, so it runs as a read; only the writes take the lock. + with get_db_curs() as curs: + curs.execute(_UNCLAIMED_VIDEOS_SQL) + claims = _first_claim_per_video(curs.fetchall()) + _claim_videos(claims) @@ -126,15 +186,13 @@ def claim_videos_for_channels(channel_ids: List[int]): return logger.info(f'Claiming Videos for {len(channel_ids)} channel(s)') - with get_db_curs(commit=True) as curs: - curs.execute(''' - UPDATE video AS v - SET channel_id = c.id - FROM channel c - INNER JOIN collection col ON col.id = c.collection_id - LEFT JOIN file_group fg ON fg.directory = col.directory - OR fg.directory LIKE col.directory || '/%' - WHERE v.channel_id IS NULL - AND fg.id = v.file_group_id + # Read the claims, then write them in chunks -- the join must not hold the write lock (see + # `_claim_videos`). Channel import calls this with every channel in channels.yaml, so it is + # just as slow as the full `video_cleanup` sweep. + with get_db_curs() as curs: + curs.execute(f''' + {_UNCLAIMED_VIDEOS_SQL} AND c.id IN (SELECT value FROM json_each(:channel_ids)) ''', dict(channel_ids=json.dumps(list(channel_ids)))) + claims = _first_claim_per_video(curs.fetchall()) + _claim_videos(claims) diff --git a/modules/videos/test/test_channel_directory_scenarios.py b/modules/videos/test/test_channel_directory_scenarios.py index 3bfaaad6..073a734c 100644 --- a/modules/videos/test/test_channel_directory_scenarios.py +++ b/modules/videos/test/test_channel_directory_scenarios.py @@ -14,6 +14,9 @@ These tests use the REAL modeling flow: 4. Verify the video is correctly associated with the channel """ +import contextlib +from unittest import mock + import pytest from modules.videos.models import Channel, Video @@ -810,3 +813,71 @@ class TestVideoCleanupWithYearSubdirs: # video_cleanup should claim the video via LIKE clause assert video.channel_id == channel.id, \ f'video_cleanup should claim video in year subdir, got channel_id={video.channel_id}' + + +@pytest.mark.asyncio +async def test_video_cleanup_claims_outside_a_write_transaction(async_client, test_session, test_directory, + video_file, refresh_files): + """The Channel-claiming join must not run inside the write transaction. + + Regression test for two minutes of global write starvation: `video_cleanup` ran the whole + `UPDATE video ... FROM channel ... LEFT JOIN file_group` in one `BEGIN IMMEDIATE` transaction. + The join is O(channels x file_groups), and on a library with ~1500 channels it held the write + lock for 110-130 seconds per refresh -- well past the 30s busy_timeout, so *every* other writer + failed with "database is locked", including correctly-written ones. The join must be a read; + only the resulting UPDATEs may take the lock. + """ + from modules.videos import video_cleanup, _UNCLAIMED_VIDEOS_SQL + from wrolpi.files import worker as worker_module # noqa: F401 (import parity with other tests) + import modules.videos as videos_module + + channel_dir = test_directory / 'videos' / 'LockChannel' + channel_dir.mkdir(parents=True, exist_ok=True) + collection = Collection(name='LockChannel', kind='channel', directory=channel_dir) + test_session.add(collection) + test_session.flush([collection]) + channel = Channel(collection_id=collection.id, url='https://example.com/lockchannel') + test_session.add(channel) + test_session.commit() + + video_file.rename(channel_dir / 'A Video.mp4') + await refresh_files() + + video = test_session.query(Video).one() + video.channel_id = None + test_session.commit() + + # Record whether each statement ran in a write transaction (`get_db_curs(commit=True)`). + executed = [] + real_get_db_curs = videos_module.get_db_curs + + @contextlib.contextmanager + def recording_get_db_curs(commit: bool = False): + with real_get_db_curs(commit=commit) as curs: + class _Recorder: + def __getattr__(self, name): + return getattr(curs, name) + + def execute(self, statement, *args, **kwargs): + executed.append((commit, statement)) + return curs.execute(statement, *args, **kwargs) + + def executemany(self, statement, *args, **kwargs): + executed.append((commit, statement)) + return curs.executemany(statement, *args, **kwargs) + + yield _Recorder() + + with mock.patch.object(videos_module, 'get_db_curs', recording_get_db_curs): + video_cleanup() + + test_session.expire_all() + assert test_session.query(Video).one().channel_id == channel.id, 'the Video was not claimed' + + join_statements = [(commit, s) for commit, s in executed if 'FROM video v' in s] + assert join_statements, 'the claim join did not run' + assert not any(commit for commit, _ in join_statements), \ + 'the Channel-claiming join ran inside a write transaction, holding the lock across it' + assert any(commit for commit, s in executed if s.startswith('UPDATE video SET channel_id')), \ + 'the claims were not written' + assert _UNCLAIMED_VIDEOS_SQL.strip().startswith('SELECT'), 'the claim query must be a read' diff --git a/wrolpi/api_utils.py b/wrolpi/api_utils.py index 25295bcf..d3b9d405 100644 --- a/wrolpi/api_utils.py +++ b/wrolpi/api_utils.py @@ -128,7 +128,14 @@ api_app.error_handler.add(Exception, json_error_handler) @api_app.middleware('request') async def inject_session(request: Request): - """Inject a database session into the request context.""" + """Inject a database session into the request context. + + This session is deliberately *deferred*, even for a POST: it is committed by the response + middleware, so a writer would hold the SQLite write lock for the whole handler -- including the + slow parts (a chunk being written to disk, a subprocess) and including any background task the + handler fires before returning. A handler that writes should use `get_db_session(commit=True)`, + which begins as a writer and commits before the handler returns. + """ from wrolpi.db import get_db_context engine, session = get_db_context() request.ctx.session = session diff --git a/wrolpi/conftest.py b/wrolpi/conftest.py index c7c4f479..89f04f2e 100644 --- a/wrolpi/conftest.py +++ b/wrolpi/conftest.py @@ -10,6 +10,7 @@ import logging import multiprocessing import pathlib import shutil +import sqlite3 import tempfile import threading import time @@ -399,6 +400,36 @@ def make_test_ctx(*, is_cancelled=None, can_download=None, ) +@contextlib.contextmanager +def write_lock_held_briefly(db_file: str, seconds: float = 1.0): + """Hold the SQLite write lock on `db_file` for `seconds`, from another connection. + + Simulates the concurrent writer (refresh, download, config save) that a real WROLPi runs + alongside whatever the test is doing. The lock is released while the test's work is still in + flight, so a correctly-written transaction waits (busy_timeout) and then succeeds; a deferred + one that upgrades its lock mid-transaction fails instantly instead. + + Pair with `production_like_sessions` — the shared `test_session` uses one connection, which + cannot contend with itself. + """ + holder = sqlite3.connect(db_file, timeout=30, check_same_thread=False) + holder.isolation_level = None + holder.execute('PRAGMA busy_timeout=30000') + holder.execute('BEGIN IMMEDIATE') + + def release(): + time.sleep(seconds) + holder.execute('ROLLBACK') + holder.close() + + thread = threading.Thread(target=release, daemon=True) + thread.start() + try: + yield + finally: + thread.join(timeout=30) + + @contextlib.contextmanager def production_like_sessions(test_session) -> Generator[sessionmaker, Any, None]: """Patch get_db_session to hand out a *fresh* Session per call, like production. diff --git a/wrolpi/db.py b/wrolpi/db.py index c67c095f..4d2a7c4b 100644 --- a/wrolpi/db.py +++ b/wrolpi/db.py @@ -7,6 +7,7 @@ Engines are created lazily (the media directory must be known first) and use Nul fresh session per use. Every connection gets the same PRAGMAs (WAL, busy_timeout, foreign keys) via `create_wrolpi_engine` — use that factory for any engine touching a WROLPi database. """ +import contextvars import pathlib import sqlite3 import threading @@ -25,9 +26,13 @@ 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() +# Set by `get_db_session(commit=True)` while its transaction is being opened, so the engine's `begin` +# listener knows to emit `BEGIN IMMEDIATE` instead of a deferred `BEGIN`. A ContextVar rather than a +# `threading.local` because the writers that set it include coroutines: Sanic runs every request in +# its own asyncio task on one shared thread, and a thread-local would leak the flag between +# interleaved tasks -- arming it for a reader and disarming it under a writer. Each asyncio task +# (and each thread) gets its own copy of a ContextVar, so worker-thread callers behave as before. +_immediate_txn = contextvars.ContextVar('wrolpi_immediate_txn', default=False) def _adapt_datetime(value): @@ -128,9 +133,9 @@ def create_wrolpi_engine(target: Union[str, pathlib.Path]) -> sqlalchemy.engine. @event.listens_for(engine, 'begin') def _sqlite_do_begin(conn): - # `get_immediate_db_session()` sessions take the write lock up front; everything else stays - # deferred so readers never block (in WAL; see `_immediate_txn`). - if getattr(_immediate_txn, 'active', False): + # Write sessions (`commit=True`) take the write lock up front; read sessions stay deferred + # so readers never block (in WAL; see `_immediate_txn`). + if _immediate_txn.get(): conn.execute('BEGIN IMMEDIATE') else: conn.execute('BEGIN') @@ -184,8 +189,22 @@ def get_db_context() -> Tuple[sqlalchemy.engine.Engine, Session]: def get_db_session(commit: bool = False) -> Generator[Session, Any, None]: """ Context manager that creates a DB session. This will automatically rollback changes, unless `commit` is True. + + `commit=True` declares write intent, so the transaction begins as a writer (`BEGIN IMMEDIATE`) + rather than deferred. A deferred transaction that upgrades to a writer at its first INSERT + fails *instantly* with "database is locked" if any other connection wrote in the meantime — + SQLite skips `busy_timeout` for lock upgrades, because waiting cannot help a snapshot that is + already stale. Taking the lock at BEGIN lets `busy_timeout` (30s) absorb the contention + instead. Rails and Django both made IMMEDIATE the default for the same reason; a per-call-site + opt-in only works if every author remembers, and read-then-write is the normal shape of a write. + + The cost is that writers serialize from BEGIN rather than from their first write, so a write + transaction must stay short — do the slow parts (globbing, ffprobe, indexing, big joins) in a + read session or outside a session, then write what you computed. Readers are unaffected: WAL + readers never block, and `commit=False` sessions stay deferred. """ _, session = get_db_context() + token = _immediate_txn.set(True) if commit else None try: yield session if commit: @@ -194,6 +213,8 @@ def get_db_session(commit: bool = False) -> Generator[Session, Any, None]: session.rollback() raise finally: + if token is not None: + _immediate_txn.reset(token) # Rollback only if a transaction hasn't been committed. # In tests, the test_session fixture manages the session lifecycle, # so we should not rollback here - that would undo other test operations. @@ -201,28 +222,6 @@ 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 d1b3f114..8e609712 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, get_immediate_db_session +from wrolpi.db import get_db_session, get_db_curs from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \ ValidationError, DownloadError from wrolpi.events import Events @@ -1110,9 +1110,9 @@ class DownloadManager: so a transient "database is locked" can be handled there without retrying the signal 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.""" + This is a read-then-write transaction (it reads the `new` downloads, then claims each by + setting `last_download_attempt`), so it must hold the write lock from BEGIN -- which + `get_db_session(commit=True)` does.""" # Downloads claimed this cycle: dicts of (download_id, download_url, domain). Two things are # deferred until AFTER the immediate (write) transaction below commits: # * `_add_processing_domain` — it mutates shared, non-transactional state, so adding it @@ -1126,7 +1126,7 @@ class DownloadManager: # still holds even though `processing_domains` is not mutated until after commit. to_dispatch = [] claimed = [] - with get_immediate_db_session() as session: + with get_db_session(commit=True) as session: new_downloads = list(session.query(Download).filter( Download.status == 'new', Download.domain not in self.processing_domains, @@ -1348,7 +1348,10 @@ class DownloadManager: """Mark any recurring downloads that are due for download as "new". Start a download.""" now_ = now() - with get_db_session() as session: + # `commit=True` (a writer from BEGIN): this reads the recurring Downloads and then renews + # the due ones. As a read session its commit failed with "database is locked" whenever + # anything else wrote while it was reading -- a config import or refresh, constantly. + with get_db_session(commit=True) as session: recurring = self.get_recurring_downloads(session) renewed_count = 0 for download in recurring: diff --git a/wrolpi/files/test/test_worker_move.py b/wrolpi/files/test/test_worker_move.py index 00ab0b37..a2a0a919 100644 --- a/wrolpi/files/test/test_worker_move.py +++ b/wrolpi/files/test/test_worker_move.py @@ -1,12 +1,9 @@ """Tests for the FileWorker move task.""" import contextlib -import sqlite3 -import threading -import time import pytest -from wrolpi.conftest import production_like_sessions +from wrolpi.conftest import production_like_sessions, write_lock_held_briefly from wrolpi.files import worker as worker_module from wrolpi.files.lib import _move_file_group_files from wrolpi.files.models import FileGroup, Directory @@ -539,32 +536,6 @@ async def test_file_worker_move_cleans_directory_records( assert source_dir.exists(), "source should still exist on disk (empty)" -@contextlib.contextmanager -def _write_lock_held_briefly(db_file: str, seconds: float = 1.0): - """Hold the SQLite write lock on `db_file` for `seconds`, from another connection. - - Simulates the concurrent writer (refresh, download, config save) that a real WROLPi runs - alongside a move. The lock is released while the test's move is still planning, so a - correctly-written move waits (busy_timeout) and then succeeds. - """ - holder = sqlite3.connect(db_file, timeout=30, check_same_thread=False) - holder.isolation_level = None - holder.execute('PRAGMA busy_timeout=30000') - holder.execute('BEGIN IMMEDIATE') - - def release(): - time.sleep(seconds) - holder.execute('ROLLBACK') - holder.close() - - thread = threading.Thread(target=release, daemon=True) - thread.start() - try: - yield - finally: - thread.join(timeout=30) - - @pytest.mark.asyncio async def test_build_move_plan_waits_for_concurrent_writer( async_client, test_session, test_directory, make_files_structure @@ -585,7 +556,7 @@ async def test_build_move_plan_waits_for_concurrent_writer( db_file = test_session.get_bind().url.database with production_like_sessions(test_session): - with _write_lock_held_briefly(db_file): + with write_lock_held_briefly(db_file): plan, _ = await build_move_plan_bulk([file1], dest) assert plan == {file1: dest / 'file1.txt'} @@ -622,7 +593,7 @@ async def test_file_worker_move_survives_concurrent_writer( async def build_plan_then_hold_write_lock(*args, **kwargs): plan = await real_build_move_plan_bulk(*args, **kwargs) # Planning is done; the contention now falls entirely on the execute phase. - stack.enter_context(_write_lock_held_briefly(db_file)) + stack.enter_context(write_lock_held_briefly(db_file)) return plan monkeypatch.setattr(worker_module, 'build_move_plan_bulk', build_plan_then_hold_write_lock) diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index f07887ac..650cdf1e 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -27,7 +27,7 @@ from wrolpi import flags from wrolpi.common import apply_modelers, apply_refresh_cleanup from wrolpi.common import get_media_directory, get_wrolpi_config, logger, walk, chunks, unique_by_predicate from wrolpi.dates import now -from wrolpi.db import get_db_session, get_db_curs, get_immediate_db_session +from wrolpi.db import get_db_session, get_db_curs from wrolpi.errors import NoPrimaryFile from wrolpi.events import Events from wrolpi.files.lib import ( @@ -600,7 +600,7 @@ async def build_move_plan_bulk( # must begin as a writer. A deferred transaction upgrading to a writer mid-way gets "database # is locked" *immediately* (busy_timeout is skipped for lock upgrades), so any concurrent # writer - a refresh, a download, a config save - would abort the whole move. - with get_immediate_db_session() as session: + with get_db_session(commit=True) as session: try: # Create temp table for source paths. SQLite has no ON COMMIT DROP; the table is # explicitly dropped below on success/failure (and IF NOT EXISTS + DELETE guard @@ -2045,7 +2045,7 @@ class FileWorker: # it must begin as a writer; a deferred transaction's lock upgrade fails instantly when # anything else is writing (see `build_move_plan_bulk`). with flags.file_worker_discovery: - with get_immediate_db_session() as session: + with get_db_session(commit=True) as session: await self._execute_move_chunks( plan, session, created_directories, revert_plan ) diff --git a/wrolpi/root_api.py b/wrolpi/root_api.py index 52afc5ce..76a1cc0e 100644 --- a/wrolpi/root_api.py +++ b/wrolpi/root_api.py @@ -29,6 +29,7 @@ from wrolpi.common import logger, get_wrolpi_config, wrol_mode_enabled, get_medi set_global_log_level, get_relative_to_media_directory, search_other_estimates, set_system_timezone from wrolpi.config_api import config_bp from wrolpi.dates import now +from wrolpi.db import get_db_session from wrolpi.downloader import download_manager from wrolpi.errors import WROLModeEnabled, InvalidConfig, ValidationError from wrolpi.events import get_events, Events @@ -355,16 +356,21 @@ async def post_download(request: Request, body: schema.DownloadRequest): # Raises an InvalidDownload if the Downloader cannot be found. download_manager.find_downloader_by_name(body.downloader) - session = request.ctx.session kwargs = dict(downloader_name=body.downloader, sub_downloader_name=body.sub_downloader, reset_attempts=True, destination=body.destination, tag_names=body.tag_names, settings=body.settings, override_skip=True) - if body.frequency: - download_manager.recurring_download(session, body.urls[0], body.frequency, collection_id=body.collection_id, - **kwargs) - else: - download_manager.create_downloads(session, body.urls, **kwargs) + # Not `request.ctx.session`: creating a Download reads (the Tags, then any existing Download for + # each URL) before it INSERTs, and a deferred transaction fails that lock upgrade *instantly* + # with "database is locked" if anything else wrote in between -- which is what a user creating a + # download during a refresh hit. This session also commits before the handler returns, so the + # background dispatch `create_downloads` fires does not wait on a lock this request still holds. + with get_db_session(commit=True) as session: + if body.frequency: + download_manager.recurring_download(session, body.urls[0], body.frequency, + collection_id=body.collection_id, **kwargs) + else: + download_manager.create_downloads(session, body.urls, **kwargs) if download_manager.disabled.is_set() or download_manager.stopped.is_set(): # Downloads are disabled, warn the user. Events.send_downloads_disabled('Download created. But, downloads are disabled.') diff --git a/wrolpi/test/test_db.py b/wrolpi/test/test_db.py index c17cb64d..a2eb03a4 100644 --- a/wrolpi/test/test_db.py +++ b/wrolpi/test/test_db.py @@ -1,7 +1,7 @@ import sqlite3 from wrolpi.conftest import production_like_sessions -from wrolpi.db import get_db_session, get_immediate_db_session, _configure_sqlite_connection +from wrolpi.db import get_db_session, _configure_sqlite_connection class _FakeCursor: @@ -120,39 +120,25 @@ def _probe_write_lock_is_held(db_file: str) -> bool: 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. +def test_write_session_takes_write_lock_up_front(test_session): + """`get_db_session(commit=True)` 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. + lets busy_timeout absorb the contention instead of erroring instantly. `commit=True` is + the declaration of write intent, so it is what arms this — no opt-in helper to forget. """ db_file = test_session.get_bind().url.database with production_like_sessions(test_session): with get_db_session(commit=True) 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 not _probe_write_lock_is_held(db_file), \ - 'plain write session unexpectedly holds the write lock before writing' + assert _probe_write_lock_is_held(db_file), \ + 'immediate session did not hold the write lock after BEGIN' def test_read_session_does_not_take_write_lock(test_session): diff --git a/wrolpi/test/test_downloader.py b/wrolpi/test/test_downloader.py index 06c3624f..447db344 100644 --- a/wrolpi/test/test_downloader.py +++ b/wrolpi/test/test_downloader.py @@ -1713,7 +1713,7 @@ async def test_download_dispatched_outside_immediate_transaction(test_session, t """The download signal must be dispatched AFTER the claiming write transaction commits. Regression test for a production deadlock (found on 10.0.0.9, where downloads silently stopped): - `_dispatch_new_downloads` claimed downloads inside a `get_immediate_db_session()` (BEGIN + `_dispatch_new_downloads` claimed downloads inside a write (BEGIN IMMEDIATE) transaction and `await`ed `api_app.dispatch(...)` while that transaction was still open. Under NullPool (production) the dispatched `signal_download_download` opens its OWN connection and — inheriting the `_immediate_txn` flag — also issues BEGIN IMMEDIATE, which diff --git a/wrolpi/test/test_root_api.py b/wrolpi/test/test_root_api.py index e11bb199..8ea33d9d 100644 --- a/wrolpi/test/test_root_api.py +++ b/wrolpi/test/test_root_api.py @@ -10,6 +10,7 @@ from wrolpi.api_utils import json_error_handler from wrolpi.common import get_wrolpi_config from wrolpi.downloader import Download, get_download_manager_config from wrolpi.errors import ValidationError, SearchEmpty +from wrolpi.conftest import production_like_sessions, write_lock_held_briefly from wrolpi.test.common import assert_dict_contains @@ -969,3 +970,34 @@ async def test_search_other_estimates(async_client, test_session, channel_factor request, response = await async_client.post('/api/search_other_estimates', json=body) assert response.status_code == HTTPStatus.OK assert response.json['others']['channel_count'] == 1, 'Only one Channel is tagged.' + + +@pytest.mark.asyncio +async def test_mutating_request_survives_concurrent_writer(async_client, test_session, test_downloader, + test_download_manager): + """Creating a Download must not fail because something else is writing. + + Regression test for the production failure: a user created a video download while a refresh was + running and got `database is locked` on `INSERT INTO download`. The request session begins + *deferred*, the handler reads (tag/duplicate checks) before it INSERTs, and SQLite refuses that + mid-transaction lock upgrade instantly -- busy_timeout is skipped for upgrades. Mutating + requests must begin as writers so busy_timeout absorbs the contention instead. + """ + db_file = test_session.get_bind().url.database + body = json.dumps(dict(urls=['https://example.com/during-refresh'], downloader=test_downloader.name)) + # Release anything the fixtures left open on the shared session's connection; only the lock the + # test takes below should contend with the request. + test_session.commit() + + with production_like_sessions(test_session) as maker: + with write_lock_held_briefly(db_file): + request, response = await async_client.post('/api/download', content=body) + + assert response.status_code == HTTPStatus.CREATED, response.body + + # A separate session sees the committed Download, as the download worker would. + session = maker() + try: + assert [i.url for i in session.query(Download).all()] == ['https://example.com/during-refresh'] + finally: + session.close()