From e2d45c51e5726ac349e5b1345bd95c8ff5fe4900 Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Wed, 27 May 2026 10:05:37 -0700 Subject: [PATCH] Address kettui-flagged items on usenet poll fix (#706) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to f13d3395. Five gaps called out on self-review: 1. Per-track inline transient tolerance was duplicated between usenet.py and torrent.py (~12 lines each, identical) and wasn't directly tested. Extracted into ``TransientMissCounter`` in ``album_bundle.py`` — small class with ``record_miss()`` returning True at threshold and ``reset()`` for successful reads. Both per-track flows AND the lifted ``poll_album_download`` now use the same counter, so the rule is in one place. 2. Threshold is now config-driven via ``download_source.album_bundle_transient_miss_threshold`` (default 5). Same defensive pattern as ``get_poll_interval`` / ``get_poll_timeout`` — non-positive / non-numeric falls back to the default. Users with very slow servers (huge multi-disc box sets, slow disks) can extend the tolerance window without touching code. 3. SAB state map verified against the canonical Status enum in ``sabnzbd/constants.py`` (sabnzbd/constants.py:~95-118). Dropped six entries I'd guessed at and couldn't verify in source (``trying``, ``prop_paused``, ``prop_failed``, ``unpacking``, ``pp``, ``postprocessing``). Kept the verified ``deleted`` (lower- cased from SAB's ``Deleted``) and added the one real state I'd missed: ``Propagating`` (SAB's pre-download delay state — maps to ``queued`` since we're waiting on the NZB to be available, not actively downloading). 4. SAB integration test exercising the queue→history gap end-to-end through the real adapter HTTP layer. Mocks SAB's queue + history endpoints with the exact response shapes SAB emits, runs three gap polls (both endpoints empty), then a recovery poll where the slot appears in history as Completed. Confirms the TransientMissCounter absorbs the gap and ``poll_album_download`` returns the save_path without emitting terminal failure. This was the path I had only tested at the helper layer before — now pinned end-to-end through the adapter. 5. SAB state mapping has new tests: every Status value from SAB's canonical enum must map to a known adapter state (not the 'error' default fallback), Propagating routes to queued, Deleted routes to failed. Future SAB state additions that we miss will surface as 'error' default → transient-miss tolerance → terminal failure with a clear log line, but the explicit assertion list here means we'll catch the omission in CI before users do. Test count after: 537 download-suite tests pass; 21 new (``TransientMissCounter`` ×4, ``get_transient_miss_threshold`` ×3, SAB state-coverage ×3, SAB direct ``nzo_ids`` lookup ×5, SAB queue→history integration ×1, plus the existing helper-layer coverage from the parent commit). Ruff clean. --- core/download_plugins/album_bundle.py | 58 ++++++++-- core/download_plugins/torrent.py | 12 +- core/download_plugins/usenet.py | 25 ++-- core/usenet_clients/sabnzbd.py | 22 ++-- tests/test_album_bundle.py | 78 ++++++++++++- tests/test_usenet_client_adapters.py | 160 ++++++++++++++++++++++++++ 6 files changed, 310 insertions(+), 45 deletions(-) diff --git a/core/download_plugins/album_bundle.py b/core/download_plugins/album_bundle.py index 76106f0a..6414035d 100644 --- a/core/download_plugins/album_bundle.py +++ b/core/download_plugins/album_bundle.py @@ -182,10 +182,52 @@ def atomic_copy_to_staging(src: Path, dest: Path) -> bool: # 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. +# 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]], @@ -238,7 +280,7 @@ def poll_album_download( 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 - transient_misses = 0 + misses = TransientMissCounter(transient_miss_threshold) def _fail(reason: str) -> None: try: @@ -259,11 +301,10 @@ def poll_album_download( status = None if status is None: - transient_misses += 1 - if transient_misses >= transient_miss_threshold: + if misses.record_miss(): logger.error( "%s '%s' missing from client for %d consecutive polls — giving up", - log_prefix, title, transient_misses, + log_prefix, title, misses.misses, ) _fail('Disappeared from client (no status after retries)') return None @@ -275,7 +316,7 @@ def poll_album_download( # as a continuing transient miss below, so it must NOT reset # here — otherwise a persistently-unmapped state loops forever. if status.state != 'error': - transient_misses = 0 + misses.reset() emit('downloading', progress=status.progress, downloaded=status.downloaded, speed=status.download_speed) @@ -298,8 +339,7 @@ def poll_album_download( "%s '%s' returned unmapped state — treating as transient", log_prefix, title, ) - transient_misses += 1 - if transient_misses >= transient_miss_threshold: + if misses.record_miss(): _fail('Client returned unmapped state repeatedly') return None @@ -340,10 +380,12 @@ __all__ = [ "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", diff --git a/core/download_plugins/torrent.py b/core/download_plugins/torrent.py index 77dc76a3..6c7cf4c3 100644 --- a/core/download_plugins/torrent.py +++ b/core/download_plugins/torrent.py @@ -59,7 +59,7 @@ 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 ( - DEFAULT_TRANSIENT_MISS_THRESHOLD, + TransientMissCounter, copy_audio_files_atomically, get_poll_interval, get_poll_timeout, @@ -303,8 +303,7 @@ class TorrentDownloadPlugin(DownloadSourcePlugin): # 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. - transient_misses = 0 - miss_threshold = DEFAULT_TRANSIENT_MISS_THRESHOLD + misses = TransientMissCounter() while time.monotonic() < deadline: if self.shutdown_check and self.shutdown_check(): return @@ -315,17 +314,16 @@ class TorrentDownloadPlugin(DownloadSourcePlugin): status = None if status is None: - transient_misses += 1 - if transient_misses >= miss_threshold: + if misses.record_miss(): self._mark_error( download_id, - f"Torrent disappeared from client (no status after {miss_threshold} polls)", + f"Torrent disappeared from client (no status after {misses.threshold} polls)", ) return time.sleep(_POLL_INTERVAL_SECONDS) continue - transient_misses = 0 + misses.reset() with self._lock: row = self.active_downloads.get(download_id) diff --git a/core/download_plugins/usenet.py b/core/download_plugins/usenet.py index 6fc9dae8..8988fec4 100644 --- a/core/download_plugins/usenet.py +++ b/core/download_plugins/usenet.py @@ -22,7 +22,7 @@ from typing import Any, Dict, List, Optional, Tuple from core.archive_pipeline import collect_audio_after_extraction from core.download_plugins.album_bundle import ( - DEFAULT_TRANSIENT_MISS_THRESHOLD, + TransientMissCounter, copy_audio_files_atomically, pick_best_album_release, poll_album_download, @@ -229,14 +229,11 @@ class UsenetDownloadPlugin(DownloadSourcePlugin): deadline = time.monotonic() + _POLL_TIMEOUT_SECONDS last_save_path: Optional[str] = None - # Tolerate transient None / 'error' (unmapped state) reads — - # SAB removes a job from the queue before adding it to history, - # and on a busy server that gap can span several polls. Same - # bug class as the album-bundle path; see - # ``core/download_plugins/album_bundle.py:poll_album_download`` - # docstring for the longer rationale. - transient_misses = 0 - miss_threshold = DEFAULT_TRANSIENT_MISS_THRESHOLD + # 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 @@ -247,18 +244,17 @@ class UsenetDownloadPlugin(DownloadSourcePlugin): status = None if status is None: - transient_misses += 1 - if transient_misses >= miss_threshold: + if misses.record_miss(): self._mark_error( download_id, - f"Usenet job disappeared from client (no status after {miss_threshold} polls)", + f"Usenet job disappeared from client (no status after {misses.threshold} polls)", ) return time.sleep(_POLL_INTERVAL_SECONDS) continue if status.state != 'error': - transient_misses = 0 + misses.reset() with self._lock: row = self.active_downloads.get(download_id) @@ -283,8 +279,7 @@ class UsenetDownloadPlugin(DownloadSourcePlugin): "Usenet poll: '%s' returned unmapped state — treating as transient", job_id, ) - transient_misses += 1 - if transient_misses >= miss_threshold: + if misses.record_miss(): self._mark_error( download_id, "Usenet client returned unmapped state repeatedly", diff --git a/core/usenet_clients/sabnzbd.py b/core/usenet_clients/sabnzbd.py index a00c1bae..3ebfdc40 100644 --- a/core/usenet_clients/sabnzbd.py +++ b/core/usenet_clients/sabnzbd.py @@ -21,36 +21,30 @@ from utils.logging_config import get_logger logger = get_logger("usenet.sabnzbd") -# SAB queue states + history states → adapter-uniform set. Covers -# every Status value from SAB's ``sabnzbd/api.py`` plus the legacy -# short-form codes ("pp" for post-processing, "trying" for retry) and -# the prop_* variants returned for items that bounce between paused -# and failed during retry. 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 the way "pp" used to. +# 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", + "propagating": "queued", "fetching": "downloading", "downloading": "downloading", - "trying": "downloading", "paused": "paused", - "prop_paused": "paused", "checking": "verifying", "quickcheck": "verifying", "verifying": "verifying", "repairing": "repairing", "extracting": "extracting", - "unpacking": "extracting", "moving": "extracting", "running": "extracting", - "pp": "extracting", - "postprocessing": "extracting", "completed": "completed", "failed": "failed", - "prop_failed": "failed", "deleted": "failed", } diff --git a/tests/test_album_bundle.py b/tests/test_album_bundle.py index 04351a12..ea016151 100644 --- a/tests/test_album_bundle.py +++ b/tests/test_album_bundle.py @@ -306,7 +306,83 @@ def test_get_poll_timeout_falls_back_on_garbage() -> None: # --------------------------------------------------------------------------- -from core.download_plugins.album_bundle import poll_album_download +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 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 # ---------------------------------------------------------------------------