From a93630576f65e7e04eda6a549033eff4f01f9be9 Mon Sep 17 00:00:00 2001 From: lrnselfreliance <33263708+lrnselfreliance@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:09:22 -0600 Subject: [PATCH] Tests fail when a request writes outside request.ctx.db.write() (#864) * Tests fail when a request writes outside request.ctx.db.write() A request's transaction begins deferred. A write in it after a read fails instantly ("database is locked") when another connection wrote in between, and a write left for the response middleware's commit can be lost after the response was sent. #846-#853 and #863 moved handlers' writes into request.ctx.db.write(); this keeps it that way. Under pytest, a listener records every INSERT/UPDATE/DELETE/REPLACE made while a request runs (or in a task it started) outside RequestDB.write() or get_db_session(commit=True), and the autouse no_deferred_request_writes fixture fails the test. allow_deferred_request_writes opts a test out. The test-only response listener now runs switches outside the request, as production's perpetual process does. It found three more: - DELETE /api/zim/ deleted the .zim file, returned 204, then lost the row's delete when the middleware's commit failed under contention. It now writes in request.ctx.db.write(). - Viewing a file recorded the view in a background task on the request's Session, after the request ended. set_file_viewed opens its own write. - tags.import_config (startup and config import) read, then wrote, in a deferred get_db_session(). It now begins as a writer. Co-Authored-By: Claude Opus 5.5 * Tasks don't inherit a declared write; tags import change backed out From review of #864: - background_task's task copied its creator's Context, including _write_intent when started inside get_db_session(commit=True) or RequestDB.write(), so the guard ignored the task's own writes for its lifetime. The task now starts with no declared write. That exposes fetch_channel_source_id (started by the channels config import): it read the Channel, asked the remote site for its source_id, then committed in the same deferred transaction. It now reads, fetches with no transaction open, then writes in a short get_db_session(commit=True). - tags.import_config: commit=True held the write lock through its directory walk, and its exit commit saved changes the need_commit gate used to drop. Reverted to master; splitting its reads from a short write is its own change. The two tests that run it through a request are marked allow_deferred_request_writes until then. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- modules/videos/lib.py | 19 +++++---- modules/videos/video/test/test_api.py | 3 +- modules/zim/api.py | 5 +-- modules/zim/test/test_api.py | 19 +++++++++ pytest.ini | 1 + wrolpi/common.py | 4 ++ wrolpi/conftest.py | 20 ++++++++- wrolpi/db.py | 52 ++++++++++++++++++++++++ wrolpi/files/api.py | 2 +- wrolpi/files/lib.py | 19 +++++---- wrolpi/test/test_config_api.py | 3 ++ wrolpi/test/test_config_backup_import.py | 3 ++ wrolpi/test/test_db.py | 52 ++++++++++++++++++++++++ 13 files changed, 179 insertions(+), 23 deletions(-) diff --git a/modules/videos/lib.py b/modules/videos/lib.py index c5dd210f..952f6458 100644 --- a/modules/videos/lib.py +++ b/modules/videos/lib.py @@ -909,15 +909,20 @@ async def fetch_channel_source_id(channel_id: int): # If we can download from a channel, we must have its source_id. try: with get_db_session() as session: - from modules.videos.channel.lib import get_channel channel = Channel.find_by_id(session, channel_id) + source_id, url = channel.source_id, channel.url + if source_id: + return - channel.source_id = channel.source_id or get_channel_source_id(channel.url) - if not channel.source_id: - channel_import_logger.warning(f'Unable to fetch source_id for {channel.url}') - else: - session.commit() - save_channels_config.activate_switch() + # Fetched outside any transaction: this asks the remote site. + source_id = get_channel_source_id(url) + if not source_id: + channel_import_logger.warning(f'Unable to fetch source_id for {url}') + return + + with get_db_session(commit=True) as session: + Channel.find_by_id(session, channel_id).source_id = source_id + save_channels_config.activate_switch() except Exception as e: logger.error(f'Failed to get Channel source id of id={channel_id}', exc_info=e) diff --git a/modules/videos/video/test/test_api.py b/modules/videos/video/test/test_api.py index 2b2bccf2..8ff706ea 100644 --- a/modules/videos/video/test/test_api.py +++ b/modules/videos/video/test/test_api.py @@ -196,10 +196,11 @@ async def test_wrol_mode(async_client, simple_channel, simple_video, wrol_mode_f @pytest.mark.asyncio -async def test_api_video_extras(async_client, simple_channel, video_factory): +async def test_api_video_extras(async_client, test_session, simple_channel, video_factory): """Can fetch extra data about a video (comments/captions).""" info_json = {'duration': 5, 'epoch': 12345, 'comments': [{'some': 'comments'}]} video = video_factory(simple_channel.id, with_info_json=info_json, with_caption_file=True) + test_session.commit() request, response = await async_client.get(f'/api/videos/{video.file_group_id}/captions') assert response.status_code == HTTPStatus.OK diff --git a/modules/zim/api.py b/modules/zim/api.py index 56be3912..46b53692 100644 --- a/modules/zim/api.py +++ b/modules/zim/api.py @@ -38,10 +38,9 @@ async def get_zims(request: Request): summary='Delete all Zim files', ) async def delete_zims(request: Request, zim_ids: str): - session = request.ctx.session - zim_ids = [int(i) for i in zim_ids.split(',')] - lib.delete_zims(session, zim_ids) + with request.ctx.db.write() as session: + lib.delete_zims(session, zim_ids) return response.empty() diff --git a/modules/zim/test/test_api.py b/modules/zim/test/test_api.py index f15915f5..630e2acc 100644 --- a/modules/zim/test/test_api.py +++ b/modules/zim/test/test_api.py @@ -347,3 +347,22 @@ async def test_zim_writes_wait_for_a_concurrent_writer(async_client, test_sessio _, response = await async_client.delete('/api/zim/subscribe/1') assert response.status_code in (HTTPStatus.CREATED, HTTPStatus.NO_CONTENT), response.body + + +@pytest.mark.asyncio +async def test_delete_zim_waits_for_a_concurrent_writer(async_client, test_session, test_zim): + """Deleting a Zim must delete its row as well as its file, even while something else is writing. It wrote + in the request's deferred transaction, so its commit in the response middleware failed ("database is + locked") after the file was deleted and the response was sent.""" + zim_id, zim_path = test_zim.id, test_zim.path + test_session.commit() + db_file = test_session.get_bind().url.database + + with production_like_sessions(test_session) as maker: + with write_lock_held_briefly(db_file, seconds=3): + _, response = await async_client.delete(f'/api/zim/{zim_id}') + assert response.status_code == HTTPStatus.NO_CONTENT, response.body + + assert not zim_path.exists() + with maker() as session: + assert session.query(Zim).filter_by(id=zim_id).one_or_none() is None diff --git a/pytest.ini b/pytest.ini index 3d51a24a..fdeea159 100644 --- a/pytest.ini +++ b/pytest.ini @@ -6,6 +6,7 @@ testpaths = asyncio_default_fixture_loop_scope = function markers = allow_unexpected_queries: the test trips the serializer query guard on purpose (see wrolpi.db.no_db_access) + allow_deferred_request_writes: a request writes outside request.ctx.db.write() on purpose (see wrolpi.db.RequestDB) # Do not remove timeout - tests should not run forever. # # This is a hang detector, not a performance budget. It was 30s, but the plugin was never diff --git a/wrolpi/common.py b/wrolpi/common.py index feb4917f..c18a7736 100644 --- a/wrolpi/common.py +++ b/wrolpi/common.py @@ -2680,6 +2680,10 @@ def background_task(coro) -> Task: coro_name = getattr(coro, '__qualname__', None) or getattr(coro, '__name__', None) or str(coro) async def error_logger(): + # The task copied its creator's Context. A write the creator declared (`get_db_session(commit=True)`, + # `RequestDB.write()`) is not the task's: the task's own writes must declare themselves. + from wrolpi.db import _write_intent # wrolpi.db imports wrolpi.common. + _write_intent.set(False) try: await coro except Exception as e: diff --git a/wrolpi/conftest.py b/wrolpi/conftest.py index e4a8ac68..364184d4 100644 --- a/wrolpi/conftest.py +++ b/wrolpi/conftest.py @@ -45,7 +45,7 @@ from wrolpi.common import logger, await_background_tasks as await_background_tas from wrolpi.common import set_test_media_directory, Base, set_test_config from wrolpi.contexts import attach_shared_contexts, initialize_configs_contexts from wrolpi.dates import set_test_now -from wrolpi.db import create_wrolpi_engine, RequestDB, RequestSession, REQUEST_SESSION_OPTIONS +from wrolpi.db import create_wrolpi_engine, RequestDB, RequestSession, REQUEST_SESSION_OPTIONS, outside_request from wrolpi.downloader import DownloadContext, DownloadManager, DownloadResult, Download, Downloader, \ downloads_manager_config_context, download_cache_config_context from wrolpi.errors import UnrecoverableDownloadError @@ -131,6 +131,20 @@ def no_unexpected_queries(request): '\n'.join(f'{reason}: {" ".join(statement.split())[:300]}' for reason, statement in queries) +@pytest.fixture(autouse=True) +def no_deferred_request_writes(request): + """Fail any test during which an API request wrote outside `request.ctx.db.write()`.""" + from wrolpi import db + db.DEFERRED_REQUEST_WRITES.clear() + yield + if request.node.get_closest_marker('allow_deferred_request_writes'): + return + writes = list(db.DEFERRED_REQUEST_WRITES) + db.DEFERRED_REQUEST_WRITES.clear() + assert not writes, 'A request wrote outside `request.ctx.db.write()` (see `RequestDB.write`):\n' + \ + '\n'.join(f'{where}: {" ".join(statement.split())[:200]}' for statement, where in writes) + + @pytest.fixture(autouse=True) def test_debug_logger(request): """Tests default to DEBUG logging. Use pytest -vvv for TRACE level.""" @@ -213,7 +227,9 @@ async def background_task_listener(request, response): if session := getattr(request.ctx, 'session', None): # The handler is done. Background work shares the test's Session, and expires on commit like production. session.expire_on_commit = True - await await_switches_() + # Production runs switches in the perpetual process, not in a request. + with outside_request(): + await await_switches_() class AutoAwaitTestClient: diff --git a/wrolpi/db.py b/wrolpi/db.py index cf1ef34b..f5fca63f 100644 --- a/wrolpi/db.py +++ b/wrolpi/db.py @@ -12,6 +12,7 @@ import functools import pathlib import sqlite3 import threading +import traceback import types from contextlib import contextmanager from typing import Tuple, List, Union, Type, Generator, Any, ContextManager @@ -44,6 +45,15 @@ _warned_serializers = set() # broad exceptions around a serializer cannot hide one; see the `no_unexpected_queries` fixture. UNEXPECTED_QUERIES = list() +# True while an API request runs (see `RequestDB`), and in tasks it starts. +_in_request = contextvars.ContextVar('wrolpi_in_request', default=False) +# True inside `RequestDB.write()` and `get_db_session(commit=True)`: blocks that declared they write, so their +# transaction began as a writer. +_write_intent = contextvars.ContextVar('wrolpi_write_intent', default=False) +# Every write during tests that a request made outside a declared write, as (statement, where). See +# `_refuse_deferred_request_write` and the `no_deferred_request_writes` fixture. +DEFERRED_REQUEST_WRITES = list() + class UnexpectedQuery(RuntimeError): """SQL ran inside `no_db_access` (while serializing).""" @@ -91,6 +101,34 @@ def _refuse_query_while_serializing(conn, cursor, statement, parameters, context f'loaded, or a commit expired the object: {statement}') +_WRITE_STATEMENTS = ('INSERT', 'UPDATE', 'DELETE', 'REPLACE') + + +def _refuse_deferred_request_write(conn, cursor, statement, parameters, context, executemany): + """During tests, record any write a request makes outside `RequestDB.write()` (or `get_db_session(commit=True)`). + + The request's own transaction begins deferred. A write in it after a read fails instantly ("database is + locked") when another connection wrote in between -- SQLite skips `busy_timeout` for that lock upgrade -- or, + when the write is only flushed by the response middleware's commit, is lost after the response was sent.""" + if not _in_request.get() or _write_intent.get() or not statement.lstrip().upper().startswith(_WRITE_STATEMENTS): + return + frames = [i for i in traceback.extract_stack() + if ('/wrolpi/' in i.filename or '/modules/' in i.filename) and not i.filename.endswith('/db.py')] + where = ' > '.join(f'{pathlib.Path(i.filename).name}:{i.name}:{i.lineno}' for i in frames[-4:]) + DEFERRED_REQUEST_WRITES.append((statement, where)) + + +@contextmanager +def outside_request(): + """Run a block as if no request were running: test code that does the perpetual process's work (switches) from + inside a request.""" + token = _in_request.set(False) + try: + yield + finally: + _in_request.reset(token) + + def _adapt_datetime(value): """Store datetimes from raw SQL exactly like SQLAlchemy does: naive UTC with microseconds.""" from datetime import timezone @@ -188,6 +226,8 @@ def create_wrolpi_engine(target: Union[str, pathlib.Path]) -> sqlalchemy.engine. journal_state['mode'] = _configure_sqlite_connection(dbapi_conn, journal_state['mode']) event.listen(engine, 'before_cursor_execute', _refuse_query_while_serializing) + if PYTEST: + event.listen(engine, 'before_cursor_execute', _refuse_deferred_request_write) @event.listens_for(engine, 'begin') def _sqlite_do_begin(conn): @@ -323,6 +363,7 @@ def get_db_session(commit: bool = False) -> Generator[Session, Any, None]: session.connection() finally: _immediate_txn.reset(token) + intent = _write_intent.set(True) if commit else None try: yield session if commit: @@ -336,6 +377,8 @@ def get_db_session(commit: bool = False) -> Generator[Session, Any, None]: # so we should not rollback here - that would undo other test operations. if not PYTEST and session.in_transaction(): session.rollback() + if intent: + _write_intent.reset(intent) @contextmanager @@ -429,6 +472,7 @@ class RequestDB: def __init__(self): self.engine, self.session = _get_request_session() _refuse_non_test_engine(self.engine) + self._in_request = _in_request.set(True) def curs(self) -> ContextManager[sqlite3.Cursor]: """A raw `sqlite3.Cursor` on this request's connection (see `session_curs`). Its writes commit or @@ -450,12 +494,15 @@ class RequestDB: session.connection() finally: _immediate_txn.reset(token) + intent = _write_intent.set(True) try: yield session session.commit() except Exception: session.rollback() raise + finally: + _write_intent.reset(intent) def finish(self, ok: bool): """End the request: commit if `ok`, otherwise roll back; then release the connection.""" @@ -471,6 +518,11 @@ class RequestDB: # In tests the shared `test_session` fixture owns the Session's lifecycle. if not PYTEST: self.session.close() + try: + _in_request.reset(self._in_request) + except ValueError: + # Finished in another Context than the one it began in. + _in_request.set(False) def get_ranked_models(ranked_primary_keys: List, model: Type[Base], session: Session) -> List[Base]: diff --git a/wrolpi/files/api.py b/wrolpi/files/api.py index c0ee369f..9ff52ebb 100644 --- a/wrolpi/files/api.py +++ b/wrolpi/files/api.py @@ -54,7 +54,7 @@ async def get_file(request: Request, body: schema.FileRequest): raise InvalidFile() if not body.skip_tracking: - background_task(lib.set_file_viewed(session, get_media_directory() / body.file)) + background_task(lib.set_file_viewed(get_media_directory() / body.file)) return json_response({'file': file}) diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index 39df0918..3fab2892 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -145,21 +145,22 @@ def get_file_dict(session: Session, file: str) -> Dict: return _get_file_dict(session, media_directory / file) -async def set_file_viewed(session: Session, file: pathlib.Path): +async def set_file_viewed(file: pathlib.Path): """Change FileGroup.viewed to the current datetime. - Files in ignored directories are not tracked. + Files in ignored directories are not tracked. Runs after the request that viewed the file, so it writes in a + session of its own, not the request's. """ if is_ignored_file(file): return - try: - fg = FileGroup.find_by_path(session, file) - except UnknownFile: - fg = FileGroup.from_paths(session, file) - fg.do_model(session) - fg.set_viewed() - session.commit() + with get_db_session(commit=True) as session: + try: + fg = FileGroup.find_by_path(session, file) + except UnknownFile: + fg = FileGroup.from_paths(session, file) + fg.do_model(session) + fg.set_viewed() def _trackable_path(file: pathlib.Path) -> pathlib.Path | None: diff --git a/wrolpi/test/test_config_api.py b/wrolpi/test/test_config_api.py index ccb0b6a0..4a947069 100644 --- a/wrolpi/test/test_config_api.py +++ b/wrolpi/test/test_config_api.py @@ -7,6 +7,9 @@ from wrolpi.tags import Tag @pytest.mark.asyncio +# tags.import_config reads, then writes, in a deferred session; splitting its reads (and directory walk) from a +# short write is its own change. +@pytest.mark.allow_deferred_request_writes async def test_configs_crud(async_client, test_session, test_tags_config, tag_factory, test_wrolpi_config, test_channels_config, await_switches): """Configs can be imported or saved on demand.""" diff --git a/wrolpi/test/test_config_backup_import.py b/wrolpi/test/test_config_backup_import.py index 09ef6dc2..8cbf73b4 100644 --- a/wrolpi/test/test_config_backup_import.py +++ b/wrolpi/test/test_config_backup_import.py @@ -603,6 +603,9 @@ class TestBackupAPIEndpoints: assert 'preview' in response.json assert len(response.json['preview']['add']) == 1 + # tags.import_config reads, then writes, in a deferred session; splitting its reads (and directory walk) from a + # short write is its own change. + @pytest.mark.allow_deferred_request_writes async def test_import_endpoint(self, async_client, test_directory, test_tags_config, test_session): from wrolpi.tags import get_tags_config config = get_tags_config() diff --git a/wrolpi/test/test_db.py b/wrolpi/test/test_db.py index c859da12..2d071f87 100644 --- a/wrolpi/test/test_db.py +++ b/wrolpi/test/test_db.py @@ -176,6 +176,7 @@ def test_request_db_does_no_io_until_used(test_session): assert len(opened) == 0 +@pytest.mark.allow_deferred_request_writes def test_request_db_cursor_shares_the_session_connection(test_session): """`curs()` runs on the Session's connection and inside its transaction: one connection per request, and raw SQL sees the request's own ORM changes.""" @@ -232,6 +233,7 @@ def test_request_db_write_rolls_back_on_error(test_session): assert test_session.query(Tag).filter_by(name='discarded').count() == 0 +@pytest.mark.allow_deferred_request_writes def test_request_db_finish_commits_reads_and_writes(test_session): """`finish(ok=True)` commits whatever the request's session changed (the existing response-middleware contract for handlers that write through `request.ctx.session`).""" @@ -341,3 +343,53 @@ def test_no_db_access_logs_the_query_in_production(test_session, caplog): warnings = [r.getMessage() for r in caplog.records if 'a production serializer ran SQL' in r.getMessage()] assert len(warnings) == 1 assert 'SELECT' in warnings[0] and 'BEGIN' not in warnings[0] + + +def test_deferred_request_writes_are_recorded(test_session): + """During tests, a write a request makes outside `RequestDB.write()` is recorded (the + `no_deferred_request_writes` fixture fails the test); a write inside it is not.""" + from wrolpi import db as db_module + test_session.commit() + with production_like_sessions(test_session): + db = RequestDB() + with db.write() as session: + session.add(Tag(name='declared', color='#000000')) + assert db_module.DEFERRED_REQUEST_WRITES == [] + + db.session.add(Tag(name='undeclared', color='#000000')) + db.session.flush() + db.finish(ok=False) + + statements = [statement for statement, _ in db_module.DEFERRED_REQUEST_WRITES] + db_module.DEFERRED_REQUEST_WRITES.clear() + assert len(statements) == 1 and statements[0].startswith('INSERT INTO tag'), statements + + # Code outside a request is not checked. + test_session.add(Tag(name='outside', color='#000000')) + test_session.flush() + assert db_module.DEFERRED_REQUEST_WRITES == [] + + +@pytest.mark.asyncio +async def test_task_started_in_a_write_does_not_inherit_its_intent(test_session): + """A task started inside `RequestDB.write()` copies the request's Context, but not its declared write: the + task's own undeclared writes are recorded.""" + from wrolpi import db as db_module + from wrolpi.common import background_task + + async def undeclared_write(): + with get_db_session() as session: + session.add(Tag(name='from a task', color='#000000')) + session.flush() + + test_session.commit() + with production_like_sessions(test_session): + db = RequestDB() + with db.write(): + task = background_task(undeclared_write()) + await task + db.finish(ok=True) + + statements = [statement for statement, _ in db_module.DEFERRED_REQUEST_WRITES] + db_module.DEFERRED_REQUEST_WRITES.clear() + assert len(statements) == 1 and statements[0].startswith('INSERT INTO tag'), statements