rippy/docker/worker/zombies.py

"""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