feat(backend): disk-backed jobs under /tmp to survive restarts; more robust polling/results

Author: kr1shnasomani

backend/app/api/v1/routes.py

@@ -7,7 +7,8 @@
 import threading
 import uuid
 import time
-from dataclasses import dataclass, field
+import json
+from dataclasses import dataclass, field, asdict
 from typing import Dict, Optional
 
 from app.settings import settings
@@ -28,9 +29,8 @@ def _cleanup(paths: list[str]) -> None:
 
 
 # --------------------
-# Simple in-memory job queue for long-running processing
-# NOTE: This is suitable for single-instance demo deployments.
-# For production, use external storage and a proper queue.
+# Disk-backed job store (under /tmp) so jobs survive simple restarts.
+# NOTE: For real production usage, use durable storage/queue.
 
 @dataclass
 class Job:
@@ -45,18 +45,54 @@ class Job:
     updated_at: float = field(default_factory=time.time)
 
 
-JOBS: Dict[str, Job] = {}
+JOBS_DIR = Path(tempfile.gettempdir()) / "sportiq-jobs"
+JOBS_DIR.mkdir(parents=True, exist_ok=True)
 JOBS_LOCK = threading.Lock()
 
 
-def _run_job(job_id: str, model_path: Path) -> None:
+def _job_dir(job_id: str) -> Path:
+    return JOBS_DIR / job_id
+
+
+def _meta_path(job_id: str) -> Path:
+    return _job_dir(job_id) / "meta.json"
+
+
+def _write_meta(job: Job) -> None:
+    d = _job_dir(job.id)
+    d.mkdir(parents=True, exist_ok=True)
+    with open(_meta_path(job.id), "w", encoding="utf-8") as f:
+        json.dump(asdict(job), f)
+
+
+def _read_meta(job_id: str) -> Optional[Job]:
+    p = _meta_path(job_id)
+    if not p.exists():
+        return None
+    try:
+        with open(p, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        return Job(**data)
+    except Exception:
+        return None
+
+
+def _update_meta(job_id: str, **updates) -> Optional[Job]:
     with JOBS_LOCK:
-        job = JOBS.get(job_id)
+        job = _read_meta(job_id)
         if not job:
-            return
-        job.status = "processing"
+            return None
+        for k, v in updates.items():
+            setattr(job, k, v)
         job.updated_at = time.time()
+        _write_meta(job)
+        return job
+
 
+def _run_job(job_id: str, model_path: Path) -> None:
+    job = _update_meta(job_id, status="processing")
+    if not job:
+        return
     try:
         run_pipeline(
             input_path=Path(job.input_path),
@@ -65,20 +101,11 @@ def _run_job(job_id: str, model_path: Path) -> None:
             intermediate_path=Path(job.intermediate_path),
             fast=job.fast,
         )
-
         if not os.path.exists(job.output_path) or os.path.getsize(job.output_path) == 0:
             raise RuntimeError("Processing produced an empty output file")
-
-        with JOBS_LOCK:
-            job.status = "done"
-            job.updated_at = time.time()
+        _update_meta(job_id, status="done")
     except Exception as e:
-        with JOBS_LOCK:
-            job = JOBS.get(job_id)
-            if job:
-                job.status = "error"
-                job.error = str(e)
-                job.updated_at = time.time()
+        _update_meta(job_id, status="error", error=str(e))
 
 
 @router.post("/jobs")
@@ -94,21 +121,23 @@ async def create_job(video: UploadFile = File(...), fast: bool = False):
         await video.seek(0)
         shutil.copyfileobj(video.file, in_f)
 
-    inter_fd, inter_path = tempfile.mkstemp(suffix=".mp4")
-    os.close(inter_fd)
-    out_fd, out_path = tempfile.mkstemp(suffix=".mp4")
-    os.close(out_fd)
-
     job_id = str(uuid.uuid4())
+    d = _job_dir(job_id)
+    d.mkdir(parents=True, exist_ok=True)
+    # Move input to job dir
+    job_input = str(d / f"input{suffix}")
+    shutil.move(in_path, job_input)
+    job_inter = str(d / "intermediate.mp4")
+    job_out = str(d / "output.mp4")
+
     job = Job(
         id=job_id,
-        input_path=in_path,
-        output_path=out_path,
-        intermediate_path=inter_path,
+        input_path=job_input,
+        output_path=job_out,
+        intermediate_path=job_inter,
         fast=fast,
     )
-    with JOBS_LOCK:
-        JOBS[job_id] = job
+    _write_meta(job)
 
     t = threading.Thread(target=_run_job, args=(job_id, Path(settings.model_path)), daemon=True)
     t.start()
@@ -118,35 +147,38 @@ async def create_job(video: UploadFile = File(...), fast: bool = False):
 
 @router.get("/jobs/{job_id}/status")
 async def get_job_status(job_id: str):
-    with JOBS_LOCK:
-        job = JOBS.get(job_id)
-        if not job:
-            raise HTTPException(status_code=404, detail="Job not found")
-        return {"job_id": job.id, "status": job.status, "error": job.error}
+    job = _read_meta(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    return {"job_id": job.id, "status": job.status, "error": job.error}
 
 
 @router.get("/jobs/{job_id}/result")
 async def get_job_result(job_id: str, background_tasks: BackgroundTasks):
-    with JOBS_LOCK:
-        job = JOBS.get(job_id)
-        if not job:
-            raise HTTPException(status_code=404, detail="Job not found")
-        status = job.status
-        out_path = job.output_path
-        to_cleanup = [job.input_path, job.intermediate_path, job.output_path]
+    job = _read_meta(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    status = job.status
+    out_path = job.output_path
+    to_cleanup = [job.input_path, job.intermediate_path, job.output_path, str(_meta_path(job_id))]
 
     if status != "done":
         if status == "error":
-            raise HTTPException(status_code=500, detail="Job failed")
+            raise HTTPException(status_code=500, detail=job.error or "Job failed")
         return JSONResponse(status_code=202, content={"status": status})
 
     if not os.path.exists(out_path) or os.path.getsize(out_path) == 0:
         raise HTTPException(status_code=500, detail="Output not available")
 
     # Clean up files after response is sent
     background_tasks.add_task(_cleanup, to_cleanup)
-    with JOBS_LOCK:
-        JOBS.pop(job_id, None)
+    # Remove job dir
+    def _rm_dir(path: Path):
+        try:
+            shutil.rmtree(path, ignore_errors=True)
+        except Exception:
+            pass
+    background_tasks.add_task(_rm_dir, _job_dir(job_id))
 
     return FileResponse(out_path, media_type="video/mp4", filename="analyzed.mp4")