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/<ids> 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

modules/videos/lib.py

@@ -909,14 +909,19 @@ 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()
+        # 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)

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

modules/zim/api.py

@@ -38,9 +38,8 @@ 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(',')]
+    with request.ctx.db.write() as session:
         lib.delete_zims(session, zim_ids)
     return response.empty()
 

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

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

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:

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,6 +227,8 @@ 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
+    # Production runs switches in the perpetual process, not in a request.
+    with outside_request():
         await await_switches_()
 
 

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]:

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

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
 
+    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()
-    session.commit()
 
 
 def _trackable_path(file: pathlib.Path) -> pathlib.Path | None:

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."""

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

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