From d3c897fb9d2efceed8a45c5c895fbfe29c0fcb08 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Fri, 29 May 2026 17:43:40 -0700 Subject: [PATCH] Wishlist: route the manual flow through the shared engine (manual == auto) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 2: the manual 'Download Wishlist' flow now calls the same _run_wishlist_cycle engine the auto timer uses, so a manual scan runs the exact same code path as an auto scan. The old bespoke manual orchestration (build payloads + SERIAL inline dispatch) is deleted — its grouping/dispatch was a near-duplicate of auto's that had already drifted. Behavior changes (all intended, discussed): - Manual now dispatches album bundles in PARALLEL (album pool) like auto, instead of serially on one thread. A single cycle='albums' engine call covers the whole selection (albums bundled, singles/ungroupable -> per-track residual), so no 'both cycles' pass is needed. - The manual placeholder batch_id is reused as the engine's first sub-batch (first_batch_id), so the modal's existing poll target stays valid. - WishlistManualDownloadRuntime gains album_bundle_executor (wired in web_server, falls back to the shared pool when unset). - 'Don't start manual while auto is running' is unchanged — the existing route guard (is_wishlist_actually_processing -> 409) already covers it; no queue added. NOT touched: process_wishlist_automatically's behavior (proven by test_automation staying green in Stage 1) and the per-track download mechanics. test_manual_download.py rewritten to characterize the new behavior (engine dispatch via the executor, parallel, placeholder reuse, album-context). Full wishlist suite green (131); wishlist + automation = 392 passed. --- core/wishlist/processing.py | 121 +++++-------------------- tests/wishlist/test_manual_download.py | 56 ++++++++---- web_server.py | 1 + 3 files changed, 64 insertions(+), 114 deletions(-) diff --git a/core/wishlist/processing.py b/core/wishlist/processing.py index 24e252f9..7cfe4a95 100644 --- a/core/wishlist/processing.py +++ b/core/wishlist/processing.py @@ -627,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( @@ -748,110 +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']] = make_wishlist_batch_row( - playlist_id='wishlist', - playlist_name=payload['display_name'], - track_count=len(payload['tracks']), - max_concurrent=runtime.get_batch_max_concurrent(), - profile_id=runtime.profile_id, - phase='analysis', - run_id=wishlist_run_id, - is_album=bool(payload['is_album']), - album_context=payload['album_context'], - artist_context=payload['artist_context'], - ) - - 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}") 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,