Manual picks: stream results, don't auto-retry, fix stuck-at-0%
Three follow-on fixes to the manual-search candidates modal once people started actually using it: 1. NDJSON streaming. Manual search waited for every source to return before showing anything. Now streams one event per source as each completes — header line, source_results per source, done terminator. Frontend appends rows incrementally via response.body.getReader(). 2. Manual picks no longer auto-retry on failure. New _user_manual_pick flag set on the task in /download-candidate. Both monitor retry paths (not-in-live-transfers stuck + Errored state) bail on the flag. Surfaces the failure to the user instead of silently picking a different candidate via fresh search. 3. Non-Soulseek manual picks (youtube/tidal/qobuz/hifi/deezer/ soundcloud/lidarr) no longer stuck at "downloading 0%" forever. The live_transfers IF branch now marks manual-pick tasks failed directly when the engine reports Errored, instead of deferring to the monitor (which bails on manual picks). Engine fallback in else branch covers the rare race where the orchestrator's pre-populated transfer lookup is missing the entry. Plus a deadlock fix discovered along the way: the new failure path synchronously called on_download_completed while holding tasks_lock, which itself re-acquires the same Lock — non-reentrant threading.Lock self-deadlocked the polling thread. While wedged, every other endpoint that needed the lock (including /candidates → other failed rows couldn't open modals) hung waiting. Moved completion callbacks onto a daemon thread so the lock releases first. Plus failed/not_found/cancelled rows are now ALWAYS clickable (not just when the auto-search cached candidates) — the modal carries the manual search bar, which is the user's recourse for empty results. Plus manual download worker now runs on a dedicated thread instead of competing with the batch's 3-worker missing_download_executor pool — saturated batches no longer queue manual picks indefinitely. All scoped to manual picks via the _user_manual_pick flag — auto attempt flow byte-identical to before. Engine fallback gated on the flag too so auto attempts in the else branch keep the original do-nothing behavior (safety valve handles the stuck-forever case). Also dropped _handle_failed_download from web_server.py — defined but had no callers (dead code). 17 new unit tests pin the gate behavior: - engine fallback: Errored/Cancelled/Succeeded/InProgress transitions, manual-pick gate, terminal-state skip, soulseek skip, missing download_id skip, engine returning None, orchestrator exception - monitor: manual-pick skips not-in-live-transfers retry + Errored retry - IF-branch end-to-end: Errored marks failed, "Completed, Errored" hits failure branch, auto attempts defer to monitor Manual-search endpoint tests rewritten for NDJSON: 11 cases (validation, single-source dispatch, parallel "all" dispatch, one-event-per-source streaming shape, unconfigured-source skip + reject, header metadata, per-source exception isolation). Full suite 2259 passed, 1 skipped.
This commit is contained in:
Broque Thomas
parent
996575fab3
commit
e20994e1c7
9 changed files with 1176 additions and 220 deletions
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
95
tests/test_manual_pick_no_auto_retry.py
Normal file
95
tests/test_manual_pick_no_auto_retry.py
Normal file
|
|
@ -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/<id>/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'
|
||||
|
||||
|
||||
|
|
@ -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
|
||||
|
|
|
|||
469
tests/test_status_engine_fallback.py
Normal file
469
tests/test_status_engine_fallback.py
Normal file
|
|
@ -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'
|
||||
215
web_server.py
215
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/<task_id>/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": "<name>", "candidates": [...]}``
|
||||
— one per source, emitted as that source's search completes.
|
||||
- ``{"type": "source_error", "source": "<name>", "error": "<msg>"}``
|
||||
— when a source's search raised.
|
||||
- ``{"type": "done", "total": <int>}`` — 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,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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 = `
|
||||
<div class="candidates-manual-search-status" id="candidates-manual-search-status">Searching...</div>
|
||||
<div class="candidates-table-wrapper" style="display: none;" id="candidates-manual-table-wrapper">
|
||||
<table class="candidates-table">
|
||||
<thead><tr>
|
||||
<th>#</th><th>File</th><th>Quality</th><th>Size</th><th>Duration</th><th>User</th><th></th>
|
||||
</tr></thead>
|
||||
<tbody id="candidates-manual-tbody"></tbody>
|
||||
</table>
|
||||
</div>`;
|
||||
};
|
||||
|
||||
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 = `<div class="candidates-manual-search-error">${escapeHtml(msg)}</div>`;
|
||||
};
|
||||
|
||||
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 = '<div class="candidates-manual-search-loading">Searching...</div>';
|
||||
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 = `<div class="candidates-manual-search-error">${escapeHtml(payload.error || 'Search failed')}</div>`;
|
||||
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
|
||||
? `<div class="candidates-manual-search-empty-note">${errors.length} source${errors.length !== 1 ? 's' : ''} failed</div>`
|
||||
: '';
|
||||
resultsContainer.innerHTML = `
|
||||
<div class="candidates-manual-search-empty">No manual search results for "${escapeHtml(q)}"</div>
|
||||
${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 = '<div class="candidates-manual-search-error">Search request failed</div>';
|
||||
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 = `<div class="candidates-manual-search-empty">No manual search results for "${escapeHtml(query)}"</div>`;
|
||||
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 = `
|
||||
<div class="candidates-manual-search-count">${results.length} result${results.length !== 1 ? 's' : ''}</div>
|
||||
<div class="candidates-table-wrapper">
|
||||
<table class="candidates-table">
|
||||
<thead><tr>
|
||||
<th>#</th><th>File</th><th>Quality</th><th>Size</th><th>Duration</th><th>User</th><th></th>
|
||||
</tr></thead>
|
||||
<tbody>${rows}</tbody>
|
||||
</table>
|
||||
</div>`;
|
||||
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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/<id>/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/<id>/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 ---
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
Loading…
Reference in a new issue