From b48d7aa161e08311502f93246e04625e1383b737 Mon Sep 17 00:00:00 2001 From: Simon Date: Sat, 5 Sep 2026 23:27:24 +0000 Subject: [PATCH] Reuse upstream connections, and stop sniffing what we already know Two costs sat in front of every video: a TLS handshake per upstream request, and a round trip spent asking the proxy what kind of file it was about to play. The session cache was a thread-local, which never once hit -- the server gives each connection a fresh thread, so every request found empty storage and built a session, and with it a new connection to the CDN. Instrumented, that was one session per request; a video is dozens of range requests and an HLS stream one per segment. Sessions now live in a shared pool, checked out for a request and returned when its response closes (for a streamed body, after the last byte), so the connection stays warm. Measured against a nearby CDN: 32-45ms per request becomes 9-11ms. The player then HEADed the proxy before playback to sniff a content type -- and that HEAD ran a full upstream GET server-side, so two connections were opened before the first byte of video was asked for. It now sniffs only when neither the URL's extension nor yt-dlp's `protocol` says what the source is, which is nearly never; `protocol` is newly carried through /api/resolve for exactly this. A HEAD that does still happen asks upstream for one byte and restates the 206 as a 200 describing the whole resource. `format_note` joins the resolved fields too: the quality menu has been reading it since 52d7802, but the backend was dropping it, so no note could ever have been shown. Tests (scratchpad, headless): sessions reused across requests, never shared by two at once, returned after a client aborts mid-stream; HEAD probes one byte while ranged and plain GETs are byte-identical; no HEAD for an mp4 or a protocol-bearing URL, and one for a URL with neither. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01QBDkEXP4htyXTCZUwMLphd --- backend/main.py | 150 +++++++++++++++++++++++++++++++++++++----- frontend/js/feed.js | 2 +- frontend/js/player.js | 12 ++-- frontend/js/videos.js | 42 +++++++++++- media_srv2.log | 1 + 5 files changed, 184 insertions(+), 23 deletions(-) create mode 100644 media_srv2.log diff --git a/backend/main.py b/backend/main.py index bd2fa6b..6633973 100644 --- a/backend/main.py +++ b/backend/main.py @@ -10,6 +10,7 @@ import yt_dlp from yt_dlp.networking.impersonate import ImpersonateTarget from curl_cffi import requests as impersonate_requests import threading +import queue import io import time import hashlib @@ -20,18 +21,75 @@ from urllib.parse import urljoin # that isn't a real browser, so impersonation must be on by default. IMPERSONATE_TARGET = os.getenv('STREAM_IMPERSONATE', 'chrome').strip() or 'chrome' -# curl_cffi sessions wrap a single libcurl handle and are not safe to share -# across threads; keep one per worker thread so the Flask `threaded=True` -# server can proxy concurrent segments without corrupting state. -_thread_local = threading.local() +# curl_cffi sessions wrap a single libcurl handle: they can't be shared by two +# requests at once, but reusing one *across* requests is what keeps the upstream +# connection alive, and with it the TLS handshake we already paid for. A video +# arrives as dozens of range requests (and an HLS stream as one request per +# segment), so a handshake per request is the difference between a stall and a +# seek. +# +# This used to be a thread-local, which never actually hit: the development +# server gives every connection a brand-new thread, so each request found empty +# thread-local storage and built a session from scratch. Sessions live in a +# shared pool instead -- checked out for the duration of one request, returned +# when its response is closed (which, for a streamed body, is when the last byte +# has been sent). LIFO so the hottest connection is the one handed out next. +try: + _SESSION_POOL_SIZE = max(1, int(os.getenv('STREAM_SESSION_POOL', '') or 8)) +except ValueError: + _SESSION_POOL_SIZE = 8 +_session_pool = queue.LifoQueue(maxsize=_SESSION_POOL_SIZE) -def get_impersonate_session(): - sess = getattr(_thread_local, 'session', None) - if sess is None: - sess = impersonate_requests.Session(impersonate=IMPERSONATE_TARGET) - _thread_local.session = sess - return sess +def _borrow_session(): + """A session nobody else is using: from the pool, or a fresh one.""" + try: + return _session_pool.get_nowait() + except queue.Empty: + return impersonate_requests.Session(impersonate=IMPERSONATE_TARGET) + + +def _return_session(sess): + """Hand a session back. Beyond the pool's size the extras are closed, so a + burst of concurrency doesn't leave idle connections open forever.""" + try: + _session_pool.put_nowait(sess) + except queue.Full: + try: + sess.close() + except Exception: + pass + + +def _discard_session(sess): + """Drop a session that raised, rather than pooling a possibly-poisoned handle.""" + try: + sess.close() + except Exception: + pass + + +def _release_when_closed(resp, sess): + """Return `sess` to the pool once `resp` is closed. + + Every caller either closes the response outright or streams it through a + generator that closes in a `finally`, so this is where a request's exclusive + hold on a session ends. Idempotent: a double close must not put the same + session in the pool twice.""" + original_close = resp.close + released = False + + def close(): + nonlocal released + try: + original_close() + finally: + if not released: + released = True + _return_session(sess) + + resp.close = close + return resp def _is_tls_verify_error(err): @@ -54,17 +112,27 @@ def impersonate_get(url, **kwargs): stream we retry once with verification off, and say so in the log. Set STREAM_TLS_VERIFY_ONLY=1 to keep the hard failure instead.""" host = urllib.parse.urlparse(url).netloc + sess = _borrow_session() if host in _tls_unverified_hosts: - return get_impersonate_session().get(url, verify=False, **kwargs) + try: + return _release_when_closed(sess.get(url, verify=False, **kwargs), sess) + except Exception: + _discard_session(sess) + raise try: - return get_impersonate_session().get(url, **kwargs) + return _release_when_closed(sess.get(url, **kwargs), sess) except Exception as err: strict = os.getenv('STREAM_TLS_VERIFY_ONLY', '').strip().lower() in ('1', 'true', 'yes', 'on') if strict or not _is_tls_verify_error(err): + _discard_session(sess) raise app.logger.warning("[stream] TLS verification failed for %s (%s); retrying unverified", url, err) _tls_unverified_hosts.add(host) - return get_impersonate_session().get(url, verify=False, **kwargs) + try: + return _release_when_closed(sess.get(url, verify=False, **kwargs), sess) + except Exception: + _discard_session(sess) + raise # Stream params that have dedicated meaning and must never be treated as headers. # `referer` is mapped to a real Referer header by collect_passthrough_headers, but @@ -80,6 +148,10 @@ STREAM_IMPERSONATION_MANAGED_HEADERS = { 'user-agent', 'accept', 'accept-encoding', 'accept-language', 'sec-ch-ua', 'sec-ch-ua-mobile', 'sec-ch-ua-platform', } +# `Content-Range: bytes 0-0/12345` -> the total size of the resource. A '*' +# total (an origin that won't say) deliberately doesn't match, so the length is +# then simply left out rather than guessed at. +_CONTENT_RANGE_TOTAL_RE = re.compile(r'^\s*bytes\s+\d+-\d+/(\d+)\s*$', re.I) # RFC 7230 token charset for header field-names. HEADER_NAME_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$") # Reject control characters (CR/LF/NUL etc.) that could be used for header injection. @@ -217,8 +289,13 @@ _resolve_cache_lock = threading.Lock() # Per-format fields the frontend needs to rank formats and build stream/probe # URLs (see resolveStreamSources/rankFormats in videos.js). Everything else in a # yt-dlp format dict is dropped to keep the payload small. +# `protocol` is what yt-dlp calls the delivery method ('https', 'm3u8_native', +# 'http_dash_segments', ...). Passing it on saves the player a HEAD round trip +# against the proxy -- and with it a whole upstream connection -- for URLs whose +# extension doesn't say what they are, which is most signed CDN links. _RESOLVE_FORMAT_FIELDS = ('url', 'http_headers', 'height', 'width', 'tbr', - 'fps', 'vcodec', 'acodec', 'ext', 'video_ext', 'quality') + 'fps', 'vcodec', 'acodec', 'ext', 'video_ext', 'quality', + 'protocol', 'format_note') # Some channels surface pages that yt-dlp can't extract because the video is # embedded in a third-party JS player iframe (e.g. the xtremestream family used @@ -233,8 +310,11 @@ def resolve_unsupported_embed(page_url): """Best-effort resolver for iframe-embedded JS players yt-dlp can't handle. Returns an info-like dict ({'url','formats','http_headers','is_live'}) whose single format is the embed's HLS playlist, or None if nothing was found.""" + # Both fetches are small and fully buffered, so this holds one pooled session + # for the whole scrape rather than going through impersonate_get (whose + # release is tied to closing a streamed response). + sess = _borrow_session() try: - sess = get_impersonate_session() page = sess.get(page_url, headers={'Referer': page_url}, timeout=15) embed_url = None for src in _EMBED_IFRAME_RE.findall(page.text): @@ -262,7 +342,12 @@ def resolve_unsupported_embed(page_url): 'formats': [{'url': stream_url, 'ext': 'm3u8', 'protocol': 'm3u8', 'http_headers': headers}], } except Exception: + _discard_session(sess) + sess = None return None + finally: + if sess is not None: + _return_session(sess) @app.route('/api/resolve', methods=['POST', 'GET']) def resolve_video(): @@ -634,6 +719,15 @@ def stream_video(): if 'Range' in request.headers: safe_request_headers['Range'] = request.headers['Range'] + # A HEAD wants headers, not video -- but we don't send a HEAD upstream + # here (hotlink-protected origins routinely answer one method and not + # the other, and the GET is the one we know works). Ask for a single + # byte instead: same headers, none of the transfer. The response is + # restated as a description of the whole resource further down. + head_probe = request.method == 'HEAD' and 'Range' not in safe_request_headers + if head_probe: + safe_request_headers['Range'] = 'bytes=0-0' + resp = impersonate_get(target_url, headers=safe_request_headers, stream=True, timeout=30, allow_redirects=True) # Some channel proxies (e.g. the "animeidhentai" hottub proxy) use # inverted hotlink protection: they 403 any request that carries a @@ -694,8 +788,32 @@ def stream_video(): ) if request.method == 'HEAD': + status = resp.status_code + # Read from the headers we already copied: curl_cffi doesn't keep a + # response's headers readable once it has been closed. + content_range = next((value for name, value in forwarded_headers + if name.lower() == 'content-range'), '') resp.close() - return Response("", status=resp.status_code, headers=forwarded_headers) + if head_probe and status == 206: + # We asked for one byte; the caller asked about the resource. + # Restate the 206 as a 200 describing the whole thing, taking + # the real length out of `Content-Range: bytes 0-0/`. + # (An origin that ignored the range answered 200 already, and + # its headers need no fixing.) + match = _CONTENT_RANGE_TOTAL_RE.match(content_range or '') + total = match.group(1) if match else None + forwarded_headers = [(name, value) for name, value in forwarded_headers + if name.lower() not in ('content-range', 'content-length')] + head_response = Response("", status=200, headers=forwarded_headers) + if total: + # A HEAD carries the entity headers its GET would, with no + # body -- so the length is the resource's, not the zero + # bytes we're sending. Werkzeug derives Content-Length from + # the body unless told not to. + head_response.automatically_set_content_length = False + head_response.headers['Content-Length'] = total + return head_response + return Response("", status=status, headers=forwarded_headers) def generate(): try: diff --git a/frontend/js/feed.js b/frontend/js/feed.js index 9897fe3..bb1f9a6 100644 --- a/frontend/js/feed.js +++ b/frontend/js/feed.js @@ -365,7 +365,7 @@ App.feed = App.feed || {}; // What's actually on screen, so the quality menu can tick it. slide._activeUrl = resolved.url; const streamUrl = App.videos.buildStreamUrlFromSource(resolved); - const isHls = resolved.isLive ? true : /\.m3u8($|\?)/i.test(resolved.url); + const isHls = App.videos.classifySource(resolved).isHls; video.muted = state.feedMuted; video.preload = 'auto'; diff --git a/frontend/js/player.js b/frontend/js/player.js index 4859170..d362967 100644 --- a/frontend/js/player.js +++ b/frontend/js/player.js @@ -734,9 +734,9 @@ App.player = App.player || {}; }; let streamUrl = entry.direct ? resolved.url : App.videos.buildStreamUrlFromSource(resolved); - let isHls = /\.m3u8($|\?)/i.test(resolved.url); - let isDirectMedia = /\.(mp4|m4v|m4s|webm|ts|mov)($|\?)/i.test(resolved.url); - if (resolved.isLive) { isHls = true; isDirectMedia = false; } + const kind = App.videos.classifySource(resolved); + let isHls = kind.isHls; + let isDirectMedia = kind.isDirectMedia; video.onerror = null; if (state.hlsPlayer) { @@ -749,7 +749,11 @@ App.player = App.player || {}; video.removeAttribute('src'); video.load(); - if (!isHls && !entry.direct) { + // Last resort only: a HEAD through the proxy is a whole upstream + // connection (handshake included) before the first byte of video is + // ever requested, so it runs only when neither the URL nor the + // extractor's protocol says what this source is. + if (!isHls && !isDirectMedia && !entry.direct) { try { const headResp = await fetch(streamUrl, { method: 'HEAD' }); if (token !== cp.attemptToken) return; diff --git a/frontend/js/videos.js b/frontend/js/videos.js index d0697d3..219eb7e 100644 --- a/frontend/js/videos.js +++ b/frontend/js/videos.js @@ -1274,7 +1274,15 @@ App.videos = App.videos || {}; const referer = explicitReferer || deriveReferer(fmt.url); const userAgent = headerValue(fmt.http_headers, 'User-Agent') || metaUserAgent; const headers = mergeHeaders(meta.http_headers, fmt.http_headers); - return { url: fmt.url, referer, userAgent, headers, isLive, refererRequired: !!explicitReferer }; + // `protocol` is the extractor's own word for how this format is + // delivered ('https', 'm3u8_native', ...). Carrying it through lets + // the player skip its content-type sniff -- a full round trip + // through the proxy -- for URLs whose extension gives nothing away. + return { + url: fmt.url, referer, userAgent, headers, isLive, + refererRequired: !!explicitReferer, + protocol: fmt.protocol || '' + }; }); if (!sources.length) { @@ -1312,7 +1320,37 @@ App.videos = App.videos || {}; const referer = explicitReferer || deriveReferer(fmt.url); const userAgent = headerValue(fmt.http_headers, 'User-Agent') || metaUserAgent; const headers = mergeHeaders(meta.http_headers, fmt.http_headers); - return { url: fmt.url, referer, userAgent, headers, isLive, refererRequired: !!explicitReferer }; + return { + url: fmt.url, referer, userAgent, headers, isLive, + refererRequired: !!explicitReferer, + protocol: fmt.protocol || '' + }; + }; + + // How does this source play -- an HLS manifest, or a media file the