From cbf69e3013203cf3a8945c964e441ac94a42ff74 Mon Sep 17 00:00:00 2001 From: lrnselfreliance <33263708+lrnselfreliance@users.noreply.github.com> Date: Wed, 21 Oct 2026 16:50:09 -0600 Subject: [PATCH] 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 --- modules/repos/api.py | 47 ++++++++++++++--------------- modules/repos/lib.py | 49 ++++++++++++++++++------------- modules/repos/test/test_api.py | 41 +++++++++++++++++++++++++- modules/repos/test/test_import.py | 20 +++++++++++++ 4 files changed, 113 insertions(+), 44 deletions(-) diff --git a/modules/repos/api.py b/modules/repos/api.py index 88060cb0..d8b2172e 100644 --- a/modules/repos/api.py +++ b/modules/repos/api.py @@ -30,20 +30,20 @@ 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 - repo = lib.create_repository( - session, - body.url, - tag_name=body.tag_name or None, - frequency=lib.DEFAULT_REPO_FREQUENCY if body.frequency is None else body.frequency, - mode=body.mode or 'full', - branch=body.branch or None, - name=body.name or None, - description=body.description or None, - submodules=body.submodules, - directory=(body.directory or '').strip() or None, - ) - repo = Repository.find_by_id(session, repo.id, options=Repository.json_options()) + with request.ctx.db.write() as session: + repo = lib.create_repository( + session, + body.url, + tag_name=body.tag_name or None, + frequency=lib.DEFAULT_REPO_FREQUENCY if body.frequency is None else body.frequency, + mode=body.mode or 'full', + branch=body.branch or None, + name=body.name or None, + description=body.description or None, + submodules=body.submodules, + directory=(body.directory or '').strip() or None, + ) + 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 - 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()) + 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(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 - repo = Repository.find_by_id(session, repo_id) - lib.request_repository_update(session, repo) - session.commit() + with request.ctx.db.write() as session: + repo = Repository.find_by_id(session, repo_id) + lib.request_repository_update(session, repo) return response.empty(HTTPStatus.NO_CONTENT) diff --git a/modules/repos/lib.py b/modules/repos/lib.py index ecbfa40a..39a7ef40 100644 --- a/modules/repos/lib.py +++ b/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,26 +1057,30 @@ 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}') - 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) - mark_clone(source, repo.clone_token) + size = await asyncio.to_thread(get_repo_size, source) - 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) - _create_repository_download(session, repo, frequency) + 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) + 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 = size + _create_repository_download(session, repo, frequency) return repo diff --git a/modules/repos/test/test_api.py b/modules/repos/test/test_api.py index 1add704a..13b83ba5 100644 --- a/modules/repos/test/test_api.py +++ b/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() diff --git a/modules/repos/test/test_import.py b/modules/repos/test/test_import.py index bc107fe0..53c440a3 100644 --- a/modules/repos/test/test_import.py +++ b/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']