Proxy: fetch segments across 4 parallel byte-range lanes

The slow-segment logs show per-connection CDN throttling (15-40 KB/s
dribbles). yt-dlp's standard fix is -N 16 parallel fragment lanes; this
is the 4-lane version for the proxy: HEAD for content-length, then 4
concurrent Range GETs reassembled byte-exact (verified sha256-identical
to a direct CDN fetch). Small files (<512KB) still use a single GET.

ANY failure (no content-length, 416, lane error, timeout) falls back to
the existing single-stream path, so this is never worse than before.

Caveat (measured live): when the CDN caps SHARED egress rather than
per-connection, lanes just split the same pie — in that window both
1-lane and 4-lane time out. A different source is the only real fix
for those windows.

18/18 tests (new test pins lane stitch + fallback; also fixed a
pre-existing _get stub leak in the Jikan-shape test).

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")

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

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