"""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 ", file=sys.stderr) sys.exit(2) main(sys.argv[1])