diff --git a/wrolpi/collections/api.py b/wrolpi/collections/api.py index 77ab4afb..1773964e 100644 --- a/wrolpi/collections/api.py +++ b/wrolpi/collections/api.py @@ -442,10 +442,8 @@ async def get_conflicts_endpoint(request: Request, collection_id: int): """ from wrolpi.collections.reorganize import get_conflict_details_with_ffprobe - session = request.ctx.session try: - conflicts = await get_conflict_details_with_ffprobe(collection_id, session) - session.commit() + conflicts = await get_conflict_details_with_ffprobe(collection_id, request.ctx.session, request.ctx.db.write) return json_response({'conflicts': conflicts}) except UnknownCollection as e: return json_response({'error': str(e)}, status=HTTPStatus.NOT_FOUND) @@ -472,9 +470,8 @@ async def execute_reorganization_endpoint(request: Request, collection_id: int): """ from wrolpi.collections.reorganize import execute_reorganization - session = request.ctx.session try: - job_id = execute_reorganization(collection_id, session) + job_id = execute_reorganization(collection_id, request.ctx.session, request.ctx.db.write) if job_id: return json_response({ 'job_id': job_id, diff --git a/wrolpi/collections/reorganize.py b/wrolpi/collections/reorganize.py index 0e80dd21..c17a328d 100644 --- a/wrolpi/collections/reorganize.py +++ b/wrolpi/collections/reorganize.py @@ -10,7 +10,7 @@ Example: User changes archive format from '%(title)s.%(ext)s' to '%(download_yea """ import pathlib from dataclasses import dataclass -from typing import List, Tuple, Optional +from typing import Callable, ContextManager, List, Tuple, Optional from sqlalchemy import func from sqlalchemy.orm import Session, joinedload @@ -19,6 +19,14 @@ from wrolpi.common import logger, get_media_directory from wrolpi.db import get_db_session from .models import Collection +# Opens a short write transaction and yields its Session; `request.ctx.db.write` in an API request. +WriteSession = Callable[[], ContextManager[Session]] + + +def _own_write_session() -> ContextManager[Session]: + """A write transaction on a connection of its own, for callers outside an API request.""" + return get_db_session(commit=True) + logger = logger.getChild(__name__) __all__ = [ @@ -865,6 +873,7 @@ def _detect_conflicts_with_details( def execute_reorganization( collection_id: int, session: Session = None, + write: WriteSession = None, ) -> str: """Execute reorganization, returns job_id for tracking. @@ -872,7 +881,9 @@ def execute_reorganization( Args: collection_id: The collection to reorganize - session: Database session (optional) + session: Database session (optional), used only to read + write: Opens the short write transaction that stores the new file_format when nothing needs to move + (defaults to a connection of its own) Returns: job_id for tracking progress via FileWorker status @@ -883,8 +894,8 @@ def execute_reorganization( from wrolpi.files.worker import file_worker if session is None: - with get_db_session(commit=True) as session: - return execute_reorganization(collection_id, session) + with get_db_session() as session: + return execute_reorganization(collection_id, session, write) collection = Collection.find_by_id(session, collection_id) @@ -907,8 +918,10 @@ def execute_reorganization( # Still update the file_format to current config current_format = collection._get_current_file_format() if current_format: - collection.file_format = current_format - session.commit() + # The scan above only read; store the result in a short write transaction that begins as a + # writer (see `RequestDB.write`). + with (write or _own_write_session)() as write_session: + Collection.find_by_id(write_session, collection_id).file_format = current_format # Trigger config save if collection.kind == 'domain': @@ -1236,6 +1249,7 @@ def get_batch_reorganization_status(batch_job_id: str) -> dict: async def get_conflict_details_with_ffprobe( collection_id: int, session: Session, + write: WriteSession, ) -> List[dict]: """Get conflict details, processing ffprobe for any videos missing it. @@ -1243,9 +1257,13 @@ async def get_conflict_details_with_ffprobe( when resolving conflicts. The +2 audio bonus in _rank_video_quality() only applies if video.ffprobe_json exists. + ffprobe runs outside any write transaction; only its results are stored, in one short `write`. + Args: collection_id: The collection to get conflicts for - session: Database session + session: Database session, used to read + write: Opens the write transaction that stores ffprobe data; it must commit `session`'s read first + (as `RequestDB.write` does), so the re-fetched preview sees the new data Returns: List of conflict detail dicts (empty if no conflicts) @@ -1258,24 +1276,30 @@ async def get_conflict_details_with_ffprobe( if not preview.conflicts: return [] - # Process ffprobe for videos missing it (only for channel collections) + from modules.videos.common import get_or_create_ffprobe_json from modules.videos.models import Video - videos_processed = 0 - for conflict in preview.conflicts: - for file_info in conflict['conflicting_files']: - if file_info.get('file_group_id') and file_info.get('model_type') == 'channel': - video = session.query(Video).filter_by(file_group_id=file_info['file_group_id']).one_or_none() - if video and not video.ffprobe_json: - try: - await video.get_ffprobe_json() - videos_processed += 1 - except Exception as e: - logger.warning(f'Failed to process ffprobe for video {video.id}: {e}') + # Videos missing ffprobe data. (A conflicting file's model_type is its FileGroup's model, 'video'.) + file_group_ids = [file_info['file_group_id'] for conflict in preview.conflicts + for file_info in conflict['conflicting_files'] if file_info.get('file_group_id')] + video_paths = {video.id: video.video_path + for video in session.query(Video).filter(Video.file_group_id.in_(file_group_ids)) + if not video.ffprobe_json and video.video_path} - # Flush ffprobe data if any was processed (preserves caller's transaction control) - if videos_processed > 0: - session.flush() + ffprobe_results = dict() + for video_id, video_path in video_paths.items(): + try: + ffprobe_results[video_id] = await get_or_create_ffprobe_json(video_path) + except Exception as e: + logger.warning(f'Failed to process ffprobe for video {video_id}: {e}') + + if ffprobe_results: + with write() as write_session: + for video in write_session.query(Video).filter(Video.id.in_(list(ffprobe_results))): + video.ffprobe_json, written = ffprobe_results[video.id] + if written: + # Track the new cache file in the FileGroup. + video.file_group.append_files(written) # Re-fetch preview to get updated quality rankings preview = get_reorganization_preview(collection_id, session) diff --git a/wrolpi/collections/test/test_reorganize.py b/wrolpi/collections/test/test_reorganize.py index 6460c8b3..15be39cb 100644 --- a/wrolpi/collections/test/test_reorganize.py +++ b/wrolpi/collections/test/test_reorganize.py @@ -1,6 +1,10 @@ """ Tests for the collection reorganization feature. """ +import json +from http import HTTPStatus +from typing import List, Tuple + import pytest from modules.archive.lib import get_archive_downloader_config @@ -18,6 +22,7 @@ from wrolpi.collections.reorganize import ( get_batch_reorganization_status, ReorganizationPreview, ) +from wrolpi.conftest import production_like_sessions, write_lock_held_briefly from wrolpi.db import get_db_session from wrolpi.files.models import FileGroup @@ -2787,3 +2792,99 @@ def test_preview_excludes_files_already_in_correct_location(test_session, test_d yt_dlp_options = dict(config._config['yt_dlp_options']) yt_dlp_options['file_name_format'] = original_format config._config['yt_dlp_options'] = yt_dlp_options + + +def _channel_with_two_conflicting_videos(test_directory, video_factory, name: str) -> Tuple[int, List[int]]: + """A Channel whose two Videos would move to the same destination, neither with ffprobe data in the DB + (each has a cached .ffprobe.json, so the conflicts endpoint stores it without running ffprobe).""" + from datetime import datetime, timezone + + channel_dir = test_directory / 'videos' / name + channel_dir.mkdir(parents=True, exist_ok=True) + with get_db_session(commit=True) as session: + collection = Collection(name=name, kind='channel', directory=channel_dir, + file_format='%(old_format)s.%(ext)s') + session.add(collection) + session.flush() + channel = Channel(name=name, collection_id=collection.id, directory=channel_dir) + session.add(channel) + session.commit() + channel_id, collection_id = channel.id, collection.id + + same_date = datetime(2024, 1, 15, tzinfo=timezone.utc) + videos = [video_factory(channel_id=channel_id, title=title, upload_date=same_date, with_video_file=True, + with_info_json={'uploader': 'TestUploader', 'id': 'dup'}) + for title in ('video_one', 'video_two')] + video_ids = [i.id for i in videos] + with get_db_session(commit=True) as session: + for video_id in video_ids: + video = session.get(Video, video_id) + video.file_group.title = 'Duplicate Title' + video.source_id = 'dup' + video.ffprobe_json = None + video.video_path.with_suffix('.ffprobe.json').write_text(json.dumps({'streams': [], 'format': {}})) + return collection_id, video_ids + + +@pytest.mark.asyncio +async def test_conflicts_endpoint_waits_for_a_concurrent_writer(async_client, test_session, test_directory, + video_factory): + """The conflicts endpoint reads, then stores ffprobe data, so its write must begin as a writer.""" + collection_id, video_ids = _channel_with_two_conflicting_videos(test_directory, video_factory, 'LockChannel') + db_file = test_session.get_bind().url.database + + with production_like_sessions(test_session) as maker: + with write_lock_held_briefly(db_file): + _, response = await async_client.get(f'/api/collections/{collection_id}/reorganize/conflicts') + assert response.status_code == HTTPStatus.OK, response.json + assert len(response.json['conflicts']) == 1 + + with maker() as session: + assert all(session.get(Video, i).ffprobe_json is not None for i in video_ids) + + +@pytest.mark.asyncio +async def test_reorganize_endpoint_waits_for_a_concurrent_writer(async_client, test_session, test_directory): + """With nothing to move, reorganizing reads, then stores the new file_format, so its write must begin as a + writer.""" + domain_dir = test_directory / 'archive' / 'lock.com' + domain_dir.mkdir(parents=True, exist_ok=True) + with get_db_session(commit=True) as session: + collection = Collection(name='lock.com', kind='domain', directory=domain_dir, file_format='%(old)s.%(ext)s') + session.add(collection) + session.commit() + collection_id = collection.id + db_file = test_session.get_bind().url.database + + with production_like_sessions(test_session) as maker: + with write_lock_held_briefly(db_file): + _, response = await async_client.post(f'/api/collections/{collection_id}/reorganize') + assert response.status_code == HTTPStatus.OK, response.json + assert response.json['job_id'] == '' + + with maker() as session: + assert session.get(Collection, collection_id).file_format == \ + get_archive_downloader_config().file_name_format + + +@pytest.mark.asyncio +async def test_batch_reorganize_waits_for_a_concurrent_writer(async_client, test_session, test_directory): + """The batch job calls execute_reorganization with a deferred (read) session; storing the new file_format + must still wait for a concurrent writer.""" + domain_dir = test_directory / 'archive' / 'batch.com' + domain_dir.mkdir(parents=True, exist_ok=True) + with get_db_session(commit=True) as session: + collection = Collection(name='batch.com', kind='domain', directory=domain_dir, file_format='%(old)s.%(ext)s') + session.add(collection) + session.commit() + collection_id = collection.id + db_file = test_session.get_bind().url.database + + with production_like_sessions(test_session) as maker: + with write_lock_held_briefly(db_file): + with get_db_session() as session: + assert execute_reorganization(collection_id, session) == '' + + with maker() as session: + assert session.get(Collection, collection_id).file_format == \ + get_archive_downloader_config().file_name_format