Zim tagging, auto-search and subscriptions wait for a concurrent writer

POST /api/zim/<id>/auto_search, /tag, /untag, /subscribe and
DELETE /api/zim/subscribe/<id> 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/zim/api.py

@@ -52,9 +52,9 @@ async def delete_zims(request: Request, zim_ids: str):
 )
 @validate(schema.ZimAutoSearchRequest)
 async def post_set_zim_auto_search(request: Request, zim_id: int, body: schema.ZimAutoSearchRequest):
-    session = request.ctx.session
     zim_id = int(zim_id)
-    lib.set_zim_auto_search(session, zim_id, body.auto_search)
+    with request.ctx.db.write() as session:
+        lib.set_zim_auto_search(session, zim_id, body.auto_search)
     return response.empty()
 
 
@@ -116,9 +116,8 @@ async def get_zim_entry(_: Request, zim_id: int, zim_path: str):
 )
 @validate(schema.TagZimEntry)
 async def post_zim_tag(request: Request, body: schema.TagZimEntry):
-    session = request.ctx.session
-
-    await lib.add_tag(session, body.tag_name, body.zim_id, body.zim_entry)
+    with request.ctx.db.write() as session:
+        await lib.add_tag(session, body.tag_name, body.zim_id, body.zim_entry)
     return response.empty(HTTPStatus.CREATED)
 
 
@@ -129,9 +128,8 @@ async def post_zim_tag(request: Request, body: schema.TagZimEntry):
 )
 @validate(schema.TagZimEntry)
 async def post_zim_untag(request: Request, body: schema.TagZimEntry):
-    session = request.ctx.session
-
-    await lib.untag(session, body.tag_name, body.zim_id, body.zim_entry)
+    with request.ctx.db.write() as session:
+        await lib.untag(session, body.tag_name, body.zim_id, body.zim_entry)
     return response.empty(HTTPStatus.NO_CONTENT)
 
 
@@ -159,9 +157,8 @@ async def get_zim_subscriptions(request: Request):
 )
 @validate(schema.ZimSubscribeRequest)
 async def post_zim_subscribe(request: Request, body: schema.ZimSubscribeRequest):
-    session = request.ctx.session
-
-    await lib.subscribe(session, body.name, body.language)
+    with request.ctx.db.write() as session:
+        await lib.subscribe(session, body.name, body.language)
     if download_manager.disabled or download_manager.stopped:
         # Downloads are disabled, warn the user.
         Events.send_downloads_disabled('Download created. But, downloads are disabled.')
@@ -174,9 +171,8 @@ async def post_zim_subscribe(request: Request, body: schema.ZimSubscribeRequest)
     summary='Unsubscribe to a particular Kiwix Zim',
 )
 async def delete_zim_subscription(request: Request, subscription_id: int):
-    session = request.ctx.session
-
-    await lib.unsubscribe(session, subscription_id)
+    with request.ctx.db.write() as session:
+        await lib.unsubscribe(session, subscription_id)
     Events.send_deleted(f'Zim subscription deleted')
     return response.empty(HTTPStatus.NO_CONTENT)
 

modules/zim/test/test_api.py

@@ -6,6 +6,7 @@ import pytest
 from modules.zim import lib
 from modules.zim.models import Zim
 from wrolpi import flags
+from wrolpi.conftest import production_like_sessions, write_lock_held_briefly
 from wrolpi.downloader import Download
 
 
@@ -314,3 +315,35 @@ async def test_zim_auto_search(test_session, async_client, zim_factory):
     request, response = await async_client.post('/api/zim/search_estimates', json=body)
     assert response.status_code == HTTPStatus.OK
     assert len(response.json['zims_estimates']) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('step', ['auto_search', 'tag', 'untag', 'subscribe', 'unsubscribe'])
+async def test_zim_writes_wait_for_a_concurrent_writer(async_client, test_session, test_zim, tag_factory, step):
+    """These requests 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."""
+    tag = await tag_factory()
+    test_session.commit()
+    entry = dict(tag_name=tag.name, zim_id=test_zim.id, zim_entry='home')
+    if step == 'untag':
+        await lib.add_tag(test_session, tag.name, test_zim.id, 'home')
+    if step == 'unsubscribe':
+        await lib.subscribe(test_session, 'Wikipedia (mini)', 'en')
+    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 == 'auto_search':
+                _, response = await async_client.post(f'/api/zim/{test_zim.id}/auto_search',
+                                                      content=json.dumps(dict(auto_search=False)))
+            elif step in ('tag', 'untag'):
+                _, response = await async_client.post(f'/api/zim/{step}', content=json.dumps(entry))
+            elif step == 'subscribe':
+                _, response = await async_client.post(
+                    '/api/zim/subscribe', content=json.dumps(dict(name='Wikipedia (mini)', language='en')))
+            else:
+                _, response = await async_client.delete('/api/zim/subscribe/1')
+
+    assert response.status_code in (HTTPStatus.CREATED, HTTPStatus.NO_CONTENT), response.body