From dfc99783b5a35327001e44e6b1b878c87aa7f469 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Fri, 23 Oct 2026 19:47:12 -0600 Subject: [PATCH] fix(downloader): report spent segment retries in one line without the URL The session message names the host, not the signed URL, and a segment that fails every attempt prints one line with no traceback. API job errors come from the exception. --- docs/guide/troubleshooting.md | 5 ++ tests/core/test_session_retry_log.py | 8 +++ .../test_retry_exhaustion_report.py | 50 +++++++++++++++++ unshackle/commands/dl.py | 23 +++++--- unshackle/core/api/download_manager.py | 11 ++-- unshackle/core/downloaders/__init__.py | 18 +++++- unshackle/core/downloaders/requests.py | 55 +++++++++++++------ unshackle/core/session.py | 38 ++++++++++++- 8 files changed, 174 insertions(+), 34 deletions(-) create mode 100644 tests/downloaders/test_retry_exhaustion_report.py diff --git a/docs/guide/troubleshooting.md b/docs/guide/troubleshooting.md index e292ce45..bccaed80 100644 --- a/docs/guide/troubleshooting.md +++ b/docs/guide/troubleshooting.md @@ -258,6 +258,11 @@ Each command reports how many files it removed and how much space it freed, then **Inspect:** set `debug_requests: true` and operate unshackle with `-d` to show the underlying `urllib3` / `rnet` / `httpx` logs. In the JSON debug log, filter for `service_call` operations to see which request failed. If you use a proxy, also read [Proxies & VPN](proxies-and-vpn.md). +!!! example "A segment download fails after every retry" + **Symptom:** the download stops with a line like `MaxRetriesError: GET failed: RequestError (stream closed because of a broken pipe), gave up after 5 failed attempts over 30s`. + + **Inspect:** the connection to the CDN host dropped on every attempt. This is almost always the network, the VPN, or the proxy, not the service. Run the download again: unshackle keeps the segments it already has. Over HTTP/2, all segments share one connection, so one drop fails every segment in flight. Set `network.http1_only: true` (see [Network & proxy](../reference/configuration/network.md)) to spread them over separate connections. Operate unshackle with `-d` to see each retry as a `segment_retry` entry in the JSON debug log. + !!! example "The wrong titles/metadata keep coming back" **Symptom:** a service returns outdated titles even after the source changed. diff --git a/tests/core/test_session_retry_log.py b/tests/core/test_session_retry_log.py index 0b56cbf2..6dbb7793 100644 --- a/tests/core/test_session_retry_log.py +++ b/tests/core/test_session_retry_log.py @@ -27,3 +27,11 @@ def test_retry_log_counts_only_real_retries(rate_limited, caplog, max_retries): retries = [r.getMessage() for r in caplog.records if "Retrying" in r.getMessage()] assert retries == [f"429 Too Many Requests(/x). Retrying... ({n}/{max_retries})" for n in range(1, max_retries + 1)] + + +@pytest.mark.parametrize(("max_retries", "attempts"), [(0, ""), (2, " after 3 attempts")]) +def test_spent_budget_names_the_attempts_and_host_not_the_url(rate_limited, max_retries, attempts): + with pytest.raises(MaxRetriesError) as raised: + rate_limited.get("https://example.invalid/x?hmac=secret", max_retries=max_retries) + + assert str(raised.value) == f"GET example.invalid failed{attempts}: 429 Too Many Requests" diff --git a/tests/downloaders/test_retry_exhaustion_report.py b/tests/downloaders/test_retry_exhaustion_report.py new file mode 100644 index 00000000..729f759b --- /dev/null +++ b/tests/downloaders/test_retry_exhaustion_report.py @@ -0,0 +1,50 @@ +"""A segment that fails every attempt reports the attempts and the cause, without the signed URL.""" + +import sys + +import pytest + +from unshackle.commands.dl import dl +from unshackle.core.downloaders.requests import download, retries_exhausted +from unshackle.core.session import MaxRetriesError, RnetSession, failure_reason + +downloader = sys.modules[download.__module__] + +SIGNED_URL = "https://cdn.example.invalid/dvt2=exp=1~hmac=deadbeef/seg/40_000.mp4a" +RNET_BROKEN_PIPE = ( + f'is_request error: wreq::Error {{ kind: Request, url: "{SIGNED_URL}", source: hyper::Error(Io, ' + 'Custom { kind: BrokenPipe, error: "stream closed because of a broken pipe" }) }' +) + + +class RequestError(Exception): + """Stands in for rnet.RequestError, which carries the URL in its text.""" + + +def test_failure_reason_keeps_the_rnet_error_text_and_drops_the_url(): + reason = failure_reason(MaxRetriesError("m", cause=RequestError(RNET_BROKEN_PIPE))) + + assert reason == "RequestError (stream closed because of a broken pipe)" + + +def test_spent_retries_report_one_line_without_the_url(monkeypatch, tmp_path): + session = RnetSession() + calls = [] + + def get(url, **kwargs): + calls.append(kwargs.get("max_retries")) + raise MaxRetriesError("GET cdn.example.invalid failed: x", cause=RequestError(RNET_BROKEN_PIPE)) + + monkeypatch.setattr(session, "get", get) + monkeypatch.setattr(downloader, "retry_sleep", lambda exc, attempts: 0) + + with pytest.raises(MaxRetriesError) as raised: + for _ in download(url=SIGNED_URL, save_path=tmp_path / "seg", session=session): + pass + + assert calls == [0] * downloader.MAX_ATTEMPTS + assert retries_exhausted(raised.value) + assert not retries_exhausted(ValueError("boom")) + header, line = dl.failure_lines(raised.value) + assert f"gave up after {downloader.MAX_ATTEMPTS} failed attempts" in line + assert "hmac" not in line diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index 7ee5441b..e399be9e 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -53,7 +53,13 @@ from unshackle.core.constants import ( context_settings, ) from unshackle.core.credential import Credential -from unshackle.core.downloaders import default_max_workers, format_speed, parse_speed_limit, set_speed_limit +from unshackle.core.downloaders import ( + default_max_workers, + format_speed, + parse_speed_limit, + retries_exhausted, + set_speed_limit, +) from unshackle.core.drm import DRM_T, ClearKeyCENC, MonaLisa, PlayReady, Widevine, cbcs, own_kids, real_kids, verify from unshackle.core.events import events from unshackle.core.export_name import ExportNamer, move_export, plural, season_episodes, title_label @@ -1277,9 +1283,9 @@ class dl: # report which weren't available without parsing the console output. See SkippedSubtitle. self.skipped_subtitles: list[SkippedSubtitle] = [] # result() catches a download-worker failure, reports it, and returns rather than - # re-raising (so the CLI still exits cleanly). Flag it so an embedding caller (the API + # re-raising (so the CLI still exits cleanly). Keep it so an embedding caller (the API # worker) can tell the title did not complete instead of reading it as a success. - self.download_failed: bool = False + self.download_error: Optional[BaseException] = None if not config.output_template: raise click.ClickException( @@ -3309,11 +3315,11 @@ class dl: self.log_export_summary() return except Exception as e: # noqa - # Reported and swallowed (no re-raise) so the CLI exits cleanly; flag it so the + # Reported and swallowed (no re-raise) so the CLI exits cleanly; keep it so the # API worker sees the title failed rather than completing with no output. - self.download_failed = True + self.download_error = e console.print(Padding(Group(*self.failure_lines(e)), (1, 5))) - if not isinstance(e, click.ClickException): + if not isinstance(e, click.ClickException) and not retries_exhausted(e): console.print_exception() if self.debug_logger: @@ -4482,10 +4488,13 @@ class dl: """Return the lines that report a failed download. A ClickException is a deliberate stop with a message for the user, so it gets no - "unexpected error" line, and the caller prints no traceback for it. + "unexpected error" line, and the caller prints no traceback for it. Neither does a + request that failed on every retry: it gets its retry notes instead. """ if isinstance(e, click.ClickException): return [":x: Download Failed...", f" {e.format_message()}"] + if retries_exhausted(e): + return [":x: Download Failed...", f" {type(e).__name__}: {e}, {'; '.join(e.__notes__)}"] lines = [":x: Download Failed...", f" {type(e).__name__}: {e}"] if (returncode := getattr(e, "returncode", None)) is not None: lines.append(f" Binary call failed, Process exit code: {returncode}") diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py index fbf21b76..ad35c861 100644 --- a/unshackle/core/api/download_manager.py +++ b/unshackle/core/api/download_manager.py @@ -20,7 +20,7 @@ from typing import Any, Callable, Dict, List, Optional from unshackle.core.api.events import bus, publish_service_event from unshackle.core.api.sanitize import sanitize_log from unshackle.core.utils.click_types import AUDIO_CODEC_LIST, QUALITY_LIST, VIDEO_CODEC_LIST -from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_path, redact_text +from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_all, redact_path, redact_text log = logging.getLogger("download_manager") @@ -693,10 +693,11 @@ def perform_download( raise # dl.result() catches a download-worker exception, reports it, but returns normally (exit 0). - # It sets download_failed in that case, so the job isn't reported as completed with no output. - if getattr(dl_instance, "download_failed", False): - detail = (stdout_capture.getvalue() + stderr_capture.getvalue())[-200:].strip() - raise APIError(APIErrorCode.WORKER_ERROR, "download worker failed: " + (detail or "see logs")) + # It keeps the error in download_error, so the job isn't reported as completed with no output. + if (error := getattr(dl_instance, "download_error", None)) is not None: + # segment URLs carry signed CDN tokens (hmac, session IDs) that redact_text does not match + detail = redact_all(" ".join(line.strip() for line in dl_instance.failure_lines(error)[1:])) + raise APIError(APIErrorCode.WORKER_ERROR, f"download worker failed: {detail}") if progress_callback: skipped_subs = getattr(dl_instance, "skipped_subtitles", None) diff --git a/unshackle/core/downloaders/__init__.py b/unshackle/core/downloaders/__init__.py index 390fab82..635debd2 100644 --- a/unshackle/core/downloaders/__init__.py +++ b/unshackle/core/downloaders/__init__.py @@ -1,3 +1,17 @@ -from .requests import default_max_workers, format_speed, parse_speed_limit, requests, set_speed_limit +from .requests import ( + default_max_workers, + format_speed, + parse_speed_limit, + requests, + retries_exhausted, + set_speed_limit, +) -__all__ = ("requests", "default_max_workers", "format_speed", "parse_speed_limit", "set_speed_limit") +__all__ = ( + "requests", + "default_max_workers", + "format_speed", + "parse_speed_limit", + "retries_exhausted", + "set_speed_limit", +) diff --git a/unshackle/core/downloaders/requests.py b/unshackle/core/downloaders/requests.py index a40312eb..fe6a6b07 100644 --- a/unshackle/core/downloaders/requests.py +++ b/unshackle/core/downloaders/requests.py @@ -27,11 +27,12 @@ from requests.adapters import HTTPAdapter, Retry from rich import filesize from unshackle.core.constants import DOWNLOAD_CANCELLED, DownloadCancelled -from unshackle.core.utilities import get_debug_logger, get_extension +from unshackle.core.utilities import get_debug_logger, get_extension, log_event from unshackle.core.utils import run_proxy MAX_ATTEMPTS = 5 RETRY_WAIT = 2 +GAVE_UP_NOTE = "gave up after" # a cut that still delivered bytes is resumable and refreshes the attempt budget; bound it # so a server that sends a little and closes, over and over, still terminates MAX_RESUMES = 50 @@ -324,15 +325,12 @@ class AdaptiveWorkerController: def permanent_status(exc: BaseException) -> Optional[int]: """HTTP status of ``exc`` when it is a client error that no retry can fix. - Both session types raise requests' HTTPError; RnetSession wraps it in MaxRetriesError - as ``__cause__``, so unwrap one level like retry_sleep does. 408 and 429 do not count: - they are transient, and a fan-out that trips a per-client rate limit can still succeed - over one sequential connection. + 408 and 429 do not count: they are transient, and a fan-out that trips a per-client rate + limit can still succeed over one sequential connection. """ - response = getattr(exc, "response", None) - if response is None: - response = getattr(getattr(exc, "__cause__", None), "response", None) - status = getattr(response, "status_code", None) + from unshackle.core.session import failure_response + + status = getattr(failure_response(exc), "status_code", None) if isinstance(status, int) and 400 <= status < 500 and status not in (408, 429): return status return None @@ -358,12 +356,9 @@ def retry_sleep(exc: Exception, attempts: int) -> float: else exponential growth from RETRY_WAIT with jitter so concurrently rate-limited segments do not retry in lockstep. Capped at ``session.MAX_BACKOFF`` either way. """ - from unshackle.core.session import MAX_BACKOFF + from unshackle.core.session import MAX_BACKOFF, failure_response - response = getattr(exc, "response", None) - if response is None: - # RnetSession wraps forcelist failures in MaxRetriesError with the HTTPError as __cause__ - response = getattr(getattr(exc, "__cause__", None), "response", None) + response = failure_response(exc) if response is not None: try: retry_after = response.headers.get("Retry-After") @@ -387,6 +382,11 @@ def retry_sleep(exc: Exception, attempts: int) -> float: return min(backoff + random.uniform(0, backoff * 0.1), MAX_BACKOFF) +def retries_exhausted(exc: BaseException) -> bool: + """Whether ``exc`` is the error download() re-raised after it spent every attempt.""" + return any(str(note).startswith(GAVE_UP_NOTE) for note in getattr(exc, "__notes__", ())) + + def is_requests_session(session: Any) -> bool: """Whether the HTTP session is a standard requests.Session (it has ``resp.raw``).""" return isinstance(session, Session) @@ -814,6 +814,9 @@ def download( item_range = has_range_header(kwargs) attempts = 1 + # attempts resets on progress, so the give-up note counts failures on their own + failures = 0 + started = _time() written = 0 resumes = 0 request_limit: Optional[int] = None @@ -1141,11 +1144,13 @@ def download( # a superseded loser's error must not retry or kill the batch if claimed is not None and claimed(): return + failures += 1 cancelled = DOWNLOAD_CANCELLED.is_set() or (abort is not None and abort.is_set()) if cancelled or attempts == MAX_ATTEMPTS: # cancel/abort is not an error, but a genuine retry-exhaustion is, so surface it # so the caller fails fast instead of hitting a missing segment at merge time if not cancelled: + exc.add_note(f"{GAVE_UP_NOTE} {failures} failed attempts over {_time() - started:.0f}s") raise return if not part_mode: @@ -1169,6 +1174,13 @@ def download( except Exception: pass delay = retry_sleep(exc, max(1, attempts)) + from unshackle.core.session import failure_reason + + log_event( + "segment_retry", + message=f"{failure_reason(exc)}, retry {max(1, attempts)}/{MAX_ATTEMPTS - 1} in {delay:.1f}s", + context={"file": save_path.name, "failures": failures, "resumes": resumes}, + ) if abort is not None: # interruptible nap: backoff can reach MAX_BACKOFF, and teardown always # sets the batch abort, so a parked worker must wake and exit at once @@ -1315,8 +1327,13 @@ def mp_worker(queue: Any, kwargs: dict[str, Any]) -> None: for event in requests(**kwargs): queue.put(event) except Exception as exc: - summary = " ".join(f"{type(exc).__name__}: {exc}".split()) - queue.put({"__mp_error__": f"{summary}\n{traceback.format_exc()}"}) + queue.put( + { + "__mp_error__": " ".join(f"{type(exc).__name__}: {exc}".split()), + "__mp_traceback__": traceback.format_exc(), + "__mp_notes__": [str(note) for note in getattr(exc, "__notes__", ())], + } + ) finally: queue.put({"__mp_done__": True}) @@ -1432,7 +1449,11 @@ def download_multiprocess( if "__mp_error__" in event: for p in procs: p.terminate() - raise RuntimeError(f"segment download child failed: {event['__mp_error__']}") + log_event("segment_child_failed", level="ERROR", message=event.get("__mp_traceback__", "")) + child_error = RuntimeError(f"segment download child failed: {event['__mp_error__']}") + for note in event.get("__mp_notes__", ()): + child_error.add_note(note) + raise child_error if "total" in event: continue if "downloaded" in event: diff --git a/unshackle/core/session.py b/unshackle/core/session.py index 8e5e1c3a..80dc6b43 100644 --- a/unshackle/core/session.py +++ b/unshackle/core/session.py @@ -6,6 +6,7 @@ import http import logging import math import random +import re import sys import time from collections.abc import Iterator, MutableMapping @@ -531,6 +532,34 @@ class MaxRetriesError(Exception): self.__cause__ = cause +# the innermost io error text in rnet's wreq::Error debug form, e.g. "stream closed because of a broken pipe" +RNET_ERROR_TEXT_RE = re.compile(r'error: "([^"]+)"') + + +def failure_response(exc: BaseException) -> Any: + """The HTTP response of a status failure, or None. + + RnetSession wraps a forcelist HTTPError in MaxRetriesError as ``__cause__``, so this unwraps one level. + """ + response = getattr(exc, "response", None) + return response if response is not None else getattr(exc.__cause__, "response", None) + + +def failure_reason(exc: BaseException) -> str: + """Name a request failure without its URL, which can carry signed tokens. + + Gives the HTTP status of a status failure, else the exception class and, for an rnet error, + its innermost error text. + """ + inner = exc.__cause__ or exc + response = failure_response(exc) + status = getattr(response, "status_code", None) + if isinstance(status, int): + return f"{status} {getattr(response, 'reason', '') or ''}".strip() + texts = RNET_ERROR_TEXT_RE.findall(str(inner)) + return f"{type(inner).__name__} ({texts[-1]})" if texts else type(inner).__name__ + + class RnetSession: """TLS-fingerprinted HTTP session powered by rnet (Rust/BoringSSL). @@ -868,12 +897,12 @@ class RnetSession: if response.status_code not in self.status_forcelist: return response last_exception = HTTPError(f"Received status code: {response.status_code}", response=response) - failure = f"{response.status_code} {response.reason}" + failure = failure_reason(last_exception) except self.catch_exceptions as e: last_exception = e response = None - failure = e.__class__.__name__ + failure = failure_reason(e) if attempt < max_retries: self.log.warning(f"{failure}({urlparse(url).path}). Retrying... ({attempt + 1}/{max_retries})") @@ -883,7 +912,10 @@ class RnetSession: else: break - raise MaxRetriesError(f"Max retries exceeded for {method} {url}", cause=last_exception) + attempts = f" after {max_retries + 1} attempts" if max_retries else "" + raise MaxRetriesError( + f"{method_upper} {parsed_origin.hostname} failed{attempts}: {failure}", cause=last_exception + ) def get(self, url: str, **kwargs: Any) -> RnetResponse: return self.request("GET", url, **kwargs)