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.
This commit is contained in:
imSp4rky
2026-10-23 19:47:12 -06:00
parent d8663b0dad
commit dfc99783b5
8 changed files with 174 additions and 34 deletions
+5
View File
@@ -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 <host> 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.
+8
View File
@@ -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"
@@ -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
+16 -7
View File
@@ -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}")
+6 -5
View File
@@ -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)
+16 -2
View File
@@ -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",
)
+38 -17
View File
@@ -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:
+35 -3
View File
@@ -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)