Proxy: stream HLS segments through instead of buffering
fragLoadTimeOut root cause: /stream buffered the whole segment (r.content) before sending, so the browser's time-to-first-byte = the CDN's entire upload time. On slow CDN moments that trips hls.js's first-byte timer even though the segment would have completed. - upstream.get_stream(): opens the upstream response without reading the body (stream=True), same retry/timeout rules as _get - segments now flow to the browser as the CDN emits them (256KB chunks); the 12s cap becomes a total-transfer cap; cache is populated on completion only (a truncated body never poisons the LRU) - playlists keep the buffered path (body needed for URI rewriting) - cache-hit fast path restored for both (stream() checks the LRU first) Verified: segment byte-identical to direct CDN (sha256), sequential keep-alive 3/3 OK, 5 concurrent fetches 200+identical, cache hit 0.148s miss -> 0.001s hit, 17/17 API tests green. Skipped: raising hls.js retry counts — the first-byte race was the binding constraint, not the retry budget.
This commit is contained in:
1 parent
27de918cb0
commit
19bc2afb32
2 files changed
+72
-22
No files matched your search
+53
-22
@@ -15,7 +15,7 @@ from base64 import urlsafe_b64encode, urlsafe_b64decode
|
|||||||
from urllib.parse import quote, urljoin, urlparse
|
from urllib.parse import quote, urljoin, urlparse
|
||||||
|
|
||||||
from fastapi import Body, FastAPI, HTTPException, Query, Request
|
from fastapi import Body, FastAPI, HTTPException, Query, Request
|
||||||
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response
|
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response, StreamingResponse
|
||||||
from fastapi.staticfiles import StaticFiles
|
from fastapi.staticfiles import StaticFiles
|
||||||
from itsdangerous import BadSignature, TimestampSigner
|
from itsdangerous import BadSignature, TimestampSigner
|
||||||
|
|
||||||
@@ -479,33 +479,64 @@ _STREAM_MAX = 24 # segments ~1-2MB -> <= ~48MB
|
|||||||
_STREAM_CACHE_HDRS = {"Cache-Control": "public, max-age=86400"}
|
_STREAM_CACHE_HDRS = {"Cache-Control": "public, max-age=86400"}
|
||||||
|
|
||||||
|
|
||||||
@app.get("/stream/{token}")
|
def _cache_put(url, ct, body):
|
||||||
def stream(token: str):
|
_STREAM_CACHE[url] = (ct, body)
|
||||||
# ponytail: buffers whole response in memory; .ts segments are small
|
while len(_STREAM_CACHE) > _STREAM_MAX:
|
||||||
url = _url(token)
|
_STREAM_CACHE.popitem(last=False)
|
||||||
# ponytail: token is plain base64, so allowlist the host — otherwise any
|
|
||||||
# logged-in user could use this as a GET proxy to arbitrary URLs (SSRF).
|
|
||||||
# Verified 2026-08: master, variant and segment URLs all live on
|
def _seg_stream(r, url, ct):
|
||||||
# hls.anidb.app; if the CDN moves, this 400s loudly and the list updates.
|
"""Forward the CDN's bytes as they arrive; keep the LRU warm for
|
||||||
p = urlparse(url)
|
completed downloads (a truncated one would poison the cache)."""
|
||||||
if p.scheme != "https" or p.netloc != "hls.anidb.app":
|
buf, complete = [], False
|
||||||
raise HTTPException(400, "bad stream token")
|
try:
|
||||||
hit = _STREAM_CACHE.get(url)
|
for c in r.iter_content(chunk_size=256 * 1024):
|
||||||
if hit:
|
buf.append(c)
|
||||||
_STREAM_CACHE.move_to_end(url)
|
yield c
|
||||||
return Response(hit[1], media_type=hit[0], headers=_STREAM_CACHE_HDRS)
|
complete = True
|
||||||
|
finally:
|
||||||
|
r.close()
|
||||||
|
if complete and buf:
|
||||||
|
_cache_put(url, ct, b"".join(buf))
|
||||||
|
|
||||||
|
|
||||||
|
def _stream_playlist(url):
|
||||||
|
# buffered: the full body is needed to rewrite URIs through the proxy
|
||||||
# ponytail: 12s < hls.js's fragLoadingTimeOut (20s) — on a hang the proxy
|
# ponytail: 12s < hls.js's fragLoadingTimeOut (20s) — on a hang the proxy
|
||||||
# 502s first, so hls.js retries the frag (backoff) instead of freezing
|
# 502s first, so hls.js retries the frag (backoff) instead of freezing
|
||||||
# on a fatal timeout race
|
# on a fatal timeout race
|
||||||
r = _upstream(upstream._get, url, timeout=12)
|
r = _upstream(upstream._get, url, timeout=12)
|
||||||
body = r.content
|
body = r.content
|
||||||
if url.endswith(".m3u8") or "mpegurl" in r.headers.get("content-type", ""):
|
if "mpegurl" in r.headers.get("content-type", ""):
|
||||||
body = _rewrite_m3u8(body.decode(), url).encode()
|
body = _rewrite_m3u8(body.decode(), url).encode()
|
||||||
_STREAM_CACHE[url] = (r.headers.get("content-type", "application/octet-stream"), body)
|
ct = r.headers.get("content-type", "application/octet-stream")
|
||||||
while len(_STREAM_CACHE) > _STREAM_MAX:
|
_cache_put(url, ct, body)
|
||||||
_STREAM_CACHE.popitem(last=False)
|
return Response(body, media_type=ct, headers=_STREAM_CACHE_HDRS)
|
||||||
return Response(body, media_type=r.headers.get("content-type", "application/octet-stream"),
|
|
||||||
headers=_STREAM_CACHE_HDRS)
|
|
||||||
|
@app.get("/stream/{token}")
|
||||||
|
def stream(token: str):
|
||||||
|
# ponytail: token is plain base64, so allowlist the host — otherwise any
|
||||||
|
# logged-in user could use this as a GET proxy to arbitrary URLs (SSRF).
|
||||||
|
# Verified 2026-08: master, variant and segment URLs all live on
|
||||||
|
# hls.anidb.app; if the CDN moves, this 400s loudly and the list updates.
|
||||||
|
url = _url(token)
|
||||||
|
p = urlparse(url)
|
||||||
|
if p.scheme != "https" or p.netloc != "hls.anidb.app":
|
||||||
|
raise HTTPException(400, "bad stream token")
|
||||||
|
# cache hit (a playlist, or a segment downloaded earlier): serve at once
|
||||||
|
hit = _STREAM_CACHE.get(url)
|
||||||
|
if hit:
|
||||||
|
_STREAM_CACHE.move_to_end(url)
|
||||||
|
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)
|
||||||
|
|
||||||
|
|
||||||
app.mount("/", StaticFiles(directory=ROOT / "static", html=True), name="static")
|
app.mount("/", StaticFiles(directory=ROOT / "static", html=True), name="static")
|
||||||
@@ -64,6 +64,25 @@ def _get(url, **kw):
|
|||||||
time.sleep(0.3)
|
time.sleep(0.3)
|
||||||
|
|
||||||
|
|
||||||
|
def get_stream(url, timeout=12):
|
||||||
|
"""Segments: open the upstream response WITHOUT reading the body, so the
|
||||||
|
proxy can forward bytes as the CDN emits them. (A buffered fetch made the
|
||||||
|
browser's time-to-first-byte = the CDN's whole upload time, which tripped
|
||||||
|
hls.js's fragLoadTimeOut on slow CDN moments.)
|
||||||
|
ponytail: skips the 'Just a moment' body check — challenge pages arrive
|
||||||
|
as 403/503 (handled below), and reading the body would defeat streaming."""
|
||||||
|
for attempt in range(2):
|
||||||
|
r = _session().get(url, **{**H, "timeout": timeout, "stream": True})
|
||||||
|
if r.status_code in (403, 429, 503) or r.status_code >= 500:
|
||||||
|
r.close()
|
||||||
|
if attempt:
|
||||||
|
raise RuntimeError(f"upstream {r.status_code}: {url}")
|
||||||
|
time.sleep(0.3)
|
||||||
|
continue
|
||||||
|
r.raise_for_status()
|
||||||
|
return r
|
||||||
|
|
||||||
|
|
||||||
def _search(q):
|
def _search(q):
|
||||||
"""Step 1: browse page -> [{id, slug_id, name, poster}]"""
|
"""Step 1: browse page -> [{id, slug_id, name, poster}]"""
|
||||||
html = _get(f"{BASE}/browse", params={"q": q}).text
|
html = _get(f"{BASE}/browse", params={"q": q}).text
|
||||||
|
|||||||
Reference in new issue
Block a user