perf: keep-alive upstream sessions, TTL cache on upstream data, /stream LRU, cache hls.min.js

- shared thread-local curl_cffi sessions (no TLS handshake per call)
- in-memory TTL cache: search 5min, details/episodes/languages/
  playlist-URLs 1h; api_stream retries once with a cold cache on 502
  (recovers rotated playlist URLs)
- /stream LRU (24 entries) serves re-seeks and playlist re-fetches from memory
- hls.min.js: Cache-Control public max-age=30d (was no-store -> 500KB
  re-download per page load on mobile)

Measured (click->play chain, cold -> warm):
  search    0.45s -> 0.00s
  detail    0.46s -> 0.00s
  episodes  0.40s -> 0.00s
  stream    1.06s -> 0.00s
  m3u8 x2   0.64s -> 0.00s
  segment   0.20s -> 0.00s

server/app.py

@@ -9,6 +9,8 @@ import json
 import os
 import time
 from pathlib import Path
+from base64 import urlsafe_b64encode, urlsafe_b64decode
+from collections import OrderedDict
 from urllib.parse import quote, urljoin
 
 from fastapi import Body, FastAPI, HTTPException, Query, Request
@@ -179,9 +181,13 @@ def _upstream(fn, *a, **kw):
 
 @app.middleware("http")
 async def no_cache(request, call_next):
-    # ponytail: no-store on everything except /stream (segments should cache)
+    # ponytail: no-store on everything except /stream (immutable HLS
+    # data) and the vendored hls.min.js (immutable, 500KB — re-downloading
+    # it on every page load hurt on mobile)
     r = await call_next(request)
-    if not request.url.path.startswith("/stream"):
+    if request.url.path == "/hls.min.js":
+        r.headers["Cache-Control"] = "public, max-age=2592000"
+    elif not request.url.path.startswith("/stream"):
         r.headers["Cache-Control"] = "no-store"
     return r
 
@@ -203,11 +209,21 @@ def api_episodes(anime_id: int):
 
 @app.get("/api/episode/{ep_id}/stream")
 def api_stream(ep_id: int, lang: str = "eng"):
-    master = _upstream(upstream.master_m3u8, ep_id, lang)
-    if not master:
-        raise HTTPException(404, f"no {lang} stream for episode {ep_id}")
-    return {"token": _b64(master),
-            "qualities": _upstream(upstream.master_variants, master)}
+    def go():
+        master = _upstream(upstream.master_m3u8, ep_id, lang)
+        if not master:
+            raise HTTPException(404, f"no {lang} stream for episode {ep_id}")
+        return {"token": _b64(master),
+                "qualities": _upstream(upstream.master_variants, master)}
+    try:
+        return go()
+    except HTTPException as e:
+        # one-shot retry with a cold cache: recovers rotated/expired
+        # playlist URLs from the TTL cache
+        if e.status_code == 502:
+            upstream.clear_cache()
+            return go()
+        raise
 
 
 def _rewrite_m3u8(text: str, base: str) -> str:
@@ -219,14 +235,25 @@ def _rewrite_m3u8(text: str, base: str) -> str:
         for line in text.splitlines()) + "\n"
 
 
+_STREAM_CACHE = OrderedDict()  # ponytail: LRU of proxied HLS data (immutable);
+_STREAM_MAX = 24                # segments ~1-2MB -> <= ~48MB
+
+
 @app.get("/stream/{token}")
 def stream(token: str):
     # ponytail: buffers whole response in memory; .ts segments are small
     url = _url(token)
+    hit = _STREAM_CACHE.get(url)
+    if hit:
+        _STREAM_CACHE.move_to_end(url)
+        return Response(hit[1], media_type=hit[0])
     r = _upstream(upstream._get, url)
     body = r.content
     if url.endswith(".m3u8") or "mpegurl" in r.headers.get("content-type", ""):
         body = _rewrite_m3u8(body.decode(), url).encode()
+    _STREAM_CACHE[url] = (r.headers.get("content-type", "application/octet-stream"), body)
+    while len(_STREAM_CACHE) > _STREAM_MAX:
+        _STREAM_CACHE.popitem(last=False)
     return Response(body, media_type=r.headers.get("content-type", "application/octet-stream"))
 
 

server/upstream.py

@@ -7,24 +7,55 @@ breaks in one place, not across the app.
 """
 import re
 import sys
+import threading
+import time
 
 from curl_cffi import requests
 
 BASE = "https://anidb.app"
 H = dict(impersonate="chrome", timeout=20)
 
+# ponytail: thread-local sessions (uvicorn threadpool); curl_cffi Sessions
+# are not thread-safe, but per-thread keep-alive still reuses TCP/TLS across
+# the many requests a worker handles. Saves a handshake on every call.
+_tls = threading.local()
+
+def _session():
+    if not hasattr(_tls, "s"):
+        _tls.s = requests.Session()
+    return _tls.s
+
+# ponytail: in-memory TTL cache for upstream data that never changes
+# (details/episodes/langs/playlist URLs). No lock: worst case a duplicate
+# fetch. A dead/rotated playlist URL is recovered by the one-shot
+# clear_cache()+retry in api_stream.
+_CACHE = {}  # key -> (expires, value)
+
+def _cached(key, ttl, fn, *a, **kw):
+    v = _CACHE.get(key)
+    if v and v[0] > time.time():
+        return v[1]
+    val = fn(*a, **kw)
+    _CACHE[key] = (time.time() + ttl, val)
+    if len(_CACHE) > 256:
+        _CACHE.clear()
+    return val
+
+def clear_cache():
+    _CACHE.clear()
+
 _ANIME_LINK = re.compile(r'anime/([a-z0-9-]+)-(\d+)"[^>]*title="([^"]*)"')
 
 
 def _get(url, **kw):
-    r = requests.get(url, **H, **kw)
+    r = _session().get(url, **H, **kw)
     if r.status_code in (403, 503) or "Just a moment" in r.text[:500]:
         raise RuntimeError(f"blocked by cloudflare: {url}")
     r.raise_for_status()
     return r
 
 
-def search(q):
+def _search(q):
     """Step 1: browse page -> [{id, slug_id, name, poster}]"""
     html = _get(f"{BASE}/browse", params={"q": q}).text
     out, seen = [], set()
@@ -41,7 +72,7 @@ def search(q):
     return out
 
 
-def details(slug_id):
+def _details(slug_id):
     """Step 2: details page -> {name, synopsis, mal_id, poster, seasons}"""
     p = _get(f"{BASE}/anime/{slug_id}").text
     mal = re.search(r'myanimelist\.net/anime/(\d+)', p)
@@ -69,19 +100,19 @@ def details(slug_id):
             "seasons": seasons}
 
 
-def episodes(anime_id):
+def _episodes(anime_id):
     """Step 3: JSON API -> [{id, number, filler}]"""
     d = _get(f"{BASE}/api/frontend/anime/{anime_id}/episodes").json()
     return [{"id": e["id"], "number": e["number"], "filler": bool(e.get("filler"))}
             for e in d["episodes"]]
 
 
-def languages(ep_id):
+def _languages(ep_id):
     """Step 4: JSON API -> [{code, name, embed_url}] (jpn=sub, eng=dub)"""
     return _get(f"{BASE}/api/frontend/episode/{ep_id}/languages").json()["languages"]
 
 
-def master_m3u8(ep_id, lang="eng"):
+def _master_m3u8(ep_id, lang="eng"):
     """Steps 5+6: embed page -> 'file: <master m3u8>' URL (None if unavailable)"""
     for l in languages(ep_id):
         if l["code"] == lang:
@@ -91,12 +122,21 @@ def master_m3u8(ep_id, lang="eng"):
     return None
 
 
-def master_variants(master_url):
+def _master_variants(master_url):
     """Step 7: parse #EXT-X-STREAM-INF lines -> ['1920x1080', ...] (for display)"""
     text = _get(master_url).text
     return re.findall(r"RESOLUTION=(\d+x\d+)", text)
 
 
+# cached public API (TTLs in seconds)
+search = lambda q: _cached(f"search:{q.lower()}", 300, _search, q)
+details = lambda s: _cached(f"details:{s}", 3600, _details, s)
+episodes = lambda i: _cached(f"episodes:{i}", 3600, _episodes, i)
+languages = lambda e: _cached(f"languages:{e}", 3600, _languages, e)
+master_m3u8 = lambda e, l="eng": _cached(f"master:{e}:{l}", 3600, _master_m3u8, e, l)
+master_variants = lambda u: _cached(f"variants:{u}", 3600, _master_variants, u)
+
+
 if __name__ == "__main__":
     # ponytail: assert self-check instead of a test framework
     results = search("chainsaw man")