Stream: Rechenlast des Proxys senken

Ein Zuschauer erzeugt rund zwanzig Anfragen je Sekunde. Bisher baute der
Proxy dafür jedes Mal eine neue Verbindung auf, holte die Antwort komplett
in den Speicher und schickte sie durch gzip — obwohl Videosegmente bereits
komprimiert sind. Gemessen: 47-67 % CPU für einen einzigen Zuschauer.

Jetzt ein gemeinsamer Client mit offenen Verbindungen, die Daten fließen
stückweise durch, und /stream bleibt von der Kompression verschont. Dazu
protokolliert die App die Stream-Anfragen nicht mehr einzeln; das waren
410 Zeilen je Minute bei einem Zuschauer.

Ergebnis: 11,5 % statt 47-67 %. Mit fünfzehn Zuschauern liegt die Last bei
150 % von 400 % verfügbarer Rechenzeit.

Dazu am Rand:
- hlsAlwaysRemux hält den Muxer durchgehend offen. Vorher riss er 60 s nach
  dem letzten Zuschauer ab und setzte beim Neuaufbau die Zeitbasis zurück.
- hlsAllowOrigin ist veraltet und heißt jetzt hlsAllowOrigins.
- Der Player holt höchstens mit 1,1-fachem Tempo auf statt mit 1,5. Das
  reicht für normale Schwankungen und bleibt unhörbar.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013XVnRop5c3or3B8xE2hpgZ

backend/app/main.py

@@ -13,29 +13,42 @@ from . import db, events, search, streamproxy, suggestions, sync, watched
 from .config import IMAGE_BASE, STATIC_DIR
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+# Ein einzelner Zuschauer erzeugt gut zwanzig Stream-Anfragen je Sekunde.
+# Jede davon zu protokollieren füllt bei fünfzehn Leuten die Platte und
+# sagt über den Betrieb nichts aus. Fehler kommen weiterhin durch.
+logging.getLogger("httpx").setLevel(logging.WARNING)
+
+
+class OhneStreamZugriffe(logging.Filter):
+    def filter(self, eintrag):
+        return "/stream/" not in eintrag.getMessage()
+
+
+logging.getLogger("uvicorn.access").addFilter(OhneStreamZugriffe())
 log = logging.getLogger("app")
 
 app = FastAPI(title="Netflix Horror DE")
 scheduler = AsyncIOScheduler()
 sync_lock = asyncio.Lock()
 
-class GzipOhneLivekanal(GZipMiddleware):
-    """Kompression für alles außer den Ereignisstrom.
+# Der Ereignisstrom darf nicht durch die Kompression, und Video auch nicht:
+# GZipMiddleware sammelt die Antwort, bevor sie ausgeht. Bei Server-Sent
+# Events kommt dadurch gar nichts an, solange der Puffer nicht voll ist —
+# und der wird bei kurzen Ereignissen nie voll. Videosegmente sind bereits
+# komprimiert; sie noch einmal durch gzip zu schicken kostet nur Rechenzeit.
+OHNE_KOMPRESSION = ("/api/events", "/stream")
 
-    GZipMiddleware sammelt die Antwort, bevor sie ausgeht. Bei Server-Sent
-    Events heißt das: Es kommt gar nichts an, solange der Puffer nicht voll
-    ist — und der wird bei kurzen Ereignissen nie voll.
-    """
 
+class GzipMitAusnahmen(GZipMiddleware):
     async def __call__(self, scope, receive, send):
-        if scope["type"] == "http" and scope["path"].startswith("/api/events"):
+        if scope["type"] == "http" and scope["path"].startswith(OHNE_KOMPRESSION):
             await self.app(scope, receive, send)
             return
         await super().__call__(scope, receive, send)
 
 
 # Der Katalog ist reines JSON und schrumpft dabei auf etwa ein Fünftel.
-app.add_middleware(GzipOhneLivekanal, minimum_size=1024)
+app.add_middleware(GzipMitAusnahmen, minimum_size=1024)
 
 app.add_middleware(
     CORSMiddleware,
@@ -65,6 +78,11 @@ async def startup():
         asyncio.create_task(guarded_sync())
 
 
+@app.on_event("shutdown")
+async def shutdown():
+    await streamproxy.schliessen()
+
+
 @app.get("/api/health")
 def health():
     return {"ok": True}

backend/app/streamproxy.py

@@ -1,14 +1,18 @@
 """Reicht den HLS-Stream von MediaMTX unter demselben Ursprung durch.
 
 Ohne das läge der Player auf Port 8000 und der Stream auf 8890 — zwei
-Ursprünge, also CORS und eine zweite Route nach außen. Produktiv ist ein
-Reverse-Proxy davor die bessere Wahl: Traefik reicht die Segmente durch,
-ohne dass Python jedes Byte anfasst.
+Ursprünge, also CORS und eine zweite Route nach außen.
+
+Low-Latency-HLS fragt viel und oft: je Zuschauer rund zwanzig Anfragen pro
+Sekunde, für Bild und Ton getrennt. Deshalb hält dieses Modul einen
+gemeinsamen Client mit offenen Verbindungen und reicht die Daten stückweise
+weiter, statt jedes Segment erst vollständig in den Speicher zu holen.
 """
 
 import httpx
 from fastapi import HTTPException, Request
 from fastapi.responses import Response, StreamingResponse
+from starlette.background import BackgroundTask
 
 from .config import MEDIAMTX_URL
 
@@ -23,26 +27,45 @@ DURCHREICHEN = {
     "etag",
 }
 
+# Ein Client für alle Anfragen. Ein neuer je Segment hieße: bei fünfzehn
+# Zuschauern dreihundert Verbindungsaufbauten pro Sekunde.
+#
+# Der lange Timeout gehört zu Low-Latency-HLS: Der Player fragt nach einem
+# Teil, das es noch nicht gibt, und MediaMTX hält die Antwort zurück, bis
+# es fertig ist.
+_klient = httpx.AsyncClient(
+    timeout=httpx.Timeout(30.0, connect=5.0),
+    limits=httpx.Limits(max_connections=200, max_keepalive_connections=100),
+    follow_redirects=True,
+)
+
+
+async def schliessen():
+    await _klient.aclose()
+
 
 async def weiterleiten(request: Request, pfad: str):
-    ziel = f"{MEDIAMTX_URL}/{pfad}"
-    frage = dict(request.query_params)
+    anfrage = _klient.build_request(
+        "GET", f"{MEDIAMTX_URL}/{pfad}", params=dict(request.query_params)
+    )
+    try:
+        antwort = await _klient.send(anfrage, stream=True)
+    except httpx.HTTPError:
+        raise HTTPException(status_code=502, detail="Stream nicht erreichbar")
 
-    async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as http:
-        try:
-            antwort = await http.get(ziel, params=frage)
-        except httpx.HTTPError:
-            raise HTTPException(status_code=502, detail="Stream nicht erreichbar")
-
-    if antwort.status_code == 404:
-        raise HTTPException(status_code=404, detail="Kein Stream")
+    if antwort.status_code >= 400:
+        await antwort.aclose()
+        if antwort.status_code == 404:
+            raise HTTPException(status_code=404, detail="Kein Stream")
+        raise HTTPException(status_code=antwort.status_code, detail="Stream-Fehler")
 
     kopf = {k: v for k, v in antwort.headers.items() if k.lower() in DURCHREICHEN}
     if pfad.endswith(".m3u8"):
         kopf.update(NICHT_ZWISCHENSPEICHERN)
 
-    return Response(
-        content=antwort.content,
+    return StreamingResponse(
+        antwort.aiter_raw(),
         status_code=antwort.status_code,
         headers=kopf,
+        background=BackgroundTask(antwort.aclose),
     )