diff --git a/core/download_engine/engine.py b/core/download_engine/engine.py index e0bdcced..e8fb9d12 100644 --- a/core/download_engine/engine.py +++ b/core/download_engine/engine.py @@ -25,6 +25,7 @@ big-bang switchover. from __future__ import annotations +import asyncio import threading from typing import Any, Dict, Iterator, List, Optional, Tuple @@ -477,29 +478,55 @@ class DownloadEngine: excluded = {s.lower() for s in (exclude_sources or []) if s} pooled_tracks = [] pooled_albums = [] + # Per-source contribution for an honest pool log — e.g. a release-level + # source like usenet/torrent that returns nothing for a track-title + # query should read "usenet=0", not silently hide behind the chain name. + contributions = [] + # Decide which sources to actually query, recording why the rest were + # skipped. Searches then run CONCURRENTLY so the pool waits only for the + # slowest source (e.g. usenet/Prowlarr, which can be slow) rather than + # the sum of every source's latency. + to_search = [] # (source_name, plugin) for source_name in source_chain: if source_name.lower() in excluded: + contributions.append(f"{source_name}=excluded") continue plugin = self._plugins.get(source_name) if plugin is None: logger.info(f"Skipping {source_name} (not available)") + contributions.append(f"{source_name}=unavailable") continue if hasattr(plugin, 'is_configured') and not plugin.is_configured(): logger.info(f"Skipping {source_name} (not configured)") + contributions.append(f"{source_name}=unconfigured") continue - try: - tracks, albums = await plugin.search(query, timeout, progress_callback) - if tracks: - pooled_tracks.extend(tracks) - if albums: - pooled_albums.extend(albums) - except Exception as e: - logger.warning(f"{source_name} search failed: {e}") + to_search.append((source_name, plugin)) + + async def _one(plugin): + return await plugin.search(query, timeout, progress_callback) + + results = await asyncio.gather( + *[_one(plugin) for _, plugin in to_search], + return_exceptions=True, + ) + + for (source_name, _), result in zip(to_search, results): + if isinstance(result, Exception): + logger.warning(f"{source_name} search failed: {result}") + contributions.append(f"{source_name}=error") + continue + tracks, albums = result + n = len(tracks) if tracks else 0 + if tracks: + pooled_tracks.extend(tracks) + if albums: + pooled_albums.extend(albums) + contributions.append(f"{source_name}={n}") logger.info( - "Best-quality pool: %d candidates across %s for: %s", - len(pooled_tracks), ', '.join(source_chain), query, + "Best-quality pool: %d candidates [%s] for: %s", + len(pooled_tracks), ', '.join(contributions), query, ) return (pooled_tracks, pooled_albums) diff --git a/core/downloads/cancel.py b/core/downloads/cancel.py index b375f8fa..d8d0b820 100644 --- a/core/downloads/cancel.py +++ b/core/downloads/cancel.py @@ -83,9 +83,26 @@ def clear_completed_local() -> int: """ cleared = 0 with tasks_lock: + # Protect tasks belonging to a still-active batch. A batch is "active" + # while any of its queued tasks is non-terminal (still searching / + # downloading / queued / post-processing). Pruning a batch's completed + # or failed tasks mid-run would yank them out of the Downloads page — + # and failed/cancelled rows aren't recoverable from library_history — + # so the user would never see them until the batch ended. Keep the whole + # active batch intact; it gets cleaned by a later run once it finishes. + protected_task_ids: set = set() + for batch in download_batches.values(): + queue = batch.get('queue', []) if isinstance(batch, dict) else [] + batch_active = any( + download_tasks.get(tid, {}).get('status') not in _TERMINAL_STATUSES + for tid in queue if tid in download_tasks + ) + if batch_active: + protected_task_ids.update(queue) + task_ids_to_remove = [ tid for tid, task in download_tasks.items() - if task.get('status') in _TERMINAL_STATUSES + if task.get('status') in _TERMINAL_STATUSES and tid not in protected_task_ids ] for tid in task_ids_to_remove: del download_tasks[tid] diff --git a/tests/downloads/test_downloads_cancel.py b/tests/downloads/test_downloads_cancel.py index 7d90f0ee..0dfac0c5 100644 --- a/tests/downloads/test_downloads_cancel.py +++ b/tests/downloads/test_downloads_cancel.py @@ -176,15 +176,30 @@ def test_clear_completed_drops_empty_batches(): assert download_batches['b3']['queue'] == ['t3'] -def test_clear_completed_prunes_terminal_task_ids_from_batch_queues(): - """Batch with mix of terminal + active tasks gets queue trimmed, not deleted.""" +def test_clear_completed_keeps_terminal_tasks_in_active_batch(): + """A completed/failed task in a batch that still has active work is KEPT, so + the Downloads page shows it for the whole run (the 5-min Clean Completed + automation must not yank terminal tasks out from under a live batch).""" download_tasks['t1'] = {'status': 'completed'} - download_tasks['t2'] = {'status': 'downloading'} + download_tasks['t2'] = {'status': 'downloading'} # batch still active download_batches['b1'] = {'queue': ['t1', 't2']} - cancel.clear_completed_local() + cleared = cancel.clear_completed_local() + assert cleared == 0 # nothing removed — batch is live assert 'b1' in download_batches - assert download_batches['b1']['queue'] == ['t2'] + assert download_batches['b1']['queue'] == ['t1', 't2'] # queue intact + + +def test_clear_completed_clears_terminal_tasks_once_batch_is_finished(): + """When every task in a batch is terminal, the batch is done — its tasks are + cleared and the empty batch dropped.""" + download_tasks['t1'] = {'status': 'completed'} + download_tasks['t2'] = {'status': 'failed'} + download_batches['b1'] = {'queue': ['t1', 't2']} + + cleared = cancel.clear_completed_local() + assert cleared == 2 + assert 'b1' not in download_batches def test_clear_completed_drops_batch_locks_for_deleted_batches(): diff --git a/tests/quality/test_search_all_sources.py b/tests/quality/test_search_all_sources.py index 59fd0f8a..cf1f24c4 100644 --- a/tests/quality/test_search_all_sources.py +++ b/tests/quality/test_search_all_sources.py @@ -72,6 +72,37 @@ def test_skips_unconfigured_and_swallows_raising_source(): assert {t.name for t in tracks} == {'c1'} # a skipped, b raised, c survives +def test_searches_run_concurrently_and_wait_for_slow_source(): + import time + + class _SlowPlugin: + def __init__(self, name, delay): + self._name = name + self._delay = delay + self.searched = False + + def is_configured(self): + return True + + async def search(self, query, timeout=None, progress_callback=None): + self.searched = True + await asyncio.sleep(self._delay) + return ([_Cand(self._name)], []) + + slow = _SlowPlugin('usenet', 0.20) # a slow release-level source + fast = _SlowPlugin('hifi', 0.05) + eng = _engine_with({'usenet': slow, 'hifi': fast}) + + start = time.monotonic() + tracks, _ = asyncio.run(eng.search_all_sources('q', ['usenet', 'hifi'])) + elapsed = time.monotonic() - start + + # Both included — the pool waits for the slow source. + assert {t.name for t in tracks} == {'usenet', 'hifi'} + # Concurrent: total ≈ max(0.20, 0.05), well under the sequential sum 0.25. + assert elapsed < 0.20 + 0.04 + + def test_empty_when_all_sources_empty(): a = _FakePlugin([]) eng = _engine_with({'a': a})