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:
-        # Delete all Videos if the FileModel no longer contains a video or audio file.
+    # Read the FileGroups that are no longer video/audio, then unmodel them in short transactions.
+    with get_db_curs() as curs:
         curs.execute('''
-                     UPDATE file_group
-                     SET model = NULL
+                     SELECT id
+                     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()]
-        if deleted_ids:
+        stale_ids = [i['id'] for i in curs.fetchall()]
+    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)))
-        # Claim all Videos in a Channel's directory for that Channel.  But, only if they have not yet been claimed.
-        curs.execute('''
-                     UPDATE video AS v
-                     SET channel_id = c.id
-                     FROM channel c
-                              INNER JOIN collection col ON col.id = c.collection_id
-                              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
-                     ''')
+                         ''', dict(ids=ids))
+
+    # Claim all Videos in a Channel's directory for that Channel.  But, only if they have not yet
+    # been claimed.  The join is the slow part, so it runs as a read; only the writes take the lock.
+    with get_db_curs() as curs:
+        curs.execute(_UNCLAIMED_VIDEOS_SQL)
+        claims = _first_claim_per_video(curs.fetchall())
+    _claim_videos(claims)
 
 
 
@@ -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:
-        curs.execute('''
-                     UPDATE video AS v
-                     SET channel_id = c.id
-                     FROM channel c
-                              INNER JOIN collection col ON col.id = c.collection_id
-                              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
+    # Read the claims, then write them in chunks -- the join must not hold the write lock (see
+    # `_claim_videos`).  Channel import calls this with every channel in channels.yaml, so it is
+    # just as slow as the full `video_cleanup` sweep.
+    with get_db_curs() as curs:
+        curs.execute(f'''
+                     {_UNCLAIMED_VIDEOS_SQL}
                        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'