rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-25 18:40:06 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Reorganize conflicts and no-move reorganize wait for a concurrent writer

Browse source 
Follow-up to #846, from review:

- The reorganize conflicts endpoint read the preview, ran ffprobe, then
  flushed and committed in the request's deferred transaction.  It now
  reads, runs ffprobe through get_or_create_ffprobe_json (no session),
  and stores the results in one short request.ctx.db.write().
- Its ffprobe step never ran: it matched model_type == 'channel', but a
  conflicting video's model_type is its FileGroup's model, 'video'.
- execute_reorganization's no-move path stored the new file_format with
  session.commit() on the caller's deferred session, from the reorganize
  endpoint and from the batch job.  The scan stays a read; the store is a
  short write (request.ctx.db.write() in a request, its own connection
  otherwise).  Called without a session it now opens a read session, so
  that write does not wait on its own caller's lock.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Roland 2026-10-20 16:51:43 -06:00
parent 97d7eb260c
commit 14ab729101
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
3 changed files with 149 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

7
wrolpi/collections/api.py
View file

@ -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,

68
wrolpi/collections/reorganize.py
View file

@ -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)

101
wrolpi/collections/test/test_reorganize.py
View file

@ -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