Downloads: per-source exhaustive retry budget on mismatch (opt-in)
Adds an opt-in exhaustive mode to the quarantine-retry path. Default behaviour is unchanged: a single global cap (MAX_QUARANTINE_RETRIES=5). When post_processing.retry_exhaustive is on, each source gets its OWN retry budget sized as query_count x retries_per_query. Soulseek peers collapse to one 'soulseek' bucket; streaming plugins keep their name. The worker now records query_count on the task; the budget scales with the track's real query count. Loop protection is threefold: per-source cap, used_sources exhaustion (the natural terminator), and an absolute ceiling (MAX_TOTAL_QUARANTINE_RETRIES=100). New settings (config + WebUI): retry_next_candidate_on_mismatch (master), retry_exhaustive, retries_per_query (default 5). Tests: 6 new cases covering per-source budgeting, source separation, Soulseek-peer bucketing, query_count default, and the absolute ceiling. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
dev
parent
e83cf19903
commit
f3d43f385e
6 changed files with 262 additions and 9 deletions
|
|
@ -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 on.
|
||||
"retry_next_candidate_on_mismatch": True,
|
||||
# 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": {
|
||||
|
|
|
|||
|
|
@ -44,8 +44,37 @@ _RELEASE_SOURCE_NAMES = frozenset(('torrent', 'usenet'))
|
|||
# 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 _download_id_key(download_id):
|
||||
return f"download_id::{download_id}" if download_id else None
|
||||
|
|
@ -100,13 +129,63 @@ def requeue_quarantined_task_for_retry(task_id, batch_id, trigger):
|
|||
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
|
||||
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:
|
||||
logger.warning(
|
||||
f"[Retry:{trigger}] Task {task_id} exhausted its retry budget "
|
||||
f"for source '{source}' ({source_count}/{budget}) — giving up, "
|
||||
f"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, 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:
|
||||
# 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.
|
||||
|
|
@ -114,7 +193,7 @@ def requeue_quarantined_task_for_retry(task_id, batch_id, trigger):
|
|||
used_sources.add(f"{username}_{filename}")
|
||||
task['used_sources'] = used_sources
|
||||
|
||||
task['quarantine_retry_count'] = retry_count + 1
|
||||
task['quarantine_retry_count'] = total_count + 1
|
||||
# Drop the stale download identity + the prior attempt's quarantine link.
|
||||
task.pop('download_id', None)
|
||||
task.pop('username', None)
|
||||
|
|
@ -125,7 +204,7 @@ def requeue_quarantined_task_for_retry(task_id, batch_id, trigger):
|
|||
|
||||
logger.info(
|
||||
f"[Retry:{trigger}] Re-queuing task {task_id} for next-best candidate "
|
||||
f"(attempt {retry_count + 1}/{MAX_QUARANTINE_RETRIES})"
|
||||
f"(attempt {attempt_desc})"
|
||||
)
|
||||
missing_download_executor.submit(_download_track_worker, task_id, batch_id)
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -277,6 +277,11 @@ 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}')")
|
||||
|
||||
|
|
|
|||
|
|
@ -329,6 +329,20 @@ def _wire_retry_engine(monkeypatch):
|
|||
return submitted
|
||||
|
||||
|
||||
def _patch_config(monkeypatch, overrides):
|
||||
"""Override specific config keys for the monitor's config_manager reads."""
|
||||
import core.downloads.monitor as monitor
|
||||
|
||||
real_get = monitor.config_manager.get
|
||||
|
||||
def fake_get(key, default=None):
|
||||
if key in overrides:
|
||||
return overrides[key]
|
||||
return real_get(key, default)
|
||||
|
||||
monkeypatch.setattr(monitor.config_manager, "get", fake_get)
|
||||
|
||||
|
||||
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}
|
||||
|
|
@ -436,3 +450,120 @@ def test_retry_budget_exhausted_fails_task(monkeypatch):
|
|||
assert task["status"] == "failed"
|
||||
assert submitted == []
|
||||
assert completion == [("rbatch", "rtask", False)]
|
||||
|
||||
|
||||
def _acoustid_quarantine(ck, ctx, fp, runtime, metadata_runtime=None):
|
||||
ctx["_acoustid_quarantined"] = True
|
||||
ctx["_acoustid_failure_msg"] = "wrong song"
|
||||
|
||||
|
||||
def test_exhaustive_mode_uses_per_source_budget(monkeypatch):
|
||||
submitted = _wire_retry_engine(monkeypatch)
|
||||
_patch_config(monkeypatch, {
|
||||
"post_processing.retry_exhaustive": True,
|
||||
"post_processing.retries_per_query": 5,
|
||||
})
|
||||
|
||||
# query_count=2 → budget for source 'hifi' = 2 * 5 = 10; first failure retries.
|
||||
task, completion, _ = _run_wrapper_with_quarantine(
|
||||
monkeypatch, _acoustid_quarantine, task_extra={"query_count": 2},
|
||||
)
|
||||
|
||||
assert task["status"] == "searching"
|
||||
# Per-source budget tracked separately from the legacy global counter.
|
||||
assert task["quarantine_retry_counts_by_source"] == {"hifi": 1}
|
||||
assert task["quarantine_retry_count"] == 1
|
||||
assert "hifi_123||A - B" in task["used_sources"]
|
||||
assert submitted == [("rtask", "rbatch")]
|
||||
assert completion == []
|
||||
|
||||
|
||||
def test_exhaustive_source_budget_exhausted_fails(monkeypatch):
|
||||
submitted = _wire_retry_engine(monkeypatch)
|
||||
_patch_config(monkeypatch, {
|
||||
"post_processing.retry_exhaustive": True,
|
||||
"post_processing.retries_per_query": 5,
|
||||
})
|
||||
|
||||
# hifi already at its full budget (query_count 2 * 5 = 10) → fail, no retry.
|
||||
task, completion, _ = _run_wrapper_with_quarantine(
|
||||
monkeypatch, _acoustid_quarantine,
|
||||
task_extra={"query_count": 2, "quarantine_retry_counts_by_source": {"hifi": 10}},
|
||||
)
|
||||
|
||||
assert task["status"] == "failed"
|
||||
assert submitted == []
|
||||
assert completion == [("rbatch", "rtask", False)]
|
||||
|
||||
|
||||
def test_exhaustive_budget_is_separate_per_source(monkeypatch):
|
||||
submitted = _wire_retry_engine(monkeypatch)
|
||||
_patch_config(monkeypatch, {
|
||||
"post_processing.retry_exhaustive": True,
|
||||
"post_processing.retries_per_query": 5,
|
||||
})
|
||||
|
||||
# soulseek is already maxed, but the failing download is on hifi — hifi has
|
||||
# its own fresh budget, so the task still retries.
|
||||
task, completion, _ = _run_wrapper_with_quarantine(
|
||||
monkeypatch, _acoustid_quarantine,
|
||||
task_extra={"query_count": 1, "quarantine_retry_counts_by_source": {"soulseek": 5}},
|
||||
)
|
||||
|
||||
assert task["status"] == "searching"
|
||||
assert task["quarantine_retry_counts_by_source"] == {"soulseek": 5, "hifi": 1}
|
||||
assert submitted == [("rtask", "rbatch")]
|
||||
|
||||
|
||||
def test_exhaustive_soulseek_peer_resolves_to_soulseek(monkeypatch):
|
||||
submitted = _wire_retry_engine(monkeypatch)
|
||||
_patch_config(monkeypatch, {
|
||||
"post_processing.retry_exhaustive": True,
|
||||
"post_processing.retries_per_query": 5,
|
||||
})
|
||||
|
||||
# A Soulseek peer name (not a streaming source) is bucketed under 'soulseek'.
|
||||
task, completion, _ = _run_wrapper_with_quarantine(
|
||||
monkeypatch, _acoustid_quarantine,
|
||||
task_extra={"username": "DjPeer", "filename": "f.flac", "query_count": 1},
|
||||
)
|
||||
|
||||
assert task["status"] == "searching"
|
||||
assert task["quarantine_retry_counts_by_source"] == {"soulseek": 1}
|
||||
|
||||
|
||||
def test_exhaustive_budget_defaults_query_count_to_one(monkeypatch):
|
||||
submitted = _wire_retry_engine(monkeypatch)
|
||||
_patch_config(monkeypatch, {
|
||||
"post_processing.retry_exhaustive": True,
|
||||
"post_processing.retries_per_query": 1,
|
||||
})
|
||||
|
||||
# No query_count on the task → budget defaults to 1 * 1 = 1; hifi already at 1.
|
||||
task, completion, _ = _run_wrapper_with_quarantine(
|
||||
monkeypatch, _acoustid_quarantine,
|
||||
task_extra={"quarantine_retry_counts_by_source": {"hifi": 1}},
|
||||
)
|
||||
|
||||
assert task["status"] == "failed"
|
||||
assert submitted == []
|
||||
|
||||
|
||||
def test_exhaustive_absolute_ceiling_guards_runaway(monkeypatch):
|
||||
submitted = _wire_retry_engine(monkeypatch)
|
||||
import core.downloads.monitor as monitor
|
||||
monkeypatch.setattr(monitor, "MAX_TOTAL_QUARANTINE_RETRIES", 3)
|
||||
_patch_config(monkeypatch, {
|
||||
"post_processing.retry_exhaustive": True,
|
||||
"post_processing.retries_per_query": 1000, # per-source budget effectively unbounded
|
||||
})
|
||||
|
||||
# Per-source budget is huge, but the absolute total ceiling (3) still fires.
|
||||
task, completion, _ = _run_wrapper_with_quarantine(
|
||||
monkeypatch, _acoustid_quarantine,
|
||||
task_extra={"query_count": 1, "quarantine_retry_count": 3,
|
||||
"quarantine_retry_counts_by_source": {"hifi": 0}},
|
||||
)
|
||||
|
||||
assert task["status"] == "failed"
|
||||
assert submitted == []
|
||||
|
|
|
|||
|
|
@ -5563,6 +5563,25 @@
|
|||
<input type="number" id="duration-tolerance-seconds" min="0" max="60" step="0.5" value="0" style="width: 100px;">
|
||||
<small class="settings-hint">Maximum drift between the file's actual length and the metadata source's expected length before the file is quarantined. <strong>0 = auto</strong> (3s normal, 5s for tracks >10min). Raise this if tracks routinely quarantine for being a few seconds off (live recordings, alternate masterings, etc). Capped at 60s.</small>
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label class="checkbox-label">
|
||||
<input type="checkbox" id="retry-next-candidate" checked>
|
||||
Retry next-best candidate on mismatch
|
||||
</label>
|
||||
<small class="settings-hint">When a download is quarantined (AcoustID mismatch or duration/integrity failure), automatically try the next-best candidate instead of failing the track. Off = quarantine and fail immediately.</small>
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label class="checkbox-label">
|
||||
<input type="checkbox" id="retry-exhaustive">
|
||||
Exhaustive retry (separate budget per source)
|
||||
</label>
|
||||
<small class="settings-hint">Give every source (Soulseek, then HiFi/Tidal/…) its own retry budget. Each source spends <em>queries × retries-per-query</em> attempts before the track moves on. Worst case across two sources can mean many downloads — use for hard-to-match tracks (e.g. CJK artist names). Requires the option above.</small>
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label for="retries-per-query">Retries per query (per source)</label>
|
||||
<input type="number" id="retries-per-query" min="1" max="20" step="1" value="5" style="width: 100px;">
|
||||
<small class="settings-hint">In exhaustive mode, how many candidates to try per search query per source. The per-source budget is <em>number of search queries × this value</em>. Only used when Exhaustive retry is on.</small>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Tag Embedding — per-tag toggles grouped by source -->
|
||||
|
|
|
|||
|
|
@ -1168,6 +1168,9 @@ async function loadSettingsData() {
|
|||
document.getElementById('lrclib-enabled').checked = settings.metadata_enhancement?.lrclib_enabled !== false;
|
||||
document.getElementById('replaygain-enabled').checked = settings.post_processing?.replaygain_enabled === true;
|
||||
document.getElementById('duration-tolerance-seconds').value = settings.post_processing?.duration_tolerance_seconds ?? 0;
|
||||
document.getElementById('retry-next-candidate').checked = settings.post_processing?.retry_next_candidate_on_mismatch !== false;
|
||||
document.getElementById('retry-exhaustive').checked = settings.post_processing?.retry_exhaustive === true;
|
||||
document.getElementById('retries-per-query').value = settings.post_processing?.retries_per_query ?? 5;
|
||||
// Load service master toggles
|
||||
document.getElementById('embed-spotify').checked = settings.spotify?.embed_tags !== false;
|
||||
document.getElementById('embed-itunes').checked = settings.itunes?.embed_tags !== false;
|
||||
|
|
@ -2996,6 +2999,9 @@ async function saveSettings(quiet = false) {
|
|||
post_processing: {
|
||||
replaygain_enabled: document.getElementById('replaygain-enabled').checked,
|
||||
duration_tolerance_seconds: parseFloat(document.getElementById('duration-tolerance-seconds').value) || 0,
|
||||
retry_next_candidate_on_mismatch: document.getElementById('retry-next-candidate').checked,
|
||||
retry_exhaustive: document.getElementById('retry-exhaustive').checked,
|
||||
retries_per_query: Math.max(1, parseInt(document.getElementById('retries-per-query').value, 10) || 5),
|
||||
},
|
||||
library: {
|
||||
music_paths: collectMusicPaths(),
|
||||
|
|
|
|||
Loading…
Reference in a new issue