From 8022b4f86bb850c5b0dee07d009d7899f6d0c436 Mon Sep 17 00:00:00 2001 From: Roland Date: Wed, 21 Oct 2026 08:01:46 -0600 Subject: [PATCH] Channel and map subscription changes wait for a concurrent writer PUT/DELETE /api/videos/channels/, POST /api/videos/channels//tag, and POST/DELETE /api/map/subscribe read, then wrote, in the request's deferred transaction, so they failed with a 500 ("database is locked") when another connection wrote in between. They now write inside request.ctx.db.write(). Co-Authored-By: Claude Opus 5.5 --- modules/map/api.py | 6 ++-- modules/map/test/test_api.py | 29 +++++++++++++++++++ modules/videos/channel/api.py | 12 ++++---- modules/videos/channel/test/test_api.py | 38 ++++++++++++++++++++++++- 4 files changed, 76 insertions(+), 9 deletions(-) diff --git a/modules/map/api.py b/modules/map/api.py index 2e2f72d8..b4bca7e8 100644 --- a/modules/map/api.py +++ b/modules/map/api.py @@ -48,7 +48,8 @@ def get_subscriptions(request: Request): @wrol_mode_check async def post_subscribe(request: Request, body: schema.MapSubscribeRequest): try: - await lib.subscribe(request.ctx.session, body.name, body.region) + with request.ctx.db.write() as session: + await lib.subscribe(session, body.name, body.region) except ValueError as e: return response.json({'error': str(e)}, HTTPStatus.BAD_REQUEST) @@ -61,7 +62,8 @@ async def post_subscribe(request: Request, body: schema.MapSubscribeRequest): @wrol_mode_check async def delete_subscription(request: Request, region: str): try: - await lib.unsubscribe(request.ctx.session, region) + with request.ctx.db.write() as session: + await lib.unsubscribe(session, region) except ValueError as e: return response.json({'error': str(e)}, HTTPStatus.NOT_FOUND) diff --git a/modules/map/test/test_api.py b/modules/map/test/test_api.py index cd7e3e0c..9bca98df 100644 --- a/modules/map/test/test_api.py +++ b/modules/map/test/test_api.py @@ -3,6 +3,8 @@ from http import HTTPStatus import pytest +from wrolpi.conftest import production_like_sessions, write_lock_held_briefly + @pytest.mark.asyncio async def test_get_files_empty(async_client, test_directory): @@ -151,3 +153,30 @@ async def test_set_and_get_map_default_location(async_client, test_directory): assert loc['lat'] == 40.7 assert loc['lon'] == -111.9 assert loc['zoom'] == 10.5 + + +@pytest.mark.asyncio +@pytest.mark.parametrize('step', ['subscribe', 'unsubscribe']) +async def test_subscription_writes_wait_for_a_concurrent_writer(async_client, test_session, test_directory, step): + """Subscribing and unsubscribing read before they write, so each must begin as a writer: SQLite fails a + deferred transaction's lock upgrade instantly ("database is locked") if another connection wrote meanwhile.""" + if step == 'unsubscribe': + _, response = await async_client.post('/api/map/subscribe', + content=json.dumps({'name': 'Alaska', 'region': 'us-alaska'})) + assert response.status_code == HTTPStatus.CREATED + test_session.commit() + db_file = test_session.get_bind().url.database + + with production_like_sessions(test_session): + # Held past any slow work a request does before it writes. + with write_lock_held_briefly(db_file, seconds=3): + if step == 'subscribe': + _, response = await async_client.post('/api/map/subscribe', + content=json.dumps({'name': 'Alaska', 'region': 'us-alaska'})) + else: + _, response = await async_client.delete('/api/map/subscribe/us-alaska') + assert response.status_code in (HTTPStatus.CREATED, HTTPStatus.NO_CONTENT), response.body + + _, response = await async_client.get('/api/map/subscribe') + regions = [i['region'] for i in response.json['subscriptions']] + assert regions == (['us-alaska'] if step == 'subscribe' else []) diff --git a/modules/videos/channel/api.py b/modules/videos/channel/api.py index b1c1ff7b..fc2a0322 100644 --- a/modules/videos/channel/api.py +++ b/modules/videos/channel/api.py @@ -80,9 +80,9 @@ def channel_post(request: Request, body: schema.ChannelPostRequest): @validate(schema.ChannelPutRequest) @wrol_mode_check async def channel_update(request: Request, channel_id: int, body: schema.ChannelPutRequest): - session = request.ctx.session body.directory = pathlib.Path(body.directory) if body.directory else None - channel = await lib.update_channel(session, data=body, channel_id=channel_id) + with request.ctx.db.write() as session: + channel = await lib.update_channel(session, data=body, channel_id=channel_id) save_channels_config.activate_switch() return response.raw('', HTTPStatus.NO_CONTENT, headers={'Location': f'/api/videos/channels/{channel.id}'}) @@ -93,8 +93,8 @@ async def channel_update(request: Request, channel_id: int, body: schema.Channel @openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse) @wrol_mode_check def channel_delete(request: Request, channel_id: int): - session = request.ctx.session - channel = lib.delete_channel(session, channel_id=channel_id) + with request.ctx.db.write() as session: + channel = lib.delete_channel(session, channel_id=channel_id) Events.send_deleted(f'Deleted Channel: {channel["name"]}') return response.raw('', HTTPStatus.NO_CONTENT) @@ -125,13 +125,13 @@ def channel_refresh(request: Request, channel_id: int): @validate(schema.ChannelTagRequest) @wrol_mode_check async def channel_tag(request: Request, channel_id: int, body: schema.ChannelTagRequest): - session = request.ctx.session directory = None if body.directory: directory = pathlib.Path(body.directory) directory = get_media_directory() / directory - await lib.tag_channel(session, body.tag_name, directory, channel_id) + with request.ctx.db.write() as session: + await lib.tag_channel(session, body.tag_name, directory, channel_id) return response.raw('', HTTPStatus.NO_CONTENT) diff --git a/modules/videos/channel/test/test_api.py b/modules/videos/channel/test/test_api.py index c9652875..c020e299 100644 --- a/modules/videos/channel/test/test_api.py +++ b/modules/videos/channel/test/test_api.py @@ -18,7 +18,7 @@ from modules.videos.lib import save_channels_config, get_channels_config from modules.videos.models import Channel, Video from wrolpi.collections.models import Collection from wrolpi.common import get_relative_to_media_directory, walk -from wrolpi.conftest import production_like_sessions +from wrolpi.conftest import production_like_sessions, write_lock_held_briefly from wrolpi.dates import now from wrolpi.db import get_db_session from wrolpi.downloader import Download, DownloadResult @@ -752,3 +752,39 @@ async def test_search_channels_by_name(async_client, test_session, channel_facto request, response = await async_client.post('/api/search_other_estimates', json=body) assert response.status_code == HTTPStatus.OK assert response.json['others']['channel_count'] == len(await search(body)), body + + +@pytest.mark.asyncio +@pytest.mark.parametrize('step', ['update', 'tag', 'delete']) +async def test_channel_writes_wait_for_a_concurrent_writer(async_client, test_session, test_directory, + channel_factory, tag_factory, step): + """Updating, tagging and deleting a Channel read before they write, so each must begin as a writer: SQLite + fails a deferred transaction's lock upgrade instantly ("database is locked") if another connection wrote + meanwhile.""" + channel = channel_factory(name='Lock Channel') + tag = await tag_factory() + test_session.commit() + channel_id, directory = channel.id, str(channel.directory) + db_file = test_session.get_bind().url.database + + with production_like_sessions(test_session) as maker: + # Held past any slow work a request does before it writes. + with write_lock_held_briefly(db_file, seconds=3): + if step == 'update': + body = dict(name='Renamed Channel', directory=directory) + _, response = await async_client.put(f'/api/videos/channels/{channel_id}', json=body) + elif step == 'tag': + body = dict(tag_name=tag.name, directory=str(get_relative_to_media_directory(directory))) + _, response = await async_client.post(f'/api/videos/channels/{channel_id}/tag', json=body) + else: + _, response = await async_client.delete(f'/api/videos/channels/{channel_id}') + assert response.status_code == HTTPStatus.NO_CONTENT, response.body + + with maker() as session: + channel = session.query(Channel).filter_by(id=channel_id).one_or_none() + if step == 'update': + assert channel.name == 'Renamed Channel' + elif step == 'tag': + assert channel.tag_name == tag.name + else: + assert channel is None