danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Wishlist: route the manual flow through the shared engine (manual == auto)

Browse source 
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.
This commit is contained in:
BoulderBadgeDad 2026-05-29 17:43:40 -07:00
parent db1e51109c
commit d3c897fb9d
3 changed files with 64 additions and 114 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

121
core/wishlist/processing.py
View file

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

56
tests/wishlist/test_manual_download.py
View file

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

1
web_server.py
View file

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