hl-valdemar/terebi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Releases Wiki Activity

fix: avoid unbounded HTTP throttling and make buffer readout opt-in

Browse Source
This commit is contained in:
Valdemar Høgsvig Lorenzen
2026-10-11 11:52:48 +02:00
parent 13554a1574
commit 537e8e2624
6 changed files with 118 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Checks/streaming.py
+24 -11
View File
@@ -24,6 +24,7 @@ class _Gate:
self.requests = 0
self.ranges = 0
self.bytes = 0
self.connections = set()
class StreamingServer:
@@ -60,10 +61,14 @@ class StreamingServer:
pass # Media URLs/headers never appear in logs.
def do_GET(self):
owner._get(self)
def handle(self):
try:
owner._get(self)
super().handle()
except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
pass # The engine intentionally cancels active media connections.
# Cancellation can close a body read or an idle keep-alive connection.
self.close_connection = True
self.http = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
self.http.daemon_threads = True
@@ -163,19 +168,20 @@ class StreamingServer:
gate = self.gates.get(parts[2])
if gate is None:
body = {"entered": False, "released": False, "waiting": 0, "requests": 0,
"ranges": 0, "bytes": 0, "total": 0, "prefix_ends": {}}
"ranges": 0, "bytes": 0, "total": 0, "connections": 0, "prefix_ends": {}}
else:
if parts[1] == "release":
gate.release.set()
body = {"entered": gate.entered.is_set(), "released": gate.release.is_set(),
"waiting": gate.waiting, "requests": gate.requests, "ranges": gate.ranges,
"bytes": gate.bytes, "total": gate.total, "prefix_ends": gate.prefix_ends}
"bytes": gate.bytes, "total": gate.total, "connections": len(gate.connections),
"prefix_ends": gate.prefix_ends}
self._reply(request, 200, json.dumps(body).encode())
return
mode = parts[0]
token = None
if len(parts) == 3 and mode in ("stall", "stall-open"):
if len(parts) == 3 and mode in ("stall", "stall-open", "chunked-media"):
token, name = parts[1:]
if not re.fullmatch(r"[A-Za-z0-9_-]+", token):
self._reply(request, 400, b"Invalid gate token", "text/plain")
@@ -223,21 +229,27 @@ class StreamingServer:
query = parse_qs(parsed.query)
try:
gate_at = (self._timestamp_offset(name, float(query["after_seconds"][0]))
if mode == "stall" and "after_seconds" in query else
int(query.get("after", ["131072"])[0]) if mode == "stall" else start)
if mode in ("stall", "chunked-media") and "after_seconds" in query else
int(query.get("after", ["131072"])[0]) if mode in ("stall", "chunked-media") else start)
except (ValueError, OverflowError):
self._reply(request, 400, b"Invalid gate offset", "text/plain")
return
with self.lock:
if token not in self.gates:
prefix = self._prefix_ends(name, gate_at) if mode == "stall" else {}
prefix = self._prefix_ends(name, gate_at) if mode in ("stall", "chunked-media") else {}
self.gates[token] = _Gate(size, prefix)
gate = self.gates[token]
gate.requests += 1
gate.connections.add(request.client_address)
gate.ranges += int(range_header is not None)
# Only the original prefix connection is gated: a seek can create a new
# range request while the old read is still blocked and never released.
should_gate = not gate.claimed and start <= gate_at <= end
# Simulate a CDN that throttles oversized/open-ended ranges, while
# finite requests of at most 10 MiB run at full available speed.
oversized = mode == "chunked-media" and (
range_header is None or not range_header.split("-", 1)[1]
or int(range_header.split("-", 1)[1]) - start + 1 > 10 * 1024 * 1024)
should_gate = (mode != "chunked-media" or oversized) and not gate.claimed and start <= gate_at <= end
if should_gate:
gate.claimed = True
with self.lock:
@@ -249,10 +261,11 @@ class StreamingServer:
request.send_header("Content-Length", str(end - start + 1))
if range_header is not None:
request.send_header("Content-Range", f"bytes {start}-{end}/{size}")
request.send_header("Connection", "close")
persistent = mode == "chunked-media" and parse_qs(parsed.query).get("keep_alive") == ["1"]
request.send_header("Connection", "keep-alive" if persistent else "close")
request.end_headers()
request.wfile.flush()
request.close_connection = True
request.close_connection = not persistent
cursor = start
with path.open("rb") as media:
media.seek(start)