Channel and map subscription changes wait for a concurrent writer (#851)

PUT/DELETE /api/videos/channels/<id>, POST /api/videos/channels/<id>/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 <noreply@anthropic.com>

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)
 

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 [])

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)
 

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