Retry rate-limited HLS segments

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.52.12 - 2026-08-14
+
+- Added retry/backoff handling for rate-limited HLS segment and key downloads, including provider CDN `HTTP 429` responses.
+- Added environment controls for native HLS segment retries.
+- Added regression coverage for segment downloads that recover after a temporary rate-limit response.
+
 ## 0.52.11 - 2026-08-14
 
 - Updated Anikoto to `5.0.2`, improving server discovery and adding shorter provider request timeouts.

README.md

@@ -67,6 +67,7 @@ Useful environment variables:
 - `KAIZOKU_DOWNLOAD_DIR=/downloads` for the default library output path.
 - `KAIZOKU_REMOTE_PATH_ROOTS=/downloads,/jellyfin/tv,/jellyfin/movies` to limit remote filesystem browsing; comma-separated and platform path separators are accepted.
 - `KAIZOKU_MODE=sub`, `KAIZOKU_QUALITY=best`, and `KAIZOKU_DEBUG=1` for runtime defaults and diagnostics.
+- `KAIZOKU_SEGMENT_RETRIES=8`, `KAIZOKU_SEGMENT_RETRY_DELAY=1.25`, and `KAIZOKU_SEGMENT_RETRY_MAX_DELAY=15` to tune retry/backoff behavior for native HLS segment downloads.
 
 ## Docker
 
@@ -99,7 +100,7 @@ Keep `./.kaizoku` mounted for production instances. That directory contains the
 
 ## Download Flow
 
-Kaizoku stores provider-backed show IDs as `provider:id`, for example `anikoto:some-show-slug`. Queue jobs resolve the episode source through `providers/bridge.js`, then `provider_downloader.py` downloads the media with `ffmpeg` into a staging directory. If a provider returns a master HLS playlist, Kaizoku selects the highest-bandwidth variant before starting `ffmpeg`. When a media playlist uses extensionless CDN segments, Kaizoku skips direct ffmpeg and uses a StrawVerse-style segment downloader: it fetches the media playlist, downloads and concatenates segments itself, strips short PNG wrappers when present, then remuxes the local transport stream to MP4. Direct ffmpeg attempts also have a timeout guard so stalled HLS inputs can fall back cleanly. Each episode is written as a temporary `.mp4.part` file and moved into place only after the download succeeds, so failed fallback attempts do not leave broken final MP4 files behind.
+Kaizoku stores provider-backed show IDs as `provider:id`, for example `anikoto:some-show-slug`. Queue jobs resolve the episode source through `providers/bridge.js`, then `provider_downloader.py` downloads the media with `ffmpeg` into a staging directory. If a provider returns a master HLS playlist, Kaizoku selects the highest-bandwidth variant before starting `ffmpeg`. When a media playlist uses extensionless CDN segments, Kaizoku skips direct ffmpeg and uses a StrawVerse-style segment downloader: it fetches the media playlist, downloads and concatenates segments itself, strips short PNG wrappers when present, retries temporary HTTP failures such as `429 Too Many Requests`, then remuxes the local transport stream to MP4. Direct ffmpeg attempts also have a timeout guard so stalled HLS inputs can fall back cleanly. Each episode is written as a temporary `.mp4.part` file and moved into place only after the download succeeds, so failed fallback attempts do not leave broken final MP4 files behind.
 
 If the primary provider cannot list, resolve, or download a requested episode, Kaizoku searches the same title on the remaining providers and tries the matching episode there. Existing finalization code moves staged files into the configured library layout, preserving data already present in production download folders.
 

VERSION

@@ -1 +1 @@
-0.52.11
+0.52.12

provider_downloader.py

@@ -9,6 +9,7 @@ import re
 import shutil
 import subprocess
 import sys
+import time
 from urllib.parse import urljoin, urlparse
 import urllib.request
 from pathlib import Path
@@ -17,6 +18,7 @@ from pathlib import Path
 PROJECT_ROOT = Path(__file__).resolve().parent
 BRIDGE = PROJECT_ROOT / "providers" / "bridge.js"
 PROVIDERS = ("anikoto", "anineko", "pahe")
+RETRYABLE_HTTP_STATUS = {408, 425, 429, 500, 502, 503, 504}
 
 
 def clean_component(value, default="Anime"):
@@ -218,6 +220,61 @@ def fetch_bytes(url, headers=None, timeout=30):
         return response.read()
 
 
+def int_env(name, default, minimum=0):
+    try:
+        value = int(str(os.environ.get(name, default)).strip())
+    except (TypeError, ValueError):
+        return default
+    return max(minimum, value)
+
+
+def float_env(name, default, minimum=0.0):
+    try:
+        value = float(str(os.environ.get(name, default)).strip())
+    except (TypeError, ValueError):
+        return default
+    return max(minimum, value)
+
+
+def retry_after_seconds(exc):
+    try:
+        value = exc.headers.get("Retry-After")
+    except Exception:
+        return None
+    if not value:
+        return None
+    text = str(value).strip()
+    try:
+        return max(0.0, float(text))
+    except ValueError:
+        return None
+
+
+def fetch_bytes_with_retries(url, headers=None, timeout=30, retries=None, retry_label="request"):
+    attempts = int_env("KAIZOKU_SEGMENT_RETRIES", 8, minimum=0) if retries is None else max(0, int(retries))
+    base_delay = float_env("KAIZOKU_SEGMENT_RETRY_DELAY", 1.25, minimum=0.0)
+    max_delay = float_env("KAIZOKU_SEGMENT_RETRY_MAX_DELAY", 15.0, minimum=0.0)
+    for attempt in range(attempts + 1):
+        try:
+            return fetch_bytes(url, headers=headers, timeout=timeout)
+        except urllib.error.HTTPError as exc:
+            retryable = exc.code in RETRYABLE_HTTP_STATUS
+            if not retryable or attempt >= attempts:
+                raise
+            header_delay = retry_after_seconds(exc)
+            delay = header_delay if header_delay is not None else base_delay * (2 ** attempt)
+            if max_delay:
+                delay = min(delay, max_delay)
+            print(
+                f"{retry_label} got HTTP {exc.code}; retrying in {delay:.1f}s "
+                f"({attempt + 1}/{attempts}).",
+                file=sys.stderr,
+                flush=True,
+            )
+            if delay:
+                time.sleep(delay)
+
+
 def fetch_text(url, headers=None, timeout=30):
     return fetch_bytes(url, headers=headers, timeout=timeout).decode("utf-8", errors="replace")
 
@@ -393,12 +450,22 @@ def download_hls_segments(stream, input_url, target, partial, episode_number=Non
         last_percent = -1
         with ts_file.open("wb") as joined:
             for index, segment in enumerate(segments, start=1):
-                data = fetch_bytes(segment["url"], headers=stream.get("headers") or {}, timeout=60)
+                data = fetch_bytes_with_retries(
+                    segment["url"],
+                    headers=stream.get("headers") or {},
+                    timeout=60,
+                    retry_label=f"Episode {episode_number} segment {index}/{len(segments)}",
+                )
                 data = strip_png_header(data)
                 key_url = segment.get("key_url")
                 if key_url:
                     if key_url not in key_cache:
-                        key_cache[key_url] = fetch_bytes(key_url, headers=stream.get("headers") or {}, timeout=30)
+                        key_cache[key_url] = fetch_bytes_with_retries(
+                            key_url,
+                            headers=stream.get("headers") or {},
+                            timeout=30,
+                            retry_label=f"Episode {episode_number} segment key",
+                        )
                     data = decrypt_aes128_segment(data, key_cache[key_url], segment.get("iv"))
                 if not data:
                     raise RuntimeError(f"Segment {index} was empty.")

test_app.py

@@ -8,6 +8,7 @@ import sys
 import tempfile
 import threading
 import unittest
+import urllib.error
 from http import HTTPStatus
 from pathlib import Path
 from unittest import mock
@@ -4356,6 +4357,33 @@ seg-3.ts
             )
         )
 
+    def test_fetch_bytes_with_retries_recovers_from_segment_rate_limit(self):
+        error = urllib.error.HTTPError(
+            "https://cdn.example/video/raw-segment",
+            429,
+            "Too Many Requests",
+            hdrs={},
+            fp=None,
+        )
+
+        try:
+            with mock.patch.object(
+                provider_downloader,
+                "fetch_bytes",
+                side_effect=[error, b"video"],
+            ) as fetch_bytes, mock.patch.object(provider_downloader.time, "sleep") as sleep:
+                data = provider_downloader.fetch_bytes_with_retries(
+                    "https://cdn.example/video/raw-segment",
+                    retries=1,
+                    retry_label="segment",
+                )
+        finally:
+            error.close()
+
+        self.assertEqual(data, b"video")
+        self.assertEqual(fetch_bytes.call_count, 2)
+        sleep.assert_called_once()
+
 
 class DockerfilePackagingTests(unittest.TestCase):
     def test_provider_updates_module_is_copied_into_image(self):