From 2fb142dded68b8fd2cf577b8861899ac7f3ff677 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sun, 28 Jun 2026 13:41:38 -0700 Subject: [PATCH] jellyfin scan: page the bulk fetch so the no-progress watchdog can't false-stall MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Discord (DXP4800 NAS, 7148 tracks): library updates kept dying with "Update appears stuck — no progress for 300s (last phase: Fetching all tracks in bulk...)". not actually hung. root cause: the bulk track/album fetch used a single 10000-item page, so a whole library came back in ONE request that emitted NO progress while in flight. the watchdog (database_update_health) kills a job with no progress for 300s — so on a slow server that one silent request tripped it even though it was alive, not stuck. raising the timeout cap only buys the silent request more rope; a bigger library or slower disk just needs a higher number. the per-batch progress line also only ran when there was a NEXT page, so a sub-page-size library reported nothing at all. fix: extract a pure paginate_all_items seam (core/library/bulk_paginate.py) that pages in 1000s and reports progress after EVERY page — so the watchdog is fed on a cadence set by page size, not library size, and can't starve mid-fetch no matter how big the library. both Jellyfin bulk loops (tracks + albums, same defect) now route through it. preserves the failure-shrink resilience (halve to a floor, then give up). does NOT change what's fetched — same query, fields, items. note: changes nothing about WHICH tracks come back; only how they're paged + that every page reports. keep the raised cap on dev as a margin — this is the actual fix. Plex/Navidrome don't share the pattern (checked). 9 seam tests incl. the watchdog-feed invariant (progress count scales with N/page_size, never one call for the whole library) + the sub-page regression + failure-shrink. 467 jellyfin/library tests green, ruff clean. --- core/jellyfin_client.py | 102 +++++++--------------------- core/library/bulk_paginate.py | 93 +++++++++++++++++++++++++ tests/library/test_bulk_paginate.py | 95 ++++++++++++++++++++++++++ 3 files changed, 212 insertions(+), 78 deletions(-) create mode 100644 core/library/bulk_paginate.py create mode 100644 tests/library/test_bulk_paginate.py diff --git a/core/jellyfin_client.py b/core/jellyfin_client.py index fa4f61b0..663c212f 100644 --- a/core/jellyfin_client.py +++ b/core/jellyfin_client.py @@ -5,6 +5,7 @@ from datetime import datetime import json from utils.logging_config import get_logger from config.settings import config_manager +from core.library.bulk_paginate import paginate_all_items # Shared dataclasses live in the neutral media_server package — every # server client used to define a near-identical XTrackInfo / @@ -512,12 +513,8 @@ class JellyfinClient(MediaServerClient): try: # SIMPLIFIED APPROACH: Fetch all tracks, then all albums separately (robust and fast) logger.info("Fetching all tracks in bulk...") - all_tracks = [] - start_index = 0 - limit = 10000 - consecutive_failures = 0 - - while True: + + def _fetch_tracks_page(start_index, limit): params = { 'ParentId': self.music_library_id, 'IncludeItemTypes': 'Audio', @@ -528,41 +525,19 @@ class JellyfinClient(MediaServerClient): 'StartIndex': start_index, 'Limit': limit } - response = self._make_request(f'/Users/{self.user_id}/Items', params) - - if not response: - consecutive_failures += 1 - # Wait before retrying — the server may still be processing the timed-out request - time.sleep(5) - if limit > 1000: - limit = limit // 2 - consecutive_failures = 0 # Reset — give the smaller batch a fair chance - logger.warning(f"Track fetch failed - reducing batch size to {limit}") - continue - elif consecutive_failures >= 2: - logger.warning("Multiple track fetch failures at minimum batch size - stopping") - break - else: - logger.warning("Track fetch failed at minimum batch size - retrying once") - continue + return response.get('Items', []) if response else None # None = failed page + + # Page in modest chunks so progress is reported every page — a single + # huge silent request used to trip the 300s no-progress watchdog on + # slow servers even though it was alive (see bulk_paginate docstring). + all_tracks = paginate_all_items( + _fetch_tracks_page, + report_progress=self._progress_callback, + label="tracks", + on_retry_wait=lambda: time.sleep(5), + ) - consecutive_failures = 0 - batch_tracks = response.get('Items', []) - if not batch_tracks: - break - - all_tracks.extend(batch_tracks) - - if len(batch_tracks) < limit: - break - - start_index += limit - progress_msg = f"Fetched {len(all_tracks)} tracks so far..." - logger.info(f" {progress_msg} (batch size: {limit})") - if self._progress_callback: - self._progress_callback(progress_msg) - # Group tracks by album ID for instant lookup self._track_cache = {} for track_data in all_tracks: @@ -578,12 +553,8 @@ class JellyfinClient(MediaServerClient): # STEP 2: Fetch all albums in bulk (same proven pattern) logger.info("Fetching all albums in bulk...") - all_albums = [] - start_index = 0 - limit = 10000 - consecutive_failures = 0 - - while True: + + def _fetch_albums_page(start_index, limit): params = { 'ParentId': self.music_library_id, 'IncludeItemTypes': 'MusicAlbum', @@ -594,41 +565,16 @@ class JellyfinClient(MediaServerClient): 'StartIndex': start_index, 'Limit': limit } - response = self._make_request(f'/Users/{self.user_id}/Items', params) - - if not response: - consecutive_failures += 1 - # Wait before retrying — the server may still be processing the timed-out request - time.sleep(5) - if limit > 1000: - limit = limit // 2 - consecutive_failures = 0 # Reset — give the smaller batch a fair chance - logger.warning(f"Album fetch failed - reducing batch size to {limit}") - continue - elif consecutive_failures >= 2: - logger.warning("Multiple album fetch failures at minimum batch size - stopping") - break - else: - logger.warning("Album fetch failed at minimum batch size - retrying once") - continue + return response.get('Items', []) if response else None # None = failed page + + all_albums = paginate_all_items( + _fetch_albums_page, + report_progress=self._progress_callback, + label="albums", + on_retry_wait=lambda: time.sleep(5), + ) - consecutive_failures = 0 - batch_albums = response.get('Items', []) - if not batch_albums: - break - - all_albums.extend(batch_albums) - - if len(batch_albums) < limit: - break - - start_index += limit - progress_msg = f"Fetched {len(all_albums)} albums so far..." - logger.info(f" {progress_msg} (batch size: {limit})") - if self._progress_callback: - self._progress_callback(progress_msg) - # Group albums by artist ID for instant lookup self._album_cache = {} for album_data in all_albums: diff --git a/core/library/bulk_paginate.py b/core/library/bulk_paginate.py new file mode 100644 index 00000000..de423219 --- /dev/null +++ b/core/library/bulk_paginate.py @@ -0,0 +1,93 @@ +"""Paginate a bulk media-server fetch while feeding the no-progress watchdog. + +The library scan fetches every track/album by paging a server API. The DB-update +watchdog (``core/database_update_health.py``) kills a job that reports no progress +for 300s. The old Jellyfin fetch used a single 10 000-item page, so a whole +library came back in ONE request that emitted NO progress while it was in flight — +on a slow server that single request exceeded 300s and the watchdog declared the +job "stuck" even though it was alive, not hung (Discord: DXP4800 NAS, 7148 tracks, +"Fetching all tracks in bulk…"). + +``paginate_all_items`` pages at a size chosen so progress is emitted on a cadence +set by the PAGE SIZE, not the library size — the watchdog is fed every page, so it +can never starve mid-fetch regardless of how big the library is. It is pure: all +I/O lives in the injected ``fetch_page``, so the pagination + progress + failure- +shrink logic is unit-testable without a server. + +This does NOT change WHAT is fetched (same query, same fields, same items) — only +how it's paged and that every page reports progress (the old loop skipped progress +on the final/only page, which is the entire bug for a sub-page-size library). +""" + +from __future__ import annotations + +from typing import Any, Callable, List, Optional + +# Page size for bulk library fetches. Small enough that a single request stays +# well under the 300s no-progress watchdog even on a slow NAS, and that progress +# is reported every page. NOT a performance knob — a resilience/observability one. +DEFAULT_PAGE_SIZE = 1000 + +# Floor the failure-shrink can reach before giving up — a server that can't return +# even this many items in one request is genuinely struggling. +DEFAULT_MIN_PAGE_SIZE = 250 + + +def paginate_all_items( + fetch_page: Callable[[int, int], Optional[List[Any]]], + *, + report_progress: Optional[Callable[[str], None]] = None, + label: str = "items", + page_size: int = DEFAULT_PAGE_SIZE, + min_page_size: int = DEFAULT_MIN_PAGE_SIZE, + on_retry_wait: Optional[Callable[[], None]] = None, +) -> List[Any]: + """Page through ``fetch_page(start_index, limit)`` until the server is drained. + + ``fetch_page`` returns the page's items (a list, possibly empty = end), or + ``None`` to signal a FAILED request (timeout/error) — on failure the page size + is halved down to ``min_page_size`` and retried, then abandoned after two + consecutive failures at the floor. + + Progress is reported after EVERY non-empty page (including the final/only one), + so a no-progress watchdog is fed on a cadence set by ``page_size`` — never by + the total library size. Returns every item gathered. + """ + items: List[Any] = [] + start_index = 0 + limit = page_size + consecutive_failures = 0 + + while True: + batch = fetch_page(start_index, limit) + + if batch is None: # failed request + consecutive_failures += 1 + if on_retry_wait is not None: + on_retry_wait() + if limit > min_page_size: + limit = max(min_page_size, limit // 2) + consecutive_failures = 0 # give the smaller batch a fair chance + continue + if consecutive_failures >= 2: + break # struggling at the floor — stop with what we have + continue + + consecutive_failures = 0 + if not batch: + break # drained + + items.extend(batch) + # Feed the watchdog on EVERY page — this is the line the old loop only ran + # when there was a *next* page, so a sub-page-size library reported nothing. + if report_progress is not None: + report_progress(f"Fetched {len(items)} {label} so far...") + + if len(batch) < limit: + break # last (partial) page + start_index += limit + + return items + + +__all__ = ["paginate_all_items", "DEFAULT_PAGE_SIZE", "DEFAULT_MIN_PAGE_SIZE"] diff --git a/tests/library/test_bulk_paginate.py b/tests/library/test_bulk_paginate.py new file mode 100644 index 00000000..3310a5ab --- /dev/null +++ b/tests/library/test_bulk_paginate.py @@ -0,0 +1,95 @@ +"""Bulk-fetch pagination seam — pages a server fetch while feeding the no-progress +watchdog every page (the DXP4800 "Fetching all tracks in bulk… stuck 300s" bug). +Pure: a fake fetch_page stands in for the server, a list records progress calls.""" + +import math + +from core.library.bulk_paginate import ( + paginate_all_items, + DEFAULT_PAGE_SIZE, +) + + +def _server(total, *, fail_at=None, fail_times=0): + """A fake server holding `total` items. Returns the right page slice for + (start_index, limit). Optionally returns None (a failed request) the first + `fail_times` times it's called at offset `fail_at`.""" + items = list(range(total)) + state = {"fails": 0} + + def fetch_page(start_index, limit): + if fail_at is not None and start_index == fail_at and state["fails"] < fail_times: + state["fails"] += 1 + return None + return items[start_index:start_index + limit] + + return fetch_page + + +def test_returns_every_item_across_pages(): + out = paginate_all_items(_server(7148), page_size=1000) + assert out == list(range(7148)) + + +def test_progress_fed_every_page_not_once_for_whole_library(): + # The watchdog-feed invariant: progress count scales with N/page_size, so the + # gap between progress beats is one page — never the whole library. A single + # 7148-track library used to emit ZERO progress (one 10k page) and stall. + calls = [] + paginate_all_items(_server(7148), page_size=1000, report_progress=calls.append) + assert len(calls) == math.ceil(7148 / 1000) == 8 + + +def test_sub_page_library_still_reports_progress(): + # Regression: the old loop skipped progress on the final/only page, so a + # library smaller than one page reported nothing → watchdog starved. + calls = [] + out = paginate_all_items(_server(500), page_size=1000, report_progress=calls.append) + assert out == list(range(500)) + assert len(calls) == 1 # was 0 before the fix + + +def test_exact_multiple_of_page_size(): + calls = [] + out = paginate_all_items(_server(2000), page_size=1000, report_progress=calls.append) + assert out == list(range(2000)) + assert len(calls) == 2 # two full pages, both reported + + +def test_empty_library(): + calls = [] + out = paginate_all_items(_server(0), page_size=1000, report_progress=calls.append) + assert out == [] + assert calls == [] + + +def test_no_progress_callback_is_safe(): + assert paginate_all_items(_server(2500), page_size=1000) == list(range(2500)) + + +def test_failed_page_shrinks_then_succeeds(): + # First request at offset 0 fails once; the helper halves the page size, retries, + # and still returns everything — the slow-server resilience path. + waits = [] + out = paginate_all_items( + _server(1500, fail_at=0, fail_times=1), + page_size=1000, min_page_size=250, + on_retry_wait=lambda: waits.append(1), + ) + assert out == list(range(1500)) + assert waits == [1] # waited once before the retry + + +def test_gives_up_after_repeated_failures_at_floor(): + # A page that always fails at the floor must terminate (not loop forever) and + # return what was gathered — here nothing, since it fails on the first page. + def always_fail(_start, _limit): + return None + out = paginate_all_items(always_fail, page_size=250, min_page_size=250) + assert out == [] + + +def test_default_page_size_is_watchdog_safe(): + # A guard on the constant itself: the default must be far below a library size + # that would fit in one request, so progress is always paged. + assert DEFAULT_PAGE_SIZE <= 1000