Import speed: downloads pause ALL enrichment workers, discovery pauses the contention five
Measured during a live album download: ~4m15s per track in post-processing (normal is ~20s), with the time vanishing silently inside embed_source_ids — up to 5 MusicBrainz calls per track crawling against a degraded musicbrainz.org while the MB enrichment worker kept eating the same ~1 req/s per-IP budget. Only Spotify/Last.fm/Genius were in the yield set; MusicBrainz, Deezer, iTunes, Discogs etc. kept grinding through downloads. Policy (new core/enrichment/yield_policy, tested): - downloads active -> ALL enrichment workers yield (post-processing touches every metadata source). listening-stats (local-only) and repair (user-scheduled) intentionally keep running. - discovery active -> the API-contention five yield (spotify/itunes/deezer/ discogs/hydrabase) — discovery never paused anything before, despite the pause helper literally defaulting to label='discovery'. - user overrides and user-paused bookkeeping keep their existing semantics; the dashboard yield_reason label now says WHICH foreground work caused it. Observability (the 4-minute silence can never come back): - every source lookup is timed; >2s logs a warning NAMING the source and duration (core/metadata/source.py _call_source_lookup) - the pipeline always logs "Metadata enhancement took X.Xs" per track 7 policy tests (incl. the motivating case: MB yields to downloads, keeps running during discovery); 277 pipeline/enrichment tests pass.
This commit is contained in:
parent
b58d7b4dad
commit
88da265ef4
5 changed files with 169 additions and 17 deletions
57
core/enrichment/yield_policy.py
Normal file
57
core/enrichment/yield_policy.py
Normal file
|
|
@ -0,0 +1,57 @@
|
||||||
|
"""Enrichment-worker yield policy: who pauses while the user's foreground
|
||||||
|
work is running.
|
||||||
|
|
||||||
|
Background enrichment workers share external API budgets with the foreground
|
||||||
|
pipelines — most painfully MusicBrainz (~1 req/s per IP), where a worker
|
||||||
|
grinding through the library can starve the import pipeline's per-track
|
||||||
|
lookups into multi-minute crawls (measured: ~4m15s/track vs the normal ~20s).
|
||||||
|
|
||||||
|
Policy (set with Boulder, 2026-06-06):
|
||||||
|
- downloads active -> EVERYTHING yields (post-processing touches every
|
||||||
|
metadata source: MusicBrainz, Spotify, iTunes,
|
||||||
|
Deezer, Discogs, Last.fm, Genius, ...)
|
||||||
|
- discovery active -> the API-contention five yield (discovery hammers
|
||||||
|
the track-matching sources only)
|
||||||
|
Workers the user explicitly resumed mid-yield are honored upstream (the
|
||||||
|
override set lives in web_server's loop, as does the user-paused bookkeeping).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
# Everything that yields during active downloads. listening-stats (talks only
|
||||||
|
# to the local media server) and repair (user-scheduled job runner, not a
|
||||||
|
# background API drip) intentionally keep running.
|
||||||
|
ALL_YIELD_WORKERS = (
|
||||||
|
'musicbrainz', 'audiodb', 'discogs', 'deezer',
|
||||||
|
'spotify-enrichment', 'itunes-enrichment', 'lastfm-enrichment',
|
||||||
|
'genius-enrichment', 'tidal-enrichment', 'qobuz-enrichment',
|
||||||
|
'amazon-enrichment', 'similar_artists', 'hydrabase', 'soulid',
|
||||||
|
)
|
||||||
|
|
||||||
|
# The sources discovery contends with (track matching APIs).
|
||||||
|
API_CONTENTION_WORKERS = frozenset({
|
||||||
|
'spotify-enrichment', 'itunes-enrichment', 'deezer', 'discogs', 'hydrabase',
|
||||||
|
})
|
||||||
|
|
||||||
|
# Discovery state phases that mean "nothing running" (idle or terminal).
|
||||||
|
_INACTIVE_PHASES = frozenset({'', 'idle', 'discovered', 'error', 'failed', 'cancelled'})
|
||||||
|
|
||||||
|
|
||||||
|
def worker_yield_reason(name: str, downloading: bool, discovering: bool) -> Optional[str]:
|
||||||
|
"""Why ``name`` should be paused right now, or None to run.
|
||||||
|
Downloads outrank discovery so the label reflects the stronger cause."""
|
||||||
|
if name not in ALL_YIELD_WORKERS:
|
||||||
|
return None
|
||||||
|
if downloading:
|
||||||
|
return 'downloads'
|
||||||
|
if discovering and name in API_CONTENTION_WORKERS:
|
||||||
|
return 'discovery'
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def discovery_state_active(state: dict) -> bool:
|
||||||
|
"""True when a per-playlist discovery state dict represents live work."""
|
||||||
|
phase = str((state or {}).get('phase', '') or '').lower()
|
||||||
|
return phase not in _INACTIVE_PHASES
|
||||||
|
|
@ -734,7 +734,12 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
logger.info("[Metadata Input] album_info: None (single track)")
|
logger.info("[Metadata Input] album_info: None (single track)")
|
||||||
|
_enhance_started = time.time()
|
||||||
enhance_file_metadata(file_path, context, artist_context, album_info, runtime=metadata_runtime)
|
enhance_file_metadata(file_path, context, artist_context, album_info, runtime=metadata_runtime)
|
||||||
|
# The enhancement block is the pipeline's biggest variable cost
|
||||||
|
# (external source lookups) and used to be a silent multi-minute
|
||||||
|
# gap in the log — always say how long it took.
|
||||||
|
logger.info(f"Metadata enhancement took {time.time() - _enhance_started:.1f}s")
|
||||||
except Exception as meta_err:
|
except Exception as meta_err:
|
||||||
import traceback
|
import traceback
|
||||||
pp_logger.info(f"[inner] Metadata enhancement FAILED for {context_key}: {meta_err}\n{traceback.format_exc()}")
|
pp_logger.info(f"[inner] Metadata enhancement FAILED for {context_key}: {meta_err}\n{traceback.format_exc()}")
|
||||||
|
|
|
||||||
|
|
@ -90,11 +90,19 @@ def _bounded_cache_set(cache, key, value, max_entries: int) -> None:
|
||||||
|
|
||||||
|
|
||||||
def _call_source_lookup(label: str, func, *args, **kwargs):
|
def _call_source_lookup(label: str, func, *args, **kwargs):
|
||||||
|
# Timed: these lookups are where import post-processing silently stalls
|
||||||
|
# when an upstream is sick or contended (measured 4m+/track against a
|
||||||
|
# degraded MusicBrainz). A slow source must NAME itself in the log.
|
||||||
|
started = time.time()
|
||||||
try:
|
try:
|
||||||
return func(*args, **kwargs)
|
return func(*args, **kwargs)
|
||||||
except _SOURCE_NETWORK_EXCEPTIONS as exc:
|
except _SOURCE_NETWORK_EXCEPTIONS as exc:
|
||||||
logger.warning("%s lookup failed (network): %s", label, exc)
|
logger.warning("%s lookup failed (network): %s", label, exc)
|
||||||
return None
|
return None
|
||||||
|
finally:
|
||||||
|
elapsed = time.time() - started
|
||||||
|
if elapsed > 2.0:
|
||||||
|
logger.warning("%s lookup took %.1fs — upstream slow or contended", label, elapsed)
|
||||||
|
|
||||||
|
|
||||||
SOURCE_TAG_CONFIG = {
|
SOURCE_TAG_CONFIG = {
|
||||||
|
|
|
||||||
60
tests/test_yield_policy.py
Normal file
60
tests/test_yield_policy.py
Normal file
|
|
@ -0,0 +1,60 @@
|
||||||
|
"""Tests for the enrichment-worker yield policy (downloads/discovery contention)."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from core.enrichment.yield_policy import (
|
||||||
|
ALL_YIELD_WORKERS,
|
||||||
|
API_CONTENTION_WORKERS,
|
||||||
|
discovery_state_active,
|
||||||
|
worker_yield_reason,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_downloads_pause_everything():
|
||||||
|
for name in ALL_YIELD_WORKERS:
|
||||||
|
assert worker_yield_reason(name, downloading=True, discovering=False) == 'downloads'
|
||||||
|
|
||||||
|
|
||||||
|
def test_discovery_pauses_only_the_contention_five():
|
||||||
|
for name in ALL_YIELD_WORKERS:
|
||||||
|
reason = worker_yield_reason(name, downloading=False, discovering=True)
|
||||||
|
if name in API_CONTENTION_WORKERS:
|
||||||
|
assert reason == 'discovery', name
|
||||||
|
else:
|
||||||
|
assert reason is None, name
|
||||||
|
|
||||||
|
|
||||||
|
def test_downloads_outrank_discovery_for_the_label():
|
||||||
|
assert worker_yield_reason('spotify-enrichment', True, True) == 'downloads'
|
||||||
|
|
||||||
|
|
||||||
|
def test_idle_pauses_nothing():
|
||||||
|
for name in ALL_YIELD_WORKERS:
|
||||||
|
assert worker_yield_reason(name, False, False) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_unknown_and_excluded_workers_never_yield():
|
||||||
|
# listening-stats (local media server only) and repair (user-scheduled job
|
||||||
|
# runner) intentionally keep running through downloads.
|
||||||
|
for name in ('listening-stats', 'repair', 'definitely-not-a-worker'):
|
||||||
|
assert worker_yield_reason(name, True, True) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_musicbrainz_yields_for_downloads_not_discovery():
|
||||||
|
# The case that motivated all of this: the MB worker starving the import
|
||||||
|
# pipeline's per-track lookups (~4m15s/track measured). It must yield to
|
||||||
|
# downloads — but keep running during discovery, which doesn't use MB.
|
||||||
|
assert worker_yield_reason('musicbrainz', True, False) == 'downloads'
|
||||||
|
assert worker_yield_reason('musicbrainz', False, True) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_discovery_state_active_phases():
|
||||||
|
assert discovery_state_active({'phase': 'discovering'})
|
||||||
|
assert discovery_state_active({'phase': 'Matching tracks...'})
|
||||||
|
assert not discovery_state_active({'phase': 'idle'})
|
||||||
|
assert not discovery_state_active({'phase': ''})
|
||||||
|
assert not discovery_state_active({'phase': 'discovered'})
|
||||||
|
assert not discovery_state_active({'phase': 'error'})
|
||||||
|
assert not discovery_state_active({'phase': 'cancelled'})
|
||||||
|
assert not discovery_state_active({})
|
||||||
|
assert not discovery_state_active(None)
|
||||||
|
|
@ -34911,6 +34911,23 @@ def _has_active_downloads():
|
||||||
# Track whether we auto-paused workers so we only resume ones we paused (not user-paused ones)
|
# Track whether we auto-paused workers so we only resume ones we paused (not user-paused ones)
|
||||||
_download_auto_paused = set()
|
_download_auto_paused = set()
|
||||||
_download_yield_override = set() # Workers the user explicitly resumed during downloads — don't re-pause
|
_download_yield_override = set() # Workers the user explicitly resumed during downloads — don't re-pause
|
||||||
|
_auto_yield_cause = {} # name -> 'downloads' | 'discovery' (status label for the UI)
|
||||||
|
|
||||||
|
|
||||||
|
def _has_active_discovery():
|
||||||
|
"""True while any playlist discovery is actively running (any platform)."""
|
||||||
|
from core.enrichment.yield_policy import discovery_state_active
|
||||||
|
try:
|
||||||
|
for states in (tidal_discovery_states, qobuz_discovery_states,
|
||||||
|
deezer_discovery_states, youtube_playlist_states,
|
||||||
|
beatport_chart_states, listenbrainz_playlist_states,
|
||||||
|
spotify_public_discovery_states, itunes_link_discovery_states):
|
||||||
|
for state in list(states.values()):
|
||||||
|
if discovery_state_active(state):
|
||||||
|
return True
|
||||||
|
except Exception as e:
|
||||||
|
logger.debug("active discovery check failed: %s", e)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
@ -35114,41 +35131,46 @@ def _emit_enrichment_status_loop():
|
||||||
'repair': lambda: repair_worker,
|
'repair': lambda: repair_worker,
|
||||||
}
|
}
|
||||||
|
|
||||||
# Workers to auto-pause during downloads (rate-limit sensitive services)
|
# Yield policy: downloads pause EVERYTHING (post-processing touches every
|
||||||
yield_workers = {
|
# metadata source — measured 4m+/track when MusicBrainz contended with its
|
||||||
'spotify-enrichment': lambda: spotify_enrichment_worker,
|
# own worker); discovery pauses the API-contention five. The name lists +
|
||||||
'lastfm-enrichment': lambda: lastfm_worker,
|
# decision live in core.enrichment.yield_policy (tested); this loop owns
|
||||||
'genius-enrichment': lambda: genius_worker,
|
# the pause/resume side effects and the user-override bookkeeping.
|
||||||
}
|
from core.enrichment.yield_policy import ALL_YIELD_WORKERS, worker_yield_reason
|
||||||
|
yield_workers = {name: workers[name] for name in ALL_YIELD_WORKERS if name in workers}
|
||||||
|
|
||||||
while not globals().get('IS_SHUTTING_DOWN', False):
|
while not globals().get('IS_SHUTTING_DOWN', False):
|
||||||
socketio.sleep(2)
|
socketio.sleep(2)
|
||||||
|
|
||||||
# Auto-pause/resume rate-limited workers during downloads
|
# Auto-pause/resume workers while foreground work runs
|
||||||
try:
|
try:
|
||||||
downloading = _has_active_downloads()
|
downloading = _has_active_downloads()
|
||||||
if not downloading:
|
discovering = _has_active_discovery()
|
||||||
_download_yield_override.clear() # Reset overrides when downloads finish
|
if not downloading and not discovering:
|
||||||
|
_download_yield_override.clear() # Reset overrides when work finishes
|
||||||
for name, get_w in yield_workers.items():
|
for name, get_w in yield_workers.items():
|
||||||
w = get_w()
|
w = get_w()
|
||||||
if w is None:
|
if w is None or not hasattr(w, 'paused'):
|
||||||
continue
|
continue
|
||||||
if downloading and not w.paused and name not in _download_yield_override:
|
reason = worker_yield_reason(name, downloading, discovering)
|
||||||
|
if reason and not w.paused and name not in _download_yield_override:
|
||||||
w.paused = True
|
w.paused = True
|
||||||
_download_auto_paused.add(name)
|
_download_auto_paused.add(name)
|
||||||
logger.debug(f"Auto-paused {name} during active downloads")
|
_auto_yield_cause[name] = reason
|
||||||
elif not downloading and name in _download_auto_paused:
|
logger.debug(f"Auto-paused {name} during active {reason}")
|
||||||
|
elif not reason and name in _download_auto_paused:
|
||||||
# Don't override an explicit user pause. If config says the worker
|
# Don't override an explicit user pause. If config says the worker
|
||||||
# was paused via the UI, leave it paused and just drop the auto-pause
|
# was paused via the UI, leave it paused and just drop the auto-pause
|
||||||
# marker so the next auto-pause/resume cycle behaves normally.
|
# marker so the next auto-pause/resume cycle behaves normally.
|
||||||
config_key = f"{name.replace('-', '_')}_paused"
|
config_key = f"{name.replace('-', '_')}_paused"
|
||||||
user_paused = config_manager.get(config_key, False)
|
user_paused = config_manager.get(config_key, False)
|
||||||
_download_auto_paused.discard(name)
|
_download_auto_paused.discard(name)
|
||||||
|
_auto_yield_cause.pop(name, None)
|
||||||
if not user_paused:
|
if not user_paused:
|
||||||
w.paused = False
|
w.paused = False
|
||||||
logger.debug(f"Auto-resumed {name} after downloads finished")
|
logger.debug(f"Auto-resumed {name} after foreground work finished")
|
||||||
else:
|
else:
|
||||||
logger.debug(f"Downloads finished but {name} remains paused by user")
|
logger.debug(f"Foreground work finished but {name} remains paused by user")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.debug(f"Error in download-yield check: {e}")
|
logger.debug(f"Error in download-yield check: {e}")
|
||||||
|
|
||||||
|
|
@ -35158,9 +35180,9 @@ def _emit_enrichment_status_loop():
|
||||||
if worker is None:
|
if worker is None:
|
||||||
continue
|
continue
|
||||||
status = worker.get_stats()
|
status = worker.get_stats()
|
||||||
# Flag workers that were auto-paused for downloads
|
# Flag workers that were auto-paused for foreground work
|
||||||
if name in _download_auto_paused:
|
if name in _download_auto_paused:
|
||||||
status['yield_reason'] = 'downloads'
|
status['yield_reason'] = _auto_yield_cause.get(name, 'downloads')
|
||||||
socketio.emit(f'enrichment:{name}', status)
|
socketio.emit(f'enrichment:{name}', status)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.debug(f"Error emitting {name} status: {e}")
|
logger.debug(f"Error emitting {name} status: {e}")
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue