diff --git a/core/wishlist/processing.py b/core/wishlist/processing.py index 487401c8..fb6cc20e 100644 --- a/core/wishlist/processing.py +++ b/core/wishlist/processing.py @@ -7,7 +7,7 @@ from dataclasses import dataclass from datetime import datetime from contextlib import AbstractContextManager from types import SimpleNamespace -from typing import Any, Callable, Dict +from typing import Any, Callable, Dict, Optional from core.wishlist.payloads import build_failed_track_wishlist_context from core.wishlist.selection import filter_wishlist_tracks_by_category, sanitize_and_dedupe_wishlist_tracks @@ -100,6 +100,202 @@ def remove_completed_tracks_from_wishlist( return removed_count +def make_wishlist_batch_row( + *, + playlist_id: str, + playlist_name: str, + track_count: int, + max_concurrent: int, + profile_id: int, + phase: str, + run_id: str | None = None, + is_album: bool = False, + album_context: Optional[Dict[str, Any]] = None, + artist_context: Optional[Dict[str, Any]] = None, + extra_fields: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """Single source of truth for a wishlist ``download_batches`` row. + + The auto and manual wishlist flows used to build this ~20-field dict in four + separate places, which let their batch shapes silently drift apart. They now + all go through here so every wishlist batch has an IDENTICAL field shape; the + genuinely per-flow differences (initial ``phase``, the auto-only + ``auto_initiated`` / ``current_cycle`` fields, album vs residual contexts) are + explicit arguments / ``extra_fields``. + + NOTE: this builds the row only — it does NOT decide grouping, batch-id + allocation, or dispatch (parallel-submit vs serial), which legitimately + differ between the flows and stay in their callers. + """ + row: Dict[str, Any] = { + 'phase': phase, + 'playlist_id': playlist_id, + 'playlist_name': playlist_name, + 'queue': [], + 'active_count': 0, + 'max_concurrent': max_concurrent, + 'queue_index': 0, + 'analysis_total': track_count, + 'analysis_processed': 0, + 'analysis_results': [], + 'permanently_failed_tracks': [], + 'cancelled_tracks': set(), + 'force_download_all': True, + 'profile_id': profile_id, + 'is_album_download': is_album, + 'album_context': album_context, + 'artist_context': artist_context, + 'wishlist_run_id': run_id, + } + if extra_fields: + row.update(extra_fields) + return row + + +def _run_wishlist_cycle( + runtime, + *, + playlist_id: str, + cycle: str, + tracks: list, + run_id: str, + auto_initiated: bool, + first_batch_id: Optional[str] = None, +) -> Dict[str, Any]: + """THE single wishlist orchestration engine — both the auto timer and the + manual trigger call this, so a manual scan runs the exact same code path as + an auto scan (group → per-album + residual batches → register → dispatch). + + Per-flow differences are arguments, not separate code: + * ``auto_initiated`` stamps the auto-only fields (auto_initiated / + auto_processing_timestamp / current_cycle, which also drives the + once-per-run cycle toggle on completion) and selects the auto vs manual + display-name + log style. + * ``first_batch_id`` lets the manual flow reuse its synchronously-created + placeholder batch so the modal's existing poll target stays valid. + + Album batches block their worker for the whole search+download, so they run + on the dedicated album pool; the residual per-track batch runs on the shared + pool. Returns a summary dict (submitted ids + album / residual counts). + """ + logger = runtime.logger + from core.wishlist.album_grouping import group_wishlist_tracks_by_album + + # Albums cycle splits into per-album bundles; singles keep the single + # per-track batch shape (Spotify already classifies them away from albums). + grouping = ( + group_wishlist_tracks_by_album( + tracks, min_tracks_per_album=_resolve_album_bundle_threshold(), + ) + if cycle == 'albums' else None + ) + + extra_fields = None + if auto_initiated: + extra_fields = { + 'auto_initiated': True, + 'auto_processing_timestamp': runtime.current_time_fn(), + 'current_cycle': cycle, + } + + # Reuse the caller-provided placeholder id for the FIRST batch created; every + # other batch gets a fresh uuid. + _reuse_id = first_batch_id + + def _alloc_id() -> str: + nonlocal _reuse_id + if _reuse_id is not None: + bid, _reuse_id = _reuse_id, None + return bid + return str(uuid.uuid4()) + + album_executor = runtime.album_bundle_executor or runtime.missing_download_executor + submitted: list = [] + + album_groups = grouping.album_groups if grouping else [] + for album_idx, group in enumerate(album_groups): + album_batch_id = _alloc_id() + album_name = group.album_context.get('name', 'Unknown') + batch_name = ( + f"Wishlist (Auto - Album: {album_name})" if auto_initiated + else f"Wishlist (Album: {album_name})" + ) + with runtime.tasks_lock: + runtime.download_batches[album_batch_id] = make_wishlist_batch_row( + playlist_id=playlist_id, + playlist_name=batch_name, + track_count=len(group.tracks), + max_concurrent=runtime.get_batch_max_concurrent(), + profile_id=runtime.profile_id, + phase='queued', + run_id=run_id, + is_album=True, + album_context=group.album_context, + artist_context=group.artist_context, + extra_fields=extra_fields, + ) + if auto_initiated: + logger.info( + f"[Auto-Wishlist] Album sub-batch {album_idx + 1}/{len(album_groups)}: " + f"'{album_name}' by '{group.artist_context.get('name')}' " + f"({len(group.tracks)} tracks) → {album_batch_id} [run {run_id[:8]}]" + ) + else: + logger.info( + f"[Manual-Wishlist] Album sub-batch {album_idx + 1}/{len(album_groups)}: " + f"'{album_name}' ({len(group.tracks)} tracks) → {album_batch_id}" + ) + submitted.append(album_batch_id) + # Album bundles block their worker for the whole search+download → dedicated + # pool (falls back to the shared pool when unset). See #740. + album_executor.submit( + runtime.run_full_missing_tracks_process, + album_batch_id, playlist_id, group.tracks, + ) + + residual_tracks = grouping.residual_tracks if grouping is not None else tracks + residual_count = len(residual_tracks) if residual_tracks else 0 + if residual_tracks: + residual_batch_id = _alloc_id() + residual_name = ( + f"Wishlist (Auto - {cycle.capitalize()})" if auto_initiated + else "Wishlist (Residual)" + ) + with runtime.tasks_lock: + runtime.download_batches[residual_batch_id] = make_wishlist_batch_row( + playlist_id=playlist_id, + playlist_name=residual_name, + track_count=residual_count, + max_concurrent=runtime.get_batch_max_concurrent(), + profile_id=runtime.profile_id, + phase='queued', + run_id=run_id, + extra_fields=extra_fields, + ) + submitted.append(residual_batch_id) + runtime.missing_download_executor.submit( + runtime.run_full_missing_tracks_process, + residual_batch_id, playlist_id, residual_tracks, + ) + if auto_initiated: + logger.info( + f"Starting wishlist residual batch {residual_batch_id} with {residual_count} tracks " + f"({'singles' if cycle == 'singles' else 'unbucketed albums'}) " + f"[run {run_id[:8]}]" + ) + else: + logger.info( + f"[Manual-Wishlist] Residual per-track batch {residual_batch_id} " + f"with {residual_count} tracks" + ) + + return { + 'submitted': submitted, + 'album_batches': len(album_groups), + 'residual_count': residual_count, + } + + def add_cancelled_tracks_to_failed_tracks( batch: Dict[str, Any], download_tasks: Dict[str, Dict[str, Any]], @@ -431,6 +627,9 @@ class WishlistManualDownloadRuntime: active_server: str profile_id: int logger: Any = module_logger + # Dedicated album-bundle pool, shared with the auto flow via + # _run_wishlist_cycle. Falls back to missing_download_executor when unset. + album_bundle_executor: Any = None def start_manual_wishlist_download_batch( @@ -455,24 +654,16 @@ def start_manual_wishlist_download_batch( playlist_name = "Wishlist" with runtime.tasks_lock: - runtime.download_batches[batch_id] = { - 'phase': 'analysis', - 'playlist_id': playlist_id, - 'playlist_name': playlist_name, - 'queue': [], - 'active_count': 0, - 'max_concurrent': runtime.get_batch_max_concurrent(), - 'queue_index': 0, - # analysis_total starts at 0; the bg job updates it after cleanup - # finishes and the real track count is known. - 'analysis_total': 0, - 'analysis_processed': 0, - 'analysis_results': [], - 'permanently_failed_tracks': [], - 'cancelled_tracks': set(), - 'force_download_all': True, - 'profile_id': runtime.profile_id, - } + # analysis_total starts at 0; the bg job updates it after cleanup + # finishes and the real track count is known. + runtime.download_batches[batch_id] = make_wishlist_batch_row( + playlist_id=playlist_id, + playlist_name=playlist_name, + track_count=0, + max_concurrent=runtime.get_batch_max_concurrent(), + profile_id=runtime.profile_id, + phase='analysis', + ) runtime.missing_download_executor.submit( _prepare_and_run_manual_wishlist_batch, @@ -560,118 +751,34 @@ def _prepare_and_run_manual_wishlist_batch( runtime.add_activity_item("", "Wishlist Download Started", f"{len(wishlist_tracks)} tracks", "Now") - # Try to split into per-album sub-batches so each album fires - # ONE slskd / torrent / usenet album-bundle search (gates on - # ``is_album_download`` + populated album/artist context). - # When a single category was requested (or no category filter) - # we apply the same grouping the auto-wishlist path uses. - # Tracks the grouper can't bucket fall through to a residual - # batch with the classic per-track flow. - from core.wishlist.album_grouping import group_wishlist_tracks_by_album - grouping = group_wishlist_tracks_by_album( - wishlist_tracks, - min_tracks_per_album=_resolve_album_bundle_threshold(), - ) - - # Build the final payload list (batch_id, tracks, album_context, - # artist_context, is_album). The first payload re-uses the - # caller-allocated ``batch_id`` so the frontend's existing poll - # against it keeps working. Subsequent payloads get fresh ids. - payloads = [] - for group in grouping.album_groups: - payloads.append({ - 'tracks': group.tracks, - 'is_album': True, - 'album_context': group.album_context, - 'artist_context': group.artist_context, - 'display_name': f"Wishlist (Album: {group.album_context.get('name', 'Unknown')})", - }) - if grouping.residual_tracks: - payloads.append({ - 'tracks': grouping.residual_tracks, - 'is_album': False, - 'album_context': None, - 'artist_context': None, - 'display_name': "Wishlist (Residual)", - }) - - if not payloads: - # Nothing to download — clear out the original batch. + if not wishlist_tracks: + # Nothing to download — clear out the placeholder batch. with runtime.tasks_lock: if batch_id in runtime.download_batches: runtime.download_batches[batch_id]['analysis_total'] = 0 runtime.download_batches[batch_id]['phase'] = 'complete' return - # Attach the original batch_id to the first payload; allocate - # fresh batch_ids for the rest. - payloads[0]['batch_id'] = batch_id - for payload in payloads[1:]: - payload['batch_id'] = str(uuid.uuid4()) - - # Reify "wishlist run" — one shared id stamped on every sub- - # batch this manual invocation produces. Mirrors the auto - # path. Note manual wishlist completion currently doesn't - # toggle the cycle (only auto does), but the id is set anyway - # so future code + UI grouping have a consistent hook. - wishlist_run_id = str(uuid.uuid4()) - - # Materialize each sub-batch's row state up-front so the - # frontend's polling can see them all under the original - # batch's flow. - with runtime.tasks_lock: - if batch_id in runtime.download_batches: - # Re-purpose the existing row for the first payload. - first = payloads[0] - runtime.download_batches[batch_id]['analysis_total'] = len(first['tracks']) - runtime.download_batches[batch_id]['wishlist_run_id'] = wishlist_run_id - if first['is_album']: - runtime.download_batches[batch_id]['is_album_download'] = True - runtime.download_batches[batch_id]['album_context'] = first['album_context'] - runtime.download_batches[batch_id]['artist_context'] = first['artist_context'] - runtime.download_batches[batch_id]['playlist_name'] = first['display_name'] - for payload in payloads[1:]: - runtime.download_batches[payload['batch_id']] = { - 'phase': 'analysis', - 'playlist_id': 'wishlist', - 'playlist_name': payload['display_name'], - 'queue': [], - 'active_count': 0, - 'max_concurrent': runtime.get_batch_max_concurrent(), - 'queue_index': 0, - 'analysis_total': len(payload['tracks']), - 'analysis_processed': 0, - 'analysis_results': [], - 'permanently_failed_tracks': [], - 'cancelled_tracks': set(), - 'force_download_all': True, - 'profile_id': runtime.profile_id, - 'is_album_download': bool(payload['is_album']), - 'album_context': payload['album_context'], - 'artist_context': payload['artist_context'], - 'wishlist_run_id': wishlist_run_id, - } - - logger.info( - f"[Manual-Wishlist] Split into {len(payloads)} sub-batch(es) " - f"({sum(1 for p in payloads if p['is_album'])} album + " - f"{sum(1 for p in payloads if not p['is_album'])} residual)" + # Run the selection through the SHARED engine — the exact code path the + # auto timer uses (group → album bundles + per-track residual → parallel + # dispatch on the album / shared pools). cycle='albums' bundles whatever + # forms an album and drops the rest (singles / ungroupable) into the + # per-track residual, so this single call covers the whole selection. + # The placeholder batch_id is reused as the first sub-batch so the + # modal's existing poll target stays valid. + result = _run_wishlist_cycle( + runtime, + playlist_id='wishlist', + cycle='albums', + tracks=wishlist_tracks, + run_id=str(uuid.uuid4()), + auto_initiated=False, + first_batch_id=batch_id, + ) + logger.info( + f"[Manual-Wishlist] Dispatched {result['album_batches']} album batch(es) + " + f"{result['residual_count']} residual track(s) via the shared engine" ) - # Serial dispatch — each album-bundle search happens one at a - # time so the slskd / Prowlarr pipeline doesn't fan out across - # multiple parallel release searches. - for payload in payloads: - label = ( - f"album '{payload['album_context'].get('name')}'" - if payload['is_album'] else 'residual per-track' - ) - logger.info( - f"[Manual-Wishlist] Running sub-batch {payload['batch_id']} " - f"({label}, {len(payload['tracks'])} tracks)" - ) - runtime.run_full_missing_tracks_process( - payload['batch_id'], "wishlist", payload['tracks'], - ) except Exception as exc: logger.error(f"Error preparing manual wishlist batch {batch_id}: {exc}") @@ -833,142 +940,24 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom for i, track in enumerate(wishlist_tracks): track['_original_index'] = i - # When the cycle is 'albums', try to split the wishlist - # into per-album sub-batches so each album fires ONE - # album-bundle search (slskd / torrent / usenet) instead - # of N per-track searches. Residual tracks (no resolvable - # album metadata) fall through to a normal per-track - # batch. Singles cycle keeps its original single-batch - # shape — Spotify already classifies them away from - # albums. - _submitted_batches: list[str] = [] - if current_cycle == 'albums': - from core.wishlist.album_grouping import group_wishlist_tracks_by_album - grouping = group_wishlist_tracks_by_album( - wishlist_tracks, - min_tracks_per_album=_resolve_album_bundle_threshold(), - ) - else: - grouping = None - - # Reify "wishlist run" — one shared id stamped on every - # sub-batch this invocation produces. The completion - # handler uses it to gate the once-per-run cycle toggle - # (so it doesn't fire N times for N sub-batches). + # Reify one "wishlist run" id (the completion handler gates the + # once-per-run cycle toggle on it) and hand off to the SHARED + # wishlist engine — the same code path the manual trigger uses. wishlist_run_id = str(uuid.uuid4()) - - if grouping and grouping.album_groups: - for album_idx, group in enumerate(grouping.album_groups): - album_batch_id = str(uuid.uuid4()) - album_batch_name = ( - f"Wishlist (Auto - Album: {group.album_context.get('name', 'Unknown')})" - ) - with runtime.tasks_lock: - runtime.download_batches[album_batch_id] = { - # ``queued`` until the master worker - # picks the batch up from the - # ``missing_download_executor`` pool - # (max_workers=3 by default). The worker - # flips phase to ``analysis`` as its - # first action — see - # ``core/downloads/master.py:328``. - # Pre-fix the row was created with - # ``analysis`` directly, so a wishlist - # run with N > 3 sub-batches looked like - # all N were working when really only - # 3 were running. - 'phase': 'queued', - 'playlist_id': playlist_id, - 'playlist_name': album_batch_name, - 'queue': [], - 'active_count': 0, - 'max_concurrent': runtime.get_batch_max_concurrent(), - 'queue_index': 0, - 'analysis_total': len(group.tracks), - 'analysis_processed': 0, - 'analysis_results': [], - 'permanently_failed_tracks': [], - 'cancelled_tracks': set(), - 'force_download_all': True, - 'auto_initiated': True, - 'auto_processing_timestamp': runtime.current_time_fn(), - 'current_cycle': current_cycle, - 'profile_id': runtime.profile_id, - # Album-bundle dispatch gate reads these - # three. With them set, the master worker - # routes through slskd / torrent / usenet - # album-bundle search instead of per-track. - 'is_album_download': True, - 'album_context': group.album_context, - 'artist_context': group.artist_context, - 'wishlist_run_id': wishlist_run_id, - } - logger.info( - f"[Auto-Wishlist] Album sub-batch {album_idx + 1}/{len(grouping.album_groups)}: " - f"'{group.album_context.get('name')}' by '{group.artist_context.get('name')}' " - f"({len(group.tracks)} tracks) → {album_batch_id} [run {wishlist_run_id[:8]}]" - ) - _submitted_batches.append(album_batch_id) - # Album bundles block their worker thread for the whole - # search+download, so run them on the dedicated album - # pool — never the shared pool that serves analysis, - # per-track downloads and the manual wishlist (#740). - # Fall back to the shared pool if unset (older callers). - _album_executor = ( - runtime.album_bundle_executor or runtime.missing_download_executor - ) - _album_executor.submit( - runtime.run_full_missing_tracks_process, - album_batch_id, playlist_id, group.tracks, - ) - - # Residual tracks (no album group could be formed, OR - # singles cycle): one classic per-track batch as before. - residual_tracks = ( - grouping.residual_tracks if grouping is not None else wishlist_tracks + _cycle_result = _run_wishlist_cycle( + runtime, + playlist_id=playlist_id, + cycle=current_cycle, + tracks=wishlist_tracks, + run_id=wishlist_run_id, + auto_initiated=True, ) - if residual_tracks: - batch_id = str(uuid.uuid4()) - playlist_name = f"Wishlist (Auto - {current_cycle.capitalize()})" - with runtime.tasks_lock: - runtime.download_batches[batch_id] = { - # See album sub-batch above — ``queued`` - # until the master worker picks it up. - 'phase': 'queued', - 'playlist_id': playlist_id, - 'playlist_name': playlist_name, - 'queue': [], - 'active_count': 0, - 'max_concurrent': runtime.get_batch_max_concurrent(), - 'queue_index': 0, - 'analysis_total': len(residual_tracks), - 'analysis_processed': 0, - 'analysis_results': [], - 'permanently_failed_tracks': [], - 'cancelled_tracks': set(), - 'force_download_all': True, - 'auto_initiated': True, - 'auto_processing_timestamp': runtime.current_time_fn(), - 'current_cycle': current_cycle, - 'profile_id': runtime.profile_id, - 'wishlist_run_id': wishlist_run_id, - } - _submitted_batches.append(batch_id) - runtime.missing_download_executor.submit( - runtime.run_full_missing_tracks_process, - batch_id, playlist_id, residual_tracks, - ) - logger.info( - f"Starting wishlist residual batch {batch_id} with {len(residual_tracks)} tracks " - f"({'singles' if current_cycle == 'singles' else 'unbucketed albums'}) " - f"[run {wishlist_run_id[:8]}]" - ) _summary_parts: list[str] = [] - if grouping and grouping.album_groups: - _summary_parts.append(f"{len(grouping.album_groups)} album batch(es)") - if residual_tracks: - _summary_parts.append(f"{len(residual_tracks)} per-track") + if _cycle_result['album_batches']: + _summary_parts.append(f"{_cycle_result['album_batches']} album batch(es)") + if _cycle_result['residual_count']: + _summary_parts.append(f"{_cycle_result['residual_count']} per-track") _summary_text = ', '.join(_summary_parts) or 'no batches' runtime.update_automation_progress( automation_id, progress=50, diff --git a/tests/wishlist/test_batch_factory.py b/tests/wishlist/test_batch_factory.py new file mode 100644 index 00000000..034ee50f --- /dev/null +++ b/tests/wishlist/test_batch_factory.py @@ -0,0 +1,102 @@ +"""Tests for make_wishlist_batch_row — the single source of truth for a wishlist +download_batches row, shared by the auto and manual flows so their batch shapes +can't drift apart. +""" + +from __future__ import annotations + +from core.wishlist.processing import make_wishlist_batch_row + + +_CORE_KEYS = { + 'phase', 'playlist_id', 'playlist_name', 'queue', 'active_count', + 'max_concurrent', 'queue_index', 'analysis_total', 'analysis_processed', + 'analysis_results', 'permanently_failed_tracks', 'cancelled_tracks', + 'force_download_all', 'profile_id', 'is_album_download', 'album_context', + 'artist_context', 'wishlist_run_id', +} + + +def _row(**overrides): + base = dict( + playlist_id='wishlist', playlist_name='Wishlist', track_count=3, + max_concurrent=4, profile_id=1, phase='analysis', + ) + base.update(overrides) + return make_wishlist_batch_row(**base) + + +def test_core_fields_always_present_and_consistent(): + row = _row() + assert _CORE_KEYS <= set(row.keys()) + # Fresh-batch invariants. + assert row['queue'] == [] and row['active_count'] == 0 and row['queue_index'] == 0 + assert row['analysis_processed'] == 0 + assert row['analysis_results'] == [] and row['permanently_failed_tracks'] == [] + assert row['cancelled_tracks'] == set() + assert row['force_download_all'] is True + assert row['analysis_total'] == 3 + assert row['max_concurrent'] == 4 + assert row['profile_id'] == 1 + + +def test_residual_defaults_are_per_track(): + row = _row() + assert row['is_album_download'] is False + assert row['album_context'] is None and row['artist_context'] is None + assert row['wishlist_run_id'] is None + + +def test_album_batch_carries_context(): + row = _row( + phase='queued', run_id='run-1', is_album=True, + album_context={'name': 'Album One'}, artist_context={'name': 'Artist 1'}, + ) + assert row['phase'] == 'queued' + assert row['is_album_download'] is True + assert row['album_context'] == {'name': 'Album One'} + assert row['artist_context'] == {'name': 'Artist 1'} + assert row['wishlist_run_id'] == 'run-1' + + +def test_extra_fields_merged_for_auto(): + row = _row(extra_fields={ + 'auto_initiated': True, 'auto_processing_timestamp': 123.0, + 'current_cycle': 'albums', + }) + assert row['auto_initiated'] is True + assert row['auto_processing_timestamp'] == 123.0 + assert row['current_cycle'] == 'albums' + + +def test_manual_row_has_no_auto_fields(): + """Manual rows must not carry the auto-only fields (no extra_fields).""" + row = _row(phase='analysis') + assert 'auto_initiated' not in row + assert 'current_cycle' not in row + + +def test_fresh_rows_do_not_share_mutable_state(): + """Each row must get its OWN queue/list/set — not a shared reference that + one batch's tasks could leak into another's.""" + a = _row() + b = _row() + a['queue'].append('task-1') + a['cancelled_tracks'].add('x') + assert b['queue'] == [] + assert b['cancelled_tracks'] == set() + assert b['analysis_results'] == [] + + +def test_auto_and_manual_rows_share_identical_key_shape(): + """The drift-prevention guarantee: an auto album row and a manual album row + expose the same set of keys (modulo the auto-only extras), so the modal / + status code sees a consistent shape from both flows.""" + manual = _row(phase='analysis', run_id='r', is_album=True, + album_context={'name': 'A'}, artist_context={'name': 'B'}) + auto = _row(phase='queued', run_id='r', is_album=True, + album_context={'name': 'A'}, artist_context={'name': 'B'}, + extra_fields={'auto_initiated': True, 'current_cycle': 'albums'}) + # Auto is a strict superset (the auto-only extras); the shared core is identical. + assert set(manual.keys()) <= set(auto.keys()) + assert set(auto.keys()) - set(manual.keys()) == {'auto_initiated', 'current_cycle'} diff --git a/tests/wishlist/test_manual_download.py b/tests/wishlist/test_manual_download.py index 7e1386ad..bb3429c1 100644 --- a/tests/wishlist/test_manual_download.py +++ b/tests/wishlist/test_manual_download.py @@ -110,6 +110,17 @@ def _run_submitted_bg_job(executor): fn(*args, **kwargs) +def _dispatched(executor, runtime): + """The run_full_missing_tracks_process dispatches the shared engine submitted + to the executor (everything after the initial bg-job submission). Manual now + parallel-dispatches via the engine instead of running them serially inline, + so the master worker is *submitted*, not called directly.""" + return [ + s for s in executor.submissions + if s[0] is runtime.run_full_missing_tracks_process + ] + + def test_start_manual_wishlist_download_batch_returns_immediately_with_placeholder(): """Endpoint returns 200 immediately; cleanup runs in the bg job.""" runtime, service, _db, executor, _logger, activity_calls, batch_map, master_calls = _build_runtime( @@ -174,11 +185,15 @@ def test_start_manual_wishlist_download_batch_filters_track_ids_and_starts_batch _run_submitted_bg_job(executor) assert activity_calls == [("", "Wishlist Download Started", "1 tracks", "Now")] - assert len(master_calls) == 1 - master_args, _ = master_calls[0] - assert master_args[1] == "wishlist" - assert master_args[2][0]["id"] == "track-2" - assert master_args[2][0]["_original_index"] == 0 + # One track → no album group (threshold 2) → one residual batch, dispatched + # via the shared engine (submitted to the executor, not called inline). + dispatched = _dispatched(executor, runtime) + assert len(dispatched) == 1 + args = dispatched[0][1] + assert args[1] == "wishlist" + assert args[2][0]["id"] == "track-2" + assert args[2][0]["_original_index"] == 0 + assert args[0] == payload["batch_id"] # placeholder batch_id reused as first sub-batch assert batch_map[payload["batch_id"]]["analysis_total"] == 1 assert batch_map[payload["batch_id"]]["force_download_all"] is True assert any("Filtered to 1 specific tracks by ID" in msg for msg in logger.info_messages) @@ -224,10 +239,12 @@ def test_start_manual_wishlist_download_batch_does_not_run_library_cleanup(): # The library check is skipped entirely — no per-track DB lookups. assert db.track_checks == [] - # All tracks are submitted to the master worker — including the "owned" one. - assert len(master_calls) == 1 - master_args, _ = master_calls[0] - assert [track["id"] for track in master_args[2]] == ["enhance-1", "owned-1"] + # All tracks are dispatched to the master worker — including the "owned" one. + # Two single-track albums → no album groups → one residual batch of both. + dispatched = _dispatched(executor, runtime) + assert len(dispatched) == 1 + args = dispatched[0][1] + assert [track["id"] for track in args[2]] == ["enhance-1", "owned-1"] assert batch_map[payload["batch_id"]]["analysis_total"] == 2 assert activity_calls == [("", "Wishlist Download Started", "2 tracks", "Now")] @@ -293,28 +310,33 @@ def test_manual_wishlist_splits_into_per_album_sub_batches(): assert status == 200 _run_submitted_bg_job(executor) - # Two album groups → two master-worker calls. - assert len(master_calls) == 2 + # Two album groups → two album sub-batches, PARALLEL-dispatched via the shared + # engine (same as auto) — not serial inline calls. + dispatched = _dispatched(executor, runtime) + assert len(dispatched) == 2 + assert master_calls == [] # nothing run synchronously inline anymore - # First sub-batch uses the caller-allocated batch_id. - first_args, _ = master_calls[0] + # First sub-batch reuses the caller-allocated placeholder batch_id. + first_args = dispatched[0][1] assert first_args[0] == payload["batch_id"] assert batch_map[payload["batch_id"]].get("is_album_download") is True + # Both dispatched batches are album bundles. + for _fn, args, _kw in dispatched: + assert batch_map[args[0]].get("is_album_download") is True # Second sub-batch gets a fresh uuid; its row exists in batch_map. - second_args, _ = master_calls[1] + second_args = dispatched[1][1] assert second_args[0] != payload["batch_id"] assert second_args[0] in batch_map - assert batch_map[second_args[0]].get("is_album_download") is True # Track counts across the two sub-batches: 2 each at threshold=2. - counts = sorted(len(args[2]) for args, _ in master_calls) + counts = sorted(len(args[2]) for _fn, args, _kw in dispatched) assert counts == [2, 2] # Both sub-batches carry album context populated from spotify_data. album_names = { batch_map[args[0]]["album_context"]["name"] - for args, _ in master_calls + for _fn, args, _kw in dispatched } assert album_names == {"Album One", "Album Two"} diff --git a/web_server.py b/web_server.py index c6332ae5..0da4c37e 100644 --- a/web_server.py +++ b/web_server.py @@ -15155,6 +15155,7 @@ def start_wishlist_missing_downloads(): download_batches=download_batches, tasks_lock=tasks_lock, missing_download_executor=missing_download_executor, + album_bundle_executor=album_bundle_executor, run_full_missing_tracks_process=_run_full_missing_tracks_process, get_batch_max_concurrent=_get_batch_max_concurrent, add_activity_item=add_activity_item,