diff --git a/tests/core/test_dash_multi_period_init.py b/tests/core/test_dash_multi_period_init.py
new file mode 100644
index 00000000..da63b364
--- /dev/null
+++ b/tests/core/test_dash_multi_period_init.py
@@ -0,0 +1,190 @@
+"""A multi-period DASH manifest must fetch and probe the init segment of the first content
+period only.
+
+``DASH.download_track`` keeps the init data and the KID of period 0 and discards them for
+every later period, so a request and an FFprobe process per extra period gain nothing. The
+``SegmentBase`` form is the exception: its ``media_range`` is built from ``len(init_data)``,
+so the request stays for every period and only the probe is dropped.
+"""
+
+from __future__ import annotations
+
+import threading
+from functools import partial
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import pytest
+import requests
+
+from unshackle.core.constants import DOWNLOAD_LICENCE_ONLY
+from unshackle.core.manifests import DASH
+from unshackle.core.tracks.track import DownloadContext, Track
+from unshackle.core.utils.xml import load_xml
+
+MEDIA = bytes(i % 256 for i in range(4096))
+INIT_LEN = 100
+
+
+class _Handler(BaseHTTPRequestHandler):
+ def do_GET(self):
+ server = self.server
+ rng = self.headers.get("Range")
+ with server.lock:
+ server.requests.append((self.path, rng))
+ body = server.routes.get(self.path)
+ if body is None:
+ self.send_response(404)
+ self.send_header("Content-Length", "0")
+ self.end_headers()
+ return
+ if rng:
+ start, end = (int(x) for x in rng.removeprefix("bytes=").split("-"))
+ chunk = body[start : end + 1]
+ self.send_response(206)
+ self.send_header("Content-Range", f"bytes {start}-{end}/{len(body)}")
+ body = chunk
+ else:
+ self.send_response(200)
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ def log_message(self, *args):
+ pass
+
+
+@pytest.fixture
+def server():
+ srv = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+ srv.lock = threading.Lock()
+ srv.requests = []
+ srv.routes = {f"/p{i}/init.bin": MEDIA[:INIT_LEN] for i in (0, 1)}
+ srv.routes.update({f"/p{i}/1.bin": MEDIA for i in (0, 1)})
+ srv.routes.update({f"/p{i}.mp4": MEDIA for i in (0, 1)})
+ thread = threading.Thread(target=srv.serve_forever, daemon=True)
+ thread.start()
+ yield srv
+ srv.shutdown()
+ thread.join(timeout=5)
+
+
+@pytest.fixture
+def probes(monkeypatch):
+ """Count KID probes without running FFprobe."""
+ calls = []
+
+ def counted(self, init_data=None, *args, **kwargs):
+ calls.append(init_data)
+ return None
+
+ monkeypatch.setattr(Track, "get_key_id", counted)
+ return calls
+
+
+@pytest.fixture
+def licence_only():
+ """Return from download_track once the periods are parsed, before any segment download."""
+ was_set = DOWNLOAD_LICENCE_ONLY.is_set()
+ DOWNLOAD_LICENCE_ONLY.set()
+ yield
+ if not was_set:
+ DOWNLOAD_LICENCE_ONLY.clear()
+
+
+def base(srv) -> str:
+ return f"http://127.0.0.1:{srv.server_address[1]}/"
+
+
+def template_mpd(srv) -> str:
+ periods = "\n".join(
+ f"""
+
+
+
+
+
+ """
+ for i in (0, 1)
+ )
+ return f"""
+
+ {base(srv)}
+{periods}
+"""
+
+
+def base_mpd(srv) -> str:
+ periods = "\n".join(
+ f"""
+
+
+ p{i}.mp4
+
+
+
+
+
+ """
+ for i in (0, 1)
+ )
+ return f"""
+
+ {base(srv)}
+{periods}
+"""
+
+
+def make_track(mpd: str, srv):
+ return DASH.from_text(mpd, url=f"{base(srv)}manifest.mpd").to_tracks(language="en").videos[0]
+
+
+def make_ctx(tmp_path):
+ return DownloadContext(
+ save_path=tmp_path / "out.mp4",
+ save_dir=tmp_path / "v0_segments",
+ progress=partial(lambda **kw: None),
+ )
+
+
+def test_segment_template_init_is_fetched_once_for_two_periods(server, probes, licence_only, tmp_path):
+ track = make_track(template_mpd(server), server)
+
+ DASH.download_track(track, make_ctx(tmp_path))
+
+ # both periods contributed a segment, so period 1 was parsed, not skipped
+ assert len(track.data["dash"]["segment_durations"]) == 2
+ assert [path for path, _ in server.requests] == ["/p0/init.bin"]
+ assert len(probes) == 1
+
+
+def test_segment_base_keeps_each_request_but_probes_once(server, probes, licence_only, tmp_path):
+ track = make_track(base_mpd(server), server)
+
+ DASH.download_track(track, make_ctx(tmp_path))
+
+ # media_range needs the init bytes of every period, so every period keeps its request
+ assert [path for path, _ in server.requests] == ["/p0.mp4", "/p1.mp4"]
+ assert len(probes) == 1
+
+
+def test_segment_base_range_survives_a_skipped_probe(server, probes):
+ tree = load_xml(base_mpd(server).encode())
+ period = tree.findall("Period")[1]
+ aset = period.find("AdaptationSet")
+ rep = aset.find("Representation")
+
+ _, segments, _, _, track_kid = DASH.get_period_segments(
+ period,
+ aset,
+ rep,
+ tree,
+ Track(url=base(server), language="en"),
+ f"{base(server)}manifest.mpd",
+ requests.Session(),
+ probe_kid=False,
+ )
+
+ assert segments == [(f"{base(server)}p1.mp4", f"{INIT_LEN}-{len(MEDIA)}")]
+ assert track_kid is None
+ assert not probes
diff --git a/tests/core/test_hls_segment_io.py b/tests/core/test_hls_segment_io.py
new file mode 100644
index 00000000..53a0d018
--- /dev/null
+++ b/tests/core/test_hls_segment_io.py
@@ -0,0 +1,143 @@
+"""File I/O rules for HLS segments.
+
+Two faults that Linux never shows:
+
+* The FFmpeg concat list is a UTF-8 file by specification. Written in the OS locale
+ encoding it either raises ``UnicodeEncodeError`` on a non-ASCII path, which kills the
+ track after every segment is already on disk, or it silently disables the fast path.
+* A subtitle segment that the UTF-8 transform leaves unchanged, the common case, needs
+ no write back. Each rewrite is a file operation an on-access virus scanner has to
+ clear.
+"""
+
+from __future__ import annotations
+
+import codecs
+import subprocess
+import threading
+from functools import partial
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+import pytest
+
+from unshackle.core.manifests import HLS
+from unshackle.core.tracks.track import DownloadContext
+
+SEGMENT_ONE = b"WEBVTT\n\n00:00:00.000 --> 00:00:04.000\nhello\n"
+SEGMENT_TWO = b"WEBVTT\n\n00:00:04.000 --> 00:00:08.000\nworld\n"
+
+
+def test_concat_list_is_written_as_utf8(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+ """A non-ASCII segment path must not depend on the OS codepage."""
+ save_dir = tmp_path / "Ünshackle тест" / "Subtitle_segments"
+ save_dir.mkdir(parents=True)
+ segments = []
+ for i in range(2):
+ segment = save_dir / f"{i:05}.vtt"
+ segment.write_bytes(b"\x00" * 8)
+ segments.append(segment)
+
+ encodings: list[object] = []
+ real_write_text = Path.write_text
+
+ def spy_write_text(self, data, encoding=None, *args, **kwargs):
+ encodings.append(encoding)
+ return real_write_text(self, data, encoding=encoding, *args, **kwargs)
+
+ seen: list[bytes] = []
+
+ def fake_run(args, **kwargs):
+ seen.append(Path(args[args.index("-i") + 1]).read_bytes())
+ raise subprocess.CalledProcessError(1, args, output="", stderr="")
+
+ monkeypatch.setattr(Path, "write_text", spy_write_text)
+ monkeypatch.setattr("unshackle.core.manifests.hls.binaries.FFMPEG", Path("ffmpeg"))
+ monkeypatch.setattr("unshackle.core.manifests.hls.subprocess.run", fake_run)
+
+ HLS.merge_segments(segments=segments, save_path=save_dir.parent / "Subtitle.vtt")
+
+ assert encodings, "the concat list was never written"
+ assert all(e is not None and codecs.lookup(e).name == "utf-8" for e in encodings), (
+ f"the concat list was written in the OS locale encoding: {encodings}"
+ )
+ assert seen and seen[0].decode("utf-8") == "\n".join(f"file '{s.absolute()}'" for s in segments)
+
+
+class _Handler(BaseHTTPRequestHandler):
+ def do_GET(self):
+ body = self.server.routes.get(self.path)
+ if body is None:
+ self.send_response(404)
+ self.send_header("Content-Length", "0")
+ self.end_headers()
+ return
+ self.send_response(200)
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ def log_message(self, *args):
+ pass
+
+
+@pytest.fixture
+def server():
+ srv = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+ srv.routes = {}
+ thread = threading.Thread(target=srv.serve_forever, daemon=True)
+ thread.start()
+ yield srv
+ srv.shutdown()
+ thread.join(timeout=5)
+
+
+MASTER = """#EXTM3U
+#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="subs",NAME="English",LANGUAGE="en",DEFAULT=YES,URI="subs.m3u8"
+#EXT-X-STREAM-INF:BANDWIDTH=1000000,RESOLUTION=1920x1080,CODECS="avc1.640028",SUBTITLES="subs"
+media.m3u8
+"""
+
+SUBS_PLAYLIST = b"""#EXTM3U
+#EXT-X-VERSION:3
+#EXT-X-TARGETDURATION:4
+#EXTINF:4.0,
+part0.vtt
+#EXTINF:4.0,
+part1.vtt
+#EXT-X-ENDLIST
+"""
+
+
+def test_unchanged_subtitle_segments_are_not_rewritten(server, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+ """Plain UTF-8 WebVTT survives the transform untouched, so nothing needs writing back."""
+ port = server.server_address[1]
+ server.routes["/subs.m3u8"] = SUBS_PLAYLIST
+ server.routes["/part0.vtt"] = SEGMENT_ONE
+ server.routes["/part1.vtt"] = SEGMENT_TWO
+
+ track = HLS.from_text(MASTER, url=f"http://127.0.0.1:{port}/master.m3u8").to_tracks(language="en").subtitles[0]
+
+ save_dir = tmp_path / "s0_segments"
+ rewritten: list[str] = []
+ real_write_bytes = Path.write_bytes
+
+ def spy_write_bytes(self, data):
+ if save_dir in self.parents:
+ rewritten.append(self.name)
+ return real_write_bytes(self, data)
+
+ monkeypatch.setattr(Path, "write_bytes", spy_write_bytes)
+
+ HLS.download_track(
+ track,
+ DownloadContext(
+ save_path=tmp_path / "out.vtt",
+ save_dir=save_dir,
+ progress=partial(lambda **kw: None),
+ max_workers=1,
+ ),
+ )
+
+ assert (tmp_path / "out.vtt").read_bytes() == SEGMENT_ONE + SEGMENT_TWO
+ assert rewritten == [], f"unchanged subtitle segments were rewritten: {rewritten}"
diff --git a/tests/core/test_remote_cdm_pssh_reuse.py b/tests/core/test_remote_cdm_pssh_reuse.py
new file mode 100644
index 00000000..e25de0bb
--- /dev/null
+++ b/tests/core/test_remote_cdm_pssh_reuse.py
@@ -0,0 +1,65 @@
+"""A reused CustomRemoteCDM must not carry one title's PlayReady PSSH into the next.
+
+``dl.get_cdm`` builds each CDM once per run, so one instance serves every title.
+``PlayReady.get_content_keys`` calls ``set_pssh_b64`` only when the DRM has a PSSH, so
+an instance-level value left over from an earlier title would be sent as the init data
+of a title that has none.
+"""
+
+from __future__ import annotations
+
+import base64
+
+import pytest
+
+from unshackle.core.cdm.custom_remote_cdm import CustomRemoteCDM
+
+pytestmark = pytest.mark.unit
+
+
+class FakePssh:
+ """A PSSH object shaped like the ones the manifests hand to the CDM."""
+
+ def __init__(self, raw: bytes) -> None:
+ self.raw = raw
+
+
+def make_cdm() -> CustomRemoteCDM:
+ return CustomRemoteCDM(host="https://cdm.test", device={"name": "SL3000", "type": "PLAYREADY"})
+
+
+def set_pssh(cdm: CustomRemoteCDM, session_id: bytes, pssh_b64: str) -> None:
+ """Store a PSSH the way ``PlayReady.get_content_keys`` does."""
+ try:
+ cdm.set_pssh_b64(pssh_b64, session_id=session_id)
+ except TypeError: # CDM predating the per-session signature
+ cdm.set_pssh_b64(pssh_b64)
+
+
+def test_a_later_session_does_not_inherit_the_earlier_pssh() -> None:
+ cdm = make_cdm()
+ assert cdm.is_playready
+
+ first = cdm.open()
+ set_pssh(cdm, first, "FIRST_TITLE_PSSH")
+ assert cdm.get_init_data_from_pssh(FakePssh(b"ignored")) == "FIRST_TITLE_PSSH"
+ cdm.close(first)
+
+ # The second title has no PSSH of its own, so nothing calls set_pssh_b64.
+ cdm.open()
+ init_data = cdm.get_init_data_from_pssh(FakePssh(b"second"))
+
+ assert init_data == base64.b64encode(b"second").decode("utf-8")
+
+
+def test_two_open_sessions_keep_their_own_pssh() -> None:
+ """Tracks decrypt in parallel on the one shared instance, so the store is per session."""
+ cdm = make_cdm()
+
+ one = cdm.open()
+ two = cdm.open()
+ set_pssh(cdm, one, "PSSH_ONE")
+ set_pssh(cdm, two, "PSSH_TWO")
+
+ assert cdm.get_init_data_from_pssh(FakePssh(b"x"), cdm._sessions[one]["pssh_b64"]) == "PSSH_ONE"
+ assert cdm.get_init_data_from_pssh(FakePssh(b"x"), cdm._sessions[two]["pssh_b64"]) == "PSSH_TWO"
diff --git a/tests/downloaders/test_empty_subrange.py b/tests/downloaders/test_empty_subrange.py
new file mode 100644
index 00000000..1997296f
--- /dev/null
+++ b/tests/downloaders/test_empty_subrange.py
@@ -0,0 +1,110 @@
+"""A bounded sub-range that returns no bytes must fail, not spin.
+
+After a cut, the sequential path asks for bounded sub-ranges and treats a whole
+sub-range as progress rather than a retry. A response with an empty body of
+unknown length (a chunked 206, a 204, a 304) makes no progress, so it must
+raise and spend an attempt, or the same request repeats forever.
+"""
+
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import pytest
+import requests as rq
+
+from unshackle.core.constants import DOWNLOAD_CANCELLED
+from unshackle.core.downloaders.requests import download
+
+# the package re-exports the `requests` function, shadowing the module of the same name
+downloader = sys.modules[download.__module__]
+
+BODY = bytes(range(256)) * 256 # 64 KiB
+CUT_AFTER = 10 * 1024
+CHUNK = 16 * 1024
+DEADLINE = 30
+
+
+class _EmptyRangeServer(ThreadingHTTPServer):
+ """Cuts the first body, then answers every bounded range with an empty chunked 206."""
+
+ allow_reuse_address = True
+ daemon_threads = True
+
+ def __init__(self):
+ self.ranges: list[str] = []
+ super().__init__(("127.0.0.1", 0), _EmptyRangeHandler)
+ self.url = f"http://127.0.0.1:{self.server_address[1]}/seg.mp4"
+
+ def handle_error(self, request, client_address): # the client hanging up is expected here
+ pass
+
+
+class _EmptyRangeHandler(BaseHTTPRequestHandler):
+ protocol_version = "HTTP/1.1"
+
+ def log_message(self, *_):
+ pass
+
+ def do_GET(self):
+ header = self.headers.get("Range")
+ self.server.ranges.append(header)
+ start, end, bounded = 0, len(BODY) - 1, False
+ if header:
+ start_s, _, end_s = header.removeprefix("bytes=").partition("-")
+ start, end = int(start_s), (int(end_s) if end_s else len(BODY) - 1)
+ bounded = bool(end_s)
+ if bounded:
+ # a valid response: the chunked terminator closes a zero-length body, so requests raises nothing
+ self.send_response(206)
+ self.send_header("Content-Range", f"bytes {start}-{end}/{len(BODY)}")
+ self.send_header("Transfer-Encoding", "chunked")
+ self.end_headers()
+ self.wfile.write(b"0\r\n\r\n")
+ return
+ body = BODY[start : end + 1]
+ self.send_response(206 if header else 200)
+ if header:
+ self.send_header("Content-Range", f"bytes {start}-{end}/{len(BODY)}")
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ # the cut teaches the downloader to bound the requests that follow
+ self.wfile.write(body[:CUT_AFTER])
+ self.close_connection = True
+
+
+@pytest.fixture
+def server(monkeypatch):
+ DOWNLOAD_CANCELLED.clear()
+ monkeypatch.setattr(downloader, "RETRY_WAIT", 0)
+ monkeypatch.setattr(downloader, "REQUEST_CHUNK_SIZE", CHUNK)
+ monkeypatch.setattr(downloader, "MIN_REQUEST_CHUNK", 4 * 1024)
+ monkeypatch.setattr(downloader, "MIN_RESUME_PROGRESS", 1024)
+ server = _EmptyRangeServer()
+ threading.Thread(target=server.serve_forever, daemon=True).start()
+ yield server
+ server.shutdown()
+ server.server_close()
+ DOWNLOAD_CANCELLED.clear()
+
+
+def test_empty_sub_range_fails_instead_of_looping(tmp_path, server):
+ save_path = tmp_path / "0.mp4"
+ result: list[BaseException] = []
+
+ def run() -> None:
+ try:
+ list(download(url=server.url, save_path=save_path, session=rq.Session()))
+ except BaseException as exc:
+ result.append(exc)
+
+ # a hard deadline: a regression hangs this thread, and the suite must fail, not stall
+ worker = threading.Thread(target=run, daemon=True)
+ worker.start()
+ worker.join(timeout=DEADLINE)
+
+ assert not worker.is_alive(), f"download still running after {DEADLINE}s ({len(server.ranges)} requests)"
+ assert result and isinstance(result[0], IOError), f"expected an IOError, got {result!r}"
+ assert "no bytes" in str(result[0])
+ assert len(server.ranges) <= downloader.MAX_ATTEMPTS + 1
+ assert not save_path.exists()
diff --git a/tests/downloaders/test_no_retry_session.py b/tests/downloaders/test_no_retry_session.py
new file mode 100644
index 00000000..b54056dd
--- /dev/null
+++ b/tests/downloaders/test_no_retry_session.py
@@ -0,0 +1,55 @@
+"""Connection pooling survives the no-retry adapter view.
+
+``download()`` builds this view once per segment. ``copy()`` runs
+``HTTPAdapter.__setstate__``, which blanks ``proxy_manager`` and re-runs
+``init_poolmanager()``, so a view that keeps those fresh objects hands every segment an
+empty pool: one TCP connection and one TLS handshake each. That is invisible on Linux and
+costs a native Windows host roughly 7x the wall time and 4x the CPU on a manifest of a few
+thousand segments.
+
+The second test pins the reason sharing one pool between the two adapters is safe:
+``HTTPAdapter.send`` passes ``retries=self.max_retries`` per call, so the view's ``Retry(0)``
+travels with the view and never reaches the caller's own session.
+"""
+
+from requests import Session
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+from unshackle.core.downloaders.requests import no_retry_session
+
+
+def pooled_session(total: int = 5) -> Session:
+ session = Session()
+ adapter = HTTPAdapter(pool_connections=16, pool_maxsize=16, pool_block=True, max_retries=Retry(total=total))
+ session.mount("https://", adapter)
+ session.mount("http://", adapter)
+ return session
+
+
+def test_view_shares_the_pools_it_was_copied_from() -> None:
+ session = pooled_session()
+ view = no_retry_session(session)
+
+ for prefix, adapter in session.adapters.items():
+ twin = view.adapters[prefix]
+ assert twin is not adapter, "the view must be its own adapter, or Retry(0) lands on the caller"
+ assert twin.poolmanager is adapter.poolmanager
+ assert twin.proxy_manager is adapter.proxy_manager
+
+
+def test_view_never_retries_and_leaves_the_caller_retrying() -> None:
+ session = pooled_session(total=5)
+ view = no_retry_session(session)
+
+ assert view.adapters["https://"].max_retries.total == 0
+ assert session.adapters["https://"].max_retries.total == 5, "the caller's retry policy must survive"
+
+
+def test_repeated_views_keep_reusing_one_pool() -> None:
+ """``download()`` rebuilds the view per segment; every rebuild must land on the same pool."""
+ session = pooled_session()
+ origin = session.adapters["https://"].poolmanager
+
+ for _ in range(3):
+ assert no_retry_session(session).adapters["https://"].poolmanager is origin
diff --git a/tests/remote/e2e/conftest.py b/tests/remote/e2e/conftest.py
index d676d921..0506665f 100644
--- a/tests/remote/e2e/conftest.py
+++ b/tests/remote/e2e/conftest.py
@@ -18,7 +18,7 @@ def load_scenarios(config: pytest.Config) -> dict:
import yaml
path = Path(__file__).parent / "fixtures" / "fixtures.yaml"
- data = yaml.safe_load(path.read_text()) if path.exists() else {}
+ data = yaml.safe_load(path.read_text(encoding="utf-8")) if path.exists() else {}
selected = (config.getoption("--services") or "").strip()
services = (data or {}).get("services", {}) or {}
if selected:
diff --git a/tests/remote/unit/test_handler_event_loop.py b/tests/remote/unit/test_handler_event_loop.py
new file mode 100644
index 00000000..2bb9c5ab
--- /dev/null
+++ b/tests/remote/unit/test_handler_event_loop.py
@@ -0,0 +1,101 @@
+"""Blocking service calls must not run on the aiohttp event loop.
+
+get_titles and get_tracks do network work that can take a whole manifest round trip. Run on
+the loop, they stop the server: the progress pump and the SSE keep-alive stall with them. An
+interactive prompt from the service can never be answered either, because the endpoint that
+answers it needs the same loop. Each test here holds the service call in a two-event
+handshake that only completes while the loop still runs another coroutine.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import threading
+from types import SimpleNamespace
+from typing import Any, Awaitable, Callable, List
+
+import pytest
+from aiohttp import web
+
+from unshackle.core.api import handlers
+from unshackle.core.api.session_store import get_session_store
+from unshackle.core.titles.movie import Movie
+
+pytestmark = pytest.mark.unit
+
+# Bounds the wait when the loop is blocked, so a regression fails the test instead of hanging.
+HANDSHAKE_TIMEOUT = 5.0
+
+
+class Handshake:
+ """A blocking call that reports whether the event loop kept running during it."""
+
+ def __init__(self) -> None:
+ self.entered = threading.Event()
+ self.resume = threading.Event()
+ self.loop_was_blocked = False
+
+ def block(self) -> None:
+ self.entered.set()
+ self.loop_was_blocked = not self.resume.wait(timeout=HANDSHAKE_TIMEOUT)
+
+ async def drive(self, call: Callable[[], Awaitable[web.Response]]) -> web.Response:
+ task = asyncio.create_task(call())
+ # Reached only while the loop is free.
+ await asyncio.to_thread(self.entered.wait, HANDSHAKE_TIMEOUT)
+ self.resume.set()
+ return await asyncio.wait_for(task, HANDSHAKE_TIMEOUT)
+
+
+def a_movie() -> Movie:
+ return Movie(id_="movie-0001", service=object, name="Film", year=2024, language=None)
+
+
+class BlockingTitles:
+ def __init__(self, handshake: Handshake) -> None:
+ self.handshake = handshake
+
+ def get_titles(self) -> List[Movie]:
+ self.handshake.block()
+ return [a_movie()]
+
+
+class BlockingTracks:
+ def __init__(self, handshake: Handshake) -> None:
+ self.handshake = handshake
+
+ def get_titles(self) -> List[Movie]:
+ return [a_movie()]
+
+ def get_tracks(self, title: Any) -> Any:
+ self.handshake.block()
+ return SimpleNamespace(videos=[], audio=[], subtitles=[])
+
+
+@pytest.mark.asyncio
+async def test_session_titles_leaves_the_loop_free() -> None:
+ handshake = Handshake()
+ store = get_session_store()
+ await store.create("EXAMPLE", BlockingTitles(handshake), session_id="loop-test-titles")
+ try:
+ response = await handshake.drive(lambda: handlers.session_titles_handler("loop-test-titles"))
+ finally:
+ await store.delete("loop-test-titles")
+
+ assert not handshake.loop_was_blocked, "get_titles blocked the event loop"
+ assert [t["name"] for t in json.loads(response.body)["titles"]] == ["Film"]
+
+
+@pytest.mark.asyncio
+async def test_list_tracks_leaves_the_loop_free(monkeypatch: pytest.MonkeyPatch) -> None:
+ handshake = Handshake()
+ service = BlockingTracks(handshake)
+ monkeypatch.setattr(handlers, "validate_service", lambda tag, request=None: "EXAMPLE")
+ monkeypatch.setattr(handlers, "setup_list_service", lambda *args, **kwargs: service)
+
+ data = {"service": "EXAMPLE", "title_id": "movie-0001"}
+ response = await handshake.drive(lambda: handlers.list_tracks_handler(data))
+
+ assert not handshake.loop_was_blocked, "get_tracks blocked the event loop"
+ assert json.loads(response.body)["title"]["name"] == "Film"
diff --git a/tests/remote/unit/test_progress_throttle.py b/tests/remote/unit/test_progress_throttle.py
new file mode 100644
index 00000000..3629ecba
--- /dev/null
+++ b/tests/remote/unit/test_progress_throttle.py
@@ -0,0 +1,126 @@
+"""Unit tests for the progress write rate and for concurrent progress writers.
+
+The aggregate sink rewrites a whole JSON file per call, and one track sends thousands of ticks
+while the reader polls twice a second. These tests pin the two guards on that: repeat ticks are
+throttled, a tick that changes the job shape is not, and two threads writing the same file at the
+same time do not lose a write."""
+
+from __future__ import annotations
+
+import json
+import threading
+from pathlib import Path
+
+import pytest
+
+from unshackle.core.api import progress as progress_module
+from unshackle.core.api.download_worker import write_result
+from unshackle.core.api.progress import build_job_progress_callables
+
+pytestmark = pytest.mark.unit
+
+
+class Video:
+ def __init__(self, height=1080, bitrate=4_000_000):
+ self.height = height
+ self.range = None
+ self.bitrate = bitrate
+
+
+def noop(**kwargs):
+ pass
+
+
+def test_repeat_ticks_are_throttled():
+ updates: list[dict] = []
+ cbs = build_job_progress_callables([Video()], [noop], updates.append)
+
+ for completed in range(1, 501):
+ cbs[0](total=1000, completed=completed)
+
+ # One emit for the first tick (the track starts); the rest say the same thing and coalesce.
+ assert len(updates) == 1
+
+
+def test_track_finish_bypasses_the_throttle():
+ updates: list[dict] = []
+ cbs = build_job_progress_callables([Video(), Video()], [noop, noop], updates.append)
+
+ cbs[0](total=100, completed=1)
+ cbs[0](total=100, completed=2)
+ cbs[0](total=100, completed=3)
+ assert len(updates) == 1
+
+ cbs[0](downloaded="Downloaded")
+ assert len(updates) == 2
+ assert updates[-1]["completed_tracks"] == 1
+
+
+def test_phase_change_bypasses_the_throttle():
+ updates: list[dict] = []
+ cbs = build_job_progress_callables([Video(2160), Video(720)], [noop, noop], updates.append)
+
+ cbs[0](total=100, completed=1)
+ first_phase = updates[-1]["phase"]
+
+ cbs[1](total=100, completed=1)
+ assert len(updates) == 2
+ assert updates[-1]["phase"] != first_phase
+
+
+def test_filled_track_bypasses_the_throttle_without_a_terminal_string():
+ """A track can reach its segment total and never send "Downloaded". That last tick must land,
+ or the job reports a percentage that is short for good."""
+ updates: list[dict] = []
+ cbs = build_job_progress_callables([Video()], [noop], updates.append)
+
+ cbs[0](total=100, completed=10)
+ cbs[0](total=100, completed=50)
+ assert len(updates) == 1
+
+ cbs[0](total=100, completed=100)
+ assert len(updates) == 2
+ assert updates[-1]["progress"] == pytest.approx(progress_module.DOWNLOAD_PROGRESS_CEILING)
+
+
+def test_throttle_releases_after_the_interval(monkeypatch: pytest.MonkeyPatch):
+ clock = {"now": 1000.0}
+ monkeypatch.setattr(progress_module.time, "monotonic", lambda: clock["now"])
+
+ updates: list[dict] = []
+ cbs = build_job_progress_callables([Video()], [noop], updates.append)
+
+ cbs[0](total=100, completed=1)
+ cbs[0](total=100, completed=2)
+ assert len(updates) == 1
+
+ clock["now"] += progress_module.PROGRESS_EMIT_INTERVAL + 0.01
+ cbs[0](total=100, completed=3)
+ assert len(updates) == 2
+
+
+def test_concurrent_writers_do_not_clobber_each_other(tmp_path: Path):
+ """Two threads can write the same progress file. A temp name shared by two writers lets one
+ replace away the file the other is about to move, which fails that write."""
+ target = tmp_path / "progress.json"
+ errors: list[BaseException] = []
+ start = threading.Barrier(8)
+
+ def writer(index: int) -> None:
+ start.wait()
+ try:
+ for tick in range(40):
+ write_result(target, {"writer": index, "tick": tick})
+ except BaseException as exc: # noqa: BLE001 - reported to the test thread
+ errors.append(exc)
+
+ threads = [threading.Thread(target=writer, args=(i,)) for i in range(8)]
+ for thread in threads:
+ thread.start()
+ for thread in threads:
+ thread.join()
+
+ assert errors == []
+ payload = json.loads(target.read_text(encoding="utf-8"))
+ assert set(payload) == {"writer", "tick"}
+ assert [p.name for p in tmp_path.iterdir()] == ["progress.json"]
diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py
index 9595b463..1c99baac 100644
--- a/unshackle/commands/dl.py
+++ b/unshackle/commands/dl.py
@@ -4879,4 +4879,13 @@ class dl:
from unshackle.core.cdm import load_cdm
- return load_cdm(cdm_name, service_name=service, vaults=self.vaults)
+ if not isinstance(cdm_name, str):
+ return load_cdm(cdm_name, service_name=service, vaults=self.vaults)
+
+ cdm_cache: dict[tuple[str, Optional[str], str], Any] = getattr(self, "_cdm_cache", {})
+ self._cdm_cache = cdm_cache
+
+ cache_key = (service, profile, cdm_name)
+ if cache_key not in cdm_cache:
+ cdm_cache[cache_key] = load_cdm(cdm_name, service_name=service, vaults=self.vaults)
+ return cdm_cache[cache_key]
diff --git a/unshackle/core/api/download_worker.py b/unshackle/core/api/download_worker.py
index fb64abe8..f57042ac 100644
--- a/unshackle/core/api/download_worker.py
+++ b/unshackle/core/api/download_worker.py
@@ -6,10 +6,12 @@ import json
import logging
import os
import sys
+import threading
import time
import traceback
from pathlib import Path
from typing import Any, Dict
+from uuid import uuid4
from .download_manager import perform_download
@@ -24,18 +26,25 @@ def read_payload(path: Path) -> Dict[str, Any]:
def write_result(path: Path, payload: Dict[str, Any]) -> None:
"""Write the payload with an atomic replace, because the parent polls this file during the write.
+ Each call gets its own temp name, so two threads writing this destination cannot replace away
+ the temp file the other is about to move.
+
Windows denies the replace while the parent holds the destination open, so retry with a backoff.
"""
path.parent.mkdir(parents=True, exist_ok=True)
- tmp = path.with_name(path.name + ".tmp")
- tmp.write_text(json.dumps(payload), encoding="utf-8")
- for attempt in range(5):
- try:
- os.replace(tmp, path)
- return
- except PermissionError:
- time.sleep(0.02 * (attempt + 1))
- os.replace(tmp, path)
+ tmp = path.with_name(f"{path.name}.{uuid4().hex}.tmp")
+ try:
+ tmp.write_text(json.dumps(payload), encoding="utf-8")
+ for attempt in range(5):
+ try:
+ os.replace(tmp, path)
+ return
+ except PermissionError:
+ time.sleep(0.02 * (attempt + 1))
+ os.replace(tmp, path)
+ except OSError:
+ tmp.unlink(missing_ok=True)
+ raise
def main(argv: list[str]) -> int:
@@ -64,14 +73,16 @@ def main(argv: list[str]) -> int:
# Merged so sparse keys (current_title, output_files) survive later writes.
progress_state: Dict[str, Any] = {}
+ progress_lock = threading.Lock()
def progress_callback(progress_data: Dict[str, Any]) -> None:
"""Write progress updates to file for main process to read."""
if progress_path:
try:
- progress_state.update(progress_data)
- log.info(f"Writing progress update: {progress_data}")
- write_result(progress_path, progress_state)
+ with progress_lock:
+ progress_state.update(progress_data)
+ log.info(f"Writing progress update: {progress_data}")
+ write_result(progress_path, progress_state)
log.info(f"Progress update written to {progress_path}")
except Exception as e:
log.error(f"Failed to write progress update: {e}")
diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py
index 82f8fbe7..113d1d2e 100644
--- a/unshackle/core/api/handlers.py
+++ b/unshackle/core/api/handlers.py
@@ -278,6 +278,72 @@ def setup_list_service(
return service_instance
+def run_service_search(
+ data: Dict[str, Any],
+ normalized_service: str,
+ query: str,
+ request: Optional[web.Request] = None,
+) -> List[Dict[str, Any]]:
+ """Assemble and authenticate a service instance, then run its search.
+
+ The same preamble as :func:`setup_list_service`, without the service option extras,
+ plus the search itself. Every step blocks on disk or network, so the caller runs the
+ whole function in one worker thread and keeps the event loop free for other clients.
+ """
+ from unshackle.commands.dl import dl
+
+ profile = data.get("profile")
+ no_proxy = data.get("no_proxy", False)
+
+ service_config = load_service_yaml(normalized_service)
+ proxy_param, proxy_providers = resolve_handler_proxy(data, normalized_service, request)
+
+ account = resolve_server_account(request, normalized_service, data.get("client_region"))
+ if account:
+ profile = None if account == "default" else account
+
+ cdm = load_full_cdm(normalized_service, profile, data.get("cdm_type"))
+ parent_ctx = build_parent_ctx(profile, cdm, proxy_param, no_proxy, proxy_providers, service_config)
+ service_module = Services.load(normalized_service)
+
+ try:
+ service_instance = instantiate_service(parent_ctx, service_module, query)
+ except Exception as exc:
+ raise APIError(
+ APIErrorCode.SERVICE_ERROR,
+ f"Failed to initialize service: {exc}",
+ details={"service": normalized_service},
+ )
+
+ if account:
+ service_instance.cache = Cacher(f"_accounts/{normalized_service}/{account}")
+ cookies = server_account_cookies(normalized_service, profile)
+ else:
+ cookies = dl.get_cookie_jar(normalized_service, profile)
+ credential = dl.get_credentials(normalized_service, profile)
+ service_instance.authenticate(cookies, credential)
+
+ results: List[Dict[str, Any]] = []
+ try:
+ for result in service_instance.search():
+ results.append(
+ {
+ "id": result.id,
+ "title": result.title,
+ "description": result.description,
+ "label": result.label,
+ "url": result.url,
+ }
+ )
+ except NotImplementedError:
+ raise APIError(
+ APIErrorCode.SERVICE_ERROR,
+ f"Search is not supported by {normalized_service}",
+ details={"service": normalized_service},
+ )
+ return results
+
+
def allowed_services_for_key(secret_key: Optional[str]) -> Optional[List[str]]:
"""Effective service allowlist for an API key: the global list intersected with the key's.
@@ -559,7 +625,7 @@ async def dashboard_authentication(request: web.Request, handler: Any) -> web.St
def rate_limit_response(secret_key: str) -> Optional[web.Response]:
"""A 429 when *secret_key* is over its hourly limit, else None after counting the request.
- Call only once the key is known to be valid, so an invalid key cannot burn someone else's
+ Call only once the API key is known to be valid, so an invalid API key cannot burn someone else's
window. Both auth middlewares go through here: the limit has to hold in every server mode.
"""
from unshackle.core.api.stats import stats
@@ -576,7 +642,7 @@ def rate_limit_response(secret_key: str) -> Optional[web.Response]:
@web.middleware
async def api_key_authentication(request: web.Request, handler: Any) -> web.StreamResponse:
- """Gate every non-dashboard route behind a key in ``app["config"]["users"]``, then rate limit it.
+ """Gate every non-dashboard route behind an API key in ``app["config"]["users"]``, then rate limit it.
Runs after :func:`dashboard_authentication`, which has already answered the dashboard
routes, so the dashboard key never reaches the limiter.
@@ -1084,8 +1150,6 @@ def serialize_attachment(attachment: Any) -> Optional[Dict[str, Any]]:
async def search_handler(data: Dict[str, Any], request: Optional[web.Request] = None) -> web.Response:
"""Answer the request to find titles."""
- from unshackle.commands.dl import dl
-
service_tag = data.get("service")
query = data.get("query")
@@ -1104,56 +1168,7 @@ async def search_handler(data: Dict[str, Any], request: Optional[web.Request] =
details={"service": service_tag},
)
- profile = data.get("profile")
- no_proxy = data.get("no_proxy", False)
-
- service_config = load_service_yaml(normalized_service)
-
- proxy_param, proxy_providers = await asyncio.to_thread(resolve_handler_proxy, data, normalized_service, request)
-
- account = resolve_server_account(request, normalized_service, data.get("client_region"))
- if account:
- profile = None if account == "default" else account
-
- cdm = load_full_cdm(normalized_service, profile, data.get("cdm_type"))
- parent_ctx = build_parent_ctx(profile, cdm, proxy_param, no_proxy, proxy_providers, service_config)
- service_module = Services.load(normalized_service)
-
- try:
- service_instance = instantiate_service(parent_ctx, service_module, query)
- except Exception as exc:
- raise APIError(
- APIErrorCode.SERVICE_ERROR,
- f"Failed to initialize service: {exc}",
- details={"service": normalized_service},
- )
-
- if account:
- service_instance.cache = Cacher(f"_accounts/{normalized_service}/{account}")
- cookies = server_account_cookies(normalized_service, profile)
- else:
- cookies = dl.get_cookie_jar(normalized_service, profile)
- credential = dl.get_credentials(normalized_service, profile)
- service_instance.authenticate(cookies, credential)
-
- results = []
- try:
- for result in service_instance.search():
- results.append(
- {
- "id": result.id,
- "title": result.title,
- "description": result.description,
- "label": result.label,
- "url": result.url,
- }
- )
- except NotImplementedError:
- raise APIError(
- APIErrorCode.SERVICE_ERROR,
- f"Search is not supported by {normalized_service}",
- details={"service": normalized_service},
- )
+ results = await asyncio.to_thread(run_service_search, data, normalized_service, query, request)
return web.json_response({"results": results, "count": len(results)})
@@ -1177,7 +1192,7 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
service_instance = await asyncio.to_thread(
setup_list_service, data, normalized_service, profile, title_id, request
)
- titles = service_instance.get_titles()
+ titles = await asyncio.to_thread(service_instance.get_titles)
if hasattr(titles, "__iter__") and not isinstance(titles, str):
title_list = [stamp_service_flags(serialize_title(t), service_instance) for t in titles]
@@ -1217,7 +1232,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
service_instance = await asyncio.to_thread(
setup_list_service, data, normalized_service, profile, title_id, request
)
- titles = service_instance.get_titles()
+ titles = await asyncio.to_thread(service_instance.get_titles)
wanted_param = data.get("wanted")
season = data.get("season")
@@ -1296,7 +1311,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
for title in sorted_titles:
try:
- tracks = service_instance.get_tracks(title)
+ tracks = await asyncio.to_thread(service_instance.get_tracks, title)
video_tracks = sorted(tracks.videos, key=lambda t: t.bitrate or 0, reverse=True)
audio_tracks = sorted(tracks.audio, key=lambda t: t.bitrate or 0, reverse=True)
@@ -1343,7 +1358,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
else:
first_title = titles
- tracks = service_instance.get_tracks(first_title)
+ tracks = await asyncio.to_thread(service_instance.get_tracks, first_title)
video_tracks = sorted(tracks.videos, key=lambda t: t.bitrate or 0, reverse=True)
audio_tracks = sorted(tracks.audio, key=lambda t: t.bitrate or 0, reverse=True)
@@ -1878,7 +1893,7 @@ async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
async def dashboard_keys_handler(request: web.Request) -> web.Response:
"""Every configured API key: which key it is, what it may do, and what it has done.
- Every key that ``configured_key`` counts gets a row here, so each ``requests_by_key``
+ Every API key that ``configured_key`` counts gets a row here, so each ``requests_by_key``
bucket except ``anonymous`` has a row to attribute it to. Without the dashboard key's own
row, a dashboard would render its own polling as traffic from an unknown caller.
"""
@@ -1924,7 +1939,7 @@ async def dashboard_keys_handler(request: web.Request) -> web.Response:
def user_grant(secret_key: str, name: str) -> Any:
"""A per-key grant as configured: False, True, or the list of service tags it covers.
- A key with no ``serve.users`` entry keeps implicit access, matching the resolvers.
+ An API key with no ``serve.users`` entry keeps implicit access, matching the resolvers.
"""
user_config = (config.serve.get("users") or {}).get(secret_key)
if not isinstance(user_config, dict):
@@ -1987,7 +2002,7 @@ _health_lock: Optional[asyncio.Lock] = None
async def dashboard_health_handler(request: web.Request) -> web.Response:
- """Preflight: can this instance actually finish a download?
+ """Preflight: can this instance finish a download?
A panel, not a liveness probe - the result is cached for 30 s and every probe is shallow,
so reading it never costs the operator a proxy session or a licence.
@@ -2915,7 +2930,8 @@ async def session_titles_handler(session_id: str, request: Optional[web.Request]
try:
service_instance = session.service_instance
- titles = service_instance.get_titles()
+ async with session.lock:
+ titles = await asyncio.to_thread(service_instance.get_titles)
session.titles = titles
if hasattr(titles, "__iter__") and not isinstance(titles, str):
@@ -2973,40 +2989,41 @@ async def session_tracks_handler(
try:
service_instance = session.service_instance
- tracks = service_instance.get_tracks(title)
+ async with session.lock:
+ tracks = await asyncio.to_thread(service_instance.get_tracks, title)
- title_tracks: Dict[str, Any] = {}
- for track in tracks.videos:
- title_tracks[str(track.id)] = track
- session.tracks[str(track.id)] = track
- for track in tracks.audio:
- title_tracks[str(track.id)] = track
- session.tracks[str(track.id)] = track
- for track in tracks.subtitles:
- title_tracks[str(track.id)] = track
- session.tracks[str(track.id)] = track
- session.tracks_by_title[str(title_id)] = title_tracks
- SessionStore.publish_update(session)
+ title_tracks: Dict[str, Any] = {}
+ for track in tracks.videos:
+ title_tracks[str(track.id)] = track
+ session.tracks[str(track.id)] = track
+ for track in tracks.audio:
+ title_tracks[str(track.id)] = track
+ session.tracks[str(track.id)] = track
+ for track in tracks.subtitles:
+ title_tracks[str(track.id)] = track
+ session.tracks[str(track.id)] = track
+ session.tracks_by_title[str(title_id)] = title_tracks
+ SessionStore.publish_update(session)
- try:
- chapters = service_instance.get_chapters(title)
- session.chapters_by_title[str(title_id)] = chapters if chapters else []
- except (NotImplementedError, Exception):
- session.chapters_by_title[str(title_id)] = []
+ try:
+ chapters = await asyncio.to_thread(service_instance.get_chapters, title)
+ session.chapters_by_title[str(title_id)] = chapters if chapters else []
+ except (NotImplementedError, Exception):
+ session.chapters_by_title[str(title_id)] = []
- video_tracks = sorted(tracks.videos, key=lambda t: t.bitrate or 0, reverse=True)
- audio_tracks = sorted(tracks.audio, key=lambda t: t.bitrate or 0, reverse=True)
+ video_tracks = sorted(tracks.videos, key=lambda t: t.bitrate or 0, reverse=True)
+ audio_tracks = sorted(tracks.audio, key=lambda t: t.bitrate or 0, reverse=True)
- manifests = extract_manifests(tracks)
- track_manifests = extract_track_manifests(tracks)
+ manifests = extract_manifests(tracks)
+ track_manifests = extract_track_manifests(tracks)
- svc_session = session.service_instance.session
- session_headers = dict(svc_session.headers) if hasattr(svc_session, "headers") else {}
- session_cookies = {}
- if hasattr(svc_session, "cookies"):
- for cookie in svc_session.cookies:
- if hasattr(cookie, "name") and hasattr(cookie, "value"):
- session_cookies[cookie.name] = cookie.value
+ svc_session = session.service_instance.session
+ session_headers = dict(svc_session.headers) if hasattr(svc_session, "headers") else {}
+ session_cookies = {}
+ if hasattr(svc_session, "cookies"):
+ for cookie in svc_session.cookies:
+ if hasattr(cookie, "name") and hasattr(cookie, "value"):
+ session_cookies[cookie.name] = cookie.value
from unshackle.core.config import config as app_config
diff --git a/unshackle/core/api/progress.py b/unshackle/core/api/progress.py
index ce47a9bb..5351aca2 100644
--- a/unshackle/core/api/progress.py
+++ b/unshackle/core/api/progress.py
@@ -6,11 +6,13 @@ a bitrate-weighted percentage, track counts, and the labels of the tracks downlo
from __future__ import annotations
+import time
from typing import Any, Callable, Optional
from unshackle.core.constants import AnyTrack
JOB_PROGRESS_TERMINAL_STATES = {"Downloaded", "Decrypted", "[yellow]SKIPPED"}
+PROGRESS_EMIT_INTERVAL = 0.25
# Weight for a track with no bitrate (subtitles); small vs media bitrates so subs barely move the bar.
SUBTITLE_PROGRESS_WEIGHT = 50_000.0
@@ -62,6 +64,10 @@ def build_job_progress_callables(
The sink gets a bitrate-weighted mean completion across all tracks, ``completed_tracks`` /
``total_tracks`` counts, and ``active_tracks`` labels. Each track's fraction is monotonic, so
the percentage only climbs. The original ``inner`` callable is always invoked.
+
+ A tick that only moves the numbers reaches the sink once per ``PROGRESS_EMIT_INTERVAL``.
+ A tick that changes the phase, the started or finished tracks, or fills a track goes to the
+ sink at once, because dropping one leaves the job looking stuck.
"""
total = len(inner_callables)
weights = [track_progress_weight(t) for t in tracks]
@@ -73,8 +79,11 @@ def build_job_progress_callables(
seg_done = [0.0] * total
seg_total = [0.0] * total
speeds: list[Optional[str]] = [None] * total
+ last_emit = 0.0
+ last_shape: Optional[tuple[Any, ...]] = None
def emit() -> None:
+ nonlocal last_emit, last_shape
completed = sum(done)
# Downloads fill 0..DOWNLOAD_PROGRESS_CEILING; dl.result drives muxing up to 100.
progress = sum(w * f for w, f in zip(weights, fractions)) * DOWNLOAD_PROGRESS_CEILING / total_weight
@@ -86,6 +95,12 @@ def build_job_progress_callables(
else:
phase = f"downloading {completed}/{total} tracks"
active_i = next((i for i in range(total) if started[i] and not done[i]), None)
+ shape = (phase, tuple(started), tuple(done), tuple(f >= 1.0 for f in fractions))
+ now = time.monotonic()
+ if shape == last_shape and now - last_emit < PROGRESS_EMIT_INTERVAL:
+ return
+ last_emit = now
+ last_shape = shape
sink(
{
"progress": progress,
diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py
index 107bed8c..8206ee6f 100644
--- a/unshackle/core/api/routes.py
+++ b/unshackle/core/api/routes.py
@@ -2136,10 +2136,10 @@ async def dashboard_session_logs(request: web.Request) -> web.Response:
description: >
The service's own log output for a remote session, mirrored at INFO regardless of the
server's log level. This is where the real reason for a failed authentication sits, in
- full, while the session summary carries only a truncated `auth_error`.
- Reading this does not refresh the session's idle timer and does not take records from
- the client draining the same buffer through `/api/session/{session_id}/logs`.
- Poll when the session summary's `log_seq` changes.
+ full, while the remote session summary carries only a truncated `auth_error`.
+ Reading this does not refresh the remote session's idle timer and does not take records
+ from the client draining the same buffer through `/api/session/{session_id}/logs`.
+ Poll when the remote session summary's `log_seq` changes.
tags: [Dashboard]
parameters:
- name: session_id
@@ -2193,14 +2193,14 @@ async def dashboard_keys(request: web.Request) -> web.Response:
---
summary: Dashboard API keys
description: >
- Every key in `serve.users`, plus `serve.api_secret` and the dashboard key when they are
- configured, with the grants that decide what it may do and the counters for what it has
- done. A key listed in more than one of those places still gets exactly one row.
- `id` is a hash prefix, stable across restarts and carrying no key material, so two
- unnamed keys never merge the way they do in the `requests_by_key` labels.
- Every key the server counts has a row here, so a `requests_by_key` bucket other than
+ Every API key in `serve.users`, plus `serve.api_secret` and the dashboard API key when
+ they are configured, with the grants that decide what it may do and the counters for what
+ it has done. An API key listed in more than one of those places still gets exactly one row.
+ `id` is a hash prefix, stable across restarts and carrying no API key material, so two
+ unnamed API keys never merge the way they do in the `requests_by_key` labels.
+ Every API key the server counts has a row here, so a `requests_by_key` bucket other than
`anonymous` always matches one.
- `bytes_out` counts response bodies only: an SSE stream reports no content length and
+ `bytes_out` counts response bodies only: an SSE stream reports no `Content-Length` and
contributes nothing.
tags: [Dashboard]
responses:
@@ -2229,9 +2229,9 @@ async def dashboard_keys(request: web.Request) -> web.Response:
items:
type: string
description: >
- Effective allowlist; null when nothing restricts the key, and an empty
- list when the key reaches no service route at all, as a dashboard key
- with no `serve.users` entry does
+ Effective allowlist; null when nothing restricts the API key, and an
+ empty list when the API key reaches no service route at all, as a
+ dashboard API key with no `serve.users` entry does
server_cdm:
description: false, true, or the list of service tags it covers
server_accounts:
@@ -2329,10 +2329,10 @@ async def dashboard_health(request: web.Request) -> web.Response:
---
summary: Dashboard health preflight
description: >
- Whether this instance could actually finish a download: the binaries on PATH, the CDM
- device files, each configured key vault and the proxy providers.
+ Whether this instance could finish a download: the binaries on PATH, the CDM device
+ files, each configured key vault and the proxy providers.
A panel, not a liveness probe. The result is cached for 30 seconds and every probe is
- shallow, so reading it never spends a proxy session or a licence. A provider that only
+ shallow, so reading it never spends a proxy session or a licence. A dependency that only
fails on first use is therefore not caught here.
tags: [Dashboard]
responses:
diff --git a/unshackle/core/api/session_store.py b/unshackle/core/api/session_store.py
index 2d5b16cd..2989d0a0 100644
--- a/unshackle/core/api/session_store.py
+++ b/unshackle/core/api/session_store.py
@@ -37,6 +37,7 @@ class SessionEntry:
tracks: Dict[str, Track] = field(default_factory=dict)
tracks_by_title: Dict[str, Dict[str, Track]] = field(default_factory=dict)
chapters_by_title: Dict[str, List[Any]] = field(default_factory=dict)
+ lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False, compare=False)
creator_ip: Optional[str] = None
owner_key: Optional[str] = None # X-Secret-Key that owns this session
cache_tag: Optional[str] = None
@@ -162,10 +163,10 @@ class SessionStore:
return entry
def peek(self, session_id: str) -> Optional[SessionEntry]:
- """A session entry without touching it, for read-only observers.
+ """A remote session entry without touching it, for read-only observers.
``get`` refreshes ``last_accessed`` and expires stale entries, so an observer polling
- through it would keep an idle session alive and always report it as active.
+ through it would keep an idle remote session alive and always report it as active.
"""
return self._sessions.get(session_id)
diff --git a/unshackle/core/api/stats.py b/unshackle/core/api/stats.py
index 4dd0a030..8a5a6b11 100644
--- a/unshackle/core/api/stats.py
+++ b/unshackle/core/api/stats.py
@@ -29,7 +29,7 @@ def _user_config(key: Optional[str]) -> Optional[Dict[str, Any]]:
def configured_key(key: Optional[str]) -> Optional[str]:
- """*key* when the config knows it (a user key, the master secret or the dashboard key), else None.
+ """``key`` when the config knows it (a user API key, the master secret or the dashboard API key), else None.
Open routes answer 200 to any header value, so attributing by the raw header would let an
unauthenticated caller grow ``stats.keys`` without bound.
@@ -63,7 +63,7 @@ def key_id(key: Optional[str]) -> str:
def key_tier(key: Optional[str]) -> Optional[str]:
- """The tier name a key references, when it names one that exists."""
+ """The tier name an API key references, when it names one that exists."""
user = _user_config(key)
tier = user.get("tier") if user else None
return str(tier) if tier else None
@@ -141,7 +141,7 @@ class ServerStats:
return counter
def check_rate_limit(self, key: str) -> Optional[int]:
- """Seconds to wait when *key* is over its hourly limit, else None after counting the request.
+ """Seconds to wait when ``key`` is over its hourly limit, else None after counting the request.
A fixed window, not a sliding one: cheap, and an operator cap does not need the precision.
"""
diff --git a/unshackle/core/cdm/custom_remote_cdm.py b/unshackle/core/cdm/custom_remote_cdm.py
index 6fac8dd2..ab48e0e0 100644
--- a/unshackle/core/cdm/custom_remote_cdm.py
+++ b/unshackle/core/cdm/custom_remote_cdm.py
@@ -237,8 +237,10 @@ class CustomRemoteCDM:
"""Mock certificate chain for PlayReady compatibility."""
return MockCertificateChain(f"{self.device_name}_Custom_Remote")
- def set_pssh_b64(self, pssh_b64: str) -> None:
+ def set_pssh_b64(self, pssh_b64: str, session_id: Optional[bytes] = None) -> None:
"""Store base64-encoded PSSH data for PlayReady compatibility."""
+ if session_id is not None and session_id in self._sessions:
+ self._sessions[session_id]["pssh_b64"] = pssh_b64
self._pssh_b64 = pssh_b64
def set_required_kids(self, kids: List[Union[str, UUID]], session_id: Optional[bytes] = None) -> None:
@@ -266,10 +268,11 @@ class CustomRemoteCDM:
"""Make a unique session ID."""
return secrets.token_bytes(16)
- def get_init_data_from_pssh(self, pssh: Any) -> str:
+ def get_init_data_from_pssh(self, pssh: Any, pssh_b64: Optional[str] = None) -> str:
"""Extract init data from various PSSH formats."""
- if self.is_playready and self._pssh_b64:
- return self._pssh_b64
+ effective_pssh_b64 = pssh_b64 or self._pssh_b64
+ if self.is_playready and effective_pssh_b64:
+ return effective_pssh_b64
if hasattr(pssh, "dumps"):
dumps_result = pssh.dumps()
@@ -653,10 +656,12 @@ class CustomRemoteCDM:
Session identifier as bytes
"""
session_id = self.generate_session_id()
+ self._pssh_b64 = None
self._sessions[session_id] = {
"service_certificate": None,
"keys": [],
"pssh": None,
+ "pssh_b64": None,
"challenge": None,
"remote_session_id": None,
"tried_cache": False,
@@ -781,7 +786,7 @@ class CustomRemoteCDM:
session = self._sessions[session_id]
session["pssh"] = pssh_or_wrm
- init_data = self.get_init_data_from_pssh(pssh_or_wrm)
+ init_data = self.get_init_data_from_pssh(pssh_or_wrm, session.get("pssh_b64"))
required_kids_list = session.get("required_kids") or self._required_kids
@@ -930,7 +935,7 @@ class CustomRemoteCDM:
license_message = license_message.encode("utf-8")
pssh = session["pssh"]
- init_data = self.get_init_data_from_pssh(pssh)
+ init_data = self.get_init_data_from_pssh(pssh, session.get("pssh_b64"))
license_request_b64 = base64.b64encode(session["challenge"]).decode("utf-8")
license_response_b64 = base64.b64encode(license_message).decode("utf-8")
diff --git a/unshackle/core/drm/monalisa.py b/unshackle/core/drm/monalisa.py
index 22ed4463..e0c92838 100644
--- a/unshackle/core/drm/monalisa.py
+++ b/unshackle/core/drm/monalisa.py
@@ -182,7 +182,7 @@ class MonaLisa:
@property
def key_pair(self) -> str:
- """Get formatted KID:KEY pair for CLI decryptors (32hex:32hex)."""
+ """The formatted ``KID:KEY`` for CLI decryptors (32hex:32hex)."""
if self._kid and self._key:
return f"{self._kid.hex}:{self._key}"
return ""
diff --git a/unshackle/core/manifests/dash.py b/unshackle/core/manifests/dash.py
index 5ca43bb9..db6a1f57 100644
--- a/unshackle/core/manifests/dash.py
+++ b/unshackle/core/manifests/dash.py
@@ -390,6 +390,7 @@ class DASH:
track=track,
track_url=track.url,
session=session,
+ probe_kid=period_idx == 0,
)
if period_idx == 0:
@@ -397,9 +398,6 @@ class DASH:
init_data = p_init
track_kid = p_kid
segment_timescale = p_timescale
- else:
- if p_kid and track_kid and p_kid != track_kid:
- log.debug(f"Period {content_period.get('id', period_idx)} has different KID: {p_kid}")
for seg in p_segments:
if seg not in seen_segments:
@@ -749,6 +747,7 @@ class DASH:
track: AnyTrack,
track_url: str,
session: Union[Session, RnetSession],
+ probe_kid: bool = True,
) -> tuple[
Optional[bytes],
list[tuple[str, Optional[str]]],
@@ -759,6 +758,12 @@ class DASH:
"""
Extract segments from a single period's representation.
+ Parameters:
+ probe_kid: Probe the initialization segment for the Key ID. Set it to False when
+ the caller needs neither the init data nor the Key ID of this period. False
+ drops the FFprobe call, and also the init request where the rest of the parse
+ does not need those bytes.
+
Returns:
A tuple of (init_data, segments, segment_timescale, segment_durations, track_kid).
"""
@@ -810,7 +815,7 @@ class DASH:
segment_template.set(item, value)
init_url = segment_template.get("initialization")
- if init_url:
+ if init_url and probe_kid:
res = session.get(
DASH.replace_fields(
init_url, Bandwidth=representation.get("bandwidth"), RepresentationID=representation.get("id")
@@ -875,7 +880,7 @@ class DASH:
init_data = None
initialization = segment_list.find("Initialization")
- if initialization is not None:
+ if initialization is not None and probe_kid:
source_url = initialization.get("sourceURL")
if not source_url:
source_url = rep_base_url
@@ -915,7 +920,8 @@ class DASH:
res = session.get(url=rep_base_url, headers=init_range_header)
res.raise_for_status()
init_data = res.content
- track_kid = track.get_key_id(init_data)
+ if probe_kid:
+ track_kid = track.get_key_id(init_data)
total_size = res.headers.get("Content-Range", "").split("/")[-1]
if total_size:
media_range = f"{len(init_data)}-{total_size}"
diff --git a/unshackle/core/manifests/hls.py b/unshackle/core/manifests/hls.py
index 7dc701a5..25a2651e 100644
--- a/unshackle/core/manifests/hls.py
+++ b/unshackle/core/manifests/hls.py
@@ -1022,7 +1022,8 @@ class HLS:
if isinstance(track, Subtitle):
segment_file_ext = get_extension(segment.uri) or ""
segment_file_path = segment_save_dir / f"{str(i).zfill(name_len)}{segment_file_ext}"
- segment_data = try_ensure_utf8(segment_file_path.read_bytes())
+ original_data = segment_file_path.read_bytes()
+ segment_data = try_ensure_utf8(original_data)
if track.codec not in (Subtitle.Codec.fVTT, Subtitle.Codec.fTTML):
segment_data = (
segment_data.decode("utf8")
@@ -1030,7 +1031,8 @@ class HLS:
.replace("", html.unescape(""))
.encode("utf8")
)
- segment_file_path.write_bytes(segment_data)
+ if segment_data != original_data:
+ segment_file_path.write_bytes(segment_data)
if segment.discontinuity and i != 0:
if encryption_data:
@@ -1210,7 +1212,11 @@ class HLS:
if binaries.FFMPEG:
try:
demuxer_file = save_path.parent / f"ffmpeg_concat_demuxer_{save_path.stem}.txt"
- demuxer_file.write_text("\n".join([f"file '{segment.absolute()}'" for segment in segments]))
+ demuxer_file.write_text(
+ "\n".join([f"file '{segment.absolute()}'" for segment in segments]),
+ encoding="utf-8",
+ newline="",
+ )
concat_start = time.monotonic()
subprocess.run(
diff --git a/unshackle/core/proxies/resolve.py b/unshackle/core/proxies/resolve.py
index 4b49e42c..0d9a5eb0 100644
--- a/unshackle/core/proxies/resolve.py
+++ b/unshackle/core/proxies/resolve.py
@@ -19,9 +19,9 @@ log = logging.getLogger("proxies")
def initialize_proxy_providers(raise_errors: bool = False, quiet: bool = False) -> List[Any]:
"""Initialise and return available proxy providers from config.
- A provider that fails to build is logged and skipped, unless *raise_errors* is set.
- *quiet* drops the per-provider summary lines: rendering a provider asks some of them for
- their server catalog over the network, which a repeated caller must not pay for.
+ A proxy provider that fails to build is logged and skipped, unless *raise_errors* is set.
+ *quiet* drops the per-provider summary lines: rendering a proxy provider asks some of them
+ for their server catalogue over the network, which a repeated caller must not pay for.
"""
proxy_providers: list = []
try:
diff --git a/unshackle/core/service.py b/unshackle/core/service.py
index 6e3f1cd9..e99af558 100644
--- a/unshackle/core/service.py
+++ b/unshackle/core/service.py
@@ -74,13 +74,15 @@ class TimeoutHTTPAdapter(HTTPAdapter):
cannot distinguish the two, and rnet has no unbounded mode either.
"""
+ __attrs__ = [*HTTPAdapter.__attrs__, "default_timeout"]
+
def __init__(self, *args: Any, timeout: Any = DEFAULT_TIMEOUT, **kwargs: Any) -> None:
self.default_timeout = timeout
super().__init__(*args, **kwargs)
def send(self, request: Any, **kwargs: Any) -> Any:
if kwargs.get("timeout") is None:
- kwargs["timeout"] = self.default_timeout
+ kwargs["timeout"] = getattr(self, "default_timeout", DEFAULT_TIMEOUT)
return super().send(request, **kwargs)
@@ -90,23 +92,20 @@ def grow_session_pool(session: Any, size: int) -> None:
The worker threads of every track draw on this one pool, because the downloader never
remounts an HTTP session the caller passes in (see downloaders/requests.py). The pool must hold
``downloads * workers`` connections, or threads queue for a slot instead of reading.
- Call this before any download thread exists: a remount races with other threads that call
+ Call this before any download thread exists: the rebuild races with other threads that call
``get_adapter``. RnetSession does not block on its idle-pool cap, so this function skips it.
+
+ Every mounted adapter grows in place, through its own ``init_poolmanager``. A service can
+ mount an :class:`HTTPAdapter` subclass, such as ``SSLCiphers``, on any prefix. Mounting a new
+ adapter over it would drop that subclass state, and its TLS context with it.
"""
if not isinstance(session, requests.Session):
return
- adapter = session.get_adapter("https://")
- if not isinstance(adapter, HTTPAdapter) or getattr(adapter, "_pool_maxsize", 0) >= size:
- return
- grown = TimeoutHTTPAdapter(
- max_retries=adapter.max_retries,
- pool_connections=size,
- pool_maxsize=size,
- pool_block=True,
- timeout=getattr(adapter, "default_timeout", DEFAULT_TIMEOUT),
- )
- session.mount("https://", grown)
- session.mount("http://", grown)
+ for adapter in {id(a): a for a in session.adapters.values()}.values():
+ if not isinstance(adapter, HTTPAdapter) or getattr(adapter, "_pool_maxsize", 0) >= size:
+ continue
+ adapter.init_poolmanager(size, size, block=True)
+ adapter.proxy_manager.clear()
@dataclass
@@ -410,7 +409,7 @@ class Service(metaclass=ABCMeta):
def get_binaries() -> list[dict]:
"""
Declare custom binary dependencies required by this service.
- :returns: List of dicts specifying name, candidates, desc, etc.
+ :returns: List of dicts, each with a ``name`` and optional ``candidates`` and ``desc``.
"""
return []