Repo changes wait for a concurrent writer (#853)

Adding, importing, changing, deleting and updating a Repo 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().

Importing reads the clone with git and walks it for its size; those run
before the write transaction, so it is not held for them.  Everything
that can refuse the import still runs before the clone is changed.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

modules/repos/api.py

@@ -30,7 +30,7 @@ async def get_repos(request: Request):
 @wrol_mode_check
 async def post_repo(request: Request, body: schema.RepoCreateRequest):
     lib.get_repos_config().check_imported()
-    session = request.ctx.session
+    with request.ctx.db.write() as session:
         repo = lib.create_repository(
             session,
             body.url,
@@ -43,7 +43,7 @@ async def post_repo(request: Request, body: schema.RepoCreateRequest):
             submodules=body.submodules,
             directory=(body.directory or '').strip() or None,
         )
-    repo = Repository.find_by_id(session, repo.id, options=Repository.json_options())
+    repo = Repository.find_by_id(request.ctx.session, repo.id, options=Repository.json_options())
     return json_response(dict(repo=repo.__json__()), status=HTTPStatus.CREATED)
 
 
@@ -82,6 +82,7 @@ async def post_import(request: Request, body: schema.RepoImportRequest):
         submodules=body.submodules,
         name=body.name or None,
         description=body.description or None,
+        write=request.ctx.db.write,
     )
     repo = Repository.find_by_id(session, repo.id, options=Repository.json_options())
     return json_response(dict(repo=repo.__json__()), status=HTTPStatus.CREATED)
@@ -116,10 +117,10 @@ async def get_repo(request: Request, repo_id: int):
 @wrol_mode_check
 async def put_repo(request: Request, repo_id: int, body: schema.RepoUpdateRequest):
     lib.get_repos_config().check_imported()
-    session = request.ctx.session
+    with request.ctx.db.write() as session:
         lib.update_repository(session, repo_id, description=body.description, frequency=body.frequency,
                               mode=body.mode, branch=body.branch, submodules=body.submodules)
-    repo = Repository.find_by_id(session, repo_id, options=Repository.json_options())
+    repo = Repository.find_by_id(request.ctx.session, repo_id, options=Repository.json_options())
     return json_response(dict(repo=repo.__json__()))
 
 
@@ -129,7 +130,8 @@ async def put_repo(request: Request, repo_id: int, body: schema.RepoUpdateReques
 async def delete_repo(request: Request, repo_id: int):
     lib.get_repos_config().check_imported()
     delete_files = request.args.get('delete_files', '').lower() in ('1', 'true', 'yes')
-    await lib.delete_repository(request.ctx.session, repo_id, delete_files=delete_files)
+    with request.ctx.db.write() as session:
+        await lib.delete_repository(session, repo_id, delete_files=delete_files)
     return response.empty()
 
 
@@ -137,10 +139,9 @@ async def delete_repo(request: Request, repo_id: int):
 @openapi.definition(summary='Update a git Repo now')
 @wrol_mode_check
 async def post_update_repo(request: Request, repo_id: int):
-    session = request.ctx.session
+    with request.ctx.db.write() as session:
         repo = Repository.find_by_id(session, repo_id)
         lib.request_repository_update(session, repo)
-    session.commit()
     return response.empty(HTTPStatus.NO_CONTENT)
 
 

modules/repos/lib.py

@@ -14,7 +14,7 @@ import subprocess
 import tempfile
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
-from typing import List, NamedTuple, Optional, Tuple
+from typing import Callable, ContextManager, List, NamedTuple, Optional, Tuple
 from urllib.parse import urlparse
 
 from sqlalchemy import func, text
@@ -1037,9 +1037,14 @@ def inspect_import(session: Session, directory: str, name: str = None, tag_name:
 async def import_repository(session: Session, directory: str, url: str = None, confirm: bool = False,
                             tag_name: str = None, frequency: int = DEFAULT_REPO_FREQUENCY, mode: str = 'full',
                             branch: str = None, submodules: bool = False, name: str = None,
-                            description: str = None) -> Repository:
+                            description: str = None,
+                            write: Callable[[], ContextManager[Session]] = None) -> Repository:
     """Import the git clone in `directory` as a Repo.  It stays where it is, and from then on is a mirror of `url`
-    (which must be the clone's origin).  Local changes and commits are discarded by its updates."""
+    (which must be the clone's origin).  Local changes and commits are discarded by its updates.
+
+    `session` is used to read.  The Repo is created in `write()` (`request.ctx.db.write` in a request), a short
+    write transaction; the slow reads of the clone (git, its size) happen before it.  Without `write`, the Repo is
+    written in `session`."""
     _validate_frequency(frequency)
     info = inspect_import(session, directory, name, tag_name, url)
     source = resolve_import_directory(session, directory)
@@ -1052,25 +1057,29 @@ async def import_repository(session: Session, directory: str, url: str = None, c
     if not confirm:
         raise ValidationError('Confirm that this clone will become a mirror, and its local changes discarded')
 
-    # Like a new Repo, an imported one follows the remote's default branch unless a branch is chosen.
-    repo = _create_repository_records(session, parsed.url, tag_name, mode, branch, info['name'], description, source,
-                                      submodules)
-
-    # Everything which can refuse the import happens before the clone is changed.
+    # Everything which can refuse the import happens before the clone is changed.  These read the clone, and run
+    # before the write transaction so it is not held for them.
     if error := await asyncio.to_thread(working_tree_git_error, source):
         raise InvalidRepo(f'{directory} cannot be imported: {error}')
+    size = await asyncio.to_thread(get_repo_size, source)
+
+    with (write or (lambda: contextlib.nullcontext(session)))() as session:
+        # Like a new Repo, an imported one follows the remote's default branch unless a branch is chosen.
+        repo = _create_repository_records(session, parsed.url, tag_name, mode, branch, info['name'], description,
+                                          source, submodules)
         try:
             await asyncio.to_thread(sanitize_git_config, source)
         except ValueError as e:
             raise InvalidRepo(f'{directory} cannot be imported: {e}')
-    await asyncio.to_thread(_git_output_sync, source, 'config', '--replace-all', 'remote.origin.url', parsed.url)
+        await asyncio.to_thread(_git_output_sync, source, 'config', '--replace-all', 'remote.origin.url',
+                                parsed.url)
         mark_clone(source, repo.clone_token)
 
         for key, value in _read_head_commit(source).items():
             setattr(repo, key, value)
         repo.readme_path = find_readme(source)
         repo.readme_text = read_readme_text(source / repo.readme_path) if repo.readme_path else None
-    repo.size = await asyncio.to_thread(get_repo_size, source)
+        repo.size = size
         _create_repository_download(session, repo, frequency)
     return repo
 

modules/repos/test/test_api.py

@@ -5,7 +5,7 @@ import pytest
 
 from modules.repos import lib
 from modules.repos.models import Repository
-from wrolpi.conftest import await_file_worker
+from wrolpi.conftest import await_file_worker, production_like_sessions, write_lock_held_briefly
 from wrolpi.downloader import Download, DownloadFrequency, DownloadStatus
 from wrolpi.files.models import FileGroup
 
@@ -233,3 +233,42 @@ async def test_repo_changes_wait_for_config_import(test_session, test_directory,
     lib.import_repos_config()
     request, response = await async_client.put(f'/api/repos/{repo.id}', content=json.dumps(dict(description='x')))
     assert response.status_code == HTTPStatus.OK, response.json
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('step', ['create', 'change', 'delete', 'update_now'])
+async def test_repo_writes_wait_for_a_concurrent_writer(test_session, async_client, test_wrolpi_config,
+                                                        git_remote_factory, repos_download_manager, 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."""
+    remote = git_remote_factory()
+    repo_id = None
+    if step != 'create':
+        repo_id = lib.create_repository(test_session, remote.url).id
+        await repos_download_manager.wait_for_all_downloads()
+    test_session.commit()
+    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 == 'create':
+                _, response = await async_client.post('/api/repos', content=json.dumps(dict(url=remote.url)))
+            elif step == 'change':
+                _, response = await async_client.put(f'/api/repos/{repo_id}',
+                                                     content=json.dumps(dict(description='Waited')))
+            elif step == 'delete':
+                _, response = await async_client.delete(f'/api/repos/{repo_id}')
+            else:
+                _, response = await async_client.post(f'/api/repos/{repo_id}/update')
+        assert response.status_code in (HTTPStatus.OK, HTTPStatus.CREATED, HTTPStatus.NO_CONTENT), response.json
+
+        with maker() as session:
+            repos = session.query(Repository).all()
+            if step == 'delete':
+                assert repos == []
+            elif step == 'change':
+                assert repos[0].collection.description == 'Waited'
+            else:
+                assert len(repos) == 1
+    await repos_download_manager.wait_for_all_downloads()

modules/repos/test/test_import.py

@@ -7,6 +7,7 @@ import pytest
 from modules.repos import lib
 from modules.repos.errors import InvalidRepo, RepoConflict
 from modules.repos.models import Repository
+from wrolpi.conftest import production_like_sessions, write_lock_held_briefly
 from wrolpi.downloader import Download, DownloadStatus
 from wrolpi.errors import UnknownDirectory, ValidationError
 from wrolpi.files.models import FileGroup
@@ -285,3 +286,22 @@ async def test_create_repository_directory_refused(test_session, test_directory,
     assert response.status_code in (HTTPStatus.BAD_REQUEST, HTTPStatus.CONFLICT), response.json
     assert error in response.json['error'], response.json
     assert (test_directory / 'not-empty/file.txt').read_text() == 'mine'
+
+
+@pytest.mark.asyncio
+async def test_import_waits_for_a_concurrent_writer(test_session, test_directory, async_client, test_wrolpi_config,
+                                                    user_clone, repos_download_manager):
+    """Importing reads (and runs git) before it writes, so its write must begin as a writer: SQLite fails a
+    deferred transaction's lock upgrade instantly ("database is locked") if another connection wrote meanwhile."""
+    user_clone()
+    test_session.commit()
+    db_file = test_session.get_bind().url.database
+
+    with production_like_sessions(test_session) as maker:
+        with write_lock_held_briefly(db_file, seconds=3):
+            body = dict(directory='code/kiwix-tools', confirm=True, frequency=0)
+            _, response = await async_client.post('/api/repos/import', content=json.dumps(body))
+        assert response.status_code == HTTPStatus.CREATED, response.json
+
+        with maker() as session:
+            assert [i.name for i in session.query(Repository)] == ['kiwix-tools']