Collection create/update/tag/delete wait for a concurrent writer

These endpoints read, then wrote, in the request's deferred transaction.
SQLite fails a deferred transaction's lock upgrade instantly ("database
is locked") when another connection wrote in between, so a collection
change during a refresh or download could fail with a 500.  They now
write inside request.ctx.db.write(), which begins as a writer and lets
busy_timeout absorb the contention; errors are mapped outside the block
so a 4xx rolls back.

Fixes #674

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

wrolpi/collections/api.py

@@ -66,15 +66,16 @@ async def get_collections_endpoint(request: Request):
 @wrol_mode_check
 async def create_collection_endpoint(request: Request, body: schema.CollectionCreateRequest):
     """Create a new Collection.  Defaults to kind='playlist' (a manually-curated, ordered list)."""
-    session = request.ctx.session
     try:
-        collection = lib.create_collection(
-            session, name=body.name, description=body.description, kind=body.kind or 'playlist',
-            tag_name=body.tag_name)
+        # Read-then-write, so the transaction must begin as a writer (see `RequestDB.write`).
+        with request.ctx.db.write() as session:
+            collection = lib.create_collection(
+                session, name=body.name, description=body.description, kind=body.kind or 'playlist',
+                tag_name=body.tag_name)
+            data = lib.get_collection_with_stats(session, collection.id)
     except ValidationError as e:
         return json_response({'error': str(e)}, status=HTTPStatus.BAD_REQUEST)
 
-    data = lib.get_collection_with_stats(session, collection.id)
     return json_response({'collection': data}, status=HTTPStatus.CREATED)
 
 
@@ -194,19 +195,20 @@ async def put_collection_endpoint(request: Request, collection_id: int, body: sc
 
     Note: To clear a field, pass an empty string. To leave unchanged, omit the field.
     """
-    session = request.ctx.session
     try:
-        collection = lib.update_collection(
-            session,
-            collection_id=collection_id,
-            directory=body.directory,
-            tag_name=body.tag_name,
-            description=body.description,
-            name=body.name,
-        )
+        # Read-then-write, so the transaction must begin as a writer (see `RequestDB.write`).
+        with request.ctx.db.write() as session:
+            lib.update_collection(
+                session,
+                collection_id=collection_id,
+                directory=body.directory,
+                tag_name=body.tag_name,
+                description=body.description,
+                name=body.name,
+            )
 
-        # Return updated collection data
-        collection_data = lib.get_collection_with_stats(session, collection_id)
+            # Return updated collection data
+            collection_data = lib.get_collection_with_stats(session, collection_id)
         return json_response({'collection': collection_data})
 
     except UnknownCollection as e:
@@ -265,14 +267,16 @@ async def tag_collection_endpoint(request: Request, collection_id: int, body: sc
 
     If no directory is specified, the collection must already have one.
     """
-    session = request.ctx.session
     try:
-        result = await lib.tag_collection(
-            session,
-            collection_id=collection_id,
-            tag_name=body.tag_name,
-            directory=body.directory
-        )
+        # Read-then-write, so the transaction must begin as a writer (see `RequestDB.write`).  A file move runs
+        # in a background task with its own session, so the write lock is not held for it.
+        with request.ctx.db.write() as session:
+            result = await lib.tag_collection(
+                session,
+                collection_id=collection_id,
+                tag_name=body.tag_name,
+                directory=body.directory
+            )
         return json_response(result)
 
     except UnknownCollection as e:
@@ -331,9 +335,10 @@ async def delete_collection_endpoint(request: Request, collection_id: int):
     - Archives remain in the database but are no longer associated with a domain
     - Triggers domain config save
     """
-    session = request.ctx.session
     try:
-        collection = lib.delete_collection(session, collection_id=collection_id)
+        # Read-then-write, so the transaction must begin as a writer (see `RequestDB.write`).
+        with request.ctx.db.write() as session:
+            collection = lib.delete_collection(session, collection_id=collection_id)
         from wrolpi.events import Events
         Events.send_deleted(f'Deleted {collection["kind"]} collection: {collection["name"]}')
         return response.raw('', HTTPStatus.NO_CONTENT)

wrolpi/collections/test/test_playlist_items.py

@@ -188,6 +188,35 @@ async def test_item_write_endpoints_wait_for_a_concurrent_writer(async_client, t
         assert response.status_code == HTTPStatus.NO_CONTENT
 
 
+@pytest.mark.asyncio
+async def test_collection_write_endpoints_wait_for_a_concurrent_writer(async_client, test_session):
+    """Create, update, tag and delete each read before they write, so each must begin as a writer
+    (see `test_item_write_endpoints_wait_for_a_concurrent_writer`)."""
+    db_file = test_session.get_bind().url.database
+
+    with production_like_sessions(test_session):
+        with write_lock_held_briefly(db_file):
+            _, response = await async_client.post('/api/collections', json={'name': 'Locks'})
+        assert response.status_code == HTTPStatus.CREATED, response.json
+        cid = response.json['collection']['id']
+
+        with write_lock_held_briefly(db_file):
+            _, response = await async_client.put(f'/api/collections/{cid}', json={'description': 'Waited'})
+        assert response.status_code == HTTPStatus.OK, response.json
+        assert response.json['collection']['description'] == 'Waited'
+
+        with write_lock_held_briefly(db_file):
+            _, response = await async_client.post(f'/api/collections/{cid}/tag', json={'tag_name': 'Cooking'})
+        assert response.status_code == HTTPStatus.OK, response.json
+
+        with write_lock_held_briefly(db_file):
+            _, response = await async_client.delete(f'/api/collections/{cid}')
+        assert response.status_code == HTTPStatus.NO_CONTENT, response.json
+
+        _, response = await async_client.get(f'/api/collections/{cid}')
+        assert response.status_code == HTTPStatus.NOT_FOUND
+
+
 @pytest.mark.asyncio
 async def test_reorder_and_remove_items(async_client, test_session):
     """Items can be reordered by id and removed, with positions kept contiguous."""