diff --git a/core/downloads/monitor.py b/core/downloads/monitor.py index 94b4a727..3252f324 100644 --- a/core/downloads/monitor.py +++ b/core/downloads/monitor.py @@ -332,6 +332,13 @@ class WebUIDownloadMonitor: live_info = live_transfers_lookup.get(lookup_key) if not live_info: + # User-initiated manual pick — skip auto-retry. The status + # engine fallback owns the terminal transition for non-Soulseek + # manual downloads. Yanking the task back to 'searching' here + # would defeat the user's explicit selection. + if task.get('_user_manual_pick'): + return False + # Task not in live transfers but status is downloading/queued - likely stuck if current_time - task.get('status_change_time', current_time) > 90: retry_count = task.get('stuck_retry_count', 0) @@ -396,6 +403,11 @@ class WebUIDownloadMonitor: # IMMEDIATE ERROR RETRY: Check for errored/rejected/timed-out downloads first (no timeout needed) if 'Errored' in state_str or 'Failed' in state_str or 'Rejected' in state_str or 'TimedOut' in state_str: + # Same manual-pick guard as the not-in-live-transfers path — + # user explicitly selected this candidate, surface the failure. + if task.get('_user_manual_pick'): + return False + retry_count = task.get('error_retry_count', 0) last_retry = task.get('last_error_retry_time', 0) diff --git a/core/downloads/status.py b/core/downloads/status.py index a3359272..c1599c5a 100644 --- a/core/downloads/status.py +++ b/core/downloads/status.py @@ -21,6 +21,7 @@ are passed via `StatusDeps` so the module is web_server-import-free. from __future__ import annotations import logging +import threading import time from dataclasses import dataclass from typing import Any, Callable, Optional @@ -34,6 +35,37 @@ from core.runtime_state import ( logger = logging.getLogger(__name__) +def _schedule_completion_callback(deps, batch_id: str, task_id: str, success: bool) -> None: + """Fire ``deps.on_download_completed`` on a one-shot daemon thread so + the caller can hold ``tasks_lock`` without deadlocking. + + ``on_download_completed`` re-acquires ``tasks_lock`` (it removes the + completed task from the batch's active set, decrements active_count, + and may submit the next queued worker). Calling it synchronously + from within ``build_batch_status_data`` — which is invoked under the + same Lock — would self-deadlock since ``threading.Lock`` is not + reentrant. A daemon thread defers the call until after the lock is + released. + """ + if deps.on_download_completed is None: + return + + def _run(): + try: + deps.on_download_completed(batch_id, task_id, success) + except Exception as exc: + logger.error( + "[Status] deferred on_download_completed raised for task %s: %s", + task_id, exc, + ) + + threading.Thread( + target=_run, + name=f"on-completed-{task_id[:8]}", + daemon=True, + ).start() + + @dataclass class StatusDeps: """Cross-cutting deps the status helpers need.""" @@ -43,6 +75,162 @@ class StatusDeps: make_context_key: Callable[[str, str], str] submit_post_processing: Callable[[str, str], None] # (task_id, batch_id) -> None get_cached_transfer_data: Callable[[], dict] + # Engine-state fallback for non-Soulseek (streaming) downloads. + # Without these, YouTube/Tidal/Qobuz/HiFi/Deezer/SoundCloud/Lidarr + # tasks never appear in live_transfers_lookup so their status never + # advances out of 'downloading 0%'. + download_orchestrator: Any = None + run_async: Optional[Callable] = None + on_download_completed: Optional[Callable[[str, str, bool], None]] = None + + +# Streaming sources the engine fallback applies to. Soulseek goes through +# slskd's live_transfers path and must NOT hit the engine fallback. +_STREAMING_SOURCE_NAMES = frozenset(( + 'youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', +)) + +# Keep these in sync with the engine plugins' state strings. +_ENGINE_FAILURE_STATES = ('Errored', 'Failed', 'Rejected', 'TimedOut', 'Aborted') +_ENGINE_CANCELLED_STATES = ('Cancelled', 'Canceled') +_ENGINE_SUCCESS_STATES = ('Succeeded', 'Completed, Succeeded') + + +def _engine_state_str(record: Any) -> str: + if record is None: + return '' + state = getattr(record, 'state', None) + if state is None and isinstance(record, dict): + state = record.get('state') + return str(state) if state is not None else '' + + +def _engine_progress_pct(record: Any) -> float: + if record is None: + return 0 + progress = getattr(record, 'progress', None) + if progress is None and isinstance(record, dict): + progress = record.get('progress') + try: + progress = float(progress) + except (TypeError, ValueError): + return 0 + if progress <= 1.0: + progress *= 100 + return progress + + +def _apply_engine_state_fallback( + task_id: str, + task: dict, + task_status: dict, + batch_id: str, + deps: StatusDeps, +) -> None: + """Populate ``task_status`` from the download engine's per-source + record when the task isn't in ``live_transfers_lookup`` — i.e. it's + a non-Soulseek streaming source. Mirrors the Soulseek branch's + Cancelled → Failed → Succeeded → InProgress priority order so + compound states like ``"Completed, Errored"`` hit the failure branch + first. + + Mutates ``task`` in place (status / error_message) the same way the + Soulseek branch does, so the next status poll sees the new state. + Submits post-processing on terminal success and fires + ``on_download_completed`` on terminal failure to free the worker + slot. + """ + if deps.download_orchestrator is None or deps.run_async is None: + return + if task.get('status') in ('completed', 'failed', 'cancelled', 'not_found', 'post_processing'): + return + # Scope this fallback to user-initiated manual picks. Auto attempts + # already flow through the live_transfers_lookup IF branch (the engine + # pre-populates non-Soulseek records via get_all_downloads), and on + # failure the monitor's existing retry path picks the next candidate. + # Marking auto attempts failed here would short-circuit that fallback. + if not task.get('_user_manual_pick'): + return + download_id = task.get('download_id') + if not download_id: + return + ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} + username = task.get('username') or ti.get('username') + if username not in _STREAMING_SOURCE_NAMES: + return + + try: + record = deps.run_async( + deps.download_orchestrator.get_download_status(download_id) + ) + except Exception as exc: + logger.debug( + "[Engine Fallback] get_download_status(%s) raised: %s", + download_id, exc, + ) + return + + if record is None: + return + + state_str = _engine_state_str(record) + if not state_str: + return + + if any(s in state_str for s in _ENGINE_CANCELLED_STATES): + if task['status'] != 'cancelled': + task['status'] = 'cancelled' + err = getattr(record, 'error_message', None) or getattr(record, 'error', None) or '' + if err: + task['error_message'] = str(err) + _schedule_completion_callback(deps, batch_id, task_id, False) + task_status['status'] = 'cancelled' + task_status['progress'] = _engine_progress_pct(record) + return + + if any(s in state_str for s in _ENGINE_FAILURE_STATES): + if task['status'] != 'failed': + task['status'] = 'failed' + err = getattr(record, 'error_message', None) or getattr(record, 'error', None) or '' + task['error_message'] = ( + str(err) if err + else f'{username} download failed (engine state: {state_str})' + ) + logger.info( + "[Engine Fallback] Task %s engine reports '%s' — marking failed", + task_id, state_str, + ) + _schedule_completion_callback(deps, batch_id, task_id, False) + task_status['status'] = 'failed' + task_status['error_message'] = task.get('error_message') + task_status['progress'] = _engine_progress_pct(record) + return + + if any(s in state_str for s in _ENGINE_SUCCESS_STATES): + if task['status'] != 'post_processing': + task['status'] = 'post_processing' + logger.info( + "[Engine Fallback] Task %s engine reports '%s' — starting post-processing verification", + task_id, state_str, + ) + try: + deps.submit_post_processing(task_id, batch_id) + except Exception as exc: + logger.error( + "[Engine Fallback] submit_post_processing raised for task %s: %s", + task_id, exc, + ) + task_status['status'] = 'post_processing' + task_status['progress'] = 95 + return + + if 'InProgress' in state_str: + task_status['status'] = 'downloading' + if task['status'] in ('searching', 'queued'): + task['status'] = 'downloading' + elif 'Queued' in state_str: + task_status['status'] = 'queued' + task_status['progress'] = _engine_progress_pct(record) def build_batch_status_data(batch_id: str, batch: dict, live_transfers_lookup: dict, deps: StatusDeps) -> dict: @@ -144,17 +332,41 @@ def build_batch_status_data(batch_id: str, batch: dict, live_transfers_lookup: d task_status['status'] = 'cancelled' task['status'] = 'cancelled' elif 'Failed' in state_str or 'Errored' in state_str or 'Rejected' in state_str or 'TimedOut' in state_str: - # UNIFIED ERROR HANDLING: Let monitor handle errors for consistency - # Monitor will detect errored state and trigger retry within 5 seconds - logger.error(f"Task {task_id} API shows error state: {state_str} - letting monitor handle retry") - - # Keep task in current status (downloading/queued) so monitor can detect error - # Don't mark as failed here - let the unified retry system handle it - if task['status'] in ['searching', 'downloading', 'queued']: - task_status['status'] = task['status'] # Keep current status for monitor + # User-initiated manual pick — surface the failure + # immediately. The monitor's auto-retry path is gated + # on `_user_manual_pick` and won't fire, so deferring + # to it would leave the task stuck at 'downloading 0%' + # forever. Mark failed here and free the worker slot. + if task.get('_user_manual_pick'): + err_msg = live_info.get('errorMessage') or live_info.get('error') or '' + task['status'] = 'failed' + task['error_message'] = ( + str(err_msg) if err_msg + else f'Manual pick failed (state: {state_str})' + ) + task_status['status'] = 'failed' + task_status['error_message'] = task['error_message'] + logger.info( + f"[Manual Pick] Task {task_id} engine reports '{state_str}' — marking failed" + ) + # NOTE: caller (build_batched_status) holds + # tasks_lock. on_download_completed re-acquires + # the same Lock — synchronous call would + # deadlock. Spawn a thread so it runs after we + # release the lock. + _schedule_completion_callback(deps, batch_id, task_id, False) else: - task_status['status'] = 'downloading' # Default to downloading for error detection - task['status'] = 'downloading' + # UNIFIED ERROR HANDLING: Let monitor handle errors for consistency + # Monitor will detect errored state and trigger retry within 5 seconds + logger.error(f"Task {task_id} API shows error state: {state_str} - letting monitor handle retry") + + # Keep task in current status (downloading/queued) so monitor can detect error + # Don't mark as failed here - let the unified retry system handle it + if task['status'] in ['searching', 'downloading', 'queued']: + task_status['status'] = task['status'] # Keep current status for monitor + else: + task_status['status'] = 'downloading' # Default to downloading for error detection + task['status'] = 'downloading' elif 'Completed' in state_str or 'Succeeded' in state_str: # Verify bytes actually transferred before trusting state string expected_size = live_info.get('size', 0) @@ -193,6 +405,15 @@ def build_batch_status_data(batch_id: str, batch: dict, live_transfers_lookup: d task_status['progress'] = 100 elif task['status'] == 'post_processing': task_status['progress'] = 95 # Nearly complete, just verifying + else: + # Non-Soulseek (streaming) sources don't appear in + # slskd's live_transfers_lookup — poll the engine + # directly so YouTube/Tidal/Qobuz/HiFi/Deezer/ + # SoundCloud/Lidarr tasks actually advance out of + # 'downloading 0%' instead of staying there forever. + _apply_engine_state_fallback( + task_id, task, task_status, batch_id, deps, + ) batch_tasks.append(task_status) batch_tasks.sort(key=lambda x: x['track_index']) response_data['tasks'] = batch_tasks diff --git a/tests/test_manual_pick_no_auto_retry.py b/tests/test_manual_pick_no_auto_retry.py new file mode 100644 index 00000000..c5b23cfe --- /dev/null +++ b/tests/test_manual_pick_no_auto_retry.py @@ -0,0 +1,95 @@ +"""Pin the ``_user_manual_pick`` flag — auto-retry monitor must not +yank a manually-picked download back to 'searching' when it fails. + +User report: "I manually searched and selected a candidate. It went +to 0% downloading, then suddenly switched back to searching." The +download monitor's ``_should_retry_task`` was treating the failed +manual pick the same as a normal auto-attempt failure — fresh search, +new query, new candidates. From the user's perspective: "I picked +THIS one, why is it searching for something else now?" + +Fix: when the user explicitly selects a candidate via +``/api/downloads/task//download-candidate``, the task is flagged +``_user_manual_pick=True``. The monitor's retry decision checks that +flag and bails — letting the failure surface to the user instead of +auto-falling-back. + +These tests exercise ``_should_retry_task`` directly with the flag +set + various engine/transfer states. +""" + +from __future__ import annotations + +import time + +import pytest + +from core.downloads import monitor as dm + + +@pytest.fixture +def fake_monitor(monkeypatch): + """Monitor with patched make_context_key so tests don't pull in the + real one.""" + monkeypatch.setattr(dm, '_make_context_key', lambda u, f: f"{u}::{f}") + return dm.WebUIDownloadMonitor() + + +def _task(*, username='youtube', filename='vid.mp3', status='downloading', + download_id='dl-1', manual_pick=False, status_change_time=None): + return { + 'track_info': {'name': 'Test Track'}, + 'username': username, + 'filename': filename, + 'status': status, + 'download_id': download_id, + '_user_manual_pick': manual_pick, + 'status_change_time': status_change_time or (time.time() - 1000), + } + + +def test_manual_pick_skips_retry_when_not_in_live_transfers(fake_monitor): + """The not-in-live-transfers stuck-task path normally triggers + retry after 90s. Manual picks must bypass it — no retry submitted, + no status reset to 'searching'.""" + task = _task(manual_pick=True) + deferred_ops = [] + + result = fake_monitor._should_retry_task( + task_id='t1', + task=task, + live_transfers_lookup={}, + current_time=time.time(), + deferred_ops=deferred_ops, + ) + + assert result is False + assert deferred_ops == [] + assert task['status'] == 'downloading' + + +def test_manual_pick_skips_retry_on_errored_state(fake_monitor): + """When the task IS in live_transfers but the engine reports + Errored, the monitor would normally retry. Manual picks bypass + that path too.""" + task = _task(manual_pick=True) + deferred_ops = [] + live_lookup = { + 'youtube::vid.mp3': { + 'state': 'Completed, Errored', + 'percentComplete': 0, + } + } + + fake_monitor._should_retry_task( + task_id='t1', + task=task, + live_transfers_lookup=live_lookup, + current_time=time.time(), + deferred_ops=deferred_ops, + ) + + assert deferred_ops == [] + assert task['status'] == 'downloading' + + diff --git a/tests/test_manual_search_endpoint.py b/tests/test_manual_search_endpoint.py index f89e61ad..3ab85f99 100644 --- a/tests/test_manual_search_endpoint.py +++ b/tests/test_manual_search_endpoint.py @@ -5,6 +5,12 @@ a failed download. Manual search adds a second avenue — type a query, hit search, get fresh results from the configured download source(s) without having to leave the modal. +The endpoint streams results as NDJSON — one JSON object per line — so +the modal can render rows from each source as that source's search +completes, instead of blocking the whole UI on the slowest source. The +``_consume_ndjson`` helper below replays the stream as a list of message +dicts so test assertions stay readable. + These tests cover the new endpoint's validation + dispatch behavior: - Query length / source whitelist validation @@ -12,12 +18,8 @@ These tests cover the new endpoint's validation + dispatch behavior: - Per-source request hits only the named source - Unconfigured sources in hybrid mode are silently skipped - Task lookup gates the endpoint with a 404 when the task isn't known - -The endpoint constructs candidate JSON via the same helpers the -``/candidates`` endpoint uses, so the response shape carries the same -fields (track_info + candidates) plus a ``source`` tag on each candidate -so the manual-search frontend can show a per-row source badge in 'all' -mode. +- Per-source exceptions emit ``source_error`` events but don't fail the + overall stream The existing ``/download-candidate`` retry path is unchanged — manual- search results POST through the same endpoint so AcoustID + post-download @@ -26,11 +28,37 @@ safety nets stay in the loop. from __future__ import annotations +import json from unittest.mock import MagicMock, patch import pytest +def _consume_ndjson(resp) -> list: + """Parse a Flask test-client streaming response into a list of + NDJSON message dicts. The endpoint emits one JSON object per line, + each terminated by ``\\n``. + """ + raw = resp.get_data(as_text=True) + msgs = [] + for line in raw.splitlines(): + line = line.strip() + if not line: + continue + msgs.append(json.loads(line)) + return msgs + + +def _flatten_candidates(msgs: list) -> list: + """Pull all candidate dicts out of every ``source_results`` message — + same flat list the old single-shot response used to return.""" + out = [] + for m in msgs: + if m.get('type') == 'source_results': + out.extend(m.get('candidates', [])) + return out + + # --------------------------------------------------------------------------- # Fixture — Flask test client + mocked plugin registry. # --------------------------------------------------------------------------- @@ -237,17 +265,13 @@ def test_manual_search_handles_task_not_found(manual_search_client): def test_manual_search_dispatches_to_configured_source_only(manual_search_client): """In hybrid mode, source='youtube' should hit only the youtube plugin's - search — not soulseek, hifi, etc. The candidates endpoint already - validates source against the configured list, so unconfigured plugins - aren't reachable here.""" + search — not soulseek, hifi, etc.""" client, ctx = manual_search_client plugins = ctx['plugins'] - # Configure youtube to return one result so we can verify the response. plugins['youtube'] = _make_plugin( search_results=[_fake_track_result('youtube_song.mp3', 'youtube')] ) - # Re-wire the orchestrator's client() so it returns the new mock. import web_server web_server.download_orchestrator.registry._plugins['youtube'] = plugins['youtube'] @@ -257,9 +281,9 @@ def test_manual_search_dispatches_to_configured_source_only(manual_search_client ) assert resp.status_code == 200 - body = resp.get_json() - assert len(body['candidates']) == 1 - # Only youtube should have been searched. + msgs = _consume_ndjson(resp) + candidates = _flatten_candidates(msgs) + assert len(candidates) == 1 assert plugins['youtube'].search.call_count == 1 assert plugins['soulseek'].search.call_count == 0 assert plugins['hifi'].search.call_count == 0 @@ -268,12 +292,10 @@ def test_manual_search_dispatches_to_configured_source_only(manual_search_client def test_manual_search_all_dispatches_parallel(manual_search_client): """source='all' with hybrid mode → searches every CONFIGURED source. - Tidal is unconfigured (is_configured()=False) so it's filtered out - upstream by _list_available_download_sources — 5 of 6 sources hit.""" + Tidal is unconfigured so it's filtered out — 5 of 6 sources hit.""" client, ctx = manual_search_client plugins = ctx['plugins'] - # Make each configured source return one distinct result. import web_server for src_name in ('soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'): new_plugin = _make_plugin( @@ -288,23 +310,47 @@ def test_manual_search_all_dispatches_parallel(manual_search_client): ) assert resp.status_code == 200 - body = resp.get_json() - # 5 configured sources × 1 result each - assert len(body['candidates']) == 5 - # Each configured source got searched once. + msgs = _consume_ndjson(resp) + candidates = _flatten_candidates(msgs) + assert len(candidates) == 5 for src_name in ('soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'): - assert plugins[src_name].search.call_count == 1, ( - f"{src_name} should have been searched in 'all' mode" - ) - # Tidal is unconfigured → not in available_sources → not searched. + assert plugins[src_name].search.call_count == 1 assert plugins['tidal'].search.call_count == 0 +def test_manual_search_streams_one_event_per_source(manual_search_client): + """source='all' must emit one ``source_results`` event per configured + source — not a single batched event. That's what lets the frontend + render rows as they arrive instead of waiting for the slowest source.""" + client, ctx = manual_search_client + plugins = ctx['plugins'] + + import web_server + for src_name in ('soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'): + new_plugin = _make_plugin( + search_results=[_fake_track_result(f'{src_name}_song.mp3', src_name)] + ) + plugins[src_name] = new_plugin + web_server.download_orchestrator.registry._plugins[src_name] = new_plugin + + resp = client.post( + '/api/downloads/task/task-abc/manual-search', + json={'query': 'drake feelings', 'source': 'all'}, + ) + + msgs = _consume_ndjson(resp) + source_events = [m for m in msgs if m.get('type') == 'source_results'] + seen_sources = {m['source'] for m in source_events} + assert seen_sources == {'soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'} + # One header + one per source + one done terminator + assert msgs[0]['type'] == 'header' + assert msgs[-1]['type'] == 'done' + assert msgs[-1]['total'] == 5 + + def test_manual_search_skips_unconfigured_sources(manual_search_client): """Sources whose is_configured() returns False are excluded from the - 'all' dispatch list. This is the same gate hybrid-mode fallback uses - for actual downloads — keeps manual search consistent with the rest - of the orchestrator.""" + 'all' dispatch list.""" client, ctx = manual_search_client plugins = ctx['plugins'] @@ -314,19 +360,18 @@ def test_manual_search_skips_unconfigured_sources(manual_search_client): ) assert resp.status_code == 200 - body = resp.get_json() - # Tidal is unconfigured — not in available_sources - available_ids = {s['id'] for s in body['available_sources']} + msgs = _consume_ndjson(resp) + header = msgs[0] + assert header['type'] == 'header' + available_ids = {s['id'] for s in header['available_sources']} assert 'tidal' not in available_ids assert {'soulseek', 'youtube', 'qobuz', 'hifi', 'deezer'} <= available_ids - # And tidal.search was NEVER called. assert plugins['tidal'].search.call_count == 0 def test_manual_search_rejects_unconfigured_source_explicitly(manual_search_client): """User can't bypass the 'all' filter by naming an unconfigured source - directly — endpoint validates `source` against the live configured- - sources list.""" + directly.""" client, _ctx = manual_search_client resp = client.post( @@ -334,8 +379,6 @@ def test_manual_search_rejects_unconfigured_source_explicitly(manual_search_clie json={'query': 'something', 'source': 'tidal'}, ) - # Tidal is in the registry but is_configured()=False, so it's not in - # available_sources, so the endpoint should reject the request. assert resp.status_code == 400 @@ -344,11 +387,11 @@ def test_manual_search_rejects_unconfigured_source_explicitly(manual_search_clie # --------------------------------------------------------------------------- -def test_manual_search_returns_same_shape_as_candidates(manual_search_client): - """Response includes track_info + candidates array; each candidate - carries a source field so the frontend can show per-row badges in - 'all' mode. Frontend renderer reuses the same row template for both - auto-candidates and manual results.""" +def test_manual_search_header_carries_track_and_source_metadata(manual_search_client): + """The first NDJSON line is a ``header`` event carrying track_info, + download_mode, available_sources, and the echoed query — everything + the frontend needs to render the modal shell before any results + arrive.""" client, ctx = manual_search_client plugins = ctx['plugins'] @@ -364,27 +407,20 @@ def test_manual_search_returns_same_shape_as_candidates(manual_search_client): ) assert resp.status_code == 200 - body = resp.get_json() + msgs = _consume_ndjson(resp) + header = msgs[0] + assert header['type'] == 'header' + assert header['task_id'] == 'task-abc' + assert header['track_info']['name'] == 'Test Track' + assert header['download_mode'] == 'hybrid' + assert isinstance(header['available_sources'], list) + assert header['query'] == 'drake feelings' + assert header['sources_queried'] == ['youtube'] - # Top-level shape mirrors /candidates - assert body['task_id'] == 'task-abc' - assert 'track_info' in body - assert body['track_info']['name'] == 'Test Track' - assert 'candidates' in body - assert 'candidate_count' in body - assert body['candidate_count'] == len(body['candidates']) - assert body['download_mode'] == 'hybrid' - assert isinstance(body['available_sources'], list) - # Echoed query lets frontend show "No results for X" with the same casing - # the user typed. - assert body['query'] == 'drake feelings' - - # Each candidate carries source for the frontend badge. - for candidate in body['candidates']: - assert 'source' in candidate + candidates = _flatten_candidates(msgs) + assert len(candidates) == 1 + for candidate in candidates: assert candidate['source'] == 'youtube' - # And the standard candidate fields are present (same shape as - # /candidates serialization). for field in ('username', 'filename', 'size', 'quality', 'duration', 'bitrate', 'queue_length', 'free_upload_slots'): @@ -392,12 +428,11 @@ def test_manual_search_returns_same_shape_as_candidates(manual_search_client): def test_manual_search_single_source_mode_only_offers_one_source(monkeypatch, manual_search_client): - """When download_source.mode is a single source (not hybrid), the - available_sources list should contain just that one source. Frontend - swaps the dropdown for a static label in this case.""" + """When download_source.mode is a single source, available_sources + contains just that one entry. Frontend swaps the dropdown for a static + label in this case.""" client, _ctx = manual_search_client - # Override config to single-source mode (soulseek only). from config.settings import config_manager monkeypatch.setattr( config_manager, 'get', @@ -413,22 +448,22 @@ def test_manual_search_single_source_mode_only_offers_one_source(monkeypatch, ma ) assert resp.status_code == 200 - body = resp.get_json() - assert body['download_mode'] == 'soulseek' - available_ids = [s['id'] for s in body['available_sources']] + msgs = _consume_ndjson(resp) + header = msgs[0] + assert header['download_mode'] == 'soulseek' + available_ids = [s['id'] for s in header['available_sources']] assert available_ids == ['soulseek'] def test_manual_search_handles_plugin_exception_gracefully(manual_search_client): - """If one source's .search() raises, the endpoint logs + skips it - instead of failing the whole 'all' request. Other sources' results - still come through.""" + """If one source's .search() raises, the endpoint emits a + ``source_error`` event for it but other sources' results still come + through. The whole stream doesn't fail.""" client, ctx = manual_search_client plugins = ctx['plugins'] import web_server - # Soulseek raises; youtube returns one result. flaky_plugin = MagicMock() flaky_plugin.is_configured = MagicMock(return_value=True) @@ -452,7 +487,12 @@ def test_manual_search_handles_plugin_exception_gracefully(manual_search_client) ) assert resp.status_code == 200 - body = resp.get_json() - # Failed plugin contributed 0; youtube's 1 result still comes through. - yt_results = [c for c in body['candidates'] if c.get('source') == 'youtube'] + msgs = _consume_ndjson(resp) + + error_events = [m for m in msgs if m.get('type') == 'source_error'] + assert any(m['source'] == 'soulseek' for m in error_events) + assert any('network blip' in m.get('error', '') for m in error_events) + + candidates = _flatten_candidates(msgs) + yt_results = [c for c in candidates if c.get('source') == 'youtube'] assert len(yt_results) == 1 diff --git a/tests/test_status_engine_fallback.py b/tests/test_status_engine_fallback.py new file mode 100644 index 00000000..2b2de3e8 --- /dev/null +++ b/tests/test_status_engine_fallback.py @@ -0,0 +1,469 @@ +"""Pin the engine-state fallback that drives non-Soulseek (streaming) +download status forward. + +Soulseek downloads land in slskd's ``live_transfers_lookup``, so their +status updates flow through the existing slskd-state branch. Streaming +sources (YouTube, Tidal, Qobuz, HiFi, Deezer, SoundCloud, Lidarr) never +appear there — without these tests' code path, a manually-picked +SoundCloud download stays at "downloading 0%" forever, even after the +engine logs an Errored terminal state. + +These tests exercise ``_apply_engine_state_fallback`` directly with a +fake ``download_orchestrator`` so we don't have to spin up the real +engine. The real fix relies on the per-source plugin storing the +terminal state via ``_mark_terminal`` (state='Errored' / 'Completed, +Succeeded'), which our fake mirrors. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional +from unittest.mock import MagicMock + +import pytest + +from core.downloads import status as status_mod + + +@dataclass +class _FakeDownloadStatus: + """Mirror the DownloadStatus shape that engine plugins return — only + the fields the fallback reads.""" + id: str + state: str + progress: float = 0 + error_message: Optional[str] = None + + +def _make_deps(record_for_id: dict, on_completed=None, submit_pp=None): + """StatusDeps with a fake orchestrator that returns whatever + DownloadStatus the test provides for a given download_id.""" + + fake_orch = MagicMock() + + async def _fake_get_status(download_id): + return record_for_id.get(download_id) + + fake_orch.get_download_status = _fake_get_status + + def _sync_run_async(coro): + import asyncio + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + return status_mod.StatusDeps( + config_manager=MagicMock(), + docker_resolve_path=lambda p: p, + find_completed_file=lambda *args, **kwargs: (None, None), + make_context_key=lambda u, f: f"{u}::{f}", + submit_post_processing=submit_pp or (lambda task_id, batch_id: None), + get_cached_transfer_data=lambda: {}, + download_orchestrator=fake_orch, + run_async=_sync_run_async, + on_download_completed=on_completed, + ) + + +def _task(*, status='downloading', username='soundcloud', filename='1234||https://sc/x', + download_id='dl-1', manual_pick=True): + """Default to manual_pick=True because the engine fallback is + deliberately scoped to manual picks — auto attempts go through the + live_transfers branch + monitor retry path. Tests opt out of the + flag explicitly when exercising the auto-attempt skip.""" + return { + 'status': status, + 'username': username, + 'filename': filename, + 'download_id': download_id, + '_user_manual_pick': manual_pick, + 'track_info': {'name': 'Test Track'}, + } + + +# --------------------------------------------------------------------------- +# Failure / cancel / success transitions +# --------------------------------------------------------------------------- + + +def test_errored_state_marks_task_failed(): + import time as _t + task = _task() + task_status = {'status': 'downloading', 'progress': 0} + completed_calls = [] + deps = _make_deps( + {'dl-1': _FakeDownloadStatus(id='dl-1', state='Errored', error_message='HTTP 404')}, + on_completed=lambda batch_id, task_id, success: completed_calls.append((batch_id, task_id, success)), + ) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'failed' + assert task['error_message'] == 'HTTP 404' + assert task_status['status'] == 'failed' + # on_download_completed is deferred to a daemon thread to avoid the + # tasks_lock self-deadlock — give it a beat to fire. + for _ in range(50): + if completed_calls: + break + _t.sleep(0.01) + assert completed_calls == [('b1', 't1', False)] + + +def test_compound_completed_errored_hits_failure_branch_first(): + """``"Completed, Errored"`` must be treated as failure, not success. + Order of state-substring checks matters.""" + task = _task() + task_status = {'status': 'downloading', 'progress': 0} + deps = _make_deps( + {'dl-1': _FakeDownloadStatus(id='dl-1', state='Completed, Errored')}, + on_completed=lambda *args: None, + ) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'failed' + + +def test_cancelled_state_marks_task_cancelled(): + import time as _t + task = _task() + task_status = {'status': 'downloading', 'progress': 0} + completed_calls = [] + deps = _make_deps( + {'dl-1': _FakeDownloadStatus(id='dl-1', state='Cancelled')}, + on_completed=lambda *args: completed_calls.append(args), + ) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'cancelled' + assert task_status['status'] == 'cancelled' + for _ in range(50): + if completed_calls: + break + _t.sleep(0.01) + assert len(completed_calls) == 1 + + +def test_succeeded_state_submits_post_processing(): + task = _task() + task_status = {'status': 'downloading', 'progress': 0} + pp_calls = [] + deps = _make_deps( + {'dl-1': _FakeDownloadStatus(id='dl-1', state='Completed, Succeeded', progress=100)}, + submit_pp=lambda task_id, batch_id: pp_calls.append((task_id, batch_id)), + ) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'post_processing' + assert task_status['status'] == 'post_processing' + assert pp_calls == [('t1', 'b1')] + + +def test_inprogress_reflects_progress_without_changing_status(): + task = _task() + task_status = {'status': 'downloading', 'progress': 0} + deps = _make_deps( + {'dl-1': _FakeDownloadStatus(id='dl-1', state='InProgress, Downloading', progress=42.5)}, + ) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'downloading' + assert task_status['status'] == 'downloading' + assert task_status['progress'] == 42.5 + + +# --------------------------------------------------------------------------- +# Gates — bail without mutating state +# --------------------------------------------------------------------------- + + +def test_skips_when_orchestrator_missing(): + task = _task() + task_status = {'status': 'downloading', 'progress': 0} + deps = status_mod.StatusDeps( + config_manager=MagicMock(), + docker_resolve_path=lambda p: p, + find_completed_file=lambda *a, **k: (None, None), + make_context_key=lambda u, f: f"{u}::{f}", + submit_post_processing=lambda *a: None, + get_cached_transfer_data=lambda: {}, + download_orchestrator=None, + ) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'downloading' # unchanged + + +def test_skips_terminal_states(): + """Already-failed / completed / cancelled tasks must not be touched — + they may have been marked by another path (e.g. the live_transfers + branch on a slskd Errored state for a Soulseek manual pick).""" + for terminal in ('completed', 'failed', 'cancelled', 'not_found', 'post_processing'): + task = _task(status=terminal) + task_status = {'status': terminal, 'progress': 0} + deps = _make_deps({'dl-1': _FakeDownloadStatus(id='dl-1', state='Errored')}) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == terminal + + +def test_skips_soulseek_username(): + """Soulseek goes through live_transfers_lookup — never the engine + fallback. Otherwise we'd double-process its terminal state.""" + task = _task(username='peer-username-xyz') # not in _STREAMING_SOURCE_NAMES + task_status = {'status': 'downloading', 'progress': 0} + deps = _make_deps({'dl-1': _FakeDownloadStatus(id='dl-1', state='Errored')}) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'downloading' + + +def test_skips_when_download_id_missing(): + task = _task() + task.pop('download_id') + task_status = {'status': 'downloading', 'progress': 0} + deps = _make_deps({}) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'downloading' + + +def test_engine_returning_none_leaves_task_alone(): + """Engine doesn't know about this download_id (worker hasn't registered + yet, or the record was cleaned). The fallback must not falsely mark + the task failed in this case — the safety valve covers stuck-forever.""" + task = _task() + task_status = {'status': 'downloading', 'progress': 0} + deps = _make_deps({'dl-1': None}) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'downloading' + + +def test_skips_auto_attempts_without_manual_pick_flag(): + """Auto attempts (no _user_manual_pick flag) must NOT hit the engine + fallback even if they end up in the else branch. The monitor's retry + path owns auto-attempt failure handling — short-circuiting it here + would skip the fallback-to-next-candidate behavior.""" + task = _task(manual_pick=False) + task_status = {'status': 'downloading', 'progress': 0} + deps = _make_deps({'dl-1': _FakeDownloadStatus(id='dl-1', state='Errored')}) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + # Untouched — auto retry path will handle it. + assert task['status'] == 'downloading' + + +# --------------------------------------------------------------------------- +# Live-transfers IF branch — manual-pick failure path +# --------------------------------------------------------------------------- +# +# Streaming-source records are pre-populated into ``live_transfers_lookup`` +# via ``download_orchestrator.engine.get_all_downloads(exclude=('soulseek',))``, +# so a manually-picked SoundCloud / YouTube / Tidal / etc. download whose +# engine record reports ``state='Errored'`` arrives via the IF branch +# (lookup_key IS in live_transfers_lookup), NOT the engine-fallback else +# branch. Without the manual-pick guard inside that elif, the live- +# transfers branch would defer to the monitor — which itself bails on +# manual picks — and the task would sit at "downloading 0%" forever. +# +# These tests exercise ``build_batch_status_data`` end-to-end so the +# guard is pinned by behavior rather than by the unit-level fallback +# tests above. + + +def _seed_runtime(batch_id, task_id, *, manual_pick: bool): + import time as _t + from core.runtime_state import download_batches, download_tasks, tasks_lock + + with tasks_lock: + download_tasks[task_id] = { + 'status': 'downloading', + 'username': 'soundcloud', + 'filename': '1234||https://sc/x||Display Name', + 'download_id': 'dl-1', + '_user_manual_pick': manual_pick, + 'track_info': {'name': 'Test Track'}, + 'track_index': 0, + # Recent so the safety-valve "stuck-too-long" branch doesn't fire. + 'status_change_time': _t.time(), + 'cached_candidates': [], + } + download_batches[batch_id] = { + 'phase': 'downloading', + 'queue': [task_id], + 'analysis_results': [], + 'active_count': 1, + 'max_concurrent': 3, + } + + +def _clear_runtime(): + from core.runtime_state import download_batches, download_tasks, tasks_lock + with tasks_lock: + download_tasks.clear() + download_batches.clear() + + +@pytest.fixture +def runtime(): + """Seeded download_batches + download_tasks; cleared after each test.""" + _clear_runtime() + yield + _clear_runtime() + + +def _build_batch_deps(completed_calls): + """StatusDeps with a long timeout so the safety valve doesn't fire + + a captured ``on_download_completed`` we can assert against.""" + fake_config = MagicMock() + fake_config.get = lambda key, default=None: 99999 if key == 'soulseek.download_timeout' else default + + return status_mod.StatusDeps( + config_manager=fake_config, + docker_resolve_path=lambda p: p, + find_completed_file=lambda *a, **k: (None, None), + make_context_key=lambda u, f: f"{u}::{f}", + submit_post_processing=lambda *a: None, + get_cached_transfer_data=lambda: {}, + download_orchestrator=None, # IF branch doesn't need engine + run_async=None, + on_download_completed=lambda batch_id, task_id, success: completed_calls.append( + (batch_id, task_id, success) + ), + ) + + +def test_if_branch_manual_pick_marks_failed_on_errored(runtime): + """Manual-pick task whose live_transfers entry reports Errored — + must transition to 'failed' synchronously, not defer to the monitor.""" + import time as _t + from core.runtime_state import download_batches + + batch_id = 'b1' + task_id = 't1' + _seed_runtime(batch_id, task_id, manual_pick=True) + + completed_calls = [] + deps = _build_batch_deps(completed_calls) + + live_transfers_lookup = { + 'soundcloud::1234||https://sc/x||Display Name': { + 'state': 'Errored', + 'percentComplete': 0, + 'errorMessage': 'HTTP 404 Not Found', + } + } + + response = status_mod.build_batch_status_data( + batch_id, download_batches[batch_id], live_transfers_lookup, deps, + ) + + task_status = response['tasks'][0] + assert task_status['status'] == 'failed' + assert 'HTTP 404' in (task_status.get('error_message') or '') + + from core.runtime_state import download_tasks + assert download_tasks[task_id]['status'] == 'failed' + + # on_download_completed is deferred to a daemon thread — wait briefly. + for _ in range(50): + if completed_calls: + break + _t.sleep(0.01) + assert completed_calls == [(batch_id, task_id, False)] + + +def test_if_branch_compound_completed_errored_hits_manual_pick_failure(runtime): + """``"Completed, Errored"`` must trigger the failure branch, not the + success branch. Slskd / engine pluginscan emit compound states when + a download technically completes but the file is corrupt / partial.""" + from core.runtime_state import download_batches + + batch_id = 'b1' + task_id = 't1' + _seed_runtime(batch_id, task_id, manual_pick=True) + + deps = _build_batch_deps([]) + live_transfers_lookup = { + 'soundcloud::1234||https://sc/x||Display Name': { + 'state': 'Completed, Errored', + 'percentComplete': 100, + } + } + + response = status_mod.build_batch_status_data( + batch_id, download_batches[batch_id], live_transfers_lookup, deps, + ) + + assert response['tasks'][0]['status'] == 'failed' + + +def test_if_branch_auto_attempt_defers_to_monitor(runtime): + """Auto attempts (no manual-pick flag) keep the original "let monitor + handle retry" behavior — task stays in its current pre-error status + so the monitor's retry path can detect the Errored live_info on its + next tick. This is the byte-identical pre-fix behavior; the guard is + additive.""" + from core.runtime_state import download_batches, download_tasks + + batch_id = 'b1' + task_id = 't1' + _seed_runtime(batch_id, task_id, manual_pick=False) + + deps = _build_batch_deps([]) + live_transfers_lookup = { + 'soundcloud::1234||https://sc/x||Display Name': { + 'state': 'Errored', + 'percentComplete': 0, + } + } + + status_mod.build_batch_status_data( + batch_id, download_batches[batch_id], live_transfers_lookup, deps, + ) + + # Auto retry path keeps the task in 'downloading' so the monitor can + # observe the Errored state on its own poll. NOT marked failed here. + assert download_tasks[task_id]['status'] == 'downloading' + + +def test_orchestrator_exception_swallowed(): + """If get_download_status raises, the fallback logs + bails — it must + not propagate and crash the whole status response.""" + task = _task() + task_status = {'status': 'downloading', 'progress': 0} + + fake_orch = MagicMock() + + async def _boom(_): + raise RuntimeError("network blip") + + fake_orch.get_download_status = _boom + deps = status_mod.StatusDeps( + config_manager=MagicMock(), + docker_resolve_path=lambda p: p, + find_completed_file=lambda *a, **k: (None, None), + make_context_key=lambda u, f: f"{u}::{f}", + submit_post_processing=lambda *a: None, + get_cached_transfer_data=lambda: {}, + download_orchestrator=fake_orch, + run_async=lambda coro: __import__('asyncio').new_event_loop().run_until_complete(coro), + ) + + status_mod._apply_engine_state_fallback('t1', task, task_status, 'b1', deps) + + assert task['status'] == 'downloading' diff --git a/web_server.py b/web_server.py index f327aa7f..6d6a00be 100644 --- a/web_server.py +++ b/web_server.py @@ -3015,35 +3015,6 @@ atexit.register(_atexit_silence_shutdown_logger_errors) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) -def _handle_failed_download(batch_id, task_id, task, task_status): - """Handle failed download by triggering retry logic like GUI""" - try: - with tasks_lock: - if task_id not in download_tasks: - return - - retry_count = task.get('retry_count', 0) - task['retry_count'] = retry_count + 1 - - if task['retry_count'] > 2: # Max 3 attempts total (matches GUI) - # All retries exhausted, mark as permanently failed - logger.error(f"Task {task_id} failed after 3 retry attempts") - task_status['status'] = 'failed' - task['status'] = 'failed' - return - - # Show retrying status while we process retry - task_status['status'] = 'pending' # Will show as pending until retry kicks in - logger.error(f"Triggering retry {task['retry_count']}/3 for failed task {task_id}") - - # Trigger retry with next candidate (matches GUI retry_parallel_download_with_fallback) - missing_download_executor.submit(download_monitor._retry_task_with_fallback, batch_id, task_id, task) - - except Exception as e: - logger.error(f"Error handling failed download {task_id}: {e}") - task_status['status'] = 'failed' - task['status'] = 'failed' - def _update_task_status(task_id, new_status): """Helper to update task status and timestamp for timeout tracking""" with tasks_lock: @@ -7911,6 +7882,21 @@ def download_selected_candidate(task_id): task.pop('download_id', None) task.pop('username', None) task.pop('filename', None) + # Mark this as a user-initiated manual pick. The auto-retry + # monitor (`_should_retry_task`) and the engine-state status + # fallback both check this flag and skip the "fall back to + # another candidate via fresh search" behavior. When the user + # explicitly chose THIS file, the mental model is "try this + # one and tell me if it failed", not "try this, then auto- + # pick something else if it fails". Stays set until the task + # reaches a terminal state. + task['_user_manual_pick'] = True + # Reset retry counters so previous auto-attempts don't + # immediately exhaust the manual pick. + task.pop('stuck_retry_count', None) + task.pop('error_retry_count', None) + task.pop('last_retry_time', None) + task.pop('last_error_retry_time', None) # Clear the selected candidate from used_sources so it won't be skipped used_sources = task.get('used_sources', set()) source_key = f"{username}_{filename}" @@ -7970,20 +7956,44 @@ def download_selected_candidate(task_id): popularity=0, ) - # Submit to thread pool — don't block the request + track_name = track_info.get('name', 'Unknown') + + # Run on a dedicated thread instead of `missing_download_executor` + # — that pool is shared with the batch's other in-flight tracks + # (3 workers total) and a saturated pool would queue the manual + # pick indefinitely, leaving the user stuck at "downloading 0%". + # Manual picks are user-initiated and infrequent; a fresh thread + # per pick is cheaper than starving them behind background work. def _run_manual_download(): - success = _attempt_download_with_candidates(task_id, [candidate], track, batch_id) - if not success: + logger.info(f"[Manual Download] worker started for task {task_id} ({username} / {track_name})") + try: + success = _attempt_download_with_candidates(task_id, [candidate], track, batch_id) + logger.info(f"[Manual Download] worker finished for task {task_id} success={success}") + if not success: + with tasks_lock: + if task_id in download_tasks: + download_tasks[task_id]['status'] = 'failed' + download_tasks[task_id]['error_message'] = 'Manual download failed to start — source may be unavailable' + if batch_id: + _on_download_completed(batch_id, task_id, success=False) + except Exception as exc: + logger.exception(f"[Manual Download] worker crashed for task {task_id}: {exc}") with tasks_lock: if task_id in download_tasks: download_tasks[task_id]['status'] = 'failed' - download_tasks[task_id]['error_message'] = 'Manual download failed to start — user may be offline' + download_tasks[task_id]['error_message'] = f'Manual download crashed: {exc}' if batch_id: - _on_download_completed(batch_id, task_id, success=False) + try: + _on_download_completed(batch_id, task_id, success=False) + except Exception: + logger.exception("[Manual Download] _on_download_completed cleanup also failed") - missing_download_executor.submit(_run_manual_download) + threading.Thread( + target=_run_manual_download, + name=f"manual-download-{task_id[:8]}", + daemon=True, + ).start() - track_name = track_info.get('name', 'Unknown') logger.info(f"[Manual Download] User selected candidate for '{track_name}' from {username}") return jsonify({"success": True, "message": f"Download initiated for '{track_name}'"}) @@ -7996,11 +8006,24 @@ def download_selected_candidate(task_id): @app.route('/api/downloads/task//manual-search', methods=['POST']) def manual_search_for_task(task_id): - """Run a user-driven search against one (or all) configured download sources - and return candidate results in the same shape as ``/candidates``. The - candidates modal lets the user pick one to retry the download — that retry - still goes through the existing ``/download-candidate`` endpoint, so all - AcoustID + post-download safety nets stay in the loop. + """Run a user-driven search against one (or all) configured download + sources and stream candidate results as NDJSON — one JSON object per + line, terminated by ``\\n``. Streaming lets the modal render results as + each source completes instead of blocking on the slowest source. + + The candidates modal lets the user pick a result; that retry still + goes through ``/download-candidate``, so all AcoustID + + post-download safety nets stay in the loop. + + Stream shape (one JSON object per line): + - ``{"type": "header", ...}`` — emitted first; carries ``track_info``, + ``download_mode``, ``available_sources``, ``query``, + ``sources_queried``. + - ``{"type": "source_results", "source": "", "candidates": [...]}`` + — one per source, emitted as that source's search completes. + - ``{"type": "source_error", "source": "", "error": ""}`` + — when a source's search raised. + - ``{"type": "done", "total": }`` — terminator. """ try: data = request.get_json(silent=True) or {} @@ -8011,16 +8034,12 @@ def manual_search_for_task(task_id): if len(query) < 2: return jsonify({"error": "Query must be at least 2 characters"}), 400 - # Validate task exists with tasks_lock: task = download_tasks.get(task_id) if not task: return jsonify({"error": "Task not found"}), 404 track_info = dict(task.get('track_info', {})) - # Validate source against the live configured-sources list (so users - # can't bypass hybrid_order and trigger a search against a source the - # user disabled / hasn't configured). download_mode, available_sources = _list_available_download_sources() valid_source_ids = {s['id'] for s in available_sources} @@ -8031,73 +8050,82 @@ def manual_search_for_task(task_id): }), 400 sources_to_query = [source] else: - # 'all' is only meaningful in hybrid mode, but we accept it in - # single-source mode too (degenerates to the one configured source). sources_to_query = list(valid_source_ids) - if not sources_to_query: - return jsonify({ - "task_id": task_id, - "track_info": { - "name": track_info.get('name', 'Unknown'), - "artist": _get_track_artist_name(track_info) if isinstance(track_info, dict) else 'Unknown', - }, - "candidates": [], - "candidate_count": 0, - "download_mode": download_mode, - "available_sources": available_sources, - }) + track_payload = { + "name": track_info.get('name', 'Unknown'), + "artist": _get_track_artist_name(track_info) if isinstance(track_info, dict) else 'Unknown', + } - # Dispatch parallel searches. Each plugin's `.search(query)` is a - # coroutine, so we wrap it via run_async and submit to a small thread - # pool — same pattern the rest of the app uses to fan out across - # sources without blocking the request thread. from concurrent.futures import ThreadPoolExecutor, as_completed def _search_one(src_name: str): client = download_orchestrator.client(src_name) if download_orchestrator else None if not client: - return src_name, [] + return src_name, [], None try: result = run_async(client.search(query)) if isinstance(result, tuple): tracks = result[0] if result else [] else: tracks = result or [] - return src_name, tracks + return src_name, tracks, None except Exception as exc: logger.warning(f"[Manual Search] {src_name} search failed for query '{query}': {exc}") - return src_name, [] + return src_name, [], str(exc) - merged_candidates = [] - max_workers = min(8, max(1, len(sources_to_query))) - with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix='manual-search') as executor: - futures = [executor.submit(_search_one, name) for name in sources_to_query] - for future in as_completed(futures): - src_name, tracks = future.result() - for t in tracks: - serialized = _serialize_candidate(t, source_override=src_name) - if serialized: - merged_candidates.append(serialized) + def _generate(): + yield json.dumps({ + "type": "header", + "task_id": task_id, + "track_info": track_payload, + "download_mode": download_mode, + "available_sources": available_sources, + "query": query, + "sources_queried": sources_to_query, + }) + "\n" - logger.info( - f"[Manual Search] task={task_id} query='{query}' source={source} " - f"sources_queried={sources_to_query} results={len(merged_candidates)}" + if not sources_to_query: + yield json.dumps({"type": "done", "total": 0}) + "\n" + return + + total = 0 + max_workers = min(8, max(1, len(sources_to_query))) + with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix='manual-search') as executor: + futures = [executor.submit(_search_one, name) for name in sources_to_query] + for future in as_completed(futures): + src_name, tracks, error = future.result() + if error is not None: + yield json.dumps({ + "type": "source_error", + "source": src_name, + "error": error, + }) + "\n" + continue + serialized = [] + for t in tracks: + s = _serialize_candidate(t, source_override=src_name) + if s: + serialized.append(s) + total += len(serialized) + yield json.dumps({ + "type": "source_results", + "source": src_name, + "candidates": serialized, + }) + "\n" + + logger.info( + f"[Manual Search] task={task_id} query='{query}' source={source} " + f"sources_queried={sources_to_query} results={total}" + ) + yield json.dumps({"type": "done", "total": total}) + "\n" + + return Response( + _generate(), + mimetype='application/x-ndjson', + headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}, ) - return jsonify({ - "task_id": task_id, - "track_info": { - "name": track_info.get('name', 'Unknown'), - "artist": _get_track_artist_name(track_info) if isinstance(track_info, dict) else 'Unknown', - }, - "candidates": merged_candidates, - "candidate_count": len(merged_candidates), - "download_mode": download_mode, - "available_sources": available_sources, - "query": query, - }) - except Exception as e: logger.error(f"[Manual Search] {e}") import traceback @@ -17697,6 +17725,9 @@ def _build_status_deps(): _run_post_processing_worker, task_id, batch_id ), get_cached_transfer_data=get_cached_transfer_data, + download_orchestrator=download_orchestrator, + run_async=run_async, + on_download_completed=_on_download_completed, ) diff --git a/webui/static/downloads.js b/webui/static/downloads.js index 91e2b879..f443ae44 100644 --- a/webui/static/downloads.js +++ b/webui/static/downloads.js @@ -3059,8 +3059,11 @@ function _wireManualSearch(overlay, taskId, trackName, isHybrid) { const sourceSelect = overlay.querySelector('#candidates-manual-source'); if (!input || !button || !resultsContainer) return; + // Aggregated results across all source streams for the current query. + // Cleared at the start of each new search. let currentResults = []; let inFlight = false; + let abortController = null; const updateButtonState = () => { const q = (input.value || '').trim(); @@ -3076,34 +3079,134 @@ function _wireManualSearch(overlay, taskId, trackName, isHybrid) { } }; + const _renderTableShell = (query) => { + resultsContainer.innerHTML = ` +
Searching...
+ `; + }; + + const _appendRows = (newCandidates, query) => { + if (!newCandidates || newCandidates.length === 0) return; + const startIdx = currentResults.length; + currentResults = currentResults.concat(newCandidates); + + const wrapper = resultsContainer.querySelector('#candidates-manual-table-wrapper'); + const tbody = resultsContainer.querySelector('#candidates-manual-tbody'); + const statusEl = resultsContainer.querySelector('#candidates-manual-search-status'); + if (!tbody || !wrapper) return; + + let rowsHtml = ''; + newCandidates.forEach((c, i) => { + rowsHtml += _renderCandidateRow(c, startIdx + i, 'candidates-row-manual', isHybrid); + }); + tbody.insertAdjacentHTML('beforeend', rowsHtml); + wrapper.style.display = ''; + if (statusEl) { + statusEl.textContent = `${currentResults.length} result${currentResults.length !== 1 ? 's' : ''} so far...`; + } + + // Wire newly-appended buttons + tbody.querySelectorAll('.candidates-download-btn').forEach(btn => { + if (btn._candidatesWired) return; + btn._candidatesWired = true; + btn.addEventListener('click', () => { + const idx = parseInt(btn.dataset.index); + const c = currentResults[idx]; + if (c) downloadCandidate(taskId, c, trackName); + }); + }); + }; + + const _setStatus = (text) => { + const statusEl = resultsContainer.querySelector('#candidates-manual-search-status'); + if (statusEl) statusEl.textContent = text; + }; + + const _setError = (msg) => { + resultsContainer.innerHTML = `
${escapeHtml(msg)}
`; + }; + const runSearch = async () => { const q = (input.value || '').trim(); if (q.length < 2 || inFlight) return; + + if (abortController) { + try { abortController.abort(); } catch (_) { } + } + abortController = new AbortController(); + const source = sourceSelect ? sourceSelect.value : 'all'; inFlight = true; button.disabled = true; const originalLabel = button.textContent; button.textContent = 'Searching...'; - resultsContainer.innerHTML = '
Searching...
'; + currentResults = []; + _renderTableShell(q); try { const resp = await fetch(`/api/downloads/task/${encodeURIComponent(taskId)}/manual-search`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ query: q, source: source }) + body: JSON.stringify({ query: q, source: source }), + signal: abortController.signal, }); - const payload = await resp.json().catch(() => ({})); if (!resp.ok) { - resultsContainer.innerHTML = `
${escapeHtml(payload.error || 'Search failed')}
`; - currentResults = []; + let errMsg = 'Search failed'; + try { + const payload = await resp.json(); + if (payload && payload.error) errMsg = payload.error; + } catch (_) { } + _setError(errMsg); return; } - currentResults = Array.isArray(payload.candidates) ? payload.candidates : []; - _renderManualSearchResults(resultsContainer, currentResults, q, isHybrid, taskId, trackName); + + const reader = resp.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + const errors = []; + + while (true) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + + let lineEnd; + while ((lineEnd = buffer.indexOf('\n')) >= 0) { + const line = buffer.slice(0, lineEnd).trim(); + buffer = buffer.slice(lineEnd + 1); + if (!line) continue; + let msg; + try { msg = JSON.parse(line); } catch (_) { continue; } + + if (msg.type === 'source_results') { + _appendRows(msg.candidates || [], q); + } else if (msg.type === 'source_error') { + errors.push(`${msg.source}: ${msg.error}`); + } else if (msg.type === 'done') { + if (currentResults.length === 0) { + const errorNote = errors.length + ? `
${errors.length} source${errors.length !== 1 ? 's' : ''} failed
` + : ''; + resultsContainer.innerHTML = ` +
No manual search results for "${escapeHtml(q)}"
+ ${errorNote}`; + } else { + _setStatus(`${currentResults.length} result${currentResults.length !== 1 ? 's' : ''}`); + } + } + } + } } catch (err) { + if (err.name === 'AbortError') return; console.error('Manual search failed:', err); - resultsContainer.innerHTML = '
Search request failed
'; - currentResults = []; + _setError('Search request failed'); } finally { inFlight = false; button.textContent = originalLabel; @@ -3123,38 +3226,6 @@ function _wireManualSearch(overlay, taskId, trackName, isHybrid) { updateButtonState(); } -function _renderManualSearchResults(container, results, query, isHybrid, taskId, trackName) { - if (!results || results.length === 0) { - container.innerHTML = `
No manual search results for "${escapeHtml(query)}"
`; - return; - } - let rows = ''; - results.forEach((c, i) => { - // Always show the source badge on manual-search rows when there's - // more than one possible source — covers hybrid "All" + per-source - // selections so the user always sees provenance. - rows += _renderCandidateRow(c, i, 'candidates-row-manual', isHybrid); - }); - container.innerHTML = ` -
${results.length} result${results.length !== 1 ? 's' : ''}
-
- - - - - ${rows} -
#FileQualitySizeDurationUser
-
`; - - container.querySelectorAll('.candidates-row-manual .candidates-download-btn').forEach(btn => { - btn.addEventListener('click', () => { - const idx = parseInt(btn.dataset.index); - const c = results[idx]; - if (c) downloadCandidate(taskId, c, trackName); - }); - }); -} - async function downloadCandidate(taskId, candidate, trackName) { if (!await showConfirmDialog({ title: 'Download File', message: `Download this file as "${trackName}"?\n\n${candidate.filename?.split(/[/\\]/).pop() || 'Unknown file'}\nfrom ${candidate.username || 'Unknown user'}`, confirmText: 'Download' })) return; try { @@ -3328,8 +3399,11 @@ function processModalStatusUpdate(playlistId, data) { statusEl.dataset.errorMsg = task.error_message; _ensureErrorTooltipListeners(statusEl); } - // Make not_found and failed cells clickable to review search candidates - if ((task.status === 'not_found' || task.status === 'failed') && task.has_candidates) { + // Make not_found / failed / cancelled cells clickable to open + // the candidates modal. Always bind — even when no auto-search + // candidates were cached — because the modal carries the manual + // search bar, which is the user's recourse for empty results. + if (task.status === 'not_found' || task.status === 'failed' || task.status === 'cancelled') { statusEl.classList.add('has-candidates'); statusEl.dataset.taskId = task.task_id; _ensureCandidatesClickListener(statusEl); diff --git a/webui/static/helper.js b/webui/static/helper.js index 2c4ea506..1c78c87e 100644 --- a/webui/static/helper.js +++ b/webui/static/helper.js @@ -3416,7 +3416,8 @@ const WHATS_NEW = { '2.4.4': [ // --- post-2.4.3 dev work — entries hidden by _getLatestWhatsNewVersion until the build version bumps --- { date: 'Unreleased — 2.4.4 dev cycle' }, - { title: 'Manual Search In The Failed-Track Candidates Modal', desc: 'when a download fails or returns "not found" the user can already click the status cell to open a modal showing whatever search candidates the auto-search left over and pick a different one. that modal now ALSO has a manual search bar. type any query, hit search, get a fresh round of results from the download sources without having to start the whole download flow over from the search page. solves the case where the auto-query was bad (featured artist not in title, parentheticals like "(remastered 2019)" tripping the matcher, slight artist-name variants) but the file genuinely exists on the source. source picker is smart per download mode: single-source mode (soulseek-only / youtube-only / etc) shows a "searching X" label, no dropdown; hybrid mode shows a dropdown with "all sources" default plus every configured source — picking "all" runs parallel searches across all of them and tags each result row with its source badge. only configured sources show up; unconfigured ones are hidden. clicking a result reuses the existing retry-download flow → same path, same acoustid verification on the file when it lands, no shortcut around the safety net. additive in the truest sense: the existing modal layout / candidates table / download buttons are byte-identical when the user doesn\'t use manual search. backend extends the candidates endpoint with `download_mode` + `available_sources` + a `source` field per candidate (purely additive — old fields untouched), and adds a new `POST /api/downloads/task//manual-search` that mirrors the candidates response shape so the frontend renderer is reused. 10 new tests pin endpoint validation (query length / source whitelist / task 404), single-source dispatch, parallel "all" dispatch, unconfigured-source skip + reject, response shape, and exception isolation per source.', page: 'downloads' }, + { title: 'Manual Search In The Failed-Track Candidates Modal', desc: 'when a download fails or returns "not found" the user can already click the status cell to open a modal showing whatever search candidates the auto-search left over and pick a different one. that modal now ALSO has a manual search bar. type any query, hit search, get a fresh round of results from the download sources without having to start the whole download flow over from the search page. solves the case where the auto-query was bad (featured artist not in title, parentheticals like "(remastered 2019)" tripping the matcher, slight artist-name variants) but the file genuinely exists on the source. source picker is smart per download mode: single-source mode (soulseek-only / youtube-only / etc) shows a "searching X" label, no dropdown; hybrid mode shows a dropdown with "all sources" default plus every configured source — picking "all" runs parallel searches across all of them and tags each result row with its source badge. only configured sources show up; unconfigured ones are hidden. results stream in as each source completes via NDJSON instead of blocking on the slowest source — the table starts populating the moment the first source returns. clicking a result reuses the existing retry-download flow → same path, same acoustid verification on the file when it lands, no shortcut around the safety net. additive in the truest sense: the existing modal layout / candidates table / download buttons are byte-identical when the user doesn\'t use manual search. backend extends the candidates endpoint with `download_mode` + `available_sources` + a `source` field per candidate (purely additive — old fields untouched), and adds a new `POST /api/downloads/task//manual-search` that streams NDJSON (one header line, one source_results line per source as completed, one done terminator) so the frontend renderer can append rows incrementally. 11 tests pin the streaming contract: query length / source whitelist / task 404 validation, single-source dispatch, parallel "all" dispatch, one-event-per-source streaming shape, unconfigured-source skip + reject, header metadata, and per-source exception isolation (one source raising emits a `source_error` event but doesn\'t fail the stream).', page: 'downloads' }, + { title: 'Manual Picks Don\'t Auto-Retry Anymore (And The Modal Always Opens)', desc: 'three follow-on fixes to the manual-search feature once people started actually using it. (1) when the user picked a candidate and that download failed (e.g. soundcloud 404 on a stale track url), the auto-retry monitor would treat it like any other failed auto-attempt — yank the task back to "searching" and pick a different candidate. felt completely wrong from the user\'s perspective: "i picked THIS one, why is it searching for something else?" now manual picks are tagged with a `_user_manual_pick` flag and the auto-retry path bails on it. failure surfaces to the user instead of getting silently fallen-back. (2) non-soulseek manual picks (youtube / tidal / qobuz / hifi / deezer / soundcloud / lidarr) were getting stuck at "downloading 0%" forever even after their engine reported terminal failure. cause: status polling went into a "let monitor handle retry" branch that never fired because manual picks bail on retry — task was orphaned in downloading state. fix: when the engine reports Errored on a manual pick, mark the task failed directly, don\'t defer to the monitor. plus an engine-state fallback path covers the rare race where the orchestrator\'s pre-populated transfer lookup is missing the entry. (3) failed / not_found rows were only clickable when the auto-search had cached candidates — but the whole point of opening the modal now is to RUN a manual search, which doesn\'t need pre-existing candidates. now every failed / not_found / cancelled row opens the modal regardless. (4) one nasty deadlock fix in the process: the new "mark failed" path was synchronously calling `on_download_completed` while holding `tasks_lock`, which itself re-acquires the same lock — `threading.Lock` is non-reentrant so the polling thread wedged forever. while wedged the lock was held → every other endpoint that needed it (including /candidates → can\'t open OTHER modals) hung waiting. moved completion callbacks onto a daemon thread so the lock releases first. (5) manual download worker now runs on its own dedicated thread instead of competing with the batch\'s 3-worker `missing_download_executor` pool — saturated batches no longer queue manual picks indefinitely. all changes are scoped to manual picks only via the `_user_manual_pick` flag — auto-attempt flow is byte-identical to before. 17 unit tests pin the gate behavior (status engine fallback / monitor retry skip / IF-branch failure transition / auto-attempt skip).', page: 'downloads' }, ], '2.4.3': [ // --- May 8, 2026 — patch release --- diff --git a/webui/static/style.css b/webui/static/style.css index 631046a9..34314341 100644 --- a/webui/static/style.css +++ b/webui/static/style.css @@ -40951,6 +40951,19 @@ div.artist-hero-badge { font-size: 12px; margin-bottom: 8px; } +.candidates-manual-search-status { + color: rgba(255, 255, 255, 0.6); + font-size: 12px; + margin-bottom: 8px; + font-style: italic; +} +.candidates-manual-search-empty-note { + color: rgba(255, 180, 100, 0.7); + font-size: 11px; + margin-top: 6px; + text-align: center; + font-style: italic; +} .candidates-source-badge { display: inline-block; background: rgba(255, 255, 255, 0.08);