danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Downloads: retry next-best candidate on AcoustID/integrity quarantine

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
dev 2026-06-05 02:24:00 +02:00
parent 971f2fd4f0
commit e83cf19903
3 changed files with 265 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

89
core/downloads/monitor.py
View file

@ -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')

50
core/imports/pipeline.py
View file

@ -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

132
tests/imports/test_import_pipeline.py
View file

@ -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)]