simon/jacuzzi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reuse upstream connections, and stop sniffing what we already know

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QBDkEXP4htyXTCZUwMLphd
This commit is contained in:
Simon
2026-09-05 23:27:24 +00:00
parent 52d7802491
commit b48d7aa161
5 changed files with 184 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
backend/main.py
View File

@@ -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/<total>`.
# (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: