diff --git a/config/settings.py b/config/settings.py
index 5ff13f6a..f729a2a4 100644
--- a/config/settings.py
+++ b/config/settings.py
@@ -494,6 +494,19 @@ class ConfigManager:
"album_bundle_poll_interval_seconds": 2.0,
"album_bundle_timeout_seconds": 6 * 60 * 60, # 6 hours
},
+ "post_processing": {
+ # When a download is quarantined (AcoustID mismatch, integrity /
+ # duration failure), retry the next-best candidate instead of
+ # failing outright. Default off — opt-in.
+ "retry_next_candidate_on_mismatch": False,
+ # Opt-in exhaustive retry: budget retries PER SOURCE so every
+ # source (Soulseek, then HiFi/Tidal/…) gets its own attempts
+ # before the track gives up. Default off (single global cap).
+ "retry_exhaustive": False,
+ # Retries per search query per source in exhaustive mode. The
+ # per-source budget is query_count × this value.
+ "retries_per_query": 5,
+ },
"tidal_download": {
"quality": "lossless", # Options: "low", "high", "lossless", "hires"
"session": {
diff --git a/core/downloads/monitor.py b/core/downloads/monitor.py
index 4bcb0bc3..472e521c 100644
--- a/core/downloads/monitor.py
+++ b/core/downloads/monitor.py
@@ -37,11 +37,228 @@ 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).
+#
+# Default (non-exhaustive) mode uses this single global cap. The opt-in
+# exhaustive mode (post_processing.retry_exhaustive) instead budgets retries
+# PER SOURCE — see requeue_quarantined_task_for_retry.
+MAX_QUARANTINE_RETRIES = 5
+
+# Absolute runaway guard for exhaustive mode. Per-source budgets are already
+# finite (query_count × retries_per_query, and Soulseek peers all collapse to
+# one 'soulseek' bucket), but this ceiling caps the TOTAL retries across every
+# source so a misbehaving source-resolution can never loop forever.
+MAX_TOTAL_QUARANTINE_RETRIES = 100
+
+# Streaming plugins report their source name as the download's "username"
+# (see download_orchestrator._streaming_sources). Soulseek uses the peer name
+# instead, so anything not in this set is bucketed under 'soulseek' for the
+# per-source retry budget.
+_STREAMING_SOURCE_NAMES = frozenset((
+ 'youtube', 'tidal', 'qobuz', 'hifi', 'deezer_dl', 'lidarr', 'soundcloud', 'amazon',
+))
+
+
+def _resolve_download_source(username):
+ """Map a download's username to its logical source for per-source budgeting.
+
+ Streaming sources use the source name as username; Soulseek uses the peer
+ name, so every Soulseek peer collapses to a single 'soulseek' bucket.
+ """
+ if username and username in _STREAMING_SOURCE_NAMES:
+ return 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
+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
+
+ total_count = task.get('quarantine_retry_count', 0)
+
+ if config_manager.get('post_processing.retry_exhaustive', False):
+ # Exhaustive mode: a SEPARATE budget per source. The budget scales
+ # with the track's own query count (the worker generates a variable
+ # number of search queries per track) × the configured retries per
+ # query. Soulseek candidates are walked first (one per retry), then
+ # the worker's hybrid fallback moves to the next source — each source
+ # spending its own budget. The natural terminator (used_sources
+ # exhaustion → worker clean-fail) still ends most tracks well before
+ # any budget is reached; the budget is the per-source safety ceiling.
+ source = _resolve_download_source(username)
+ retries_per_query = config_manager.get('post_processing.retries_per_query', 5)
+ try:
+ retries_per_query = int(retries_per_query)
+ except (TypeError, ValueError):
+ retries_per_query = 5
+ if retries_per_query < 1:
+ retries_per_query = 1
+
+ query_count = task.get('query_count') or 1
+ if query_count < 1:
+ query_count = 1
+ budget = query_count * retries_per_query
+
+ counts = task.get('quarantine_retry_counts_by_source')
+ if not isinstance(counts, dict):
+ counts = {}
+ source_count = counts.get(source, 0)
+
+ if source_count >= budget:
+ # 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)})"
+ )
+ 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:
+ logger.warning(
+ f"[Retry:{trigger}] Task {task_id} hit the quarantine-retry cap "
+ f"({MAX_QUARANTINE_RETRIES}) — giving up, marking failed"
+ )
+ return False
+ attempt_desc = f"{total_count + 1}/{MAX_QUARANTINE_RETRIES}"
+
+ # 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'] = total_count + 1
+ # Flag the re-run as a quarantine retry so the worker walks the
+ # already-found candidates (cached-first) before re-searching — the
+ # connection was fine, the content was just wrong. Dead-connection /
+ # stuck retries (handled elsewhere in the monitor) deliberately do NOT
+ # set this, so they re-search fresh.
+ task['_quarantine_retry'] = True
+ # 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 {attempt_desc})"
+ )
+ 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/downloads/task_worker.py b/core/downloads/task_worker.py
index cc6751dc..1e27cbc4 100644
--- a/core/downloads/task_worker.py
+++ b/core/downloads/task_worker.py
@@ -31,6 +31,65 @@ from core.spotify_client import Track as SpotifyTrack
logger = logging.getLogger(__name__)
+def _resolve_worker_source(username):
+ """Logical source bucket for a candidate's username (Soulseek peers all
+ collapse to 'soulseek'; streaming sources keep their name). Mirrors the
+ monitor's resolver — imported lazily to avoid an import cycle."""
+ try:
+ from core.downloads.monitor import _resolve_download_source
+ return _resolve_download_source(username)
+ except Exception:
+ return 'soulseek'
+
+
+def _cand_user_file(candidate):
+ """Read (username, filename) from a candidate that may be a TrackResult
+ object or a plain dict (tests / cached raw rows)."""
+ if isinstance(candidate, dict):
+ return candidate.get('username'), candidate.get('filename')
+ return getattr(candidate, 'username', None), getattr(candidate, 'filename', None)
+
+
+def _try_cached_candidates(task_id, batch_id, track, deps):
+ """Quarantine-retry fast path: attempt the already-found candidates before
+ re-searching anything.
+
+ When a verified-bad file is re-queued, the connection was fine (the file
+ downloaded, it was just the wrong/broken content) — so the next-best pick is
+ almost always already sitting in ``cached_candidates``. Walk those (skipping
+ sources already tried or budget-exhausted) and hand them to the normal
+ download path. Returns True if a download was started; False to fall through
+ to a fresh search (which only happens for a not-yet-searched source).
+ """
+ with tasks_lock:
+ task = download_tasks.get(task_id)
+ if not task:
+ return False
+ cached = list(task.get('cached_candidates') or [])
+ used = set(task.get('used_sources') or ())
+ exhausted = {str(s).lower() for s in (task.get('exhausted_download_sources') or ())}
+
+ remaining = []
+ for c in cached:
+ uname, fname = _cand_user_file(c)
+ if not uname or not fname:
+ continue
+ if f"{uname}_{fname}" in used:
+ continue
+ if _resolve_worker_source(uname).lower() in exhausted:
+ continue
+ remaining.append(c)
+
+ if not remaining:
+ return False
+
+ logger.info(
+ f"[Modal Worker] Quarantine retry: trying {len(remaining)} cached "
+ f"candidate(s) before re-searching (task {task_id})"
+ )
+ return deps.attempt_download_with_candidates(task_id, remaining, track, batch_id)
+
+
def _private_album_bundle_staging_miss_reason(batch_id: Optional[str], deps: Any) -> Optional[str]:
"""Return a user-facing miss reason when per-track search should stop.
@@ -92,6 +151,7 @@ class TaskWorkerDeps:
attempt_download_with_candidates: Callable # (task_id, candidates, track, batch_id) -> bool
on_download_completed: Callable # (batch_id, task_id, success) -> None
recover_worker_slot: Callable # (batch_id, task_id) -> None
+ try_version_mismatch_fallback: Optional[Callable] = None # (title, artist, task_id, batch_id) -> bool
def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorkerDeps) -> None:
@@ -206,6 +266,26 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
download_tasks[task_id]['used_sources'] = set()
# Else: keep existing used_sources to avoid retrying same failed hosts
+ # Cached-first quarantine retry. The monitor sets ``_quarantine_retry``
+ # when a verified-bad file is re-queued; in that case we walk the
+ # already-found candidates before re-searching (the connection was fine,
+ # just the content was wrong). A NON-quarantine entry (fresh download, or
+ # the monitor's dead-connection/stuck retry) instead starts a new search
+ # generation: clear the searched-source memory so each source can be
+ # searched fresh again.
+ with tasks_lock:
+ _t = download_tasks.get(task_id, {})
+ is_quarantine_retry = bool(_t.pop('_quarantine_retry', False))
+ if not is_quarantine_retry:
+ _t.pop('searched_queries', None)
+ if is_quarantine_retry and _try_cached_candidates(task_id, batch_id, track, deps):
+ with tasks_lock:
+ used_filename = download_tasks.get(task_id, {}).get('filename')
+ used_username = download_tasks.get(task_id, {}).get('username')
+ if used_filename and used_username:
+ deps.store_batch_source(batch_id, used_username, used_filename)
+ return
+
# 1. Generate multiple search queries (like GUI's generate_smart_search_queries)
artist_name = track.artists[0] if track.artists else None
track_name = track.name
@@ -277,12 +357,40 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
seen.add(query.lower())
search_queries = unique_queries
+ # Expose the query count so the quarantine-retry budget (exhaustive mode)
+ # can size each source's budget as query_count × retries_per_query.
+ with tasks_lock:
+ if task_id in download_tasks:
+ download_tasks[task_id]['query_count'] = len(search_queries)
logger.info(f"[Modal Worker] Generated {len(search_queries)} smart search queries for '{track.name}': {search_queries}")
logger.info(f"[Modal Worker] About to start search loop for task {task_id} (track: '{track.name}')")
# 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.
+ #
+ # On a quarantine retry we do NOT exclude a source just because it was
+ # searched once: the first run only ran ONE query before starting a
+ # download, so the later queries (e.g. "artist + album") have never hit
+ # that source yet and may surface the correct upload. Instead we remember
+ # which QUERIES already ran (``searched_queries``) and skip re-running
+ # only those — their candidates are walked via the cached-first path
+ # above. The not-yet-searched queries still search the same source, so
+ # every query is exhausted per source before the chain switches sources.
+ # Fresh / dead-connection runs cleared searched_queries above, so they
+ # search everything again.
+ with tasks_lock:
+ _t = download_tasks.get(task_id, {})
+ _exhausted_sources = [str(s) for s in (_t.get('exhausted_download_sources') or ())]
+ _searched_queries = (
+ set(_t.get('searched_queries') or ()) if is_quarantine_retry else set()
+ )
for query_index, query in enumerate(search_queries):
# Cancellation check before each query
with tasks_lock:
@@ -295,6 +403,17 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
return
download_tasks[task_id]['current_query_index'] = query_index
+ # Cached-first: a query already run last generation has its candidates
+ # sitting in cache (walked above) — re-searching it is the wasteful
+ # repeat the cached-first design removes. Skip it; the not-yet-run
+ # queries below still search this source.
+ if is_quarantine_retry and query in _searched_queries:
+ logger.debug(
+ f"[Modal Worker] Skipping already-searched query '{query}' "
+ f"(candidates served from cache) for task {task_id}"
+ )
+ continue
+
logger.debug(f"[Modal Worker] Query {query_index + 1}/{len(search_queries)}: '{query}'")
logger.debug(f"About to call soulseek search for task {task_id}")
@@ -319,9 +438,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")
@@ -330,6 +453,16 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
if task_id not in download_tasks:
logger.info(f"[Modal Worker] Task {task_id} was deleted after search returned")
return
+ # Remember this query ran so a later quarantine retry skips
+ # re-searching it (its candidates are walked via cached-first).
+ # Recorded regardless of result count: re-running a query is
+ # deterministic, so a query that returned nothing won't return
+ # anything new next time either.
+ _sq = download_tasks[task_id].get('searched_queries')
+ if not isinstance(_sq, set):
+ _sq = set()
+ _sq.add(query)
+ download_tasks[task_id]['searched_queries'] = _sq
if download_tasks[task_id]['status'] == 'cancelled':
logger.warning(f"[Modal Worker] Task {task_id} cancelled after search returned - ignoring results")
# Don't call _on_download_completed for cancelled tasks as it can stop monitoring
@@ -352,7 +485,9 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
logger.warning(f"[Modal Worker] Task {task_id} cancelled before processing candidates")
# Don't call _on_download_completed for cancelled tasks as it can stop monitoring
return
- # Store candidates for retry fallback (like GUI)
+ # Store candidates for retry fallback (like GUI). A
+ # later quarantine retry walks these via cached-first
+ # and skips re-searching this query (searched_queries).
download_tasks[task_id]['cached_candidates'] = candidates
# Try to download with these candidates
@@ -414,7 +549,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}")
@@ -433,6 +573,9 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
fb_candidates = deps.get_valid_candidates(fb_results, track, fb_query)
if fb_candidates:
logger.warning(f"[Hybrid Fallback] {fallback_source} found {len(fb_candidates)} valid candidates!")
+ with tasks_lock:
+ if task_id in download_tasks:
+ download_tasks[task_id]['cached_candidates'] = fb_candidates
success = deps.attempt_download_with_candidates(task_id, fb_candidates, track, batch_id)
if success:
return
@@ -447,6 +590,15 @@ def download_track_worker(task_id: str, batch_id: Optional[str], deps: TaskWorke
# If we get here, all search queries and hybrid fallbacks failed
logger.warning(f"[Modal Worker] No valid candidates found for '{track.name}' after trying all {len(search_queries)} queries.")
+
+ # Last-resort: quarantine retry with no new candidates — the retry search
+ # exhausted all sources. If the setting is enabled, accept the best
+ # already-quarantined candidate rather than leaving the track missing.
+ if is_quarantine_retry and deps.try_version_mismatch_fallback:
+ _fallback_artist = track.artists[0] if track.artists else ''
+ if deps.try_version_mismatch_fallback(track.name, _fallback_artist, task_id, batch_id):
+ return # fallback re-dispatched; batch completion handled by reprocess thread
+
with tasks_lock:
if task_id in download_tasks:
download_tasks[task_id]['status'] = 'not_found'
diff --git a/core/imports/pipeline.py b/core/imports/pipeline.py
index 27e0384c..fc5bdf24 100644
--- a/core/imports/pipeline.py
+++ b/core/imports/pipeline.py
@@ -35,7 +35,12 @@ from core.imports.context import (
from core.imports.file_integrity import check_audio_integrity, resolve_duration_tolerance
from core.imports.filename import extract_track_number_from_filename
from core.imports.guards import check_flac_bit_depth, move_to_quarantine
-from core.imports.quarantine import entry_id_from_quarantined_filename
+from core.imports.quarantine import (
+ approve_quarantine_entry,
+ entry_id_from_quarantined_filename,
+ list_quarantine_entries,
+)
+from core.imports.version_mismatch_fallback import try_accept_version_mismatch_fallback
from core.imports.side_effects import (
emit_track_downloaded,
record_download_provenance,
@@ -109,6 +114,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.
@@ -178,6 +207,19 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
f"{os.path.basename(existing_final)}"
)
return
+ # File was intentionally moved to quarantine by a concurrent/earlier
+ # post-process call — this is a stale duplicate dispatch, not a race.
+ # _mark_task_quarantined sets _quarantine_entry_id for every quarantine
+ # trigger (AcoustID, integrity, bit-depth). The quarantine entry and
+ # its retry are already in flight; don't overwrite the task state with
+ # a spurious race-guard failure.
+ if context.get('_quarantine_entry_id'):
+ logger.debug(
+ f"[Race Guard] Source gone but already quarantined (entry %s) — stale duplicate call, ignoring: "
+ f"{os.path.basename(file_path)}",
+ context['_quarantine_entry_id'],
+ )
+ return
logger.error(
f"[Race Guard] Source file gone and no known destination — marking as failed: "
f"{os.path.basename(file_path)}"
@@ -966,6 +1008,53 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
post_process_locks.pop(context_key, None)
+def _attempt_version_mismatch_fallback(context, task_id, batch_id, runtime, metadata_runtime):
+ """Opt-in last resort once AcoustID retries are exhausted: accept the best
+ quarantined version-mismatch candidate for this track instead of failing.
+
+ Delegates the decision + safety rules to
+ ``core.imports.version_mismatch_fallback`` (version-mismatch only, all the
+ same matched version, >= min_count, AcoustID-only bypass). Returns True when
+ a candidate was accepted and re-dispatched — the caller then skips marking
+ the task failed.
+ """
+ try:
+ download_path = docker_resolve_path(
+ config_manager.get('soulseek.download_path', './downloads')
+ )
+ quarantine_dir = os.path.join(download_path, 'ss_quarantine')
+ restore_dir = os.path.join(download_path, 'Transfer')
+ expected_title = get_import_clean_title(context, default='')
+ expected_artist = get_import_clean_artist(context, default='')
+ if not expected_title or not expected_artist:
+ return False
+
+ def _reprocess(restored_path, ctx, tid, bid):
+ new_key = f"vmfallback_{tid}_{int(time.time())}"
+ threading.Thread(
+ target=lambda: post_process_matched_download_with_verification(
+ new_key, ctx, restored_path, tid, bid, runtime, metadata_runtime
+ ),
+ daemon=True,
+ ).start()
+
+ return try_accept_version_mismatch_fallback(
+ quarantine_dir=quarantine_dir,
+ restore_dir=restore_dir,
+ expected_title=expected_title,
+ expected_artist=expected_artist,
+ task_id=task_id,
+ batch_id=batch_id,
+ config_get=config_manager.get,
+ list_entries=list_quarantine_entries,
+ approve_entry=approve_quarantine_entry,
+ reprocess=_reprocess,
+ )
+ except Exception as exc:
+ pp_logger.debug("[Version-Mismatch Fallback] skipped due to error: %s", exc)
+ return False
+
+
def post_process_matched_download_with_verification(context_key, context, file_path, task_id, batch_id, runtime, metadata_runtime=None):
on_download_completed = getattr(runtime, "on_download_completed", None)
@@ -997,6 +1086,21 @@ 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
+ # Retries exhausted. Opt-in last resort: if every quarantined
+ # candidate for this track failed the SAME version mismatch (e.g. all
+ # instrumental), accept the best one rather than leaving it missing.
+ if _attempt_version_mismatch_fallback(context, task_id, batch_id, runtime, metadata_runtime):
+ return
logger.info(f"File was quarantined by AcoustID verification (task={task_id}): {failure_msg}")
with tasks_lock:
if task_id in download_tasks:
@@ -1005,9 +1109,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
@@ -1049,6 +1150,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}"
)
@@ -1061,9 +1172,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/core/imports/version_mismatch_fallback.py b/core/imports/version_mismatch_fallback.py
new file mode 100644
index 00000000..7eaab479
--- /dev/null
+++ b/core/imports/version_mismatch_fallback.py
@@ -0,0 +1,185 @@
+"""Last-resort acceptance of a version-mismatched download.
+
+Some tracks simply don't exist on the configured sources in the wanted cut —
+every copy is, say, the instrumental. The retry engine correctly rejects each
+one (version mismatch) and eventually gives up, leaving the track missing.
+
+This module provides an OPT-IN fallback: once a track's retries are fully
+exhausted, if every quarantined candidate for it failed the *same* way (same
+matched version, e.g. all ``instrumental``) and there are at least ``min_count``
+of them, accept the best (first-tried) one rather than failing outright.
+
+Hard safety rules:
+- Only ``Version mismatch`` quarantines qualify. Audio/artist mismatches
+ (a genuinely different recording) and integrity/duration failures
+ (truncated or wrong file) never participate.
+- All qualifying entries must share the same matched version. A mix
+ (instrumental + live) is ambiguous → no acceptance.
+- The chosen candidate is re-imported with only the AcoustID gate bypassed;
+ the integrity / duration / bit-depth gates still run, so a truncated or
+ corrupt file is never let through by this path.
+
+``select_version_mismatch_fallback`` is the pure decision core (no I/O) so it
+can be tested directly. ``try_accept_version_mismatch_fallback`` wires it to the
+quarantine store + re-import dispatch via injected callables.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any, Callable, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Matches the reason string written by acoustid_verification's version gate:
+# "Version mismatch: expected '
' () but file is '' ()"
+# We only need the matched () version to test cross-entry consistency.
+_VERSION_MISMATCH_RE = re.compile(
+ r"^Version mismatch:.*\bbut file is\b.*\(([^()]+)\)\s*$"
+)
+
+
+def _norm(text: Optional[str]) -> str:
+ return (text or "").strip().casefold()
+
+
+def matched_version(reason: Optional[str]) -> Optional[str]:
+ """Return the matched version token (e.g. ``'instrumental'``) for a
+ Version-mismatch reason string, or None if the reason isn't a version
+ mismatch / can't be parsed."""
+ if not reason:
+ return None
+ m = _VERSION_MISMATCH_RE.match(reason.strip())
+ if not m:
+ return None
+ return m.group(1).strip().casefold()
+
+
+def select_version_mismatch_fallback(
+ entries: List[Dict[str, Any]],
+ expected_title: str,
+ expected_artist: str,
+ min_count: int,
+) -> Optional[Dict[str, Any]]:
+ """Pick the quarantine entry to accept as a last resort, or None.
+
+ ``entries`` are dicts as produced by
+ :func:`core.imports.quarantine.list_quarantine_entries` (needs ``id``,
+ ``reason``, ``expected_track``, ``expected_artist``, ``has_full_context``).
+
+ Returns the chosen entry (the first-tried = oldest = best, by ascending
+ ``id`` whose timestamp prefix sorts chronologically) when, for this track,
+ there are at least ``min_count`` version-mismatch entries that all share the
+ same matched version and carry full context. Otherwise None.
+ """
+ title = _norm(expected_title)
+ artist = _norm(expected_artist)
+
+ candidates = []
+ for e in entries:
+ if not e.get("has_full_context"):
+ continue
+ if _norm(e.get("expected_track")) != title:
+ continue
+ if _norm(e.get("expected_artist")) != artist:
+ continue
+ version = matched_version(e.get("reason"))
+ if version is None:
+ continue
+ candidates.append((version, e))
+
+ if len(candidates) < max(1, int(min_count or 1)):
+ return None
+
+ versions = {v for v, _ in candidates}
+ if len(versions) != 1:
+ # Inconsistent wrong versions (e.g. instrumental + live) — ambiguous,
+ # don't guess which the user wants.
+ return None
+
+ # First tried = oldest = highest-confidence (the retry walks candidates
+ # best-first). The id is a "_