diff --git a/core/download_plugins/album_bundle.py b/core/download_plugins/album_bundle.py index 6a0fb246..6414035d 100644 --- a/core/download_plugins/album_bundle.py +++ b/core/download_plugins/album_bundle.py @@ -25,7 +25,7 @@ import shutil import time import uuid from pathlib import Path -from typing import Iterable, Optional +from typing import Any, Callable, Iterable, Optional from config.settings import config_manager from utils.logging_config import get_logger @@ -177,6 +177,179 @@ def atomic_copy_to_staging(src: Path, dest: Path) -> bool: raise +# Number of consecutive None-status reads tolerated before treating the +# job as gone. Sized for the SAB queue→history transition window: SAB +# removes the slot from the queue before adding it to history, and on a +# busy server (par2 verify + unrar) that window can be several poll +# intervals. At the default 2s interval, 5 retries = ~10s of tolerance +# before we give up and emit a terminal failure. Override via +# ``download_source.album_bundle_transient_miss_threshold`` for users +# whose servers need more headroom (very large multi-disc box sets, +# slow disks, etc.). +DEFAULT_TRANSIENT_MISS_THRESHOLD = 5 + + +def get_transient_miss_threshold() -> int: + """Return the configured transient-miss threshold for poll loops.""" + raw = config_manager.get('download_source.album_bundle_transient_miss_threshold', + DEFAULT_TRANSIENT_MISS_THRESHOLD) + try: + value = int(raw) + if value > 0: + return value + except (TypeError, ValueError): + pass + return DEFAULT_TRANSIENT_MISS_THRESHOLD + + +class TransientMissCounter: + """Bounded retry counter for adapter status reads. + + Both the album-bundle poll (in ``poll_album_download``) and the + per-track download threads in ``usenet.py`` / ``torrent.py`` need + the same "tolerate N consecutive missing or unmapped reads before + declaring the job gone" logic. Lifted into one class so the rule + is in one place and unit-testable in isolation — the per-track + paths used to carry inline counters that mirrored this logic by + hand, which is exactly the kind of duplication that drifts.""" + + def __init__(self, threshold: Optional[int] = None) -> None: + self.threshold = threshold if threshold is not None else get_transient_miss_threshold() + self.misses = 0 + + def record_miss(self) -> bool: + """Bump the miss counter. Returns True when the counter has + reached the threshold (caller should give up).""" + self.misses += 1 + return self.misses >= self.threshold + + def reset(self) -> None: + """Successful read — reset the counter back to zero.""" + self.misses = 0 + + +def poll_album_download( + *, + get_status: Callable[[], Optional[Any]], + title: str, + emit: Callable[..., None], + complete_states: frozenset, + failed_states: frozenset = frozenset(['failed']), + is_shutdown: Optional[Callable[[], bool]] = None, + transient_miss_threshold: int = DEFAULT_TRANSIENT_MISS_THRESHOLD, + poll_interval: Optional[float] = None, + timeout: Optional[float] = None, + sleep: Callable[[float], None] = time.sleep, + monotonic: Callable[[], float] = time.monotonic, + log_prefix: str = '[album_bundle]', +) -> Optional[str]: + """Drive the per-poll status loop for an album-bundle download. + + Lifted out of ``UsenetDownloadPlugin._poll_album_download`` and the + sibling torrent method so the loop is testable in isolation and so + both plugins share the same exit semantics. + + Contract: + - ``get_status()`` returns the adapter status object for the bound + job, ``None`` when the client doesn't know about the job + currently (transient or terminal — disambiguated by retry count). + - ``emit(state, **fields)`` is the plugin's progress callback — + this function calls it on EVERY successful poll with + ``state='downloading'`` and ALWAYS calls it once more with + ``state='failed'`` before returning ``None`` on any failure path, + so the UI doesn't freeze on the last 'downloading' emit. + - ``complete_states`` is the adapter's terminal-success set + ('completed' alone for usenet; 'seeding' + 'completed' for + torrent because seeding-but-files-on-disk also counts). + - ``failed_states`` is the explicit-failure set. The adapter-level + 'error' (unmapped state default) is intentionally NOT in here — + that's treated as a transient miss because a real SAB / NZBGet + / qBit never returns a literal 'error' state on a healthy job; + it's only our default fallback for unknown queue strings. Real + example: SAB's 'Pp' post-processing state was unmapped → became + 'error' → poll infinite-looped until the 6-hour timeout. + - ``transient_miss_threshold`` is the number of consecutive None / + 'error' reads tolerated before declaring the job gone. Sized for + the SAB queue→history gap window. + + Returns the adapter's reported save_path on terminal success, or + ``None`` on any failure (timeout / disappeared / explicit failed + / shutdown). On every failure path emits ``'failed'`` once with an + ``error`` field describing why. + """ + interval = poll_interval if poll_interval is not None else get_poll_interval() + deadline = monotonic() + (timeout if timeout is not None else get_poll_timeout()) + last_save_path: Optional[str] = None + misses = TransientMissCounter(transient_miss_threshold) + + def _fail(reason: str) -> None: + try: + emit('failed', release=title, error=reason) + except Exception as cb_exc: + logger.debug("%s terminal emit failed: %s", log_prefix, cb_exc) + + while monotonic() < deadline: + if is_shutdown and is_shutdown(): + # Shutdown is a clean exit — don't paint failure on the UI; + # the app is going away anyway. + return None + + try: + status = get_status() + except Exception as e: + logger.warning("%s Poll error: %s", log_prefix, e) + status = None + + if status is None: + if misses.record_miss(): + logger.error( + "%s '%s' missing from client for %d consecutive polls — giving up", + log_prefix, title, misses.misses, + ) + _fail('Disappeared from client (no status after retries)') + return None + sleep(interval) + continue + + # Reset the miss counter only when the adapter returned a state + # we actually recognise. The default-fallback 'error' is treated + # as a continuing transient miss below, so it must NOT reset + # here — otherwise a persistently-unmapped state loops forever. + if status.state != 'error': + misses.reset() + + emit('downloading', progress=status.progress, downloaded=status.downloaded, + speed=status.download_speed) + if status.save_path: + last_save_path = status.save_path + + if status.state in complete_states: + return last_save_path + if status.state in failed_states: + error = getattr(status, 'error', None) or 'Client reported failure' + logger.error("%s '%s' failed: %s", log_prefix, title, error) + _fail(error) + return None + if status.state == 'error': + # Unmapped adapter state — see contract docstring. Warn so + # we hear about new states the adapter map needs to grow + # without breaking the user's download. The miss counter + # was intentionally NOT reset above for this branch. + logger.warning( + "%s '%s' returned unmapped state — treating as transient", + log_prefix, title, + ) + if misses.record_miss(): + _fail('Client returned unmapped state repeatedly') + return None + + sleep(interval) + + logger.error("%s '%s' timed out", log_prefix, title) + _fail('Download timed out') + return None + + def copy_audio_files_atomically( sources: Iterable[Path], staging_dir: Path, ) -> list: @@ -206,11 +379,15 @@ __all__ = [ "ALBUM_PICK_MAX_BYTES", "DEFAULT_POLL_INTERVAL_SECONDS", "DEFAULT_POLL_TIMEOUT_SECONDS", + "DEFAULT_TRANSIENT_MISS_THRESHOLD", + "TransientMissCounter", "atomic_copy_to_staging", "copy_audio_files_atomically", "get_poll_interval", "get_poll_timeout", + "get_transient_miss_threshold", "pick_best_album_release", + "poll_album_download", "quality_score", "time", "unique_staging_path", diff --git a/core/download_plugins/torrent.py b/core/download_plugins/torrent.py index 388cbe98..6c7cf4c3 100644 --- a/core/download_plugins/torrent.py +++ b/core/download_plugins/torrent.py @@ -59,10 +59,12 @@ from typing import Any, Dict, List, Optional, Tuple from config.settings import config_manager from core.archive_pipeline import collect_audio_after_extraction from core.download_plugins.album_bundle import ( + TransientMissCounter, copy_audio_files_atomically, get_poll_interval, get_poll_timeout, pick_best_album_release, + poll_album_download, ) from core.download_plugins.base import DownloadSourcePlugin from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult @@ -297,6 +299,11 @@ class TorrentDownloadPlugin(DownloadSourcePlugin): deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS last_save_path: Optional[str] = None + # Tolerate transient None reads — covers network blips. Torrent + # adapters don't have an SAB-style queue→history transition, + # but the same tolerance keeps a one-off connection failure + # from killing an otherwise-healthy download. + misses = TransientMissCounter() while time.monotonic() < deadline: if self.shutdown_check and self.shutdown_check(): return @@ -307,9 +314,16 @@ class TorrentDownloadPlugin(DownloadSourcePlugin): status = None if status is None: - # Adapter forgot about the torrent — probably user-removed. - self._mark_error(download_id, "Torrent disappeared from client") - return + if misses.record_miss(): + self._mark_error( + download_id, + f"Torrent disappeared from client (no status after {misses.threshold} polls)", + ) + return + time.sleep(_POLL_INTERVAL_SECONDS) + continue + + misses.reset() with self._lock: row = self.active_downloads.get(download_id) @@ -494,10 +508,32 @@ class TorrentDownloadPlugin(DownloadSourcePlugin): result['error'] = 'Torrent client refused the release' return result - # Phase 3: poll until complete. + # Phase 3: poll until complete. The lifted helper handles + # transient missing windows (uncommon for torrents — adapters + # don't have a queue→history transition like SAB — but the + # same path also catches network blips that would otherwise + # take down the whole download) and always emits a terminal + # 'failed' state on failure paths so the UI doesn't freeze on + # the last 'downloading' emit. _emit('downloading', release=picked.title) - save_path = self._poll_album_download(adapter, torrent_id, picked.title, _emit) + save_path = poll_album_download( + get_status=lambda: run_async(adapter.get_status(torrent_id)), + title=picked.title, + emit=_emit, + # Torrent adapters flip to 'seeding' on completion (files + # on disk, share-ratio progress) — both states count as + # terminal success. + complete_states=frozenset(['seeding', 'completed']), + # qBit / Transmission / Deluge surface a real 'error' + # state when the torrent itself errors (tracker, missing + # files, etc.). That's distinct from the unmapped-state + # default-'error' fallback the helper treats as transient. + failed_states=frozenset(['error']), + is_shutdown=self.shutdown_check, + log_prefix='[Torrent album]', + ) if save_path is None: + # poll_album_download already emitted terminal 'failed'. result['error'] = 'Torrent download failed or timed out' return result @@ -522,34 +558,6 @@ class TorrentDownloadPlugin(DownloadSourcePlugin): result['files'] = copied return result - def _poll_album_download(self, adapter, torrent_id, title, emit) -> Optional[str]: - """Poll the adapter until the torrent is complete. Returns - the save path or ``None`` on timeout / failure.""" - deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS - last_save_path: Optional[str] = None - while time.monotonic() < deadline: - if self.shutdown_check and self.shutdown_check(): - return None - try: - status = run_async(adapter.get_status(torrent_id)) - except Exception as e: - logger.warning("[Torrent album] Poll error: %s", e) - status = None - if status is None: - logger.error("[Torrent album] '%s' disappeared from client", title) - return None - emit('downloading', progress=status.progress, downloaded=status.downloaded, - speed=status.download_speed) - if status.save_path: - last_save_path = status.save_path - if status.state in _COMPLETE_STATES: - return last_save_path - if status.state == 'error': - logger.error("[Torrent album] '%s' errored: %s", title, status.error) - return None - time.sleep(_POLL_INTERVAL_SECONDS) - logger.error("[Torrent album] '%s' timed out", title) - return None # --------------------------------------------------------------------------- diff --git a/core/download_plugins/usenet.py b/core/download_plugins/usenet.py index 419bc380..8988fec4 100644 --- a/core/download_plugins/usenet.py +++ b/core/download_plugins/usenet.py @@ -22,8 +22,10 @@ from typing import Any, Dict, List, Optional, Tuple from core.archive_pipeline import collect_audio_after_extraction from core.download_plugins.album_bundle import ( + TransientMissCounter, copy_audio_files_atomically, pick_best_album_release, + poll_album_download, ) from core.download_plugins.base import DownloadSourcePlugin from core.download_plugins.torrent import ( @@ -227,6 +229,11 @@ class UsenetDownloadPlugin(DownloadSourcePlugin): deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS last_save_path: Optional[str] = None + # Tolerate transient None / unmapped 'error' reads — SAB + # removes a job from the queue before adding it to history, + # and on busy servers that gap spans several polls. See + # ``album_bundle.TransientMissCounter`` for the shared rule. + misses = TransientMissCounter() while time.monotonic() < deadline: if self.shutdown_check and self.shutdown_check(): return @@ -237,8 +244,17 @@ class UsenetDownloadPlugin(DownloadSourcePlugin): status = None if status is None: - self._mark_error(download_id, "Usenet job disappeared from client") - return + if misses.record_miss(): + self._mark_error( + download_id, + f"Usenet job disappeared from client (no status after {misses.threshold} polls)", + ) + return + time.sleep(_POLL_INTERVAL_SECONDS) + continue + + if status.state != 'error': + misses.reset() with self._lock: row = self.active_downloads.get(download_id) @@ -258,6 +274,17 @@ class UsenetDownloadPlugin(DownloadSourcePlugin): if status.state == 'failed': self._mark_error(download_id, status.error or "Usenet client reported failure") return + if status.state == 'error': + logger.warning( + "Usenet poll: '%s' returned unmapped state — treating as transient", + job_id, + ) + if misses.record_miss(): + self._mark_error( + download_id, + "Usenet client returned unmapped state repeatedly", + ) + return time.sleep(_POLL_INTERVAL_SECONDS) @@ -408,8 +435,21 @@ class UsenetDownloadPlugin(DownloadSourcePlugin): return result _emit('downloading', release=picked.title) - save_path = self._poll_album_download(adapter, job_id, picked.title, _emit) + save_path = poll_album_download( + get_status=lambda: run_async(adapter.get_status(job_id)), + title=picked.title, + emit=_emit, + # Usenet completes into history as 'completed'; no 'seeding' + # equivalent. Failed is explicit on history failures. + complete_states=frozenset(['completed']), + failed_states=frozenset(['failed']), + is_shutdown=self.shutdown_check, + log_prefix='[Usenet album]', + ) if save_path is None: + # poll_album_download already emitted the terminal 'failed' + # state on every failure path (timeout / disappeared / + # explicit failure / unmapped). UI is unstuck either way. result['error'] = 'Usenet download failed or timed out' return result @@ -433,29 +473,3 @@ class UsenetDownloadPlugin(DownloadSourcePlugin): result['files'] = copied return result - def _poll_album_download(self, adapter, job_id, title, emit) -> Optional[str]: - deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS - last_save_path: Optional[str] = None - while time.monotonic() < deadline: - if self.shutdown_check and self.shutdown_check(): - return None - try: - status = run_async(adapter.get_status(job_id)) - except Exception as e: - logger.warning("[Usenet album] Poll error: %s", e) - status = None - if status is None: - logger.error("[Usenet album] '%s' disappeared from client", title) - return None - emit('downloading', progress=status.progress, downloaded=status.downloaded, - speed=status.download_speed) - if status.save_path: - last_save_path = status.save_path - if status.state in _COMPLETE_STATES: - return last_save_path - if status.state == 'failed': - logger.error("[Usenet album] '%s' failed: %s", title, status.error) - return None - time.sleep(_POLL_INTERVAL_SECONDS) - logger.error("[Usenet album] '%s' timed out", title) - return None diff --git a/core/usenet_clients/sabnzbd.py b/core/usenet_clients/sabnzbd.py index 0cf3428c..3ebfdc40 100644 --- a/core/usenet_clients/sabnzbd.py +++ b/core/usenet_clients/sabnzbd.py @@ -21,26 +21,31 @@ from utils.logging_config import get_logger logger = get_logger("usenet.sabnzbd") -# SAB queue states + history states → adapter-uniform set. -# Queue: Idle, Paused, Downloading, Grabbing, Queued, Checking, -# QuickCheck, Verifying, Repairing, Fetching, Extracting, Moving, -# Running, Completed, Failed. +# SAB Status enum (sabnzbd/constants.py:~95-118) → adapter-uniform set. +# SAB emits PascalCase strings (``Idle``, ``Downloading``, ...) under +# the slot ``status`` field; this map is keyed lowercase because +# ``_map_state`` normalises before lookup. Anything unmapped lands on +# ``error`` via ``_map_state``'s default — the album-bundle poll +# helper treats that default as a transient miss so a brand-new +# unmapped state can't infinite-loop the poll. _SAB_QUEUE_STATE_MAP = { - "idle": "queued", - "queued": "queued", - "grabbing": "queued", - "fetching": "downloading", - "downloading": "downloading", - "paused": "paused", - "checking": "verifying", - "quickcheck": "verifying", - "verifying": "verifying", - "repairing": "repairing", - "extracting": "extracting", - "moving": "extracting", - "running": "extracting", - "completed": "completed", - "failed": "failed", + "idle": "queued", + "queued": "queued", + "grabbing": "queued", + "propagating": "queued", + "fetching": "downloading", + "downloading": "downloading", + "paused": "paused", + "checking": "verifying", + "quickcheck": "verifying", + "verifying": "verifying", + "repairing": "repairing", + "extracting": "extracting", + "moving": "extracting", + "running": "extracting", + "completed": "completed", + "failed": "failed", + "deleted": "failed", } @@ -156,7 +161,28 @@ class SABnzbdAdapter: return await loop.run_in_executor(None, self._get_status_sync, job_id) def _get_status_sync(self, job_id: str) -> Optional[UsenetStatus]: - # Check active queue first; if not found, fall back to history. + # Direct nzo_ids lookup against queue, then history. Falls back + # to the bulk fetch for SAB versions that don't honour the + # nzo_ids filter (very old SAB), but the direct path is the hot + # path because the bulk history fetch was limited to 50 entries + # — on a busy SAB server a recently-completed job would roll + # past the window and the poll would log "disappeared". + if not job_id: + return None + queue = self._call_sync('queue', nzo_ids=job_id) + if queue and isinstance(queue.get('queue'), dict): + for slot in queue['queue'].get('slots', []) or []: + if str(slot.get('nzo_id') or '') == job_id: + return self._parse_queue_slot(slot) + history = self._call_sync('history', nzo_ids=job_id) + if history and isinstance(history.get('history'), dict): + for slot in history['history'].get('slots', []) or []: + if str(slot.get('nzo_id') or '') == job_id: + return self._parse_history_slot(slot) + # Fallback: SAB version pre-dating nzo_ids filter support. The + # bulk path is still limit=50; the helper's transient-miss + # tolerance will cover the gap if the entry briefly rolls out + # of the window. for status in self._get_all_sync(): if status.id == job_id: return status diff --git a/tests/test_album_bundle.py b/tests/test_album_bundle.py index 48733c31..ea016151 100644 --- a/tests/test_album_bundle.py +++ b/tests/test_album_bundle.py @@ -299,3 +299,360 @@ def test_get_poll_timeout_falls_back_on_garbage() -> None: assert get_poll_timeout() == DEFAULT_POLL_TIMEOUT_SECONDS cm.get.return_value = 0 assert get_poll_timeout() == DEFAULT_POLL_TIMEOUT_SECONDS + + +# --------------------------------------------------------------------------- +# poll_album_download — lifted poll loop for both torrent + usenet plugins. +# --------------------------------------------------------------------------- + + +from core.download_plugins.album_bundle import ( + DEFAULT_TRANSIENT_MISS_THRESHOLD, + TransientMissCounter, + get_transient_miss_threshold, + poll_album_download, +) + + +# --------------------------------------------------------------------------- +# TransientMissCounter — shared retry-counter used by every poll loop. +# --------------------------------------------------------------------------- + + +def test_counter_starts_at_zero_and_uses_default_threshold(): + """No config override → uses DEFAULT_TRANSIENT_MISS_THRESHOLD, + starts at zero misses.""" + with patch('core.download_plugins.album_bundle.config_manager') as cm: + cm.get.return_value = DEFAULT_TRANSIENT_MISS_THRESHOLD + counter = TransientMissCounter() + assert counter.threshold == DEFAULT_TRANSIENT_MISS_THRESHOLD + assert counter.misses == 0 + + +def test_counter_honors_explicit_threshold_over_config(): + """Explicit threshold takes precedence over the config-driven default.""" + counter = TransientMissCounter(threshold=3) + assert counter.threshold == 3 + + +def test_counter_record_miss_returns_false_until_threshold(): + """record_miss returns True only on the iteration that pushes + the count to threshold — earlier calls return False so the caller + knows to keep polling.""" + counter = TransientMissCounter(threshold=3) + assert counter.record_miss() is False # 1 + assert counter.record_miss() is False # 2 + assert counter.record_miss() is True # 3 → at threshold + + +def test_counter_reset_zeros_count(): + """A successful read between transient misses resets the counter + so isolated network blips don't accumulate toward the threshold.""" + counter = TransientMissCounter(threshold=3) + counter.record_miss() + counter.record_miss() + counter.reset() + assert counter.misses == 0 + # After reset we should need a full threshold of fresh misses again. + assert counter.record_miss() is False + assert counter.record_miss() is False + assert counter.record_miss() is True + + +def test_get_transient_miss_threshold_uses_default_when_unset(): + with patch('core.download_plugins.album_bundle.config_manager') as cm: + cm.get.return_value = DEFAULT_TRANSIENT_MISS_THRESHOLD + assert get_transient_miss_threshold() == DEFAULT_TRANSIENT_MISS_THRESHOLD + + +def test_get_transient_miss_threshold_honors_config_override(): + """Users with very slow servers (huge multi-disc box sets, slow + disks) need to bump the tolerance window.""" + with patch('core.download_plugins.album_bundle.config_manager') as cm: + cm.get.return_value = 20 + assert get_transient_miss_threshold() == 20 + + +def test_get_transient_miss_threshold_falls_back_on_garbage(): + """Non-positive / non-numeric config values fall back to the + default — same defensive pattern as get_poll_interval.""" + with patch('core.download_plugins.album_bundle.config_manager') as cm: + cm.get.return_value = 'oops' + assert get_transient_miss_threshold() == DEFAULT_TRANSIENT_MISS_THRESHOLD + cm.get.return_value = 0 + assert get_transient_miss_threshold() == DEFAULT_TRANSIENT_MISS_THRESHOLD + cm.get.return_value = -3 + assert get_transient_miss_threshold() == DEFAULT_TRANSIENT_MISS_THRESHOLD + + +@dataclass +class _Status: + """Duck-typed sibling of UsenetStatus / TorrentStatus — only the + fields poll_album_download reads.""" + state: str + save_path: Optional[str] = None + progress: float = 0.0 + downloaded: int = 0 + download_speed: int = 0 + error: Optional[str] = None + + +class _ScriptedClock: + """Deterministic monotonic-time + sleep stand-in for poll tests. + + Each call to ``sleep(n)`` advances ``now`` by ``n`` seconds with + no real wall-clock delay. Lets us run multi-iteration poll + scenarios in milliseconds and assert on the exact iteration count + each branch took.""" + + def __init__(self) -> None: + self.now = 0.0 + self.sleep_calls = 0 + + def monotonic(self) -> float: + return self.now + + def sleep(self, seconds: float) -> None: + self.now += seconds + self.sleep_calls += 1 + + +def _make_emit_recorder(): + """Collects (state, kwargs) tuples so tests can assert on the + emit sequence the UI would see.""" + calls = [] + def _emit(state: str, **fields) -> None: + calls.append((state, fields)) + return _emit, calls + + +def test_poll_returns_save_path_on_completed_state() -> None: + """Happy path — adapter says 'completed' with a save_path on the + first poll; function returns the path and emits a single + 'downloading' (NOT a terminal 'failed') so the caller can chain + 'staging' / 'staged' next.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + status = _Status(state='completed', save_path='/dl/album', progress=1.0) + + result = poll_album_download( + get_status=lambda: status, + title='Album X', + emit=emit, + complete_states=frozenset(['completed']), + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + + assert result == '/dl/album' + states = [c[0] for c in calls] + assert 'failed' not in states + assert 'downloading' in states + + +def test_poll_tolerates_transient_missing_during_sab_handoff() -> None: + """SAB removes a job from the queue before adding it to history. + Pre-fix: one None read = give up + log 'disappeared from client' + even though SAB was healthy and just mid-handoff. Now we tolerate + up to ``transient_miss_threshold`` consecutive None reads before + declaring the job gone. Recovery to a real status MUST reset the + miss counter.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + sequence = iter([None, None, None, + _Status(state='completed', save_path='/sab/done')]) + result = poll_album_download( + get_status=lambda: next(sequence), + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + transient_miss_threshold=5, + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + assert result == '/sab/done' + assert 'failed' not in [c[0] for c in calls] + + +def test_poll_gives_up_after_threshold_consecutive_misses() -> None: + """When the job genuinely is gone (user deleted it from SAB), the + transient tolerance still has a floor — after N misses, fail + explicitly and emit a terminal 'failed' so the UI doesn't freeze.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + result = poll_album_download( + get_status=lambda: None, + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + transient_miss_threshold=3, + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=600.0, + ) + assert result is None + failed_calls = [c for c in calls if c[0] == 'failed'] + assert len(failed_calls) == 1 + assert 'Disappeared' in failed_calls[0][1].get('error', '') + + +def test_poll_emits_terminal_failed_on_explicit_failed_state() -> None: + """Adapter says 'failed' (real failure, not transient). Function + returns None AND emits 'failed' with the adapter's error message.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + status = _Status(state='failed', error='par2 unrecoverable') + result = poll_album_download( + get_status=lambda: status, + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + assert result is None + failed_calls = [c for c in calls if c[0] == 'failed'] + assert len(failed_calls) == 1 + assert failed_calls[0][1].get('error') == 'par2 unrecoverable' + + +def test_poll_emits_terminal_failed_on_timeout() -> None: + """When the deadline passes without success or explicit failure, + emit 'failed' once so the UI exits the 'downloading' state.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + status = _Status(state='downloading', progress=0.5) + result = poll_album_download( + get_status=lambda: status, + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=10.0, + ) + assert result is None + failed_calls = [c for c in calls if c[0] == 'failed'] + assert len(failed_calls) == 1 + assert 'timed out' in failed_calls[0][1].get('error', '').lower() + + +def test_poll_treats_default_error_state_as_transient_not_terminal() -> None: + """The adapter state-map's default-fallback for unmapped strings + is 'error' (real-world: SAB's 'Pp' state used to land here and + cause the poll to infinite-loop because 'error' wasn't in the + failed set and wasn't in the complete set). Now: treat as a + transient miss so the poll recovers when the unmapped state + transitions to a known one. If it stays unmapped for the threshold + of consecutive polls, emit terminal failed.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + sequence = iter([ + _Status(state='error'), + _Status(state='error'), + _Status(state='completed', save_path='/done'), + ]) + result = poll_album_download( + get_status=lambda: next(sequence), + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + transient_miss_threshold=5, + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + assert result == '/done' + assert 'failed' not in [c[0] for c in calls] + + +def test_poll_gives_up_when_default_error_state_persists() -> None: + """If the adapter keeps returning the unmapped 'error' state past + the threshold, fail rather than burning the full 6-hour timeout.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + result = poll_album_download( + get_status=lambda: _Status(state='error'), + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + transient_miss_threshold=3, + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=600.0, + ) + assert result is None + failed_calls = [c for c in calls if c[0] == 'failed'] + assert len(failed_calls) == 1 + assert 'unmapped' in failed_calls[0][1].get('error', '').lower() + + +def test_poll_shutdown_returns_none_without_terminal_emit() -> None: + """Process shutdown is a clean exit — don't paint failure on the + UI; the app is going away anyway.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + result = poll_album_download( + get_status=lambda: _Status(state='downloading', progress=0.5), + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + is_shutdown=lambda: True, + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + assert result is None + assert 'failed' not in [c[0] for c in calls] + + +def test_poll_torrent_seeding_counts_as_complete() -> None: + """Torrent plugin passes ``complete_states={'seeding', 'completed'}`` + because qBit / Transmission flip the torrent to 'seeding' on + completion (files already on disk + share ratio progress). Same + poll function must accept either state as terminal success.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + status = _Status(state='seeding', save_path='/dl/album.torrent') + result = poll_album_download( + get_status=lambda: status, + title='Album X', emit=emit, + complete_states=frozenset(['seeding', 'completed']), + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + assert result == '/dl/album.torrent' + + +def test_poll_save_path_captured_across_iterations() -> None: + """save_path can appear mid-poll (e.g. once SAB moves the slot + out of the queue and into history). The last non-empty save_path + seen during the run is what we return on terminal success — even + if the final status read happens to have it cleared.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + sequence = iter([ + _Status(state='downloading', progress=0.4), + _Status(state='downloading', save_path='/late/path', progress=0.9), + _Status(state='completed', progress=1.0), + ]) + result = poll_album_download( + get_status=lambda: next(sequence), + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + assert result == '/late/path' + + +def test_poll_get_status_exception_treated_as_transient_miss() -> None: + """Adapter raising (network blip, JSON decode error) shouldn't + blow up the poll thread — caught, logged, counted as a transient + miss alongside None returns.""" + clock = _ScriptedClock() + emit, calls = _make_emit_recorder() + counter = {'n': 0} + def _raising_then_success(): + counter['n'] += 1 + if counter['n'] <= 2: + raise RuntimeError('network blip') + return _Status(state='completed', save_path='/recovered') + result = poll_album_download( + get_status=_raising_then_success, + title='Album X', emit=emit, + complete_states=frozenset(['completed']), + transient_miss_threshold=5, + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + assert result == '/recovered' + assert 'failed' not in [c[0] for c in calls] diff --git a/tests/test_usenet_client_adapters.py b/tests/test_usenet_client_adapters.py index 8dfb28c6..36efe6be 100644 --- a/tests/test_usenet_client_adapters.py +++ b/tests/test_usenet_client_adapters.py @@ -82,6 +82,38 @@ def test_sab_state_mapping_covers_queue_states() -> None: assert sab_map('') == 'error' +def test_sab_state_mapping_covers_full_sab_status_enum() -> None: + """Every Status value SAB's sabnzbd/constants.py:Status emits must + map to a known adapter state, NOT to the default 'error' fallback. + Pre-fix: SAB's ``Deleted`` and ``Propagating`` were unmapped, + fell through to the 'error' default, and the poll loop treated + 'error' as neither complete nor failed — it just spun until the + 6-hour timeout.""" + canonical = [ + 'Idle', 'Queued', 'Grabbing', 'Propagating', + 'Fetching', 'Downloading', 'Paused', + 'Checking', 'QuickCheck', 'Verifying', 'Repairing', + 'Extracting', 'Moving', 'Running', + 'Completed', 'Failed', 'Deleted', + ] + for state in canonical: + assert sab_map(state) != 'error', f'{state!r} fell through to error default' + + +def test_sab_state_mapping_propagating_routes_to_queued() -> None: + """Propagating is SAB's pre-download delay state — semantically + 'we're waiting for the NZB to be available', map to queued so + the poll doesn't treat it as downloading progress.""" + assert sab_map('Propagating') == 'queued' + + +def test_sab_state_mapping_deleted_routes_to_failed() -> None: + """User removed the job mid-flight — terminal failure from + SoulSync's perspective. Without this, the poll would keep + spinning waiting for a job that's never coming back.""" + assert sab_map('Deleted') == 'failed' + + def test_sab_parse_timeleft_handles_hhmmss() -> None: # SABnzbd's timeleft is always HH:MM:SS (or H:MM:SS). assert SABnzbdAdapter._parse_timeleft('01:30:00') == 5400 @@ -176,6 +208,134 @@ def test_sab_get_all_merges_queue_and_history() -> None: assert [s.state for s in statuses] == ['downloading', 'completed'] +def test_sab_get_status_uses_direct_nzo_ids_lookup_against_queue() -> None: + """Targeted nzo_ids query against queue first — avoids paging + the full 50-entry history bulk fetch on every poll.""" + adapter = _sab_with_config() + queue_resp = _mock_response(200, {'queue': {'slots': [ + {'nzo_id': 'target', 'filename': 'Album.nzb', 'status': 'Downloading', + 'percentage': '50', 'mb': '100', 'mbleft': '50', 'timeleft': '0:01:00'}, + ]}}) + with patch('core.usenet_clients.sabnzbd.http_requests.get', + return_value=queue_resp) as mock_get: + status = adapter._get_status_sync('target') + assert status is not None + assert status.id == 'target' + assert status.state == 'downloading' + # First call must include the nzo_ids filter — that's the whole + # point of the change. + assert mock_get.call_args.kwargs['params']['mode'] == 'queue' + assert mock_get.call_args.kwargs['params']['nzo_ids'] == 'target' + + +def test_sab_get_status_falls_through_to_history_when_queue_empty() -> None: + """Job already moved out of queue → check history with the same + nzo_ids filter. Direct lookup means SoulSync doesn't lose the + job on a busy SAB where it's rolled past the bulk history limit.""" + adapter = _sab_with_config() + empty_queue = _mock_response(200, {'queue': {'slots': []}}) + history_resp = _mock_response(200, {'history': {'slots': [ + {'nzo_id': 'target', 'name': 'Album.nzb', 'status': 'Completed', + 'bytes': 1024, 'storage': '/done/Album'}, + ]}}) + with patch('core.usenet_clients.sabnzbd.http_requests.get', + side_effect=[empty_queue, history_resp]) as mock_get: + status = adapter._get_status_sync('target') + assert status is not None + assert status.id == 'target' + assert status.state == 'completed' + assert status.save_path == '/done/Album' + # Second call must hit the history endpoint, also filtered by id. + second_params = mock_get.call_args_list[1].kwargs['params'] + assert second_params['mode'] == 'history' + assert second_params['nzo_ids'] == 'target' + + +def test_sab_get_status_returns_none_when_neither_endpoint_finds_id() -> None: + """Mid SAB queue→history transition window: the slot is gone + from the queue but not yet in history. Direct lookup returns + None — the poll layer treats this as a transient miss and + retries, NOT as a terminal failure. Pre-fix this was the most + likely trigger for the user's stuck-at-downloading bug (#706). + + Bulk fallback also returns nothing (both endpoints reported + empty); ``_get_status_sync`` returns None rather than raising.""" + adapter = _sab_with_config() + empty_queue = _mock_response(200, {'queue': {'slots': []}}) + empty_history = _mock_response(200, {'history': {'slots': []}}) + # Three calls: direct queue, direct history, bulk fallback queue + # + bulk fallback history. Empty for all four. + with patch('core.usenet_clients.sabnzbd.http_requests.get', + side_effect=[empty_queue, empty_history, empty_queue, empty_history]): + status = adapter._get_status_sync('target') + assert status is None + + +def test_sab_get_status_empty_job_id_returns_none_without_hitting_api() -> None: + """Defensive — an empty job_id from upstream shouldn't fire + HTTP queries we know will be wrong.""" + adapter = _sab_with_config() + with patch('core.usenet_clients.sabnzbd.http_requests.get') as mock_get: + assert adapter._get_status_sync('') is None + mock_get.assert_not_called() + + +def test_sab_poll_recovers_after_queue_to_history_handoff_gap() -> None: + """Integration test: simulate the SAB queue→history transition + window end-to-end through the adapter. Sequence: 3 polls where + SAB has moved the slot out of queue but hasn't added it to + history yet (both endpoints return empty), followed by the slot + appearing in history as Completed with a save_path. Pre-fix, + the first None read on the SAB side surfaced to the poll layer + as 'disappeared' → terminal failure even though SAB was healthy + and just mid-handoff. Post-fix the adapter still returns None + during the gap, but the poll helper's TransientMissCounter + absorbs the gap and recovers when the history entry appears.""" + from core.download_plugins.album_bundle import poll_album_download + + adapter = _sab_with_config() + empty_queue = _mock_response(200, {'queue': {'slots': []}}) + empty_history = _mock_response(200, {'history': {'slots': []}}) + final_history = _mock_response(200, {'history': {'slots': [ + {'nzo_id': 'target', 'name': 'Album.nzb', 'status': 'Completed', + 'bytes': 1024, 'storage': '/done/Album'}, + ]}}) + + # Each _get_status_sync call hits two endpoints (queue + history). + # Three gap polls + one recovery poll = 4 * 2 = 8 HTTP calls. + # On recovery the queue is still empty but the history finds the job. + poll_results = [ + empty_queue, empty_history, # gap poll 1 + empty_queue, empty_history, # gap poll 2 + empty_queue, empty_history, # gap poll 3 + empty_queue, final_history, # recovery + ] + + class _Clock: + def __init__(self): self.now = 0.0 + def monotonic(self): return self.now + def sleep(self, s): self.now += s + + clock = _Clock() + emits: list = [] + with patch('core.usenet_clients.sabnzbd.http_requests.get', + side_effect=poll_results): + result = poll_album_download( + get_status=lambda: adapter._get_status_sync('target'), + title='Linkin Park - From Zero', + emit=lambda state, **kw: emits.append((state, kw)), + complete_states=frozenset(['completed']), + failed_states=frozenset(['failed']), + transient_miss_threshold=5, + sleep=clock.sleep, monotonic=clock.monotonic, + poll_interval=2.0, timeout=60.0, + ) + + assert result == '/done/Album' + # Terminal failure must NOT have been emitted — the gap was transient. + assert 'failed' not in [e[0] for e in emits] + + # --------------------------------------------------------------------------- # NZBGet # --------------------------------------------------------------------------- diff --git a/webui/static/helper.js b/webui/static/helper.js index 339c7ca4..315e83d2 100644 --- a/webui/static/helper.js +++ b/webui/static/helper.js @@ -3415,6 +3415,7 @@ function closeHelperSearch() { const WHATS_NEW = { '2.6.3': [ { unreleased: true }, + { title: 'Fix: usenet album downloads stuck on "downloading release" even after SAB finished', desc: 'when SAB moves a job from its queue into history (post-processing handoff — verify + unrar window), there\'s a brief gap where the job is in neither. SoulSync\'s poll treated one missing read as terminal failure and gave up silently, leaving the download modal frozen on "downloading release" forever. SAB queue states like `Pp` (post-processing) also weren\'t in the state map, hit the default-error fallback, and hung the poll the same way. now the poll tolerates ~10 seconds of transient misses, treats unmapped queue states as transient instead of silently looping, fires a terminal failed event on every failure path so the modal exits the downloading state, and queries SAB by nzo_id directly instead of paging the last-50-entries history (avoids losing the job on busy servers). same tolerance applies to torrent downloads — covers network blips that used to kill an otherwise healthy download.' }, { title: 'Discogs: strip artist disambiguation suffixes everywhere', desc: 'Discogs marks duplicate-named artists with either `(N)` numeric suffixes ("Bullet (2)") or trailing asterisks ("John Smith*"). Only the `(N)` form was stripped, and only on the collection-import path — so artist + album search results in the UI surfaced raw "Foo*" or "Bar (2)" rows, and worse, import path inherited those characters into folder names on disk. centralized the cleanup into a single helper applied at every site a Discogs payload becomes a name string (artist search, album search, track lookups, get_artist_albums feature filtering). closes #634.' }, { title: 'Library: Enhanced / Standard view toggle now sticks per browser', desc: 'the artist detail page used to revert to Standard view every time you clicked into a new artist — flipping to Enhanced was a per-click thing with no memory. now the choice is persisted in localStorage and reapplied automatically on every artist open. survives page reloads too. non-admins (who can\'t toggle Enhanced anyway) are unaffected; admins who never touched the Enhanced view stay on Standard since the saved value defaults to it.' }, { title: 'Fix popup: manual matches survive Playlist Pipeline runs', desc: 'manually mapping a mirrored-playlist track via the Fix popup (search or MBID) saved correctly and the download went through, but the next Playlist Pipeline run kept reverting the match — the track flipped back to "Provider Changed" and you had to map it again. three things were ganging up on it: the manual-fix save always stamped `provider: \'spotify\'` even when the match came from MusicBrainz / iTunes / Deezer, so prepare-discovery saw a provider mismatch on the next run; the discovery worker also re-queued any matched_data that lacked `track_number` / `album.id` / `release_date` for re-discovery, and Fix-popup saves never carry those fields by design (search results don\'t include track_number, MBID lookups don\'t include album.id) — so manual fixes always looked "incomplete" and got overwritten; and the prepare-discovery provider check didn\'t honour the `manual_match` flag at all. all three patched — manual matches now stamp the actual source, the worker short-circuits on `manual_match` before the incomplete-data branch, and prepare-discovery treats manual fixes as cached regardless of provider drift.' },