genrecon/backend/run_job.py

"""Detached, restart-surviving reconstruction runner.

Launched by app.py with start_new_session=True and, with `KillMode=process` on the
systemd unit, it OUTLIVES a `systemctl restart genrecon-app` — so redeploying or
restarting the web service never kills an in-flight reconstruction. Concurrent
uploads are serialized on the single GPU by an flock. app.py's startup re-attaches
to a live runner (worker.pid) or relaunches a dead one, instead of failing the job.
"""
from __future__ import annotations

import fcntl
import json
import os
import sys
import time
import traceback
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import config          # noqa: E402
import pipeline        # noqa: E402


def main(job_id: str) -> None:
    job = config.JOBS_DIR / job_id
    status = job / "status.json"
    pid_file = job / "worker.pid"

    def write(**patch) -> None:
        st: dict = {"id": job_id}
        try:
            st = json.loads(status.read_text())
        except Exception:
            pass
        st.update(patch)
        st["updated"] = time.time()
        tmp = status.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(st))
        tmp.replace(status)                       # atomic — a reader never sees a half-written status

    def log_line(line: str) -> None:
        with open(job / "pipeline.log", "a") as f:
            f.write(line + "\n")

    # Claim the job immediately (before the GPU flock) so app.py's startup sees a
    # live runner and re-attaches rather than relaunching.
    pid_file.write_text(str(os.getpid()))
    lock = open(config.JOBS_DIR / ".gpu.lock", "w")
    try:
        write(status="running", stage=None, progress=1, message="w kolejce (GPU)…", error=None)
        fcntl.flock(lock, fcntl.LOCK_EX)          # one reconstruction at a time on the single GPU
        write(status="running", stage=config.STAGES[0][0], progress=1, message="start", error=None)
        pipeline.run_pipeline(
            job,
            log=log_line,
            set_progress=lambda stage, pct, msg: write(status="running", stage=stage, progress=pct, message=msg),
        )
        write(status="done", stage="glb", progress=100, message="ready")
    except Exception as e:  # noqa: BLE001 — surface any stage failure to the client
        log_line(f"ERROR: {e}")
        traceback.print_exc()
        write(status="error", message="failed", error=str(e))
    finally:
        try:
            fcntl.flock(lock, fcntl.LOCK_UN)
        except OSError:
            pass
        try:
            pid_file.unlink()
        except OSError:
            pass


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("usage: run_job.py <job_id>", file=sys.stderr)
        sys.exit(2)
    main(sys.argv[1])