"""Erkennt Job-Leichen: Jobs, die behaupten zu laufen, an denen aber niemand arbeitet. Befund 25.07.2026 (Akira-UHD): Nach einem Absturz stand ein Job auf `transcoding` bei 96 %, obwohl weder ein Prozess lief noch etwas in den Celery-Queues stand. Folge für den Nutzer: kein Fortschritt, kein Download — und der Knopf „Neu komprimieren" fehlte, weil `_kann_neu_komprimieren` (api/main.py) `status == "failed"` verlangt. Der Job war damit unerreichbar, obwohl die Rohdateien vollständig dalagen. ## Die Leitregel: ohne vollständige Auskunft wird NICHTS angetastet Ein falsch als Leiche markierter Job ist teurer als eine stehengebliebene Leiche. Deshalb drei Sicherungen: 1. **Nur beim Worker-Start.** Da ist die Aussage eindeutig: „als ich hochkam, lief nichts". Ein periodischer Lauf könnte einen Job erwischen, der legitim in der Warteschlange wartet, weil alle Arbeitsplätze belegt sind — der ist für `active()`/`reserved()` unsichtbar. Bewusst nicht gebaut. 2. **Gnadenfrist.** Nach einem Neustart stellt Celery unbestätigte Aufgaben erneut zu. Erst abwarten, dann urteilen. 3. **Vollzähligkeit.** Es wird nur geurteilt, wenn mindestens so viele Celery-Knoten antworten, wie laut Herzschlag online sind. Antwortet ein beschäftigter Remote-Worker nicht rechtzeitig, wäre sein laufender Job sonst eine falsche Leiche. Die Kernfunktionen sind bewusst rein (kein Postgres, kein Redis), damit die Ampel sie ohne Infrastruktur prüfen kann. """ # Zustände, die behaupten: hier arbeitet gerade jemand. ARBEITS_STATI = ("ripping", "transcoding", "canceling") # Wartezeit nach dem Worker-Start, bevor geurteilt wird. Deckt die # Wiederzustellung unbestätigter Aufgaben durch Celery ab. GNADENFRIST_SEKUNDEN = 120 # Wie lange auf Antworten der Celery-Knoten gewartet wird. Großzügig, weil ein # Worker mitten in einem 4K-Encode träge antwortet. INSPEKT_TIMEOUT_SEKUNDEN = 10 def belegte_job_ids(inspektionen, kandidaten) -> set: """Welche der `kandidaten`-IDs kommen in irgendeiner Celery-Aufgabe vor? Bewusst als Textsuche über die ganze Inspektions-Antwort: `job_id` steht bei `rip_disc` an ZWEITER, bei `transcode_files` an ERSTER Stelle, und Celery liefert `args` je nach Version als Liste oder als Text. Eine Positions-Auswertung wäre an beidem zerbrechlich. Eine Job-ID ist eine 36-stellige UUID — Fehltreffer sind praktisch ausgeschlossen. `inspektionen` ist die Liste der Antworten (active/reserved/scheduled); `None`-Einträge werden übersprungen. """ vorhandene = [i for i in (inspektionen or []) if i] if not vorhandene: return set() text = repr(vorhandene) return {jid for jid in kandidaten if jid and jid in text} def antwortende_knoten(inspektionen) -> set: """Namen aller Celery-Knoten, die auf die Inspektion geantwortet haben.""" knoten = set() for antwort in (inspektionen or []): if antwort: knoten.update(antwort.keys()) return knoten def auskunft_vollstaendig(anzahl_antworten: int, anzahl_bekannt: int) -> bool: """Darf aus dieser Auskunft überhaupt ein Urteil folgen? Nein, wenn niemand geantwortet hat (dann wissen wir nichts), und nein, wenn weniger Knoten antworten als laut Herzschlag online sind (dann fehlt uns genau der Knoten, der den Job vielleicht gerade bearbeitet). """ if anzahl_antworten < 1: return False return anzahl_antworten >= anzahl_bekannt def finde_zombies(offene_jobs, belegte) -> list: """Jobs aus `offene_jobs`, zu denen keine Celery-Aufgabe existiert.""" return [job for job in offene_jobs if job.get("id") not in belegte] def fehlertext(job) -> str: """Ehrlicher Klartext für die Job-Karte — was war, und was jetzt hilft.""" zuletzt = job.get("status") or "unbekannt" return ( f"Abgebrochen: Der Vorgang lief nicht mehr (zuletzt gemeldet: {zuletzt}). " "Beim Start des Workers war dazu weder ein Prozess noch eine Aufgabe in " "der Warteschlange zu finden — meistens ein Absturz oder ein Neustart " "mitten im Lauf. Die Rohdateien wurden NICHT gelöscht: mit " "'Neu komprimieren' läuft die Kompression erneut, ohne die Disc noch " "einmal zu rippen." ) def raeume_zombies_auf(celery_app, db) -> dict: """Sucht Leichen und setzt sie ehrlich auf `failed`. Wirft nie. Rückgabe (auch für den Log): was geprüft und was getan wurde. """ bericht = {"geprueft": 0, "aufgeraeumt": [], "uebersprungen": ""} try: offene = db.list_jobs_mit_status(ARBEITS_STATI) bericht["geprueft"] = len(offene) if not offene: return bericht inspektor = celery_app.control.inspect(timeout=INSPEKT_TIMEOUT_SEKUNDEN) inspektionen = [inspektor.active(), inspektor.reserved(), inspektor.scheduled()] knoten = antwortende_knoten(inspektionen) bekannt = db.zaehle_online_worker() if not auskunft_vollstaendig(len(knoten), bekannt): bericht["uebersprungen"] = ( f"unvollständige Auskunft ({len(knoten)} von {bekannt} Worker " "haben geantwortet) — es wird nichts als Leiche gewertet" ) db.add_log( "info", "worker", f"Zombie-Erkennung übersprungen: {bericht['uebersprungen']}", ) return bericht belegte = belegte_job_ids(inspektionen, [j.get("id") for j in offene]) for job in finde_zombies(offene, belegte): db.update_job( job["id"], status="failed", error=fehlertext(job), finished_at=db.utcnow(), ) db.add_log( "warning", "worker", f"Job {job['id']} ({job.get('title') or 'ohne Titel'}) stand auf " f"'{job.get('status')}', es arbeitet aber niemand daran — " "ehrlich auf 'failed' gesetzt. Rohdateien bleiben liegen.", ) bericht["aufgeraeumt"].append(job["id"]) except Exception as e: # Erkennung darf den Worker-Start NIEMALS verhindern bericht["uebersprungen"] = f"Fehler: {e}" try: db.add_log("warning", "worker", f"Zombie-Erkennung fehlgeschlagen: {e}") except Exception: pass return bericht