From e21c6d9f0a71b75fa472bb660ffcbafd915c53dd Mon Sep 17 00:00:00 2001 From: Roland Date: Thu, 8 Oct 2026 16:04:18 -0600 Subject: [PATCH] Progress: write as a writer, confine and merge the import, keep FTS on downgrade - The progress endpoints begin as writers (request.ctx.db.write()), so a heartbeat during a refresh waits instead of failing a lock upgrade. The config switch fires after the commit. - `..` is collapsed before the ignored-directory check. Clear only resets an existing FileGroup; it never indexes a file. - Import confines config paths to the media directory. A newer plain view keeps its `viewed` but takes the config's position; only progress written after the config wins. Files are globbed before the write transaction. - The dump skips a row outside the media directory instead of failing. - The downgrade drops the columns with ALTER TABLE, which keeps file_group's FTS triggers. Co-Authored-By: Claude Opus 5.5 --- .../2026_10_08_1200_file_group_progress.py | 6 +- wrolpi/files/api.py | 13 +- wrolpi/files/lib.py | 69 ++++++----- wrolpi/files/progress.py | 51 ++++++-- wrolpi/files/test/test_progress.py | 114 ++++++++++++++++++ wrolpi/test/test_db_bootstrap.py | 26 ++++ 6 files changed, 232 insertions(+), 47 deletions(-) diff --git a/alembic/versions/2026_10_08_1200_file_group_progress.py b/alembic/versions/2026_10_08_1200_file_group_progress.py index cc7e6f3e..5742beee 100644 --- a/alembic/versions/2026_10_08_1200_file_group_progress.py +++ b/alembic/versions/2026_10_08_1200_file_group_progress.py @@ -19,6 +19,6 @@ def upgrade(): def downgrade(): - with op.batch_alter_table('file_group') as batch_op: - batch_op.drop_column('position') - batch_op.drop_column('progress') + # Not batch_alter_table: rebuilding file_group drops its FTS triggers (see 2026_09_20_1200). + op.execute('ALTER TABLE file_group DROP COLUMN position') + op.execute('ALTER TABLE file_group DROP COLUMN progress') diff --git a/wrolpi/files/api.py b/wrolpi/files/api.py index be7f89c1..83199c60 100644 --- a/wrolpi/files/api.py +++ b/wrolpi/files/api.py @@ -14,6 +14,7 @@ from wrolpi.common import get_media_directory, wrol_mode_check, get_relative_to_ background_task, walk, timer, TRACE_LEVEL, unique_by_predicate from wrolpi.errors import InvalidFile, UnknownDirectory, FileUploadFailed, FileConflict from wrolpi.events import Events +from wrolpi.files.progress import save_recently_viewed_config from . import lib, schema from .worker import file_worker from ..api_utils import json_response, api_app @@ -65,7 +66,12 @@ async def get_file(request: Request, body: schema.FileRequest): @validate(schema.FileProgressRequest) async def post_file_progress(request: Request, body: schema.FileProgressRequest): path = get_media_directory() / body.file - progress, position = lib.set_file_progress(request.ctx.session, path, body.progress, body.position, body.final) + # Begin as a writer; a playback heartbeat often overlaps a refresh or download. + with request.ctx.db.write() as session: + progress, position = lib.set_file_progress(session, path, body.progress, body.position) + if body.final and progress is not None: + # The User stopped (paused, closed). Heartbeats during playback only write the DB. + save_recently_viewed_config.activate_switch() return json_response({'progress': progress, 'position': position}) @@ -76,7 +82,10 @@ async def post_file_progress(request: Request, body: schema.FileProgressRequest) ) @validate(schema.FileProgressClearRequest) async def post_file_progress_clear(request: Request, body: schema.FileProgressClearRequest): - lib.clear_file_progress(request.ctx.session, get_media_directory() / body.file) + with request.ctx.db.write() as session: + changed = lib.clear_file_progress(session, get_media_directory() / body.file) + if changed: + save_recently_viewed_config.activate_switch() return response.empty(HTTPStatus.NO_CONTENT) diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index ee7eb071..ae0de494 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -36,7 +36,7 @@ from wrolpi.errors import InvalidFile, UnknownDirectory, UnknownFile, UnknownTag NoPrimaryFile, InvalidDirectory, IgnoredDirectoryError, UnsupportedArchive, InvalidArchiveMember from wrolpi.events import Events from wrolpi.files.models import FileGroup, Directory -from wrolpi.files.progress import normalize_progress, save_recently_viewed_config +from wrolpi.files.progress import normalize_progress from wrolpi.lang import ISO_639_CODES, ISO_3166_CODES from wrolpi.tags import TagFile, Tag, tag_append_sub_select_where, save_tags_config, sync_tags_directory from wrolpi.vars import PYTEST, IS_MACOS @@ -160,51 +160,62 @@ async def set_file_viewed(session: Session, file: pathlib.Path): session.commit() -def _get_or_create_file_group(session: Session, file: pathlib.Path) -> FileGroup: +def _trackable_path(file: pathlib.Path) -> pathlib.Path | None: + """Return `file` with any `..` collapsed, or None if it is in an ignored directory. + + `..` is collapsed first so it cannot step into an ignored directory past the check. + + @raise InvalidFile: The file is not in the media directory. + """ + file = pathlib.Path(os.path.normpath(file)) if not get_paths_in_media_directory([file]): raise InvalidFile('File must be in the media directory') + if not remove_files_in_ignored_directories([file]): + return None + return file + + +def set_file_progress(session: Session, file: pathlib.Path, progress: float, position: dict = None) \ + -> Tuple[float | None, dict | None]: + """Record how far the User is through a file (see wrolpi.files.progress). This is also a view. + + The caller owns the (write) transaction, and saves the recently viewed config after it commits. + + @return: The (progress, position) that was stored, (None, None) if the file is not tracked. + """ + progress, position = normalize_progress(progress, position) + + if not (file := _trackable_path(file)): + return None, None + try: - return FileGroup.find_by_path(session, file) + fg = FileGroup.find_by_path(session, file) except UnknownFile: if not file.is_file(): raise InvalidFile(f'File does not exist: {get_relative_to_media_directory(file)}') fg = FileGroup.from_paths(session, file) fg.do_model(session) - return fg - - -def set_file_progress(session: Session, file: pathlib.Path, progress: float, position: dict = None, - final: bool = False) -> Tuple[float | None, dict | None]: - """Record how far the User is through a file (see wrolpi.files.progress). This is also a view. - - @param final: The User stopped (paused, closed); save the recently viewed config. - @return: The (progress, position) that was stored. - """ - progress, position = normalize_progress(progress, position) - - # Files in ignored directories are not tracked. - if not remove_files_in_ignored_directories([file]): - return None, None - - fg = _get_or_create_file_group(session, file) # Not `set_viewed`, which saves the config; a heartbeat during playback only writes the DB. fg.viewed = now() fg.progress = progress fg.position = position - session.commit() - - if final: - save_recently_viewed_config.activate_switch() return progress, position -def clear_file_progress(session: Session, file: pathlib.Path): - """Forget how far the User is through a file, so it starts from the beginning. It remains viewed.""" - fg = _get_or_create_file_group(session, file) +def clear_file_progress(session: Session, file: pathlib.Path) -> bool: + """Forget how far the User is through a file, so it starts from the beginning. It remains viewed. + + A file that was never tracked is left alone. The caller owns the (write) transaction. + + @return: True if a FileGroup was changed. + """ + if not (file := _trackable_path(file)): + return False + if not (fg := FileGroup.get_by_path(session, file)): + return False fg.progress = None fg.position = None - session.commit() - save_recently_viewed_config.activate_switch() + return True @cachetools.func.ttl_cache(10_000, 30.0) diff --git a/wrolpi/files/progress.py b/wrolpi/files/progress.py index 1028c0e2..c0cdf342 100644 --- a/wrolpi/files/progress.py +++ b/wrolpi/files/progress.py @@ -14,6 +14,8 @@ The database is the working store. `recently_viewed.yaml` holds the most recent history and positions survive a database rebuild. """ import math +import os +import pathlib from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional, Tuple @@ -21,7 +23,7 @@ from typing import List, Optional, Tuple import pytz from wrolpi import dates -from wrolpi.common import ConfigFile, logger, get_media_directory +from wrolpi.common import ConfigFile, logger, get_media_directory, get_paths_in_media_directory from wrolpi.db import get_db_session from wrolpi.errors import ValidationError, NoPrimaryFile from wrolpi.events import Events @@ -155,6 +157,11 @@ class RecentlyViewedConfig(ConfigFile): .limit(RECENTLY_VIEWED_LIMIT) files = list() for file_group in file_groups: + if not get_paths_in_media_directory([file_group.primary_path]): + # One bad row must not stop every later save. + logger.warning(f'Not saving recently viewed file outside the media directory:' + f' {file_group.primary_path}') + continue entry = dict( path=str(file_group.primary_path.relative_to(media_directory)), viewed=file_group.viewed.isoformat(), @@ -172,8 +179,8 @@ class RecentlyViewedConfig(ConfigFile): Events.send_config_save_failed(message) def import_config(self, file=None, send_events=False): - """Apply the config to the database. The newer `viewed` wins, so a file viewed while WROLPi was - starting keeps its newer position.""" + """Apply the config to the database. Progress written after the config (e.g. while WROLPi was starting) + is kept; a newer plain view keeps its `viewed`, but takes the config's position.""" super().import_config(file, send_events) try: # wrolpi.files.models imports this module. @@ -181,33 +188,51 @@ class RecentlyViewedConfig(ConfigFile): from wrolpi.files.models import FileGroup media_directory = get_media_directory() - entries = [i for i in map(_parse_entry, self.files) if i] + entries = list() + for path, viewed, progress, position in filter(None, map(_parse_entry, self.files)): + # An absolute `path` replaces the media directory in the join, and `..` can leave it. + absolute_path = pathlib.Path(os.path.normpath(media_directory / path)) + if not get_paths_in_media_directory([absolute_path]): + logger.warning(f'Not importing recently viewed file outside the media directory: {path}') + continue + entries.append((absolute_path, viewed, progress, position)) + absolute_paths = [i[0] for i in entries] + + # Glob the files which are not yet in the DB before the write transaction; stat'ing a thousand files on + # a slow disk would block every other writer. + with get_db_session() as session: + known = {i for (i,) in session.query(FileGroup.primary_path) + .filter(FileGroup.primary_path.in_(absolute_paths))} + new_files = {i: glob_shared_stem(i) for i in absolute_paths if i not in known and i.is_file()} + with get_db_session(commit=True) as session: - absolute_paths = [media_directory / path for path, *_ in entries] file_groups = session.query(FileGroup).filter(FileGroup.primary_path.in_(absolute_paths)) file_groups_by_path = {i.primary_path: i for i in file_groups} applied = 0 - for path, viewed, progress, position in entries: - absolute_path = media_directory / path + for absolute_path, viewed, progress, position in entries: file_group = file_groups_by_path.get(absolute_path) if not file_group: - if not absolute_path.is_file(): + if not (files := new_files.pop(absolute_path, None)): # The file was deleted; its history goes with it. - logger.debug(f'Not importing recently viewed file which does not exist: {path}') + logger.debug(f'Not importing recently viewed file which does not exist: {absolute_path}') continue try: - file_group = FileGroup.from_paths(session, *glob_shared_stem(absolute_path)) + file_group = FileGroup.from_paths(session, *files) except NoPrimaryFile: - logger.warning(f'Cannot import recently viewed file: {path}') + logger.warning(f'Cannot import recently viewed file: {absolute_path}') continue session.add(file_group) session.flush([file_group]) file_groups_by_path[absolute_path] = file_group - if file_group.viewed and file_group.viewed >= viewed: + db_is_newer = file_group.viewed is not None and file_group.viewed >= viewed + if db_is_newer and file_group.progress is not None: + # Progress was written after the config (e.g. while WROLPi was starting). continue - file_group.viewed = viewed + # A newer plain view is kept, but it is not a newer place to resume. + if not db_is_newer: + file_group.viewed = viewed file_group.progress = progress file_group.position = position applied += 1 diff --git a/wrolpi/files/test/test_progress.py b/wrolpi/files/test/test_progress.py index d5a0369a..06c3c5ba 100644 --- a/wrolpi/files/test/test_progress.py +++ b/wrolpi/files/test/test_progress.py @@ -301,3 +301,117 @@ async def test_recently_viewed_follows_move(test_session, async_client, test_dir entry, = read_recently_viewed(test_directory) assert entry['path'] == 'watched/movie.mp4' assert entry['progress'] == 0.4 + + +@pytest.mark.asyncio +async def test_progress_survives_concurrent_writer(async_client, test_session, test_directory, make_files_structure): + """A progress save waits for another writer (a refresh, a download) instead of failing.""" + from wrolpi.conftest import production_like_sessions, write_lock_held_briefly + movie, = make_files_structure(['videos/movie.mp4']) + FileGroup.from_paths(test_session, movie) + test_session.commit() + db_file = test_session.get_bind().url.database + + body = json.dumps(dict(file='videos/movie.mp4', progress=0.4, position=dict(kind='time', seconds=480))) + with production_like_sessions(test_session) as maker: + with write_lock_held_briefly(db_file): + request, response = await async_client.post('/api/files/progress', content=body) + assert response.status_code == HTTPStatus.OK, response.body + + with write_lock_held_briefly(db_file): + request, response = await async_client.post('/api/files/progress/clear', + content=json.dumps(dict(file='videos/movie.mp4'))) + assert response.status_code == HTTPStatus.NO_CONTENT, response.body + + session = maker() + try: + fg, = session.query(FileGroup).all() + assert fg.viewed and fg.progress is None + finally: + session.close() + + +@pytest.mark.asyncio +async def test_progress_ignored_directory_dot_dot(test_session, async_client, test_directory, make_files_structure, + test_wrolpi_config, await_switches): + """`..` cannot step into an ignored directory.""" + from wrolpi.common import get_wrolpi_config + make_files_structure(['private/movie.mp4', 'videos/other.mp4']) + get_wrolpi_config().ignored_directories = [str(test_directory / 'private')] + await await_switches() + + body = dict(file='videos/../private/movie.mp4', progress=0.4, position=dict(kind='time', seconds=480)) + await async_client.post('/api/files/progress', content=json.dumps(body)) + assert test_session.query(FileGroup).count() == 0 + + +@pytest.mark.asyncio +async def test_progress_clear_creates_nothing(test_session, async_client, test_directory, make_files_structure, + test_wrolpi_config, await_switches): + """Starting over a file that was never tracked (or is ignored) does not index it.""" + from wrolpi.common import get_wrolpi_config + make_files_structure(['private/movie.mp4', 'videos/movie.mp4']) + get_wrolpi_config().ignored_directories = [str(test_directory / 'private')] + await await_switches() + + for file in ('private/movie.mp4', 'videos/movie.mp4'): + request, response = await async_client.post('/api/files/progress/clear', content=json.dumps(dict(file=file))) + assert response.status_code == HTTPStatus.NO_CONTENT + assert test_session.query(FileGroup).count() == 0 + + +@pytest.mark.asyncio +async def test_recently_viewed_import_after_plain_view(test_session, async_client, test_directory, + make_files_structure): + """A plain view during startup is newer, but it is not a newer resume point; the config's position applies.""" + movie, = make_files_structure(['videos/movie.mp4']) + fg = FileGroup.from_paths(test_session, movie) + viewed_now = fg.set_viewed() + test_session.commit() + + config_file = test_directory / 'config/recently_viewed.yaml' + config_file.parent.mkdir(exist_ok=True) + config_file.write_text(yaml.dump(dict(version=1, files=[ + dict(path='videos/movie.mp4', viewed=(viewed_now - timedelta(days=1)).isoformat(), progress=0.5, + position=dict(kind='time', seconds=600))]))) + + get_recently_viewed_config().import_config() + + test_session.expire_all() + fg, = test_session.query(FileGroup).all() + assert fg.progress == 0.5 + assert fg.position == dict(kind='time', seconds=600.0) + assert fg.viewed == viewed_now, 'The newer view is kept' + + +@pytest.mark.asyncio +async def test_recently_viewed_import_outside_media_directory(test_session, async_client, test_directory, + make_files_structure): + """A hand-edited entry cannot create a FileGroup outside the media directory, and one bad row in the DB + does not stop the config from being saved.""" + make_files_structure(['videos/movie.mp4']) + outside = test_directory.parent / f'{test_directory.name}-outside.mp4' + outside.touch() + try: + now = dates.now() + config_file = test_directory / 'config/recently_viewed.yaml' + config_file.parent.mkdir(exist_ok=True) + config_file.write_text(yaml.dump(dict(version=1, files=[ + dict(path=str(outside), viewed=now.isoformat()), + dict(path=f'../{outside.name}', viewed=now.isoformat()), + dict(path='videos/movie.mp4', viewed=(now - timedelta(hours=1)).isoformat()), + ]))) + config = get_recently_viewed_config() + config.import_config() + test_session.expire_all() + assert [i.primary_path.name for i in test_session.query(FileGroup)] == ['movie.mp4'] + + # A row outside the media directory (however it got there) is skipped by the dump. + bad = FileGroup.from_paths(test_session, outside) + bad.viewed = now + test_session.commit() + config.dump_config() + config.save() + assert [i['path'] for i in read_recently_viewed(test_directory)] == ['videos/movie.mp4'] + finally: + outside.unlink() diff --git a/wrolpi/test/test_db_bootstrap.py b/wrolpi/test/test_db_bootstrap.py index 44649ebd..fad811ba 100644 --- a/wrolpi/test/test_db_bootstrap.py +++ b/wrolpi/test/test_db_bootstrap.py @@ -173,6 +173,32 @@ def test_drop_file_group_idempotency_preserves_fts(test_directory): assert len(hits) == 1 +def test_file_group_progress_downgrade_preserves_fts(test_directory): + """Dropping file_group.progress/position must not rebuild file_group, which would drop its FTS triggers.""" + from alembic.config import Config + from alembic import command + from wrolpi.db import get_db_uri + from wrolpi.vars import PROJECT_DIR + + (test_directory / 'config').mkdir(parents=True, exist_ok=True) + config = Config(str(PROJECT_DIR / 'alembic.ini')) + config.set_main_option('sqlalchemy.url', get_db_uri()) + command.upgrade(config, '2026_10_08_1200') + command.downgrade(config, '2026_10_06_1500') + + with sqlite3.connect(get_db_file()) as conn: + cols = {r[1] for r in conn.execute('PRAGMA table_info(file_group)')} + assert not {'progress', 'position'} & cols + triggers = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='trigger'")} + assert {'file_group_fts_ai', 'file_group_fts_ad', 'file_group_fts_au'} <= triggers + conn.execute( + 'INSERT INTO file_group (directory, primary_path, files, stem, indexed, a_text) VALUES (?,?,?,?,0,?)', + (str(test_directory), str(test_directory / 'searchable.txt'), '[]', 'searchable', 'alpha uniquephrase'), + ) + hits = conn.execute("SELECT rowid FROM file_group_fts WHERE file_group_fts MATCH 'uniquephrase'").fetchall() + assert len(hits) == 1 + + def test_domain_collection_summary_migration_backfills(test_directory): """The domain-summary migration installs the archive triggers and backfills the summary columns of existing Domain Collections from their Archives."""