diff --git a/core/downloads/monitor.py b/core/downloads/monitor.py index dcdfcd77..273d1d45 100644 --- a/core/downloads/monitor.py +++ b/core/downloads/monitor.py @@ -76,6 +76,25 @@ def _resolve_download_source(username): return 'soulseek' +def _remaining_fallback_sources(exhausted): + """Sources in the configured hybrid chain that haven't exhausted their + per-source budget yet. + + When a source spends its whole budget (exhaustive mode), the task switches + to the next source instead of failing — but only if there *is* another + source. Single-source mode has nothing to fall back to, so this returns + empty there (and when the orchestrator isn't wired). The returned list + drives both the give-up decision here and the worker's search-exclusion on + the next attempt (see task_worker: exhausted_download_sources). + """ + orch = download_orchestrator + if orch is None or getattr(orch, 'mode', None) != 'hybrid': + return [] + chain = getattr(orch, 'hybrid_order', None) or [] + blocked = {str(s).lower() for s in exhausted} + return [s for s in chain if str(s).lower() not in blocked] + + def _download_id_key(download_id): return f"download_id::{download_id}" if download_id else None @@ -160,23 +179,47 @@ def requeue_quarantined_task_for_retry(task_id, batch_id, trigger): source_count = counts.get(source, 0) if source_count >= budget: - logger.warning( - f"[Retry:{trigger}] Task {task_id} exhausted its retry budget " - f"for source '{source}' ({source_count}/{budget}) — giving up, " - f"marking failed" + # This source spent its whole budget. Rather than fail the + # track outright, mark the source exhausted and fall through to + # the next source in the hybrid chain (the worker excludes + # exhausted sources from its next search). Only give up once no + # fallback source remains — or the absolute ceiling trips. + exhausted = set(task.get('exhausted_download_sources') or ()) + exhausted.add(source) + remaining = _remaining_fallback_sources(exhausted) + if not remaining: + logger.warning( + f"[Retry:{trigger}] Task {task_id} exhausted its retry " + f"budget for source '{source}' ({source_count}/{budget}) " + f"and no fallback source remains — giving up, marking failed" + ) + return False + if total_count >= MAX_TOTAL_QUARANTINE_RETRIES: + logger.warning( + f"[Retry:{trigger}] Task {task_id} hit the absolute retry " + f"ceiling ({MAX_TOTAL_QUARANTINE_RETRIES}) — giving up, " + f"marking failed" + ) + return False + task['exhausted_download_sources'] = exhausted + # Don't push this source's counter past its budget — it's done. + # The next source starts spending its own fresh budget when its + # first candidate fails verification. + attempt_desc = ( + f"source '{source}' budget spent ({source_count}/{budget}) " + f"— switching sources (remaining: {', '.join(remaining)})" ) - return False - if total_count >= MAX_TOTAL_QUARANTINE_RETRIES: - logger.warning( - f"[Retry:{trigger}] Task {task_id} hit the absolute retry " - f"ceiling ({MAX_TOTAL_QUARANTINE_RETRIES}) — giving up, marking " - f"failed" - ) - return False - - counts[source] = source_count + 1 - task['quarantine_retry_counts_by_source'] = counts - attempt_desc = f"source '{source}' {source_count + 1}/{budget}" + else: + if total_count >= MAX_TOTAL_QUARANTINE_RETRIES: + logger.warning( + f"[Retry:{trigger}] Task {task_id} hit the absolute retry " + f"ceiling ({MAX_TOTAL_QUARANTINE_RETRIES}) — giving up, " + f"marking failed" + ) + return False + counts[source] = source_count + 1 + task['quarantine_retry_counts_by_source'] = counts + attempt_desc = f"source '{source}' {source_count + 1}/{budget}" else: # Default mode: a single global cap, conservative and predictable. if total_count >= MAX_QUARANTINE_RETRIES: diff --git a/core/downloads/task_worker.py b/core/downloads/task_worker.py index cf349381..092d589f 100644 --- a/core/downloads/task_worker.py +++ b/core/downloads/task_worker.py @@ -288,6 +288,16 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke # 2. Sequential Query Search (matches GUI's start_search_worker_parallel logic) search_diagnostics = [] # Track what happened per query for detailed error messages all_raw_results = [] # Collect raw results across queries for candidate review modal + # Sources whose per-source quarantine-retry budget is spent (exhaustive + # mode). The monitor sets this when a source gives up; we exclude those + # sources from the hybrid search so the chain falls through to the next + # source instead of re-fetching the same exhausted one (e.g. Soulseek + # keeps returning fresh wrong peers — once its budget is gone, switch to + # HiFi/Tidal/…). See monitor.requeue_quarantined_task_for_retry. + with tasks_lock: + _exhausted_sources = [ + str(s) for s in (download_tasks.get(task_id, {}).get('exhausted_download_sources') or ()) + ] for query_index, query in enumerate(search_queries): # Cancellation check before each query with tasks_lock: @@ -324,9 +334,13 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke _exclude_for_hybrid_album = ['torrent', 'usenet'] except Exception as _exc_filter_err: logger.debug("[Modal Worker] album-source-exclusion check failed: %s", _exc_filter_err) + # Fold in budget-exhausted sources (per-source quarantine retry). + _exclude_sources = list(_exhausted_sources) + if _exclude_for_hybrid_album: + _exclude_sources.extend(_exclude_for_hybrid_album) # Perform search with timeout tracks_result, _ = deps.run_async(deps.download_orchestrator.search( - query, timeout=30, exclude_sources=_exclude_for_hybrid_album, + query, timeout=30, exclude_sources=_exclude_sources or None, )) logger.debug(f"Search completed for task {task_id}, got {len(tracks_result) if tracks_result else 0} results") @@ -419,7 +433,12 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke # (which was definitely tried). If the first was skipped (unconfigured), # the orchestrator would have tried the second — but trying it again is # harmless (streaming sources return fast). - remaining_sources = [s for s in hybrid_order[1:] if s in source_clients and source_clients[s]] + _exhausted_lower = {s.lower() for s in _exhausted_sources} + remaining_sources = [ + s for s in hybrid_order[1:] + if s in source_clients and source_clients[s] + and s.lower() not in _exhausted_lower + ] if remaining_sources: logger.warning(f"[Hybrid Fallback] Primary source had no valid matches. Trying fallback sources: {remaining_sources}") diff --git a/tests/imports/test_import_pipeline.py b/tests/imports/test_import_pipeline.py index e6b6891a..a3651254 100644 --- a/tests/imports/test_import_pipeline.py +++ b/tests/imports/test_import_pipeline.py @@ -567,3 +567,79 @@ def test_exhaustive_absolute_ceiling_guards_runaway(monkeypatch): assert task["status"] == "failed" assert submitted == [] + + +def _wire_orchestrator(monkeypatch, mode, hybrid_order): + """Wire monitor's download_orchestrator so per-source budget exhaustion can + decide whether another source remains to fall back to.""" + import core.downloads.monitor as monitor + orch = types.SimpleNamespace(mode=mode, hybrid_order=list(hybrid_order)) + monkeypatch.setattr(monitor, "download_orchestrator", orch) + return orch + + +def test_exhaustive_exhausted_source_switches_in_hybrid(monkeypatch): + submitted = _wire_retry_engine(monkeypatch) + _wire_orchestrator(monkeypatch, "hybrid", ["soulseek", "hifi"]) + _patch_config(monkeypatch, { + "post_processing.retry_exhaustive": True, + "post_processing.retries_per_query": 5, + }) + + # soulseek's budget (query_count 2 * 5 = 10) is spent. In hybrid mode the + # task switches to the next source instead of failing the whole track. + task, completion, _ = _run_wrapper_with_quarantine( + monkeypatch, _acoustid_quarantine, + task_extra={"username": "DjPeer", "query_count": 2, + "quarantine_retry_counts_by_source": {"soulseek": 10}}, + ) + + assert task["status"] == "searching" + # The spent source is flagged so the worker excludes it from the next search. + assert task["exhausted_download_sources"] == {"soulseek"} + # Its per-source counter is NOT pushed past budget — the source is simply done. + assert task["quarantine_retry_counts_by_source"]["soulseek"] == 10 + assert submitted == [("rtask", "rbatch")] + assert completion == [] + + +def test_exhaustive_all_sources_exhausted_fails_in_hybrid(monkeypatch): + submitted = _wire_retry_engine(monkeypatch) + _wire_orchestrator(monkeypatch, "hybrid", ["soulseek", "hifi"]) + _patch_config(monkeypatch, { + "post_processing.retry_exhaustive": True, + "post_processing.retries_per_query": 5, + }) + + # soulseek was exhausted on an earlier attempt; now hifi spends its last + # budget too — no fallback source remains, so the task finally fails. + task, completion, _ = _run_wrapper_with_quarantine( + monkeypatch, _acoustid_quarantine, + task_extra={"username": "hifi", "query_count": 2, + "exhausted_download_sources": {"soulseek"}, + "quarantine_retry_counts_by_source": {"hifi": 10}}, + ) + + assert task["status"] == "failed" + assert submitted == [] + assert completion == [("rbatch", "rtask", False)] + + +def test_exhaustive_single_source_exhausted_fails(monkeypatch): + submitted = _wire_retry_engine(monkeypatch) + # Single-source mode: nothing to fall back to once the budget is spent. + _wire_orchestrator(monkeypatch, "soulseek", []) + _patch_config(monkeypatch, { + "post_processing.retry_exhaustive": True, + "post_processing.retries_per_query": 5, + }) + + task, completion, _ = _run_wrapper_with_quarantine( + monkeypatch, _acoustid_quarantine, + task_extra={"username": "DjPeer", "query_count": 2, + "quarantine_retry_counts_by_source": {"soulseek": 10}}, + ) + + assert task["status"] == "failed" + assert submitted == [] + assert completion == [("rbatch", "rtask", False)] diff --git a/webui/index.html b/webui/index.html index cfed5f5f..9570cbda 100644 --- a/webui/index.html +++ b/webui/index.html @@ -5579,7 +5579,7 @@
- + In exhaustive mode, how many candidates to try per search query per source. The per-source budget is number of search queries × this value. Only used when Exhaustive retry is on.