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()