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