fix: a concurrent writer no longer aborts a file move with "database is locked"

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) <noreply@anthropic.com>

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):

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'

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
                     )