From c894ba1fddebe5ae60546fd0dfa1c383fff081a8 Mon Sep 17 00:00:00 2001 From: Roland Date: Wed, 21 Oct 2026 07:47:10 -0600 Subject: [PATCH] Download and tag requests wait for a concurrent writer Deleting, restarting, clearing and batch-editing Downloads, and creating, renaming and deleting Tags, read then wrote in the request's deferred transaction. SQLite fails that lock upgrade instantly ("database is locked") when another connection wrote in between, so these failed with a 500 while a download or refresh was writing. They now write inside request.ctx.db.write(), which begins as a writer and lets busy_timeout absorb the contention. Co-Authored-By: Claude Opus 5.5 --- wrolpi/root_api.py | 40 +++++++++++++++++------------------ wrolpi/test/test_root_api.py | 41 ++++++++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 20 deletions(-) diff --git a/wrolpi/root_api.py b/wrolpi/root_api.py index 31f344c6..59cd8036 100644 --- a/wrolpi/root_api.py +++ b/wrolpi/root_api.py @@ -443,16 +443,16 @@ async def put_download(request: Request, download_id: int, body: schema.Download @openapi.description('Delete a download') @wrol_mode_check async def delete_download(request: Request, download_id: int): - session = request.ctx.session - deleted = download_manager.delete_download(session, download_id) + with request.ctx.db.write() as session: + deleted = download_manager.delete_download(session, download_id) return response.empty(HTTPStatus.NO_CONTENT if deleted else HTTPStatus.NOT_FOUND) @api_bp.post('/download//restart') @openapi.description('Restart a download.') async def restart_download(request: Request, download_id: int): - session = request.ctx.session - download_manager.restart_download(session, download_id) + with request.ctx.db.write() as session: + download_manager.restart_download(session, download_id) return response.empty() @@ -498,16 +498,16 @@ async def enable_downloads(_: Request): @api_bp.post('/download/clear_completed') @openapi.description('Clear completed downloads') async def clear_completed(request: Request): - session = request.ctx.session - download_manager.delete_completed(session) + with request.ctx.db.write() as session: + download_manager.delete_completed(session) return response.empty() @api_bp.post('/download/clear_failed') @openapi.description('Clear failed downloads') async def clear_failed(request: Request): - session = request.ctx.session - download_manager.delete_failed(session) + with request.ctx.db.write() as session: + download_manager.delete_failed(session) return response.empty() @@ -522,8 +522,8 @@ async def retry_once(request: Request): @api_bp.post('/download/delete_once') @openapi.description('Delete all once downloads') async def delete_once(request: Request): - session = request.ctx.session - download_manager.delete_once(session) + with request.ctx.db.write() as session: + download_manager.delete_once(session) return response.empty() @@ -535,8 +535,8 @@ async def delete_once(request: Request): ) @wrol_mode_check async def batch_delete_downloads(request: Request, body: schema.DownloadBatchRequest): - session = request.ctx.session - deleted_ids = download_manager.delete_downloads_by_ids(session, body.download_ids) + with request.ctx.db.write() as session: + deleted_ids = download_manager.delete_downloads_by_ids(session, body.download_ids) return json_response({'deleted_ids': deleted_ids}) @@ -548,8 +548,8 @@ async def batch_delete_downloads(request: Request, body: schema.DownloadBatchReq ) @wrol_mode_check async def batch_retry_downloads(request: Request, body: schema.DownloadBatchRequest): - session = request.ctx.session - count = download_manager.retry_downloads_by_ids(session, body.download_ids) + with request.ctx.db.write() as session: + count = download_manager.retry_downloads_by_ids(session, body.download_ids) return json_response({'retried_count': count}) @@ -561,8 +561,8 @@ async def batch_retry_downloads(request: Request, body: schema.DownloadBatchRequ ) @wrol_mode_check async def batch_clear_downloads(request: Request, body: schema.DownloadBatchRequest): - session = request.ctx.session - deleted_ids = download_manager.clear_completed_by_ids(session, body.download_ids) + with request.ctx.db.write() as session: + deleted_ids = download_manager.clear_completed_by_ids(session, body.download_ids) return json_response({'deleted_ids': deleted_ids}) @@ -682,8 +682,8 @@ async def get_recent_tags_request(request: Request): validate=True, ) async def post_tag(request: Request, body: schema.TagRequest, tag_id: int = None): - session = request.ctx.session - await tags.upsert_tag(session, body.name, body.color, tag_id) + with request.ctx.db.write() as session: + await tags.upsert_tag(session, body.name, body.color, tag_id) if tag_id: return response.empty(HTTPStatus.OK) else: @@ -692,8 +692,8 @@ async def post_tag(request: Request, body: schema.TagRequest, tag_id: int = None @api_bp.delete('/tag/') async def delete_tag_request(request: Request, tag_id: int): - session = request.ctx.session - Tag.find_by_id(session, tag_id).delete() + with request.ctx.db.write() as session: + Tag.find_by_id(session, tag_id).delete() return response.empty() diff --git a/wrolpi/test/test_root_api.py b/wrolpi/test/test_root_api.py index 4b8eff7a..bca4c016 100644 --- a/wrolpi/test/test_root_api.py +++ b/wrolpi/test/test_root_api.py @@ -1084,3 +1084,44 @@ async def test_mutating_request_survives_concurrent_writer(async_client, test_se assert [i.url for i in session.query(Download).all()] == ['https://example.com/during-refresh'] finally: session.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('method,path,body,status', [ + # Deleting a Download saves the skip list before it deletes, so the lock is held longer (see `hold` below). + ('delete', '/api/download/{failed}', None, HTTPStatus.NO_CONTENT), + ('post', '/api/download/{failed}/restart', None, HTTPStatus.NO_CONTENT), + ('post', '/api/download/clear_completed', None, HTTPStatus.NO_CONTENT), + ('post', '/api/download/clear_failed', None, HTTPStatus.NO_CONTENT), + ('post', '/api/download/delete_once', None, HTTPStatus.NO_CONTENT), + ('post', '/api/download/batch/delete', {'download_ids': ['{failed}']}, HTTPStatus.OK), + ('post', '/api/download/batch/retry', {'download_ids': ['{failed}']}, HTTPStatus.OK), + ('post', '/api/download/batch/clear', {'download_ids': ['{complete}']}, HTTPStatus.OK), + ('post', '/api/tag', {'name': 'New', 'color': '#000000'}, HTTPStatus.CREATED), + ('post', '/api/tag/{tag}', {'name': 'Renamed', 'color': '#ffffff'}, HTTPStatus.OK), + ('delete', '/api/tag/{tag}', None, HTTPStatus.NO_CONTENT), +]) +async def test_download_and_tag_writes_wait_for_a_concurrent_writer(async_client, test_session, test_download_manager, + tag_factory, method, path, body, status): + """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 + (see `test_mutating_request_survives_concurrent_writer`).""" + failed = Download(url='https://example.com/failed', status='failed') + complete = Download(url='https://example.com/complete', status='complete') + test_session.add_all([failed, complete]) + tag = await tag_factory('Old') + test_session.commit() + ids = dict(failed=failed.id, complete=complete.id, tag=tag.id) + path = path.format(**ids) + if body and 'download_ids' in body: + body = {'download_ids': [int(i.format(**ids)) for i in body['download_ids']]} + db_file = test_session.get_bind().url.database + + with production_like_sessions(test_session): + # Hold the lock until the request's first write, even when it does slow work first. + hold = 3 if method == 'delete' and path.startswith('/api/download/') else 1 + with write_lock_held_briefly(db_file, seconds=hold): + kwargs = {'content': json.dumps(body)} if body else {} + _, response = await getattr(async_client, method)(path, **kwargs) + + assert response.status_code == status, response.body