diff --git a/server/app.py b/server/app.py index dec07d2..11de78b 100644 --- a/server/app.py +++ b/server/app.py @@ -545,12 +545,27 @@ def stream(token: str): return Response(hit[1], media_type=hit[0], headers=_STREAM_CACHE_HDRS) if url.endswith(".m3u8"): return _stream_playlist(url) - # segment miss: stream through (12s = total transfer cap), so the browser's - # first byte no longer waits for the CDN's whole upload - r = _upstream(upstream.get_stream, url, timeout=12) - ct = r.headers.get("content-type", "application/octet-stream") - return StreamingResponse(_seg_stream(r, url, ct), media_type=ct, - headers=_STREAM_CACHE_HDRS) + # segment miss: try parallel byte-range lanes first (4 concurrent fetches + # beat per-connection CDN throttling); ANY failure falls back to the + # single streaming path, so this can never be worse than before + t0 = time.time() + try: + body, ct = _upstream(upstream.fetch_parallel, url) + except HTTPException: + r = _upstream(upstream.get_stream, url, timeout=12) + ct = r.headers.get("content-type", "application/octet-stream") + if time.time() - t0 > 8: + name = url.rsplit("/", 1)[-1][:40] + print(f"[stream-slow] {name} parallel-fallback elapsed={time.time()-t0:.1f}s", + flush=True) + return StreamingResponse(_seg_stream(r, url, ct), media_type=ct, + headers=_STREAM_CACHE_HDRS) + if time.time() - t0 > 8: + name = url.rsplit("/", 1)[-1][:40] + print(f"[stream-slow] {name} parallel total={time.time()-t0:.1f}s " + f"bytes={len(body)}", flush=True) + _cache_put(url, ct, body) + return Response(body, media_type=ct, headers=_STREAM_CACHE_HDRS) app.mount("/", StaticFiles(directory=ROOT / "static", html=True), name="static") diff --git a/server/upstream.py b/server/upstream.py index 189e047..b7e3350 100644 --- a/server/upstream.py +++ b/server/upstream.py @@ -83,6 +83,46 @@ def get_stream(url, timeout=12): return r +def fetch_parallel(url, lanes=4, lane=256 * 1024, timeout=12): + """One upstream file across N parallel byte-range lanes. + + Per-connection CDN throttling is the common slow-segment cause; N lanes + turn one 60s crawl into N concurrent crawls (I/O-bound, GIL irrelevant). + Returns (body, content-type); raises on ANY failure so the caller falls + back to the single-request path — never worse than before. + Verified 2026-08 against hls.anidb.app: HEAD -> content-length, + ranges -> 206, 4 lanes -> byte-identical stitch (sha256).""" + from concurrent.futures import ThreadPoolExecutor + head = _session().head(url, **{**H, "timeout": timeout}) + head.raise_for_status() + total = int(head.headers.get("content-length") or 0) + ct = head.headers.get("content-type", "application/octet-stream") + if total == 0: + raise RuntimeError("no content-length") + if total <= 2 * lane: # too small to be worth N connections + r = _get(url, timeout=timeout) + return r.content, ct + n = min(lanes, total // lane) + size = (total + n - 1) // n + parts = [None] * n + + def grab(i): + a = i * size + b = total - 1 if i == n - 1 else (i + 1) * size - 1 + r = _get(url, headers={"Range": f"bytes={a}-{b}"}, timeout=timeout) + if r.status_code not in (200, 206) or len(r.content) != b - a + 1: + raise RuntimeError(f"lane {i}: http {r.status_code} " + f"len={len(r.content)} want={b - a + 1}") + parts[i] = r.content + + with ThreadPoolExecutor(max_workers=n) as ex: + list(ex.map(grab, range(n))) + body = b"".join(parts) + if len(body) != total: + raise RuntimeError(f"stitch mismatch {len(body)} != {total}") + return body, ct + + def _search(q): """Step 1: browse page -> [{id, slug_id, name, poster}]""" html = _get(f"{BASE}/browse", params={"q": q}).text diff --git a/tests/test_api.py b/tests/test_api.py index 3b3afdc..510bbf6 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -330,6 +330,7 @@ def test_intros_parses_jikan_shapes(): ]} u._CACHE.clear() + real_get = u._get u._get = lambda url: R() # stub the network assert u._intros(92573) == [ {"title": "Episode 1", "start": 111, "length": 77}, @@ -343,4 +344,62 @@ def test_intros_parses_jikan_shapes(): ]}} u._get = lambda url: W() assert u._intros(92573)[0]["start"] == 90 + u._get = real_get # restore: later tests need the real _get u._CACHE.clear() + + +def test_fetch_parallel_stitches_range_lanes(): + # fetch_parallel across N byte-range lanes must reassemble byte-identical, + # and must raise (so the app falls back to the single-request path) when + # the CDN refuses ranges or HEAD fails. + import server.upstream as u + + body = bytes((i * 7 + 3) % 256 for i in range(1024 * 1024)) # 1MiB > 2*lane, forces the lane path + + class Resp: + def __init__(self, content, status, clen=None): + self.content, self.status_code = content, status + self.headers = {"content-type": "video/mp2t"} + if clen is not None: + self.headers["content-length"] = str(clen) + def raise_for_status(self): + if self.status_code >= 400: + raise RuntimeError(f"http {self.status_code}") + @property + def text(self): + return "" + + class Ranges: + def head(self, url, **kw): + return Resp(b"", 200, len(body)) + def get(self, url, **kw): + rng = (kw.get("headers") or {}).get("Range") + if rng: + a, b = rng[len("bytes="):].split("-") + return Resp(body[int(a):int(b) + 1], 206) + return Resp(body, 200) + + class NoRanges: # 416 on every range -> must raise, app falls back + def head(self, url, **kw): + return Resp(b"", 200, len(body)) + def get(self, url, **kw): + if (kw.get("headers") or {}).get("Range"): + return Resp(b"", 416) + return Resp(body, 200) + + real = u._session + real_get = u._get + try: + u._session = lambda: Ranges() + out, ct = u.fetch_parallel("http://cdn/x.xls", lanes=4) + assert out == body and ct == "video/mp2t" + + u._session = lambda: NoRanges() + try: + u.fetch_parallel("http://cdn/x.xls", lanes=4) + raise AssertionError("should have raised") + except RuntimeError: + pass + finally: + u._session = real + u._get = real_get