From dbbcab4d908286e1dc279ddb183dc622e0f3fb11 Mon Sep 17 00:00:00 2001 From: Roland Date: Sat, 1 Aug 2026 19:18:58 -0600 Subject: [PATCH] fix: a concurrent writer no longer aborts a file move with "database is locked" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tagging a Channel with "move files" checked failed the whole move: the Collection's directory was updated, but its files stayed put. `build_move_plan_bulk` reads FileGroups and then INSERTs the ones missing from the DB, inside a transaction that begins *deferred*. SQLite refuses a mid-transaction lock upgrade immediately — busy_timeout is skipped for upgrades to avoid deadlock — so any concurrent writer (a refresh, a download, a config save) aborted planning outright. The same pattern in `_execute_move_chunks` fails the same way on its Directory INSERT. Both transactions now begin as writers, so busy_timeout absorbs the contention. That requires the writes nested inside the move to share its transaction rather than open a second connection, which would deadlock against the lock the move already holds: `_bulk_update_file_groups_db` now runs on the caller's session, and the old-directory cleanup deletes its Directory record through that session instead of via `delete_directory`. As a side effect, a failed move no longer leaves earlier chunks' path updates committed after their files have been moved back. Co-Authored-By: Claude Opus 5 (1M context) --- wrolpi/files/lib.py | 36 +++++----- wrolpi/files/test/test_worker_move.py | 94 ++++++++++++++++++++++++++- wrolpi/files/worker.py | 22 +++++-- 3 files changed, 127 insertions(+), 25 deletions(-) diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index 7f6c3761..7a2a6983 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -18,7 +18,7 @@ from pathlib import Path from typing import Callable, List, Tuple, Union, Dict, Generator, Iterable, Set import cachetools.func -from sqlalchemy import asc, or_ +from sqlalchemy import asc, or_, text as sa_text from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -1594,23 +1594,25 @@ def _bulk_update_file_groups_db(session: Session, chunk_plan: Dict[pathlib.Path, if not chunk_plan: return - with get_db_curs(commit=True) as curs: - # Build the update rows: [old_path, new_dir, new_path] - updates = json.dumps([ - [str(old_path), str(new_path.parent), str(new_path)] - for old_path, new_path in chunk_plan.items() - ]) + # Build the update rows: [old_path, new_dir, new_path] + updates = json.dumps([ + [str(old_path), str(new_path.parent), str(new_path)] + for old_path, new_path in chunk_plan.items() + ]) - # Use UPDATE FROM json_each pattern for efficient batch update - sql = ''' - UPDATE file_group AS fg SET - directory = json_extract(v.value, '$[1]'), - primary_path = json_extract(v.value, '$[2]'), - indexed = CASE WHEN fg.data IS NULL THEN FALSE ELSE fg.indexed END - FROM json_each(?) AS v - WHERE fg.primary_path = json_extract(v.value, '$[0]') - ''' - curs.execute(sql, (updates,)) + # Use UPDATE FROM json_each pattern for efficient batch update. This runs on the caller's + # session (rather than a second connection) so it shares the move's transaction: a second + # connection would deadlock against the write lock the caller already holds, and its separate + # commit would survive a rollback of the rest of the move. + sql = ''' + UPDATE file_group AS fg SET + directory = json_extract(v.value, '$[1]'), + primary_path = json_extract(v.value, '$[2]'), + indexed = CASE WHEN fg.data IS NULL THEN FALSE ELSE fg.indexed END + FROM json_each(:updates) AS v + WHERE fg.primary_path = json_extract(v.value, '$[0]') + ''' + session.execute(sa_text(sql), dict(updates=updates)) def _json_serial(obj): diff --git a/wrolpi/files/test/test_worker_move.py b/wrolpi/files/test/test_worker_move.py index b12c7386..b6ee6284 100644 --- a/wrolpi/files/test/test_worker_move.py +++ b/wrolpi/files/test/test_worker_move.py @@ -1,9 +1,15 @@ """Tests for the FileWorker move task.""" +import contextlib +import sqlite3 +import threading +import time + import pytest +from wrolpi.conftest import production_like_sessions from wrolpi.files.lib import _move_file_group_files from wrolpi.files.models import FileGroup, Directory -from wrolpi.files.worker import file_worker, FileTask, FileTaskType +from wrolpi.files.worker import file_worker, FileTask, FileTaskType, build_move_plan_bulk @pytest.mark.asyncio @@ -530,3 +536,89 @@ async def test_file_worker_move_cleans_directory_records( ).one_or_none() assert source_record is not None, "source Directory record should remain (directory still exists)" assert source_dir.exists(), "source should still exist on disk (empty)" + + +@contextlib.contextmanager +def _write_lock_held_briefly(db_file: str, seconds: float = 1.0): + """Hold the SQLite write lock on `db_file` for `seconds`, from another connection. + + Simulates the concurrent writer (refresh, download, config save) that a real WROLPi runs + alongside a move. The lock is released while the test's move is still planning, so a + correctly-written move waits (busy_timeout) and then succeeds. + """ + holder = sqlite3.connect(db_file, timeout=30, check_same_thread=False) + holder.isolation_level = None + holder.execute('PRAGMA busy_timeout=30000') + holder.execute('BEGIN IMMEDIATE') + + def release(): + time.sleep(seconds) + holder.execute('ROLLBACK') + holder.close() + + thread = threading.Thread(target=release, daemon=True) + thread.start() + try: + yield + finally: + thread.join(timeout=30) + + +@pytest.mark.asyncio +async def test_build_move_plan_waits_for_concurrent_writer( + async_client, test_session, test_directory, make_files_structure +): + """`build_move_plan_bulk` must not fail instantly when another connection is writing. + + Regression test for the tag-a-channel move that died with `database is locked` while + INSERTing a FileGroup for an unindexed file. The plan's transaction begins *deferred* + (a read), then upgrades to a writer at that INSERT; SQLite refuses a lock upgrade + immediately, skipping busy_timeout, so any concurrent writer aborts the whole move. + Beginning the transaction as a writer lets busy_timeout absorb the contention instead. + """ + # An unindexed file, so planning must INSERT a FileGroup (the write that failed in production). + file1, = make_files_structure(['source/file1.txt']) + dest = test_directory / 'destination' + dest.mkdir() + + db_file = test_session.get_bind().url.database + + with production_like_sessions(test_session): + with _write_lock_held_briefly(db_file): + plan, _ = await build_move_plan_bulk([file1], dest) + + assert plan == {file1: dest / 'file1.txt'} + + +@pytest.mark.asyncio +async def test_file_worker_move_survives_concurrent_writer( + async_client, test_session, test_directory, make_files_structure +): + """A move of already-indexed files must survive a concurrent writer holding the write lock. + + Planning writes nothing here (every file is already indexed), so the first write is the + `FileGroup` UPDATE inside `_execute_move_chunks`. That transaction must begin as a writer, + or SQLite refuses the mid-transaction lock upgrade instantly and the move aborts. + """ + file1, = make_files_structure(['source/file1.txt']) + FileGroup.from_paths(test_session, file1) + test_session.commit() + + dest = test_directory / 'destination' + dest.mkdir() + + db_file = test_session.get_bind().url.database + + task = FileTask(FileTaskType.move, [file1], destination=dest) + file_worker.private_queue.put_nowait(task) + + with production_like_sessions(test_session) as maker: + with _write_lock_held_briefly(db_file): + await file_worker.process_queue() + + assert (dest / 'file1.txt').is_file(), 'file was not moved' + assert not file1.exists(), 'file was left at its old path' + + session = maker() + fg = session.query(FileGroup).one() + assert fg.primary_path == dest / 'file1.txt', 'FileGroup was not updated to the new path' diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index dc88c9bc..6b508efb 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -27,7 +27,7 @@ from wrolpi import flags from wrolpi.common import apply_modelers, apply_refresh_cleanup from wrolpi.common import get_media_directory, get_wrolpi_config, logger, walk, chunks, unique_by_predicate from wrolpi.dates import now -from wrolpi.db import get_db_session, get_db_curs +from wrolpi.db import get_db_session, get_db_curs, get_immediate_db_session from wrolpi.errors import NoPrimaryFile from wrolpi.events import Events from wrolpi.files.lib import ( @@ -556,7 +556,11 @@ async def build_move_plan_bulk( if progress_callback and i % 100 == 0: progress_callback(i, total_sources * 2) # *2 because file collection is second half - with get_db_session() as session: + # Planning reads FileGroups and then INSERTs the ones missing from the DB, so the transaction + # must begin as a writer. A deferred transaction upgrading to a writer mid-way gets "database + # is locked" *immediately* (busy_timeout is skipped for lock upgrades), so any concurrent + # writer - a refresh, a download, a config save - would abort the whole move. + with get_immediate_db_session() as session: try: # Create temp table for source paths. SQLite has no ON COMMIT DROP; the table is # explicitly dropped below on success/failure (and IF NOT EXISTS + DELETE guard @@ -1616,13 +1620,15 @@ class FileWorker: revert_plan[new_path] = old_file chunk_plan[old_file] = new_path - # Delete old directories + # Delete old directories. `delete_directory` is not used here because it opens its own + # session; a second connection deadlocks against the write lock this transaction holds. for old_dir in old_dirs: if old_dir.is_dir(): try: - delete_directory(old_dir) + old_dir.rmdir() except OSError: - pass # Directory not empty yet + continue # Directory not empty yet + session.query(Directory).filter_by(path=str(old_dir)).delete(synchronize_session=False) # Insert new Directory records missing_dirs = new_directories - existing_directories - inserted_directories @@ -2037,9 +2043,11 @@ class FileWorker: operation_total=len(plan), ) - # Execute the plan in chunks + # Execute the plan in chunks. This reads FileGroups/Directories then updates them, so + # it must begin as a writer; a deferred transaction's lock upgrade fails instantly when + # anything else is writing (see `build_move_plan_bulk`). with flags.file_worker_discovery: - with get_db_session(commit=True) as session: + with get_immediate_db_session() as session: await self._execute_move_chunks( plan, session, created_directories, revert_plan )