From e83cf199039341ea6f6782b3ff69b99bffda97c9 Mon Sep 17 00:00:00 2001 From: dev Date: Fri, 5 Jun 2026 02:24:00 +0200 Subject: [PATCH] Downloads: retry next-best candidate on AcoustID/integrity quarantine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a downloaded file is quarantined because AcoustID verification or the integrity/duration check fails, the task no longer dead-ends as failed — it re-runs the worker on the next-best candidate, skipping the quarantined source. Reuses the monitor's existing transfer-error retry machinery (used_sources + cached_candidates + worker re-dispatch), just triggered from the post-process verification wrapper's two quarantine branches instead of only on transfer errors. Universal across sources (Soulseek, HiFi, Tidal, etc.) since all batch/sync downloads funnel through post_process_matched_download_with_verification. - monitor.requeue_quarantined_task_for_retry(): marks bad source used, resets task to searching, resubmits worker. Guards: manual picks, cancelled tasks, missing source id, and a MAX_QUARANTINE_RETRIES=5 loop cap. - Opt-out via post_processing.retry_next_candidate_on_mismatch (default on). - Manual quarantine approve is unaffected (_skip_quarantine_check='all' bypasses the checks, so no quarantine flag, so no retry). Co-Authored-By: Claude Opus 4.8 --- core/downloads/monitor.py | 89 +++++++++++++++++ core/imports/pipeline.py | 50 ++++++++-- tests/imports/test_import_pipeline.py | 132 ++++++++++++++++++++++++++ 3 files changed, 265 insertions(+), 6 deletions(-) diff --git a/core/downloads/monitor.py b/core/downloads/monitor.py index 4bcb0bc3..c99028a3 100644 --- a/core/downloads/monitor.py +++ b/core/downloads/monitor.py @@ -37,11 +37,100 @@ missing_download_executor = None download_orchestrator = None _RELEASE_SOURCE_NAMES = frozenset(('torrent', 'usenet')) +# Hard ceiling on automatic next-candidate retries after a download was +# quarantined (AcoustID mismatch / integrity / duration). The natural +# terminator is used_sources exhaustion — once every candidate the worker can +# find has been tried, attempt_download_with_candidates returns False and the +# worker reports a clean failure. This cap is a safety net against a pathological +# quarantine→retry→quarantine loop (e.g. a source that keeps returning fresh +# wrong files). +MAX_QUARANTINE_RETRIES = 5 + def _download_id_key(download_id): return f"download_id::{download_id}" if download_id else None +def requeue_quarantined_task_for_retry(task_id, batch_id, trigger): + """Re-queue a task whose download was just quarantined so the worker tries + the NEXT best candidate instead of failing outright. + + Called from the post-processing verification wrapper when AcoustID + verification or the integrity/duration check quarantines a file. It mirrors + the monitor's transfer-error retry path: mark the bad source as used, clear + the stale download identity, reset the task to ``searching`` and resubmit + the download worker. Because ``used_sources`` is preserved across the + re-run, the worker skips the quarantined source and picks the next-best + candidate (see ``attempt_download_with_candidates``). + + Returns True if a retry was queued — the caller must then NOT mark the task + failed or notify batch completion, since the task is going around again. + Returns False when no retry is possible (retry engine unwired, manual pick, + cancelled, or retry budget exhausted); the caller falls through to its + existing failure handling. + """ + # Opt-out escape hatch — default on. Lets users restore the old + # quarantine-and-fail behaviour without a code change. + if not config_manager.get('post_processing.retry_next_candidate_on_mismatch', True): + return False + + # Retry engine not wired (e.g. manual-import path that never started a + # download worker). Nothing to re-run. + if missing_download_executor is None or _download_track_worker is None: + return False + + with tasks_lock: + task = download_tasks.get(task_id) + if not task: + return False + # The user explicitly picked this candidate via the candidates modal — + # honour their choice rather than silently swapping in another file. + # (Matches the monitor's transfer-retry guards.) + if task.get('_user_manual_pick'): + return False + if task.get('status') == 'cancelled': + return False + + username = task.get('username') + filename = task.get('filename') + # No source identity means this wasn't a worker-dispatched download we + # can retry — without the "{username}_{filename}" key we can't flag the + # bad source as used, so a re-run could re-pick the same file and loop. + # Bail and let the caller fail it normally. + if not username or not filename: + return False + + retry_count = task.get('quarantine_retry_count', 0) + if retry_count >= MAX_QUARANTINE_RETRIES: + logger.warning( + f"[Retry:{trigger}] Task {task_id} hit the quarantine-retry cap " + f"({MAX_QUARANTINE_RETRIES}) — giving up, marking failed" + ) + return False + + # Mark the quarantined source as used so the re-run won't pick it again. + # Uses the same "{username}_{filename}" key the worker dedups against. + used_sources = task.get('used_sources', set()) + used_sources.add(f"{username}_{filename}") + task['used_sources'] = used_sources + + task['quarantine_retry_count'] = retry_count + 1 + # Drop the stale download identity + the prior attempt's quarantine link. + task.pop('download_id', None) + task.pop('username', None) + task.pop('filename', None) + task.pop('quarantine_entry_id', None) + task['status'] = 'searching' + task['status_change_time'] = time.time() + + logger.info( + f"[Retry:{trigger}] Re-queuing task {task_id} for next-best candidate " + f"(attempt {retry_count + 1}/{MAX_QUARANTINE_RETRIES})" + ) + missing_download_executor.submit(_download_track_worker, task_id, batch_id) + return True + + def _is_release_task(task): ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} username = task.get('username') or ti.get('username') diff --git a/core/imports/pipeline.py b/core/imports/pipeline.py index 067e4559..a2890519 100644 --- a/core/imports/pipeline.py +++ b/core/imports/pipeline.py @@ -109,6 +109,30 @@ def _mark_task_quarantined(context: dict, quarantine_path: str | None) -> None: download_tasks[task_id]['quarantine_entry_id'] = entry_id +def _requeue_quarantined_task_for_retry(task_id, batch_id, trigger) -> bool: + """Ask the download monitor to re-run this task on its next-best candidate. + + Thin lazy-import wrapper around + ``core.downloads.monitor.requeue_quarantined_task_for_retry``. Imported + lazily (and defensively) so the post-processing pipeline stays importable on + its own — the monitor's retry globals are wired by web_server at startup, and + manual-import callers that never started a download worker simply get False. + Returns True when a retry was queued (caller must not mark the task failed). + """ + if not task_id: + return False + try: + from core.downloads.monitor import requeue_quarantined_task_for_retry + except Exception as exc: # pragma: no cover - defensive + logger.debug(f"next-candidate retry unavailable ({trigger}): {exc}") + return False + try: + return requeue_quarantined_task_for_retry(task_id, batch_id, trigger) + except Exception as exc: # pragma: no cover - defensive + logger.error(f"next-candidate retry failed ({trigger}): {exc}") + return False + + def import_rejection_reason(context: dict) -> str | None: """Human-readable reason if post-processing terminally rejected the file (quarantine or race-guard), else ``None`` for a clean import. @@ -992,6 +1016,16 @@ def post_process_matched_download_with_verification(context_key, context, file_p if context.get('_acoustid_quarantined'): failure_msg = context.get('_acoustid_failure_msg', 'AcoustID verification failed') + # Before failing outright, try the next-best candidate. The wrong + # file was just quarantined; re-running the worker (with the bad + # source flagged used) picks the runner-up match instead. + with matched_context_lock: + matched_downloads_context.pop(context_key, None) + if _requeue_quarantined_task_for_retry(task_id, batch_id, 'acoustid'): + logger.info( + f"AcoustID mismatch for task {task_id} — retrying next-best candidate: {failure_msg}" + ) + return logger.info(f"File was quarantined by AcoustID verification (task={task_id}): {failure_msg}") with tasks_lock: if task_id in download_tasks: @@ -1000,9 +1034,6 @@ def post_process_matched_download_with_verification(context_key, context, file_p _eid = context.get('_quarantine_entry_id') if _eid: download_tasks[task_id]['quarantine_entry_id'] = _eid - with matched_context_lock: - if context_key in matched_downloads_context: - del matched_downloads_context[context_key] _notify_download_completed(batch_id, task_id, success=False) return @@ -1044,6 +1075,16 @@ def post_process_matched_download_with_verification(context_key, context, file_p # source files failed integrity and were quarantined. if context.get('_integrity_failure_msg'): failure_msg = context.get('_integrity_failure_msg', 'unknown') + # Integrity/duration mismatch (truncated transfer, wrong-length cut, + # etc). Same treatment as an AcoustID mismatch: quarantine the bad + # file and retry the next-best candidate before failing. + with matched_context_lock: + matched_downloads_context.pop(context_key, None) + if _requeue_quarantined_task_for_retry(task_id, batch_id, 'integrity'): + logger.info( + f"Integrity check failed for task {task_id} — retrying next-best candidate: {failure_msg}" + ) + return logger.error( f"Task {task_id} failed integrity check — marking failed: {failure_msg}" ) @@ -1056,9 +1097,6 @@ def post_process_matched_download_with_verification(context_key, context, file_p _eid = context.get('_quarantine_entry_id') if _eid: download_tasks[task_id]['quarantine_entry_id'] = _eid - with matched_context_lock: - if context_key in matched_downloads_context: - del matched_downloads_context[context_key] _notify_download_completed(batch_id, task_id, success=False) return diff --git a/tests/imports/test_import_pipeline.py b/tests/imports/test_import_pipeline.py index b283cb7e..6667e985 100644 --- a/tests/imports/test_import_pipeline.py +++ b/tests/imports/test_import_pipeline.py @@ -304,3 +304,135 @@ def test_verification_wrapper_applies_quarantine_entry_id_on_integrity_failure(m runtime_state.download_tasks.update(original) runtime_state.matched_downloads_context.clear() runtime_state.matched_downloads_context.update(original_ctx) + + +# --------------------------------------------------------------------------- +# Next-best-candidate retry on AcoustID / integrity quarantine. When a +# verification or integrity check quarantines the wrong/broken file, the wrapper +# asks the monitor to re-run the worker on the next candidate (skipping the bad +# source) instead of failing the task outright. +# --------------------------------------------------------------------------- + +def _wire_retry_engine(monkeypatch): + """Wire monitor's retry globals to capture the worker re-submission.""" + import core.downloads.monitor as monitor + + submitted = [] + + class _Exec: + def submit(self, fn, *args): + submitted.append(args) + + monkeypatch.setattr(monitor, "missing_download_executor", _Exec()) + monkeypatch.setattr(monitor, "_download_track_worker", lambda task_id, batch_id: None) + monkeypatch.setattr(monitor, "MAX_QUARANTINE_RETRIES", 5) + return submitted + + +def _run_wrapper_with_quarantine(monkeypatch, flag_setter, task_extra=None): + task_id, batch_id, context_key = "rtask", "rbatch", "rctx" + context = {"track_info": {}, "task_id": task_id, "batch_id": batch_id} + + monkeypatch.setattr(import_pipeline, "post_process_matched_download", flag_setter) + + original = dict(runtime_state.download_tasks) + original_ctx = dict(runtime_state.matched_downloads_context) + try: + runtime_state.download_tasks.clear() + task = { + "track_info": {}, "status": "downloading", + "username": "hifi", "filename": "123||A - B", "used_sources": set(), + } + if task_extra: + task.update(task_extra) + runtime_state.download_tasks[task_id] = task + runtime_state.matched_downloads_context.clear() + runtime_state.matched_downloads_context[context_key] = context + + completion = [] + runtime = types.SimpleNamespace( + automation_engine=None, + on_download_completed=lambda b, t, success: completion.append((b, t, success)), + web_scan_manager=None, + repair_worker=None, + ) + import_pipeline.post_process_matched_download_with_verification( + context_key, context, "/tmp/source.flac", task_id, batch_id, runtime, + ) + return dict(runtime_state.download_tasks[task_id]), completion, context_key + finally: + runtime_state.download_tasks.clear() + runtime_state.download_tasks.update(original) + runtime_state.matched_downloads_context.clear() + runtime_state.matched_downloads_context.update(original_ctx) + + +def test_acoustid_mismatch_requeues_next_candidate(monkeypatch): + submitted = _wire_retry_engine(monkeypatch) + + def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None): + ctx["_acoustid_quarantined"] = True + ctx["_acoustid_failure_msg"] = "wrong song" + + task, completion, context_key = _run_wrapper_with_quarantine(monkeypatch, _fake_inner) + + # Task goes back to searching for the next candidate — NOT failed. + assert task["status"] == "searching" + assert task["quarantine_retry_count"] == 1 + # The quarantined source is flagged so the re-run won't re-pick it. + assert "hifi_123||A - B" in task["used_sources"] + # Stale download identity cleared; worker re-submitted; no batch failure. + assert "download_id" not in task and "username" not in task + assert submitted == [("rtask", "rbatch")] + assert completion == [] + # Old context cleaned up (the re-run builds a fresh one for the new pick). + assert context_key not in runtime_state.matched_downloads_context + + +def test_integrity_mismatch_requeues_next_candidate(monkeypatch): + submitted = _wire_retry_engine(monkeypatch) + + def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None): + ctx["_integrity_failure_msg"] = "Duration mismatch: file is 231.0s, expected 271.0s" + + task, completion, _ = _run_wrapper_with_quarantine(monkeypatch, _fake_inner) + + assert task["status"] == "searching" + assert task["quarantine_retry_count"] == 1 + assert submitted == [("rtask", "rbatch")] + assert completion == [] + + +def test_manual_pick_does_not_requeue_on_mismatch(monkeypatch): + submitted = _wire_retry_engine(monkeypatch) + + def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None): + ctx["_integrity_failure_msg"] = "Duration mismatch" + + task, completion, _ = _run_wrapper_with_quarantine( + monkeypatch, _fake_inner, task_extra={"_user_manual_pick": True}, + ) + + # User explicitly chose this file — fail it, don't silently swap. + assert task["status"] == "failed" + assert submitted == [] + assert completion == [("rbatch", "rtask", False)] + + +def test_retry_budget_exhausted_fails_task(monkeypatch): + submitted = _wire_retry_engine(monkeypatch) + import core.downloads.monitor as monitor + monkeypatch.setattr(monitor, "MAX_QUARANTINE_RETRIES", 2) + + def _fake_inner(ck, ctx, fp, runtime, metadata_runtime=None): + ctx["_acoustid_quarantined"] = True + ctx["_acoustid_failure_msg"] = "wrong song" + + task, completion, _ = _run_wrapper_with_quarantine( + monkeypatch, _fake_inner, task_extra={"quarantine_retry_count": 2}, + ) + + # Cap reached — fall through to normal failure handling. + assert task["status"] == "failed" + assert submitted == [] + assert completion == [("rbatch", "rtask", False)]