"""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. # # ⚠️ „running" FEHLTE bis zum 26.07.2026 — und das war der ganze Witz: Die # Zombie-Erkennung aus v3.14 wurde gebaut, um genau einen abgestürzten RIP zu # finden, und hat ihn nie gesehen. Der Rip setzt `status="running"` # (tasks.py, db.update_job), gesucht wurde aber „ripping". Dieser Wert steht # ausschließlich in Celerys Task-Meta und NIE in einer Job-Zeile. # # Die Folge war besonders tückisch: Bei jedem Worker-Start meldete die Erkennung # `{'geprueft': 0, ...}` — und das sah wie „nachgesehen, alles gesund" aus, # während sie in Wahrheit nach einem Status suchte, den es nicht gibt. Gefunden # am 26.07.2026, als ein Rip mitten im Lauf abbrach und der Job danach dauerhaft # auf „processing 12 %" stand, mit einer Restzeit-Schätzung obendrauf. # # „ripping" bleibt bewusst drin: Es schadet nicht und deckt eine etwaige # Bestandsinstallation ab, in der es doch gesetzt wurde. ARBEITS_STATI = ("running", "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] # In welcher Phase starb der Job? Entscheidet, was danach hilft. RIP_STATI = ("running", "ripping") def war_im_rip(job) -> bool: """Starb der Job WÄHREND des Rippens? (pure Funktion) Der Unterschied ist für den Nutzer der ganze Punkt: Nach einem toten TRANSCODE liegt eine vollständige Roh-MKV da — „Neu komprimieren" genügt und spart eine Stunde Rippen. Nach einem toten RIP liegt ein ABGESCHNITTENES Bruchstück da (im Vorfall 26.07.2026: 5,1 GB von rund 40), und Komprimieren ergäbe einen Film, der bei 12 % aufhört. """ return (job.get("status") or "") in RIP_STATI def fehlertext(job) -> str: """Ehrlicher Klartext für die Job-Karte — was war, und was jetzt hilft. ⚠️ Der Text unterscheidet die Phase (26.07.2026). Vorher stand hier für JEDEN Fall „mit 'Neu komprimieren' läuft die Kompression erneut, ohne die Disc noch einmal zu rippen" — geschrieben für einen toten Transcode. Seit „running" mit zu den Arbeitsstati gehört, traf derselbe Satz auch abgebrochene RIPS, und da ist er schlicht falsch: Wer ihn befolgt, bekommt ein Bruchstück komprimiert. """ zuletzt = job.get("status") or "unbekannt" kopf = ( 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. " ) if war_im_rip(job): fortschritt = job.get("progress") or 0 return kopf + ( f'Der Rip war bei {fortschritt} % — die Roh-Datei ist damit ' 'UNVOLLSTÄNDIG. Sie wurde nicht gelöscht, aber „Neu komprimieren" ' 'ergäbe einen Film, der mitten drin aufhört. Richtig ist: Disc ' 'wieder einlegen und neu rippen.' ) return kopf + ( 'Der Rip war fertig, nur die Kompression nicht. Die Roh-Dateien wurden ' 'NICHT gelöscht: mit „Neu komprimieren" läuft sie 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