rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 19:01:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

files: serialize stem backfill so a cancelled compare cannot lock SQLite

Browse source 
compare_file_groups runs ensure_file_group_stems in a thread.  Cancelling
the task does not stop that thread, and a second compare then writes
wrolpi_kv on the same connection (tests share one session).  A lock
plus a version re-check makes the second call a no-op.
This commit is contained in:
Roland 2026-09-18 10:53:09 -06:00
parent d0820e1bc5
commit 2f5197cfbc
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
1 changed files with 23 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

38
wrolpi/files/lib.py
View file

@ -11,6 +11,7 @@ import re
import shutil
import subprocess
import tarfile
import threading
import time
import urllib.parse
import zipfile
@ -493,6 +494,9 @@ _HARDCODED_SUFFIXES = frozenset(
# before any compare when this version does not match the value in wrolpi_kv.
STEM_ALGORITHM_VERSION = 1
STEM_ALGORITHM_VERSION_KEY = 'stem_algorithm_version'
# Cancelled compare_file_groups leaves ensure_file_group_stems running in a thread.
# Tests (and overlapping refreshes) share one SQLite writer; serialize the backfill.
_stem_backfill_lock = threading.Lock()
PART_PARSER = re.compile(r'(.+?)(\.f[\d]{2,3})?(\.info)?(\.\w{3,4})(\.part)', re.IGNORECASE)
@ -850,22 +854,26 @@ def ensure_file_group_stems():
After a successful backfill the kv version matches, and later refreshes skip the table scan.
Upsert, from_paths, move, and reorganize must keep stem current.
A lock serializes callers: cancelling compare_file_groups does not stop the executor
thread, and a second compare would otherwise write wrolpi_kv concurrently (SQLITE_BUSY).
"""
stored = get_stem_algorithm_version()
if stored == STEM_ALGORITHM_VERSION:
return
with get_db_curs() as curs:
curs.execute('SELECT id, primary_path FROM file_group')
rows = list(curs.fetchall())
if rows:
updates = []
for row in rows:
stem, _ = split_path_stem_and_suffix(row['primary_path'])
updates.append((stem, row['id']))
with get_db_curs(commit=True) as curs:
for chunk in chunks(updates, 500):
curs.executemany('UPDATE file_group SET stem = ? WHERE id = ?', chunk)
set_stem_algorithm_version(STEM_ALGORITHM_VERSION)
with _stem_backfill_lock:
stored = get_stem_algorithm_version()
if stored == STEM_ALGORITHM_VERSION:
return
with get_db_curs() as curs:
curs.execute('SELECT id, primary_path FROM file_group')
rows = list(curs.fetchall())
if rows:
updates = []
for row in rows:
stem, _ = split_path_stem_and_suffix(row['primary_path'])
updates.append((stem, row['id']))
with get_db_curs(commit=True) as curs:
for chunk in chunks(updates, 500):
curs.executemany('UPDATE file_group SET stem = ? WHERE id = ?', chunk)
set_stem_algorithm_version(STEM_ALGORITHM_VERSION)
async def apply_indexers(progress_callback: Callable[[int, int], None] = None):