db: write intent is the opt-in, not a second session helper

Three "database is locked" failures on the docker WROLPi, one cause each:

  * `INSERT INTO download` from POST /api/download -- a user creating a
    video download while a refresh ran (2026-08-06 14:06:46);
  * six `INSERT INTO file_group` from `upsert_file`, uploads landing in
    the same three-minute window;
  * `UPDATE download SET status` from `renew_recurring_downloads`.

All three are the deferred read-then-write trap: a transaction that
begins as a reader and upgrades at its first write is refused
*instantly* when another connection wrote in between, because
busy_timeout is skipped for lock upgrades.  `get_immediate_db_session`
existed to avoid it, but an opt-in helper only works if every author
remembers, and read-then-write is the ordinary shape of a write.  So it
is gone: `get_db_session(commit=True)` -- the existing declaration of
write intent -- now begins IMMEDIATE.  Rails 7.1 and Django 5.1 made the
same call, defaulting SQLite transactions to IMMEDIATE rather than
asking application code to opt in.

That default is only safe if write transactions stay short, and one did
not.  `video_cleanup` ran its whole Channel-claiming join --
`UPDATE video ... FROM channel ... LEFT JOIN file_group`, O(channels x
file_groups) -- inside one write transaction: 110 to 130 seconds each of
the eight times it appears in the log, on a library with ~1500 channels.
While it ran, *every* writer on the box failed, including ones that
began IMMEDIATE, because the hold far exceeds the 30s busy_timeout.  The
14:04-14:06 failures above sit inside its 130-second window.  The join
is now a read -- WAL readers never block writers -- and only the
resulting UPDATEs take the lock, 500 rows per transaction.
`claim_videos_for_channels` had the same join and the same fix; channel
import calls it with every channel in channels.yaml.

Also here:

  * `renew_recurring_downloads` was a read session with an inner commit;
    it is a write session now.  Its computed `next_download` values now
    persist even when nothing was renewed -- that is the point of
    computing them.
  * POST /api/download no longer writes through `request.ctx.session`.
    That session is committed by the response middleware, so a writer
    would hold the lock across the whole request; a write session
    commits before the handler returns, ahead of the background dispatch
    it fires.
  * `_immediate_txn` is a ContextVar, not a threading.local.  The
    writers that set it include coroutines, and one asyncio thread runs
    many of them interleaved.
  * `write_lock_held_briefly` moved from the move tests into conftest,
    since two suites now need it.

Both regression tests were verified by mutation: reverting either site
alone fails its own test, and the API one fails with the production
error verbatim.

Sources: fractaledmind.com/2023/12/11/sqlite-on-rails-improving-concurrency,
tenthousandmeters.com/blog/sqlite-concurrent-writes-and-database-is-locked-errors,
berthub.eu/articles/posts/a-brief-post-on-sqlite3-database-locked-despite-timeout

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

modules/videos/__init__.py

@@ -76,38 +76,98 @@ async def video_modeler(progress_callback: Callable[[int], None] = None):
         await asyncio.sleep(0)
 
 
+# Rows written per transaction when claiming Videos for their Channels.  Small enough that the
+# write lock is held for milliseconds at a time; large enough that a big library does not pay a
+# transaction per row.
+CLAIM_CHUNK_SIZE = 500
+
+
+def _chunks(items: list, size: int = CLAIM_CHUNK_SIZE):
+    for start in range(0, len(items), size):
+        yield items[start:start + size]
+
+
+def _first_claim_per_video(rows) -> List[Tuple[int, int]]:
+    """One `(video_id, channel_id)` per Video.
+
+    A Video nested under two Channel directories matches both; the single `UPDATE ... FROM` this
+    replaced also picked one arbitrarily, so keep the first row and drop the rest rather than
+    writing the same Video twice."""
+    claims = dict()
+    for video_id, channel_id in rows:
+        claims.setdefault(video_id, channel_id)
+    return list(claims.items())
+
+
+def _claim_videos(claims: List[Tuple[int, int]]):
+    """Assign `(video_id, channel_id)` pairs, a chunk per write transaction.
+
+    The pairs are computed by a *read* — WAL readers never block writers — and only these short
+    UPDATEs take the write lock.  Doing the whole thing as one `UPDATE ... FROM channel` held the
+    lock for the entire join: 110-130 seconds per refresh on a library with ~1500 channels, which
+    is well past the 30s `busy_timeout`, so every other writer on the box failed with "database is
+    locked" for two solid minutes.
+
+    `channel_id IS NULL` is re-checked in the UPDATE because the read ran in an earlier
+    transaction: a download may have claimed the Video since, and it should win.
+    """
+    if not claims:
+        return
+    claimed = 0
+    for chunk in _chunks(claims):
+        with get_db_curs(commit=True) as curs:
+            curs.executemany('UPDATE video SET channel_id = ? WHERE id = ? AND channel_id IS NULL',
+                             [(channel_id, video_id) for video_id, channel_id in chunk])
+        claimed += len(chunk)
+    logger.debug(f'Claimed {claimed} Videos for their Channels')
+
+
+# Videos in a Channel's directory (or below it) that no Channel has claimed yet.
+_UNCLAIMED_VIDEOS_SQL = '''
+                        SELECT v.id, c.id
+                        FROM video v
+                                 INNER JOIN file_group fg ON fg.id = v.file_group_id
+                                 INNER JOIN collection col ON fg.directory = col.directory
+                            OR fg.directory LIKE col.directory || '/%'
+                                 INNER JOIN channel c ON c.collection_id = col.id
+                        WHERE v.channel_id IS NULL
+                        '''
+
+
 @register_refresh_cleanup
 @limit_concurrent(1)
 def video_cleanup():
     logger.info('Claiming Videos for their Channels')
-    with get_db_curs(commit=True) as curs:
+    # Read the FileGroups that are no longer video/audio, then unmodel them in short transactions.
-        # Delete all Videos if the FileModel no longer contains a video or audio file.
+    with get_db_curs() as curs:
         curs.execute('''
-                     UPDATE file_group
+                     SELECT id
-                     SET model = NULL
+                     FROM file_group
                      WHERE model = 'video'
                        AND mimetype NOT LIKE 'video/%'
                        AND mimetype NOT LIKE 'audio/%'
-                     RETURNING id
                      ''')
-        deleted_ids = [i['id'] for i in curs.fetchall()]
+        stale_ids = [i['id'] for i in curs.fetchall()]
-        if deleted_ids:
+    for chunk in _chunks(stale_ids):
+        ids = json.dumps(chunk)
+        with get_db_curs(commit=True) as curs:
+            curs.execute('''
+                         UPDATE file_group
+                         SET model = NULL
+                         WHERE id IN (SELECT value FROM json_each(:ids))
+                         ''', dict(ids=ids))
             curs.execute('''
                          DELETE
                          FROM video
                          WHERE file_group_id IN (SELECT value FROM json_each(:ids))
-                         ''', dict(ids=json.dumps(deleted_ids)))
+                         ''', dict(ids=ids))
-        # Claim all Videos in a Channel's directory for that Channel.  But, only if they have not yet been claimed.
+
-        curs.execute('''
+    # Claim all Videos in a Channel's directory for that Channel.  But, only if they have not yet
-                     UPDATE video AS v
+    # been claimed.  The join is the slow part, so it runs as a read; only the writes take the lock.
-                     SET channel_id = c.id
+    with get_db_curs() as curs:
-                     FROM channel c
+        curs.execute(_UNCLAIMED_VIDEOS_SQL)
-                              INNER JOIN collection col ON col.id = c.collection_id
+        claims = _first_claim_per_video(curs.fetchall())
-                              LEFT JOIN file_group fg ON fg.directory = col.directory
+    _claim_videos(claims)
-                                                      OR fg.directory LIKE col.directory || '/%'
-                     WHERE v.channel_id IS NULL
-                       AND fg.id = v.file_group_id
-                     ''')
 
 
 
@@ -126,15 +186,13 @@ def claim_videos_for_channels(channel_ids: List[int]):
         return
 
     logger.info(f'Claiming Videos for {len(channel_ids)} channel(s)')
-    with get_db_curs(commit=True) as curs:
+    # Read the claims, then write them in chunks -- the join must not hold the write lock (see
-        curs.execute('''
+    # `_claim_videos`).  Channel import calls this with every channel in channels.yaml, so it is
-                     UPDATE video AS v
+    # just as slow as the full `video_cleanup` sweep.
-                     SET channel_id = c.id
+    with get_db_curs() as curs:
-                     FROM channel c
+        curs.execute(f'''
-                              INNER JOIN collection col ON col.id = c.collection_id
+                     {_UNCLAIMED_VIDEOS_SQL}
-                              LEFT JOIN file_group fg ON fg.directory = col.directory
-                                                      OR fg.directory LIKE col.directory || '/%'
-                     WHERE v.channel_id IS NULL
-                       AND fg.id = v.file_group_id
                        AND c.id IN (SELECT value FROM json_each(:channel_ids))
                      ''', dict(channel_ids=json.dumps(list(channel_ids))))
+        claims = _first_claim_per_video(curs.fetchall())
+    _claim_videos(claims)

modules/videos/test/test_channel_directory_scenarios.py

@@ -14,6 +14,9 @@ These tests use the REAL modeling flow:
 4. Verify the video is correctly associated with the channel
 """
 
+import contextlib
+from unittest import mock
+
 import pytest
 
 from modules.videos.models import Channel, Video
@@ -810,3 +813,71 @@ class TestVideoCleanupWithYearSubdirs:
         # video_cleanup should claim the video via LIKE clause
         assert video.channel_id == channel.id, \
             f'video_cleanup should claim video in year subdir, got channel_id={video.channel_id}'
+
+
+@pytest.mark.asyncio
+async def test_video_cleanup_claims_outside_a_write_transaction(async_client, test_session, test_directory,
+                                                                video_file, refresh_files):
+    """The Channel-claiming join must not run inside the write transaction.
+
+    Regression test for two minutes of global write starvation: `video_cleanup` ran the whole
+    `UPDATE video ... FROM channel ... LEFT JOIN file_group` in one `BEGIN IMMEDIATE` transaction.
+    The join is O(channels x file_groups), and on a library with ~1500 channels it held the write
+    lock for 110-130 seconds per refresh -- well past the 30s busy_timeout, so *every* other writer
+    failed with "database is locked", including correctly-written ones.  The join must be a read;
+    only the resulting UPDATEs may take the lock.
+    """
+    from modules.videos import video_cleanup, _UNCLAIMED_VIDEOS_SQL
+    from wrolpi.files import worker as worker_module  # noqa: F401  (import parity with other tests)
+    import modules.videos as videos_module
+
+    channel_dir = test_directory / 'videos' / 'LockChannel'
+    channel_dir.mkdir(parents=True, exist_ok=True)
+    collection = Collection(name='LockChannel', kind='channel', directory=channel_dir)
+    test_session.add(collection)
+    test_session.flush([collection])
+    channel = Channel(collection_id=collection.id, url='https://example.com/lockchannel')
+    test_session.add(channel)
+    test_session.commit()
+
+    video_file.rename(channel_dir / 'A Video.mp4')
+    await refresh_files()
+
+    video = test_session.query(Video).one()
+    video.channel_id = None
+    test_session.commit()
+
+    # Record whether each statement ran in a write transaction (`get_db_curs(commit=True)`).
+    executed = []
+    real_get_db_curs = videos_module.get_db_curs
+
+    @contextlib.contextmanager
+    def recording_get_db_curs(commit: bool = False):
+        with real_get_db_curs(commit=commit) as curs:
+            class _Recorder:
+                def __getattr__(self, name):
+                    return getattr(curs, name)
+
+                def execute(self, statement, *args, **kwargs):
+                    executed.append((commit, statement))
+                    return curs.execute(statement, *args, **kwargs)
+
+                def executemany(self, statement, *args, **kwargs):
+                    executed.append((commit, statement))
+                    return curs.executemany(statement, *args, **kwargs)
+
+            yield _Recorder()
+
+    with mock.patch.object(videos_module, 'get_db_curs', recording_get_db_curs):
+        video_cleanup()
+
+    test_session.expire_all()
+    assert test_session.query(Video).one().channel_id == channel.id, 'the Video was not claimed'
+
+    join_statements = [(commit, s) for commit, s in executed if 'FROM video v' in s]
+    assert join_statements, 'the claim join did not run'
+    assert not any(commit for commit, _ in join_statements), \
+        'the Channel-claiming join ran inside a write transaction, holding the lock across it'
+    assert any(commit for commit, s in executed if s.startswith('UPDATE video SET channel_id')), \
+        'the claims were not written'
+    assert _UNCLAIMED_VIDEOS_SQL.strip().startswith('SELECT'), 'the claim query must be a read'

wrolpi/api_utils.py

@@ -128,7 +128,14 @@ api_app.error_handler.add(Exception, json_error_handler)
 
 @api_app.middleware('request')
 async def inject_session(request: Request):
-    """Inject a database session into the request context."""
+    """Inject a database session into the request context.
+
+    This session is deliberately *deferred*, even for a POST: it is committed by the response
+    middleware, so a writer would hold the SQLite write lock for the whole handler -- including the
+    slow parts (a chunk being written to disk, a subprocess) and including any background task the
+    handler fires before returning.  A handler that writes should use `get_db_session(commit=True)`,
+    which begins as a writer and commits before the handler returns.
+    """
     from wrolpi.db import get_db_context
     engine, session = get_db_context()
     request.ctx.session = session

wrolpi/conftest.py

@@ -10,6 +10,7 @@ import logging
 import multiprocessing
 import pathlib
 import shutil
+import sqlite3
 import tempfile
 import threading
 import time
@@ -399,6 +400,36 @@ def make_test_ctx(*, is_cancelled=None, can_download=None,
     )
 
 
+@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 whatever the test is doing.  The lock is released while the test's work is still in
+    flight, so a correctly-written transaction waits (busy_timeout) and then succeeds; a deferred
+    one that upgrades its lock mid-transaction fails instantly instead.
+
+    Pair with `production_like_sessions` — the shared `test_session` uses one connection, which
+    cannot contend with itself.
+    """
+    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)
+
+
 @contextlib.contextmanager
 def production_like_sessions(test_session) -> Generator[sessionmaker, Any, None]:
     """Patch get_db_session to hand out a *fresh* Session per call, like production.

wrolpi/db.py

@@ -7,6 +7,7 @@ Engines are created lazily (the media directory must be known first) and use Nul
 fresh session per use.  Every connection gets the same PRAGMAs (WAL, busy_timeout, foreign keys)
 via `create_wrolpi_engine` — use that factory for any engine touching a WROLPi database.
 """
+import contextvars
 import pathlib
 import sqlite3
 import threading
@@ -25,9 +26,13 @@ from wrolpi.vars import PYTEST
 
 logger = logger.getChild(__name__)
 
-# Toggled only by `get_immediate_db_session()` (below) while such a session is open on this thread,
+# Set by `get_db_session(commit=True)` while its transaction is being opened, so the engine's `begin`
-# so the engine's `begin` listener knows to emit `BEGIN IMMEDIATE` instead of a deferred `BEGIN`.
+# listener knows to emit `BEGIN IMMEDIATE` instead of a deferred `BEGIN`.  A ContextVar rather than a
-_immediate_txn = threading.local()
+# `threading.local` because the writers that set it include coroutines: Sanic runs every request in
+# its own asyncio task on one shared thread, and a thread-local would leak the flag between
+# interleaved tasks -- arming it for a reader and disarming it under a writer.  Each asyncio task
+# (and each thread) gets its own copy of a ContextVar, so worker-thread callers behave as before.
+_immediate_txn = contextvars.ContextVar('wrolpi_immediate_txn', default=False)
 
 
 def _adapt_datetime(value):
@@ -128,9 +133,9 @@ def create_wrolpi_engine(target: Union[str, pathlib.Path]) -> sqlalchemy.engine.
 
     @event.listens_for(engine, 'begin')
     def _sqlite_do_begin(conn):
-        # `get_immediate_db_session()` sessions take the write lock up front; everything else stays
+        # Write sessions (`commit=True`) take the write lock up front; read sessions stay deferred
-        # deferred so readers never block (in WAL; see `_immediate_txn`).
+        # so readers never block (in WAL; see `_immediate_txn`).
-        if getattr(_immediate_txn, 'active', False):
+        if _immediate_txn.get():
             conn.execute('BEGIN IMMEDIATE')
         else:
             conn.execute('BEGIN')
@@ -184,8 +189,22 @@ def get_db_context() -> Tuple[sqlalchemy.engine.Engine, Session]:
 def get_db_session(commit: bool = False) -> Generator[Session, Any, None]:
     """
     Context manager that creates a DB session.  This will automatically rollback changes, unless `commit` is True.
+
+    `commit=True` declares write intent, so the transaction begins as a writer (`BEGIN IMMEDIATE`)
+    rather than deferred.  A deferred transaction that upgrades to a writer at its first INSERT
+    fails *instantly* with "database is locked" if any other connection wrote in the meantime —
+    SQLite skips `busy_timeout` for lock upgrades, because waiting cannot help a snapshot that is
+    already stale.  Taking the lock at BEGIN lets `busy_timeout` (30s) absorb the contention
+    instead.  Rails and Django both made IMMEDIATE the default for the same reason; a per-call-site
+    opt-in only works if every author remembers, and read-then-write is the normal shape of a write.
+
+    The cost is that writers serialize from BEGIN rather than from their first write, so a write
+    transaction must stay short — do the slow parts (globbing, ffprobe, indexing, big joins) in a
+    read session or outside a session, then write what you computed.  Readers are unaffected: WAL
+    readers never block, and `commit=False` sessions stay deferred.
     """
     _, session = get_db_context()
+    token = _immediate_txn.set(True) if commit else None
     try:
         yield session
         if commit:
@@ -194,6 +213,8 @@ def get_db_session(commit: bool = False) -> Generator[Session, Any, None]:
         session.rollback()
         raise
     finally:
+        if token is not None:
+            _immediate_txn.reset(token)
         # Rollback only if a transaction hasn't been committed.
         # In tests, the test_session fixture manages the session lifecycle,
         # so we should not rollback here - that would undo other test operations.
@@ -201,28 +222,6 @@ def get_db_session(commit: bool = False) -> Generator[Session, Any, None]:
             session.rollback()
 
 
-@contextmanager
-def get_immediate_db_session() -> Generator[Session, Any, None]:
-    """A committing session that takes the SQLite write lock up front (`BEGIN IMMEDIATE`).
-
-    Use this for read-then-write transactions (read some rows, then UPDATE/INSERT them).  A plain
-    `get_db_session(commit=True)` begins *deferred* and only upgrades to a writer at its first
-    write; if another connection holds the write lock at that moment, SQLite returns "database is
-    locked" *immediately* — `busy_timeout` is skipped for lock upgrades to avoid deadlock.  Taking
-    the write lock at BEGIN instead lets `busy_timeout` (30s) absorb the contention.
-
-    The `_immediate_txn` flag is read by the engine's `begin` listener when the transaction opens
-    (on the caller's first statement), so keep queries inside this `with` block.
-    """
-    previous = getattr(_immediate_txn, 'active', False)
-    _immediate_txn.active = True
-    try:
-        with get_db_session(commit=True) as session:
-            yield session
-    finally:
-        _immediate_txn.active = previous
-
-
 @contextmanager
 def get_db_curs(commit: bool = False) -> Generator[sqlite3.Cursor, Any, None]:
     """

wrolpi/downloader.py

@@ -35,7 +35,7 @@ from wrolpi.common import Base, ModelHelper, logger, wrol_mode_check, zig_zag, C
     wrol_mode_enabled, background_task, get_absolute_media_path, timer, aiohttp_get, \
     get_download_info, trim_file_name, get_wrolpi_config, TRACE_LEVEL, normalize_domain
 from wrolpi.dates import TZDateTime, now, Seconds
-from wrolpi.db import get_db_session, get_db_curs, get_immediate_db_session
+from wrolpi.db import get_db_session, get_db_curs
 from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \
     ValidationError, DownloadError
 from wrolpi.events import Events
@@ -1110,9 +1110,9 @@ class DownloadManager:
         so a transient "database is locked" can be handled there without retrying the signal
         dispatches inside this transaction.
 
-        Uses `get_immediate_db_session` because this is a read-then-write transaction (it reads the
+        This is a read-then-write transaction (it reads the `new` downloads, then claims each by
-        `new` downloads, then claims each by setting `last_download_attempt`); a deferred lock
+        setting `last_download_attempt`), so it must hold the write lock from BEGIN -- which
-        upgrade here fails instantly with "database is locked" under concurrency."""
+        `get_db_session(commit=True)` does."""
         # Downloads claimed this cycle: dicts of (download_id, download_url, domain).  Two things are
         # deferred until AFTER the immediate (write) transaction below commits:
         #   * `_add_processing_domain` — it mutates shared, non-transactional state, so adding it
@@ -1126,7 +1126,7 @@ class DownloadManager:
         # still holds even though `processing_domains` is not mutated until after commit.
         to_dispatch = []
         claimed = []
-        with get_immediate_db_session() as session:
+        with get_db_session(commit=True) as session:
             new_downloads = list(session.query(Download).filter(
                 Download.status == 'new',
                 Download.domain not in self.processing_domains,
@@ -1348,7 +1348,10 @@ class DownloadManager:
         """Mark any recurring downloads that are due for download as "new".  Start a download."""
         now_ = now()
 
-        with get_db_session() as session:
+        # `commit=True` (a writer from BEGIN): this reads the recurring Downloads and then renews
+        # the due ones.  As a read session its commit failed with "database is locked" whenever
+        # anything else wrote while it was reading -- a config import or refresh, constantly.
+        with get_db_session(commit=True) as session:
             recurring = self.get_recurring_downloads(session)
             renewed_count = 0
             for download in recurring:

wrolpi/files/test/test_worker_move.py

@@ -1,12 +1,9 @@
 """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.conftest import production_like_sessions, write_lock_held_briefly
 from wrolpi.files import worker as worker_module
 from wrolpi.files.lib import _move_file_group_files
 from wrolpi.files.models import FileGroup, Directory
@@ -539,32 +536,6 @@ async def test_file_worker_move_cleans_directory_records(
     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
@@ -585,7 +556,7 @@ async def test_build_move_plan_waits_for_concurrent_writer(
     db_file = test_session.get_bind().url.database
 
     with production_like_sessions(test_session):
-        with _write_lock_held_briefly(db_file):
+        with write_lock_held_briefly(db_file):
             plan, _ = await build_move_plan_bulk([file1], dest)
 
     assert plan == {file1: dest / 'file1.txt'}
@@ -622,7 +593,7 @@ async def test_file_worker_move_survives_concurrent_writer(
         async def build_plan_then_hold_write_lock(*args, **kwargs):
             plan = await real_build_move_plan_bulk(*args, **kwargs)
             # Planning is done; the contention now falls entirely on the execute phase.
-            stack.enter_context(_write_lock_held_briefly(db_file))
+            stack.enter_context(write_lock_held_briefly(db_file))
             return plan
 
         monkeypatch.setattr(worker_module, 'build_move_plan_bulk', build_plan_then_hold_write_lock)

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, get_immediate_db_session
+from wrolpi.db import get_db_session, get_db_curs
 from wrolpi.errors import NoPrimaryFile
 from wrolpi.events import Events
 from wrolpi.files.lib import (
@@ -600,7 +600,7 @@ async def build_move_plan_bulk(
     # 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:
+    with get_db_session(commit=True) 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
@@ -2045,7 +2045,7 @@ class FileWorker:
             # 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_immediate_db_session() as session:
+                with get_db_session(commit=True) as session:
                     await self._execute_move_chunks(
                         plan, session, created_directories, revert_plan
                     )

wrolpi/root_api.py

@@ -29,6 +29,7 @@ from wrolpi.common import logger, get_wrolpi_config, wrol_mode_enabled, get_medi
     set_global_log_level, get_relative_to_media_directory, search_other_estimates, set_system_timezone
 from wrolpi.config_api import config_bp
 from wrolpi.dates import now
+from wrolpi.db import get_db_session
 from wrolpi.downloader import download_manager
 from wrolpi.errors import WROLModeEnabled, InvalidConfig, ValidationError
 from wrolpi.events import get_events, Events
@@ -355,16 +356,21 @@ async def post_download(request: Request, body: schema.DownloadRequest):
     # Raises an InvalidDownload if the Downloader cannot be found.
     download_manager.find_downloader_by_name(body.downloader)
 
-    session = request.ctx.session
     kwargs = dict(downloader_name=body.downloader,
                   sub_downloader_name=body.sub_downloader, reset_attempts=True,
                   destination=body.destination, tag_names=body.tag_names,
                   settings=body.settings, override_skip=True)
-    if body.frequency:
+    # Not `request.ctx.session`: creating a Download reads (the Tags, then any existing Download for
-        download_manager.recurring_download(session, body.urls[0], body.frequency, collection_id=body.collection_id,
+    # each URL) before it INSERTs, and a deferred transaction fails that lock upgrade *instantly*
-                                            **kwargs)
+    # with "database is locked" if anything else wrote in between -- which is what a user creating a
-    else:
+    # download during a refresh hit.  This session also commits before the handler returns, so the
-        download_manager.create_downloads(session, body.urls, **kwargs)
+    # background dispatch `create_downloads` fires does not wait on a lock this request still holds.
+    with get_db_session(commit=True) as session:
+        if body.frequency:
+            download_manager.recurring_download(session, body.urls[0], body.frequency,
+                                                collection_id=body.collection_id, **kwargs)
+        else:
+            download_manager.create_downloads(session, body.urls, **kwargs)
     if download_manager.disabled.is_set() or download_manager.stopped.is_set():
         # Downloads are disabled, warn the user.
         Events.send_downloads_disabled('Download created. But, downloads are disabled.')

wrolpi/test/test_db.py

@@ -1,7 +1,7 @@
 import sqlite3
 
 from wrolpi.conftest import production_like_sessions
-from wrolpi.db import get_db_session, get_immediate_db_session, _configure_sqlite_connection
+from wrolpi.db import get_db_session, _configure_sqlite_connection
 
 
 class _FakeCursor:
@@ -120,39 +120,25 @@ def _probe_write_lock_is_held(db_file: str) -> bool:
         probe.close()
 
 
-def test_immediate_session_takes_write_lock_up_front(test_session):
+def test_write_session_takes_write_lock_up_front(test_session):
-    """`get_immediate_db_session` must acquire the SQLite write lock at BEGIN.
+    """`get_db_session(commit=True)` must acquire the SQLite write lock at BEGIN.
 
     Regression test for `database is locked` on `UPDATE download SET last_download_attempt`:
     a deferred (read-then-write) transaction that tries to upgrade its lock while another
     connection holds the write lock gets SQLITE_BUSY *immediately* — busy_timeout is ignored
     for lock upgrades to avoid deadlock.  Taking the write lock up front (BEGIN IMMEDIATE)
-    lets busy_timeout absorb the contention instead of erroring instantly.
+    lets busy_timeout absorb the contention instead of erroring instantly.  `commit=True` is
-    """
+    the declaration of write intent, so it is what arms this — no opt-in helper to forget.
-    db_file = test_session.get_bind().url.database
-
-    with production_like_sessions(test_session):
-        with get_immediate_db_session() as session:
-            # Trigger the transaction's BEGIN with a trivial read (as the download dispatcher
-            # does before writing).  This session already holds the write lock.
-            session.execute('SELECT 1')
-            assert _probe_write_lock_is_held(db_file), \
-                'immediate session did not hold the write lock after BEGIN'
-
-
-def test_plain_write_session_stays_deferred(test_session):
-    """A plain `get_db_session(commit=True)` stays deferred (no write lock until it actually writes).
-
-    Only `get_immediate_db_session` opts into the up-front write lock, so ordinary write sessions
-    keep WAL read-concurrency and don't serialize behind each other at BEGIN.
     """
     db_file = test_session.get_bind().url.database
 
     with production_like_sessions(test_session):
         with get_db_session(commit=True) as session:
+            # Trigger the transaction's BEGIN with a trivial read (as the download dispatcher
+            # does before writing).  This session already holds the write lock.
             session.execute('SELECT 1')
-            assert not _probe_write_lock_is_held(db_file), \
+            assert _probe_write_lock_is_held(db_file), \
-                'plain write session unexpectedly holds the write lock before writing'
+                'immediate session did not hold the write lock after BEGIN'
 
 
 def test_read_session_does_not_take_write_lock(test_session):

wrolpi/test/test_downloader.py

@@ -1713,7 +1713,7 @@ async def test_download_dispatched_outside_immediate_transaction(test_session, t
     """The download signal must be dispatched AFTER the claiming write transaction commits.
 
     Regression test for a production deadlock (found on 10.0.0.9, where downloads silently stopped):
-    `_dispatch_new_downloads` claimed downloads inside a `get_immediate_db_session()` (BEGIN
+    `_dispatch_new_downloads` claimed downloads inside a write (BEGIN
     IMMEDIATE) transaction and `await`ed `api_app.dispatch(...)` while that transaction was still
     open.  Under NullPool (production) the dispatched `signal_download_download` opens its OWN
     connection and — inheriting the `_immediate_txn` flag — also issues BEGIN IMMEDIATE, which

wrolpi/test/test_root_api.py

@@ -10,6 +10,7 @@ from wrolpi.api_utils import json_error_handler
 from wrolpi.common import get_wrolpi_config
 from wrolpi.downloader import Download, get_download_manager_config
 from wrolpi.errors import ValidationError, SearchEmpty
+from wrolpi.conftest import production_like_sessions, write_lock_held_briefly
 from wrolpi.test.common import assert_dict_contains
 
 
@@ -969,3 +970,34 @@ async def test_search_other_estimates(async_client, test_session, channel_factor
     request, response = await async_client.post('/api/search_other_estimates', json=body)
     assert response.status_code == HTTPStatus.OK
     assert response.json['others']['channel_count'] == 1, 'Only one Channel is tagged.'
+
+
+@pytest.mark.asyncio
+async def test_mutating_request_survives_concurrent_writer(async_client, test_session, test_downloader,
+                                                           test_download_manager):
+    """Creating a Download must not fail because something else is writing.
+
+    Regression test for the production failure: a user created a video download while a refresh was
+    running and got `database is locked` on `INSERT INTO download`.  The request session begins
+    *deferred*, the handler reads (tag/duplicate checks) before it INSERTs, and SQLite refuses that
+    mid-transaction lock upgrade instantly -- busy_timeout is skipped for upgrades.  Mutating
+    requests must begin as writers so busy_timeout absorbs the contention instead.
+    """
+    db_file = test_session.get_bind().url.database
+    body = json.dumps(dict(urls=['https://example.com/during-refresh'], downloader=test_downloader.name))
+    # Release anything the fixtures left open on the shared session's connection; only the lock the
+    # test takes below should contend with the request.
+    test_session.commit()
+
+    with production_like_sessions(test_session) as maker:
+        with write_lock_held_briefly(db_file):
+            request, response = await async_client.post('/api/download', content=body)
+
+    assert response.status_code == HTTPStatus.CREATED, response.body
+
+    # A separate session sees the committed Download, as the download worker would.
+    session = maker()
+    try:
+        assert [i.url for i in session.query(Download).all()] == ['https://example.com/during-refresh']
+    finally:
+        session.close()