From 05c0f6ef93549f09c3b7101d7d18afc3e0471bb2 Mon Sep 17 00:00:00 2001 From: jtrzupek Date: Sun, 31 May 2026 11:19:13 +0200 Subject: [PATCH] fix(scheduler): per-connector hard timeout + reorder mangoporn-first MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug-report 2026-05-30 "ingest znów się zawiesił". streamporn/pandamovies wieszały się intermittentnie mid-run (zależnie od live-contentu danego dnia), blokując sekwencyjny _job_movie_ingest → mangoporn (jedyny mirror z realnym new-content: 72 nowych 05-28) nigdy nie startował. try/except chronił przed wyjątkiem, NIE przed hangiem. Fix: - _job_movie_ingest: każdy connector w ThreadPoolExecutor z future.result (timeout=360s). Hang jednego źródła → log + shutdown(wait=False) + kolejka leci dalej. Healthy run ~50s, cap 6min = zapas. - get_movie_connectors: reorder paradisehill, MANGOPORN, streamporn, pandamovies — mangoporn zaraz po canonical primary, przed wolniejszymi/wieszającymi się. Zweryfikowane: pełny _job_movie_ingest przeszedł wszystkie 4 success w nowej kolejności (mangoporn 2nd, 23s). 33 osierocone "running" rows (worker ubity mid-run przy deployach) wyczyszczone osobno. Co-Authored-By: Claude Opus 4.8 (1M context) --- app/connectors/__init__.py | 8 +++++++- app/scheduler/jobs.py | 25 ++++++++++++++++++++++++- 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/app/connectors/__init__.py b/app/connectors/__init__.py index db3d9df..9818b87 100644 --- a/app/connectors/__init__.py +++ b/app/connectors/__init__.py @@ -40,9 +40,15 @@ def get_movie_connectors() -> list[tuple[str, type]]: ) from app.connectors.paradisehill import ParadisehillConnector + # Kolejność ingestu: paradisehill FIRST (canonical primary, mirrory się do + # niego przyklejają), potem mangoporn (jedyny mirror z realnym new-content — + # 72 nowych 2026-05-28; streamporn/pandamovies zwracają stale 0 new), na końcu + # streamporn + pandamovies. Powód reorderu (2026-05-30): gdy streamporn wiesza + # się intermittentnie, mangoporn musi zdążyć przed nim — patrz per-connector + # timeout w _job_movie_ingest. return [ ("paradisehill", ParadisehillConnector), + ("mangoporn", MangopornConnector), ("streamporn", StreampornConnector), ("pandamovies", PandamoviesConnector), - ("mangoporn", MangopornConnector), ] diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 873ee13..f5b241e 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -98,11 +98,34 @@ def _job_movie_ingest() -> None: Kolejność: paradisehill FIRST (żeby mirrory miały do czego się przykleić), potem mirrory. Pojedynczy failed connector NIE zatrzymuje pozostałych — każdy w osobnym try/except. + + HARD TIMEOUT per-connector (bug-report 2026-05-30 "ingest znów się zawiesił"): + sam try/except chroni przed *wyjątkiem*, ale NIE przed *hangiem* (CPU-bound + ReDoS na patologicznej stronie / thread-stall) — wtedy jeden mirror blokuje + resztę i mangoporn (jedyny z realnym new-content) nigdy nie startuje. + Każdy connector leci w osobnym wątku z `future.result(timeout)`; po + przekroczeniu logujemy i idziemy dalej (osierocony wątek dożywa do restartu + workera — OK, bo loop się odblokowuje). Healthy run ~50s, cap 6 min = zapas. """ + from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout + + PER_CONNECTOR_TIMEOUT = 360 # sekundy + for name, cls in get_movie_connectors(): log.info("[scheduler] movie ingest %s starting", name) try: - ingest_movies_from_connector(cls(), use_delta=True) + ex = ThreadPoolExecutor(max_workers=1) + fut = ex.submit(ingest_movies_from_connector, cls(), use_delta=True) + try: + fut.result(timeout=PER_CONNECTOR_TIMEOUT) + except FutureTimeout: + log.error( + "[scheduler] movie ingest %s HUNG > %ds — skip, kolejka leci dalej", + name, PER_CONNECTOR_TIMEOUT, + ) + finally: + # shutdown(wait=False): nie blokuj na join osieroconego wątku. + ex.shutdown(wait=False) except Exception: log.exception("[scheduler] movie ingest %s failed", name)