From a2bef48c6c3716ec32d2c1180674e22d8b8b7a2f Mon Sep 17 00:00:00 2001 From: lrnselfreliance <33263708+lrnselfreliance@users.noreply.github.com> Date: Sat, 3 Oct 2026 15:42:06 -0600 Subject: [PATCH] db: one request-scoped database handle per API request (#773) * db: one request-scoped database handle per API request Every API request now gets `request.ctx.db` (`wrolpi.db.RequestDB`): a Session plus raw cursors on that Session's connection, so a request needs at most one SQLite connection. Creating it does no I/O; requests that never touch the database never connect. `request.ctx.session` remains the handle's Session. A `RequestSession` binds itself to one lazily-opened Connection. An engine-bound Session releases its connection at every commit, and NullPool closes it, so a request that read and then wrote reconnected. Handlers write in `request.ctx.db.write()`: BEGIN IMMEDIATE, committed when the block exits, so the write lock is not held for the rest of the handler or by background work it starts. This replaces the handler-level `get_db_session(commit=True)` blocks (download create, collection item add/remove/reorder, channel create). The zim search handlers use the request's Session in their worker thread. Tests count the connections a request opens with production's per-call sessions (`count_request_connections`). Endpoints whose library functions still open their own connection are strict xfails, to be removed as those functions take the request's session. Co-Authored-By: Claude Opus 5.5 * tests: name the real extra connections in the request connection xfails Channel detail: get_channel already uses the request session; the extra connection is Channel.get_statistics opening a raw cursor (get_db_curs). Search: count_file_groups opens a cursor for the total, then handle_file_group_search_results opens a cursor and a session. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- modules/videos/channel/api.py | 13 +- modules/zim/api.py | 38 +++--- wrolpi/api_utils.py | 40 +++--- wrolpi/collections/api.py | 9 +- .../collections/test/test_playlist_items.py | 39 +++--- wrolpi/conftest.py | 62 ++++++++- wrolpi/db.py | 126 +++++++++++++++++- wrolpi/root_api.py | 13 +- wrolpi/test/test_db.py | 100 +++++++++++++- wrolpi/test/test_request_db.py | 112 ++++++++++++++++ 10 files changed, 455 insertions(+), 97 deletions(-) create mode 100644 wrolpi/test/test_request_db.py diff --git a/modules/videos/channel/api.py b/modules/videos/channel/api.py index 92e98ac7..5169132b 100644 --- a/modules/videos/channel/api.py +++ b/modules/videos/channel/api.py @@ -9,7 +9,6 @@ from sanic_ext.extensions.openapi import openapi from wrolpi.api_utils import json_response from wrolpi.common import logger, wrol_mode_check, get_media_directory, get_relative_to_media_directory -from wrolpi.db import get_db_session from wrolpi.events import Events from wrolpi.schema import JSONErrorResponse from wrolpi.vars import PYTEST @@ -54,12 +53,12 @@ def channel_get(request: Request, channel_id: int = None): def channel_post(request: Request, body: schema.ChannelPostRequest): body.directory = get_media_directory() / body.directory - # Not `request.ctx.session`: creating a Channel reads (the URL/name/directory conflict check) - # 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 - # Channel while background workers saved configs hit. This session also commits before the - # refresh below is fired, so the refresh does not wait on a lock this request still holds. - with get_db_session(commit=True) as session: + # A write block: creating a Channel reads (the URL/name/directory conflict check) 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 Channel while + # background workers saved configs hit. The block also commits before the refresh below is + # fired, so the refresh does not wait on a lock this request still holds. + with request.ctx.db.write() as session: channel = lib.create_channel(session, data=body, return_dict=False) channel_id, channel_name = channel.id, channel.name diff --git a/modules/zim/api.py b/modules/zim/api.py index 17403e66..d0ebae22 100644 --- a/modules/zim/api.py +++ b/modules/zim/api.py @@ -9,7 +9,6 @@ from sanic_ext.extensions.openapi import openapi from wrolpi import lang from wrolpi.api_utils import json_response from wrolpi.common import logger -from wrolpi.db import get_db_session from wrolpi.downloader import download_manager from wrolpi.events import Events from . import lib, schema @@ -65,13 +64,12 @@ async def post_set_zim_auto_search(request: Request, zim_id: int, body: schema.Z body=schema.ZimSearchRequest, ) @validate(schema.ZimSearchRequest) -async def search_all_zims(_: Request, body: schema.ZimSearchRequest): - # libzim searches are synchronous and slow; run off the event loop. The session is opened in the - # worker thread: a SQLAlchemy Session must not be shared across threads. +async def search_all_zims(request: Request, body: schema.ZimSearchRequest): + # libzim searches are synchronous and slow; run off the event loop. The request's Session is used + # in the worker thread; the handler awaits the thread, so the Session has one user at a time. def search(): - with get_db_session() as session: - return lib.search_all_zims(session, body.search_str, tag_names=body.tag_names, offset=body.offset, - limit=body.limit) + return lib.search_all_zims(request.ctx.session, body.search_str, tag_names=body.tag_names, + offset=body.offset, limit=body.limit) results = await asyncio.to_thread(search) return json_response({'zims': results}) @@ -84,11 +82,10 @@ async def search_all_zims(_: Request, body: schema.ZimSearchRequest): response=schema.ZimSearchResponse, ) @validate(schema.ZimSearchRequest) -async def search_zim(_: Request, zim_id: int, body: schema.ZimSearchRequest): +async def search_zim(request: Request, zim_id: int, body: schema.ZimSearchRequest): def search(): - with get_db_session() as session: - return lib.headline_zim(session, body.search_str, zim_id, tag_names=body.tag_names, offset=body.offset, - limit=body.limit) + return lib.headline_zim(request.ctx.session, body.search_str, zim_id, tag_names=body.tag_names, + offset=body.offset, limit=body.limit) headlines = await asyncio.to_thread(search) return json_response({'zim': headlines}) @@ -219,17 +216,14 @@ async def post_search_estimates(request: Request, body: schema.SearchEstimateReq return response.empty(HTTPStatus.BAD_REQUEST) def estimate(): - zims_estimates = list() - with get_db_session() as session: - if body.tag_names: - # Get actual count of entries tagged with the tag names. - counts = Zims.entries_with_tags(session, body.tag_names) - else: - # Get estimates using libzim. - counts = Zims.estimate(session, body.search_str) - for zim, count in counts.items(): - zims_estimates.append(dict(estimate=count, **zim.__json__())) - return zims_estimates + session = request.ctx.session + if body.tag_names: + # Get actual count of entries tagged with the tag names. + counts = Zims.entries_with_tags(session, body.tag_names) + else: + # Get estimates using libzim. + counts = Zims.estimate(session, body.search_str) + return [dict(estimate=count, **zim.__json__()) for zim, count in counts.items()] zims_estimates = await asyncio.to_thread(estimate) diff --git a/wrolpi/api_utils.py b/wrolpi/api_utils.py index c23cc2b2..195c119f 100644 --- a/wrolpi/api_utils.py +++ b/wrolpi/api_utils.py @@ -12,6 +12,7 @@ from pathlib import Path from sanic import response, HTTPResponse, Request, Sanic, SanicException from wrolpi.common import Base, get_media_directory, logger, LOGGING_CONFIG +from wrolpi.db import RequestDB from wrolpi.errors import APIError from wrolpi.perpetual import PERPETUAL_LOOPS, PER_WORKER_TASKS, PerpetualLoop, run_loop from wrolpi.vars import PYTEST @@ -141,37 +142,26 @@ 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. + """Give the request its database handle, `request.ctx.db` (see `wrolpi.db.RequestDB`). - 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. + `request.ctx.session` is the handle's Session. It begins *deferred*, even for a POST, because it + is committed by the response middleware: 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 any + background task the handler fires before returning. A handler that writes does so in + `request.ctx.db.write()`, which begins as a writer and commits when its block exits. + + Nothing connects until the handler's first statement, so requests that never touch the database + cost nothing here. """ - from wrolpi.db import get_db_context - engine, session = get_db_context() - request.ctx.session = session - request.ctx._db_engine = engine + request.ctx.db = RequestDB() + request.ctx.session = request.ctx.db.session @api_app.middleware('response') async def cleanup_session(request: Request, response_: HTTPResponse): - """Cleanup session after request completes.""" - if hasattr(request.ctx, 'session'): - session = request.ctx.session - try: - if response_.status < 400: - session.commit() - else: - session.rollback() - except Exception: - session.rollback() - raise - finally: - # Don't close session during tests - the test_session fixture manages it. - if not PYTEST: - session.close() + """Commit (or, for an error response, roll back) the request's transaction and release its connection.""" + if hasattr(request.ctx, 'db'): + request.ctx.db.finish(ok=response_.status < 400) return response_ diff --git a/wrolpi/collections/api.py b/wrolpi/collections/api.py index 8e497bf7..0e3a786a 100644 --- a/wrolpi/collections/api.py +++ b/wrolpi/collections/api.py @@ -10,7 +10,6 @@ from sanic_ext.extensions.openapi import openapi from wrolpi.api_utils import json_response from wrolpi.common import wrol_mode_check -from wrolpi.db import get_db_session from wrolpi.errors import ValidationError from wrolpi.schema import JSONErrorResponse from . import lib, schema @@ -91,8 +90,8 @@ async def create_collection_endpoint(request: Request, body: schema.CollectionCr async def add_collection_item_endpoint(request: Request, collection_id: int, body: schema.AddItemRequest): """Append (or insert at ``position``) a file/zim/url item into the collection.""" try: - # Read-then-write, so the session must begin as a writer (see `get_db_session`). - with get_db_session(commit=True) as session: + # Read-then-write, so the transaction must begin as a writer (see `RequestDB.write`). + with request.ctx.db.write() as session: item, created = lib.add_collection_item( session, collection_id, item_kind=body.item_kind, file_group_id=body.file_group_id, zim_id=body.zim_id, zim_entry=body.zim_entry, @@ -114,7 +113,7 @@ async def add_collection_item_endpoint(request: Request, collection_id: int, bod async def remove_collection_item_endpoint(request: Request, collection_id: int, item_id: int): """Remove a single item by its id and close the ordering gap.""" try: - with get_db_session(commit=True) as session: + with request.ctx.db.write() as session: removed = lib.remove_collection_item(session, collection_id, item_id) except UnknownCollection as e: return json_response({'error': str(e)}, status=HTTPStatus.NOT_FOUND) @@ -139,7 +138,7 @@ async def reorder_collection_items_endpoint(request: Request, collection_id: int body: schema.ReorderItemsRequest): """Set the item order from a full list of item ids (used by drag-and-drop).""" try: - with get_db_session(commit=True) as session: + with request.ctx.db.write() as session: lib.reorder_collection_items(session, collection_id, body.item_ids) data = lib.get_collection_with_stats(session, collection_id) except UnknownCollection as e: diff --git a/wrolpi/collections/test/test_playlist_items.py b/wrolpi/collections/test/test_playlist_items.py index b25f417f..db221afb 100644 --- a/wrolpi/collections/test/test_playlist_items.py +++ b/wrolpi/collections/test/test_playlist_items.py @@ -4,6 +4,7 @@ from http import HTTPStatus import pytest from wrolpi.collections.models import Collection +from wrolpi.conftest import production_like_sessions, write_lock_held_briefly @pytest.mark.asyncio @@ -157,39 +158,33 @@ async def test_add_file_item(async_client, test_session, test_directory): @pytest.mark.asyncio -async def test_item_write_endpoints_use_write_intent_sessions(async_client, test_session): - """The item write endpoints (add/remove/reorder) each open their session with - `get_db_session(commit=True)` — the deferred `request.ctx.session` reads before its first - write, which SQLite fails instantly on concurrent writes (see `get_db_session`).""" - from unittest import mock - from wrolpi.collections import api as collections_api - from wrolpi.db import get_db_session +async def test_item_write_endpoints_wait_for_a_concurrent_writer(async_client, test_session): + """Add, reorder and remove each read before they write, so each must begin as a writer. + SQLite fails a deferred transaction's read-to-write lock upgrade *instantly* ("database is locked") + if another connection wrote meanwhile; a transaction that begins as a writer waits out the lock + (busy_timeout) instead. Here a concurrent writer holds the lock while each request starts.""" _, response = await async_client.post('/api/collections', json={'name': 'Locks'}) cid = response.json['collection']['id'] + test_session.commit() + db_file = test_session.get_bind().url.database - commit_kwargs = list() - - def recording_get_db_session(*args, **kwargs): - commit_kwargs.append(kwargs.get('commit', args[0] if args else False)) - return get_db_session(*args, **kwargs) - - with mock.patch.object(collections_api, 'get_db_session', recording_get_db_session): - _, response = await async_client.post(f'/api/collections/{cid}/items', - json={'item_kind': 'url', 'url': '/u/a', 'title': 'a'}) + with production_like_sessions(test_session): + with write_lock_held_briefly(db_file): + _, response = await async_client.post(f'/api/collections/{cid}/items', + json={'item_kind': 'url', 'url': '/u/a', 'title': 'a'}) assert response.status_code == HTTPStatus.CREATED, response.json item_id = response.json['item']['id'] - _, response = await async_client.put(f'/api/collections/{cid}/items/order', - json={'item_ids': [item_id]}) + with write_lock_held_briefly(db_file): + _, response = await async_client.put(f'/api/collections/{cid}/items/order', + json={'item_ids': [item_id]}) assert response.status_code == HTTPStatus.OK, response.json - _, response = await async_client.delete(f'/api/collections/{cid}/items/{item_id}') + with write_lock_held_briefly(db_file): + _, response = await async_client.delete(f'/api/collections/{cid}/items/{item_id}') assert response.status_code == HTTPStatus.NO_CONTENT - assert len(commit_kwargs) == 3, 'Each item write endpoint must open its own session' - assert all(commit is True for commit in commit_kwargs) - @pytest.mark.asyncio async def test_reorder_and_remove_items(async_client, test_session): diff --git a/wrolpi/conftest.py b/wrolpi/conftest.py index da6fbdcb..544b3666 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 +from wrolpi.db import create_wrolpi_engine, RequestSession from wrolpi.downloader import DownloadContext, DownloadManager, DownloadResult, Download, Downloader, \ downloads_manager_config_context, download_cache_config_context from wrolpi.errors import UnrecoverableDownloadError @@ -91,7 +91,8 @@ def test_session(test_directory) -> Generator[Session, Any, None]: return test_engine, session try: - with mock.patch('wrolpi.db._get_db_session', fake_get_db_session): + with mock.patch('wrolpi.db._get_db_session', fake_get_db_session), \ + mock.patch('wrolpi.db._get_request_session', fake_get_db_session): yield session finally: session.rollback() @@ -240,6 +241,11 @@ class AutoAwaitTestClient: """WebSocket connection (pass-through).""" return await self._client.websocket(*args, **kwargs) + @property + def unawaited(self) -> SanicASGITestClient: + """The underlying client: requests return without awaiting switches or background tasks.""" + return self._client + @property def sanic_app(self): """Access to the underlying Sanic app.""" @@ -460,7 +466,8 @@ def production_like_sessions(test_session) -> Generator[sessionmaker, Any, None] one `with` block stays attached for the next and exceptions never roll back. That convenience masks detached-object and transaction-rollback bugs that only bite production's per-call sessions. Use this to exercise that behavior: the yielded sessionmaker creates independent Sessions so a test can verify committed - DB state the way a separate worker would see it. + DB state the way a separate worker would see it. API requests get a fresh `RequestSession` each, as in + production. A dedicated NullPool engine is used and disposed on exit so its connections are released before the test DB is dropped during fixture teardown. @@ -474,8 +481,14 @@ def production_like_sessions(test_session) -> Generator[sessionmaker, Any, None] sessions.append(session) return engine, session + def request_factory(): + session = RequestSession(engine) + sessions.append(session) + return engine, session + try: - with mock.patch('wrolpi.db._get_db_session', factory): + with mock.patch('wrolpi.db._get_db_session', factory), \ + mock.patch('wrolpi.db._get_request_session', request_factory): yield maker finally: for session in sessions: @@ -488,6 +501,47 @@ def production_like_sessions(test_session) -> Generator[sessionmaker, Any, None] engine.dispose() +@contextlib.contextmanager +def count_connections(maker: sessionmaker) -> Generator[List[int], Any, None]: + """Record every SQLite connection opened on a `production_like_sessions` engine during the block. + + Yields a list that gains one entry per connection, so `len()` is the count. Each connection + costs a connect plus WROLPi's PRAGMAs (about 3 ms on a Raspberry Pi 4), so an API request should + open at most one.""" + engine = maker.kw['bind'] + opened = list() + + def on_connect(*_): + opened.append(1) + + sqlalchemy.event.listen(engine, 'connect', on_connect) + try: + yield opened + finally: + sqlalchemy.event.remove(engine, 'connect', on_connect) + + +async def count_request_connections(async_client: 'AutoAwaitTestClient', test_session: Session, method: str, + path: str, **kwargs) -> Tuple[Any, int]: + """Send one request with production's per-call sessions; return its response and how many SQLite + connections the request opened. + + The test's data must already be in `test_session` (it is committed here so other connections see + it). Switches and background tasks are awaited only after counting, so the count is the request's + own work: in tests, `background_task_listener` would otherwise run switches inside the request, + where production runs them in the switch worker.""" + test_session.commit() + with production_like_sessions(test_session) as maker: + with count_connections(maker) as opened, \ + mock.patch(f'{__name__}.await_switches_', mock.AsyncMock()): + _, response = await getattr(async_client.unawaited, method.lower())(path, **kwargs) + count = len(opened) + await await_switches_() + await await_background_tasks_() + test_session.expire_all() + return response, count + + @pytest.fixture def test_downloader(test_download_manager) -> Downloader: class TestDownloader(Downloader, ABC): diff --git a/wrolpi/db.py b/wrolpi/db.py index a1aa5d75..43a40cc0 100644 --- a/wrolpi/db.py +++ b/wrolpi/db.py @@ -174,17 +174,57 @@ def _get_db_session(): return engine, session +def _refuse_non_test_engine(engine: sqlalchemy.engine.Engine): + from wrolpi.common import is_tempfile + if PYTEST and not is_tempfile(engine.url.database or ''): + raise ValueError(f'Running tests, but a test database is not being used!! {engine.url=}') + + def get_db_context() -> Tuple[sqlalchemy.engine.Engine, Session]: """ Get a DB engine and session. """ - from wrolpi.common import is_tempfile local_engine, session = _get_db_session() - if PYTEST and not is_tempfile(local_engine.url.database or ''): - raise ValueError(f'Running tests, but a test database is not being used!! {local_engine.url=}') + _refuse_non_test_engine(local_engine) return local_engine, session +class RequestSession(Session): + """A Session that keeps one connection for its whole life, opened at its first statement. + + An engine-bound Session releases its connection at every commit, and NullPool then closes it, so a + request that commits and keeps working would reconnect (a connect plus PRAGMAs each time). This + Session binds itself to a single Connection, created lazily, so transactions after a commit reuse + it. `close()` releases the connection.""" + + def __init__(self, engine: sqlalchemy.engine.Engine, **kwargs): + super().__init__(**kwargs) + self._request_engine = engine + self._request_connection = None + + def get_bind(self, mapper=None, clause=None, **kwargs): + if self._request_connection is None: + self._request_connection = self._request_engine.connect() + return self._request_connection + + def close(self): + try: + super().close() + finally: + if self._request_connection is not None: + self._request_connection.close() + self._request_connection = None + + +def _get_request_session() -> Tuple[sqlalchemy.engine.Engine, Session]: + """Create the Session for one API request. + + This function allows the database to be wrapped during testing. See: wrolpi.conftest.test_session + """ + engine = get_engine() + return engine, RequestSession(engine) + + @contextmanager def get_db_session(commit: bool = False) -> Generator[Session, Any, None]: """ @@ -277,6 +317,86 @@ def get_db_curs(commit: bool = False) -> Generator[sqlite3.Cursor, Any, None]: connection.close() +def _session_has_connection(session: Session) -> bool: + """True if the Session's transaction has already connected (and so already emitted BEGIN).""" + # SQLAlchemy 1.3 has no public API for this; 2.0 replaces it with `session.in_transaction()`. + transaction = session.transaction + return transaction is not None and bool(transaction._connections) + + +class RequestDB: + """The database handle for one API request: `request.ctx.db`. + + One Session per request, and raw cursors (`curs()`) on that Session's connection, so a request + uses at most one SQLite connection. Creating the handle does no I/O; SQLAlchemy connects at the + first statement, so requests that never touch the database never connect. The response + middleware calls `finish()`. + + Reads begin deferred, so they never take the write lock. A handler that writes does so inside + `write()`, which begins the transaction as a writer and commits when the block exits; the write + lock is held only for that block, never for the rest of the handler or for background work the + handler starts afterwards. (See `get_db_session` for why a writer must take the lock at BEGIN.) + + Code that runs outside a request (workers, switches, perpetual loops) uses `get_db_session` and + `get_db_curs`; they open their own connection. + """ + + def __init__(self): + self.engine, self.session = _get_request_session() + _refuse_non_test_engine(self.engine) + + @contextmanager + def curs(self) -> Generator[sqlite3.Cursor, Any, None]: + """A raw `sqlite3.Cursor` on this request's connection, inside the Session's transaction. + + Pending ORM changes are flushed first, so raw SQL sees them. Writes through the cursor commit + or roll back with the request (or with the enclosing `write()` block). Rows are `sqlite3.Row`.""" + self.session.flush() + curs = self.session.connection().connection.cursor() + curs.row_factory = sqlite3.Row + try: + yield curs + finally: + curs.close() + + @contextmanager + def write(self) -> Generator[Session, Any, None]: + """Begin a write transaction (BEGIN IMMEDIATE) on the request's connection; commit on exit. + + If the request already read, its deferred transaction is committed first: SQLite cannot upgrade + a deferred transaction to a writer under `busy_timeout`, and its snapshot is stale for a writer + anyway. Any changes pending in the Session at that point are committed with it.""" + session = self.session + if _session_has_connection(session): + session.commit() + token = _immediate_txn.set(True) + try: + session.connection() + finally: + _immediate_txn.reset(token) + try: + yield session + session.commit() + except Exception: + session.rollback() + raise + + def finish(self, ok: bool): + """End the request: commit if `ok`, otherwise roll back; then release the connection.""" + try: + if ok: + self.session.commit() + else: + self.session.rollback() + except Exception: + self.session.rollback() + raise + finally: + # In tests the shared `test_session` fixture owns the Session's lifecycle. + if not PYTEST: + self.session.close() + + def get_ranked_models(ranked_primary_keys: List, model: Type[Base], session: Session) -> List[Base]: """Get all objects whose primary keys are in the `ranked_primary_keys`, preserve their order.""" pkey = sqlalchemy.inspect(model).primary_key[0] diff --git a/wrolpi/root_api.py b/wrolpi/root_api.py index 5b9b64e3..54c5fdb9 100644 --- a/wrolpi/root_api.py +++ b/wrolpi/root_api.py @@ -30,7 +30,6 @@ 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 @@ -369,12 +368,12 @@ async def post_download(request: Request, body: schema.DownloadRequest): sub_downloader_name=body.sub_downloader, reset_attempts=True, destination=body.destination, tag_names=body.tag_names, settings=body.settings, override_skip=True) - # 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: + # A write block: 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. The block also commits before the handler returns, so the background dispatch + # `create_downloads` fires does not wait on a lock this request still holds. + with request.ctx.db.write() as session: if body.frequency: download_manager.recurring_download(session, body.urls[0], body.frequency, collection_id=body.collection_id, **kwargs) diff --git a/wrolpi/test/test_db.py b/wrolpi/test/test_db.py index dd1a62f0..037958b1 100644 --- a/wrolpi/test/test_db.py +++ b/wrolpi/test/test_db.py @@ -3,8 +3,9 @@ import sqlite3 import pytest -from wrolpi.conftest import production_like_sessions, probe_write_lock_is_held -from wrolpi.db import get_db_session, _configure_sqlite_connection +from wrolpi.conftest import production_like_sessions, probe_write_lock_is_held, count_connections +from wrolpi.db import get_db_session, _configure_sqlite_connection, RequestDB, RequestSession +from wrolpi.tags import Tag class _FakeCursor: @@ -160,3 +161,98 @@ async def test_background_task_does_not_inherit_write_intent(test_session): assert held_during_child_read['held'] is False, \ 'a read session in a spawned task took the write lock, inheriting the parent write intent' + + +def test_request_db_does_no_io_until_used(test_session): + """Creating the request handle opens no connection; most endpoints never touch the database.""" + test_session.commit() + with production_like_sessions(test_session) as maker, count_connections(maker) as opened: + db = RequestDB() + db.finish(ok=True) + assert len(opened) == 0 + + +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.""" + test_session.commit() + with production_like_sessions(test_session) as maker, count_connections(maker) as opened: + db = RequestDB() + db.session.add(Tag(name='shared', color='#000000')) + with db.curs() as curs: + curs.execute("SELECT name FROM tag WHERE name = 'shared'") + assert [tuple(row) for row in curs.fetchall()] == [('shared',)] + db.finish(ok=False) + assert len(opened) == 1 + + # finish(ok=False) rolled the uncommitted Tag back. + assert test_session.query(Tag).filter_by(name='shared').count() == 0 + + +def test_request_db_write_holds_the_write_lock_only_inside_the_block(test_session): + """`write()` begins as a writer (BEGIN IMMEDIATE) and commits on exit, so the write lock is not held + for the rest of the handler, or by any background work the handler starts afterwards.""" + db_file = test_session.get_bind().url.database + test_session.commit() + with production_like_sessions(test_session) as maker, count_connections(maker) as opened: + db = RequestDB() + # A read first, as most write handlers do (validation, conflict checks). + db.session.query(Tag).count() + assert not probe_write_lock_is_held(db_file), 'a read must not take the write lock' + + with db.write() as session: + assert probe_write_lock_is_held(db_file), 'write() did not take the write lock at BEGIN' + session.add(Tag(name='written', color='#000000')) + + assert not probe_write_lock_is_held(db_file), 'write() held the lock after its block' + db.finish(ok=True) + assert len(opened) == 1, 'the write reused the request connection' + + assert test_session.query(Tag).filter_by(name='written').count() == 1 + + +def test_request_db_write_rolls_back_on_error(test_session): + """An exception inside `write()` rolls its changes back and releases the write lock.""" + db_file = test_session.get_bind().url.database + test_session.commit() + with production_like_sessions(test_session): + db = RequestDB() + with pytest.raises(ValueError): + with db.write() as session: + session.add(Tag(name='discarded', color='#000000')) + session.flush() + raise ValueError('handler failed') + assert not probe_write_lock_is_held(db_file) + db.finish(ok=False) + + assert test_session.query(Tag).filter_by(name='discarded').count() == 0 + + +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`).""" + test_session.commit() + with production_like_sessions(test_session): + db = RequestDB() + db.session.add(Tag(name='committed', color='#000000')) + db.finish(ok=True) + + assert test_session.query(Tag).filter_by(name='committed').count() == 1 + + +def test_request_session_keeps_one_connection_until_closed(test_session): + """A RequestSession reuses its connection across commits and releases it on close (production's + `RequestDB.finish`); a later statement reconnects.""" + test_session.commit() + with production_like_sessions(test_session) as maker, count_connections(maker) as opened: + session = RequestSession(maker.kw['bind']) + session.query(Tag).count() + session.commit() + session.query(Tag).count() + assert len(opened) == 1, 'the commit released the connection' + + session.close() + assert session._request_connection is None + session.query(Tag).count() + assert len(opened) == 2, 'a closed RequestSession did not reconnect' + session.close() diff --git a/wrolpi/test/test_request_db.py b/wrolpi/test/test_request_db.py new file mode 100644 index 00000000..c915e923 --- /dev/null +++ b/wrolpi/test/test_request_db.py @@ -0,0 +1,112 @@ +"""An API request uses at most one SQLite connection (see `wrolpi.db.RequestDB`). + +Each connection costs a connect plus WROLPi's PRAGMAs, about 3 ms on a Raspberry Pi 4. These tests count +the connections a request opens with production's per-call sessions; the shared `test_session` hides them. +""" +from http import HTTPStatus + +import pytest + +from wrolpi.conftest import count_request_connections + +# Library functions these endpoints call still open their own cursor or session, instead of using the +# request's. Remove an endpoint's mark when its count reaches one (strict: an unexpected pass fails). +OPENS_OWN_CONNECTION = pytest.mark.xfail( + strict=True, reason='a library function opens its own connection instead of using the request session') + + +@pytest.mark.asyncio +@pytest.mark.parametrize('method,path,body', [ + ('post', '/api/echo', {}), + ('get', '/api/settings', None), +]) +async def test_requests_without_database_work_open_no_connection(async_client, test_session, method, path, body): + kwargs = dict(json=body) if body is not None else dict() + response, count = await count_request_connections(async_client, test_session, method, path, **kwargs) + assert response.status_code == HTTPStatus.OK, response.body + assert count == 0 + + +@pytest.mark.asyncio +async def test_collection_item_writes_use_one_connection(async_client, test_session): + """The item endpoints read, then write in `RequestDB.write()`; both use the request's connection.""" + _, response = await async_client.post('/api/collections', json={'name': 'One Connection'}) + cid = response.json['collection']['id'] + + response, count = await count_request_connections( + async_client, test_session, 'post', f'/api/collections/{cid}/items', + json={'item_kind': 'url', 'url': '/u/a', 'title': 'a'}) + assert response.status_code == HTTPStatus.CREATED, response.body + assert count == 1 + item_id = response.json['item']['id'] + + response, count = await count_request_connections( + async_client, test_session, 'put', f'/api/collections/{cid}/items/order', json={'item_ids': [item_id]}) + assert response.status_code == HTTPStatus.OK, response.body + assert count == 1 + + response, count = await count_request_connections( + async_client, test_session, 'delete', f'/api/collections/{cid}/items/{item_id}') + assert response.status_code == HTTPStatus.NO_CONTENT, response.body + assert count == 1 + + +@pytest.mark.asyncio +async def test_create_download_uses_one_connection(async_client, test_session, test_downloader): + response, count = await count_request_connections( + async_client, test_session, 'post', '/api/download', + json={'urls': ['https://example.com/one-connection'], 'downloader': test_downloader.name}) + assert response.status_code == HTTPStatus.CREATED, response.body + assert count == 1 + + +@pytest.mark.asyncio +async def test_tags_use_one_connection(async_client, test_session, tag_factory): + await tag_factory() + response, count = await count_request_connections(async_client, test_session, 'get', '/api/tag') + assert response.status_code == HTTPStatus.OK, response.body + assert count == 1 + + +@pytest.mark.asyncio +async def test_zim_estimates_use_one_connection(async_client, test_session, zim_factory): + zim_factory('test zim') + response, count = await count_request_connections( + async_client, test_session, 'post', '/api/zim/search_estimates', json={'search_str': 'one'}) + assert response.status_code == HTTPStatus.OK, response.body + assert count == 1 + + +@OPENS_OWN_CONNECTION +@pytest.mark.asyncio +async def test_zim_search_uses_one_connection(async_client, test_session, zim_factory): + """`get_all_entries_tags` opens a cursor of its own.""" + zim_factory('test zim') + response, count = await count_request_connections( + async_client, test_session, 'post', '/api/zim/search', json={'search_str': 'one'}) + assert response.status_code == HTTPStatus.OK, response.body + assert count == 1 + + +@OPENS_OWN_CONNECTION +@pytest.mark.asyncio +@pytest.mark.parametrize('path', ['/api/files/search', '/api/videos/search']) +async def test_search_uses_one_connection(async_client, test_session, simple_channel, video_factory, path): + """`count_file_groups` opens a cursor for the total (skipped when the total is cached), then + `handle_file_group_search_results` opens another cursor and a separate session for the page.""" + video_factory(title='a video', channel_id=simple_channel.id) + response, count = await count_request_connections( + async_client, test_session, 'post', path, json={'search_str': 'video'}) + assert response.status_code == HTTPStatus.OK, response.body + assert count == 1 + + +@OPENS_OWN_CONNECTION +@pytest.mark.asyncio +async def test_channel_uses_one_connection(async_client, test_session, simple_channel): + """`get_channel` uses the request session, but `Channel.get_statistics` opens a raw cursor of its own + (`get_db_curs`).""" + response, count = await count_request_connections( + async_client, test_session, 'get', f'/api/videos/channels/{simple_channel.id}') + assert response.status_code == HTTPStatus.OK, response.body + assert count == 1