feat: background analysis jobs servisi ile fair queueing ve caching eklendi

app/services/analysis_jobs.py

@@ -1,26 +1,41 @@
 """
-Background analysis jobs with real, per-step progress.
+Background analysis jobs: a fair queue in front of the CPU, real per-step
+progress, cancellation, and a small result cache.
 
 POST /api/analyze/jobs starts a job and returns its id; the client polls
-GET /api/analyze/jobs/{id}. Each pipeline step reports when it starts and
-finishes, so the waiting screen shows what the server is actually doing
-instead of a simulated progress bar. A job also survives the client
-leaving: a reloaded page can resume polling the same id.
+GET /api/analyze/jobs/{id}. The Space has two vCPUs and one analysis
+already keeps both busy, so jobs take the CPU one at a time (CPU_SLOTS) in
+arrival order and everyone else sees their place in the queue. Network
+waits (YouTube download, the external FST model) happen outside that slot.
 
-State lives in process memory — the Space runs one replica, and a job
-only needs to outlive the request that started it, not a restart.
+A job belongs to whoever polls it. One that nobody has polled for a while
+when its turn comes (the tab was closed) is dropped instead of analysed.
+A reloaded page resumes polling the same id, so it isn't dropped.
+
+State lives in process memory: the Space runs one replica, and a job only
+has to outlive the request that started it, not a restart.
 """
 
 from __future__ import annotations
 
 import asyncio
+import os
 import time
 import uuid
+from collections import OrderedDict
+from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
-from typing import Any, Awaitable, Callable, Dict, List, Optional
+from typing import Any, AsyncIterator, Awaitable, Callable, Dict, List, Optional
 
 JOB_TTL_SEC = 15 * 60
 MAX_JOBS = 200
+# Jobs waiting or running at once; beyond this, new ones get 503.
+MAX_PENDING = int(os.getenv("AURIS_MAX_PENDING", "24"))
+CPU_SLOTS = int(os.getenv("AURIS_CPU_SLOTS", "1"))
+# Unpolled this long when its turn comes: the visitor has gone.
+ABANDON_SEC = 180.0
+CACHE_SIZE = 64
+CACHE_TTL_SEC = 6 * 3600
 
 STEP_PENDING = "pending"
 STEP_RUNNING = "running"
@@ -32,13 +47,24 @@ FILE_STEPS = ["features", "vocals", "wav2vec2", "clap", "fst", "xai", "meta"]
 URL_STEPS = ["download", *FILE_STEPS]
 
 
+class JobCancelled(Exception):
+    """Raised inside a pipeline when its job was cancelled or abandoned."""
+
+    def __init__(self, code: str = "cancelled") -> None:
+        super().__init__(code)
+        self.code = code
+
+
 class Progress:
-    """Ordered step states for one analysis. A no-op sink is fine too."""
+    """Ordered step states for one analysis, plus its cancel flag."""
 
     def __init__(self, steps: List[str]) -> None:
         self._steps: Dict[str, str] = {s: STEP_PENDING for s in steps}
         self._started: Dict[str, float] = {}
         self._took: Dict[str, float] = {}
+        self.cancelled = False
+        # Returns True when nobody is waiting for the result any more.
+        self.gone: Callable[[], Awaitable[bool]] = _never
 
     def set(self, step: str, state: str) -> None:
         if step not in self._steps:
@@ -54,6 +80,15 @@ class Progress:
         for s in steps:
             self.set(s, STEP_RUNNING)
 
+    def skip_pending(self) -> None:
+        for step, state in self._steps.items():
+            if state in (STEP_PENDING, STEP_RUNNING):
+                self._steps[step] = STEP_SKIPPED
+
+    def check(self) -> None:
+        if self.cancelled:
+            raise JobCancelled("cancelled")
+
     def snapshot(self) -> List[Dict[str, Any]]:
         out: List[Dict[str, Any]] = []
         for step, state in self._steps.items():
@@ -64,72 +99,243 @@ class Progress:
         return out
 
 
+async def _never() -> bool:
+    return False
+
+
+class CpuGate:
+    """First come, first served access to the CPU-heavy stage."""
+
+    def __init__(self, slots: int) -> None:
+        self._slots = max(1, slots)
+        self._busy = 0
+        self._waiting: List[str] = []
+        self._events: Dict[str, asyncio.Event] = {}
+        # Mean seconds a job holds the CPU; None until one has finished.
+        self._mean: Optional[float] = None
+
+    def position(self, ticket: str) -> Optional[int]:
+        """1-based place in the queue, or None when not waiting."""
+        try:
+            return self._waiting.index(ticket) + 1
+        except ValueError:
+            return None
+
+    def estimated_wait(self, position: int) -> Optional[float]:
+        if self._mean is None:
+            return None
+        # Everyone ahead, plus whatever is on the CPU now (half done on average).
+        return round(((position - 1) / self._slots + (0.5 if self._busy else 0)) * self._mean, 1)
+
+    @property
+    def waiting(self) -> int:
+        return len(self._waiting)
+
+    @property
+    def busy(self) -> int:
+        return self._busy
+
+    def _wake(self) -> None:
+        while self._waiting and self._busy < self._slots:
+            ticket = self._waiting.pop(0)
+            self._busy += 1
+            self._events.pop(ticket).set()
+
+    @asynccontextmanager
+    async def slot(self, ticket: str, progress: Optional[Progress] = None) -> AsyncIterator[None]:
+        event = asyncio.Event()
+        self._events[ticket] = event
+        self._waiting.append(ticket)
+        self._wake()
+        try:
+            await event.wait()
+        except BaseException:
+            if ticket in self._waiting:
+                self._waiting.remove(ticket)
+                self._events.pop(ticket, None)
+            else:
+                self._release()
+            raise
+        started = time.monotonic()
+        try:
+            if progress is not None:
+                progress.check()
+                if await progress.gone():
+                    raise JobCancelled("job_abandoned")
+            yield
+        finally:
+            held = time.monotonic() - started
+            if held > 1.0:
+                self._mean = held if self._mean is None else self._mean * 0.7 + held * 0.3
+            self._release()
+
+    def _release(self) -> None:
+        self._busy = max(0, self._busy - 1)
+        self._wake()
+
+
+cpu_gate = CpuGate(CPU_SLOTS)
+
+
+class ResultCache:
+    """Recent successful responses by content key (same file, same link)."""
+
+    def __init__(self, size: int, ttl: float) -> None:
+        self._size = size
+        self._ttl = ttl
+        self._items: "OrderedDict[str, tuple[float, Dict[str, Any]]]" = OrderedDict()
+
+    def get(self, key: Optional[str]) -> Optional[Dict[str, Any]]:
+        if not key or key not in self._items:
+            return None
+        stored, value = self._items[key]
+        if time.time() - stored > self._ttl:
+            del self._items[key]
+            return None
+        self._items.move_to_end(key)
+        return value
+
+    def put(self, key: Optional[str], value: Dict[str, Any]) -> None:
+        if not key or value.get("errors") or not value.get("result"):
+            return
+        self._items[key] = (time.time(), value)
+        self._items.move_to_end(key)
+        while len(self._items) > self._size:
+            self._items.popitem(last=False)
+
+
+result_cache = ResultCache(CACHE_SIZE, CACHE_TTL_SEC)
+
+
 @dataclass
 class Job:
     id: str
     source_type: str
     progress: Progress
+    key: Optional[str] = None
     created: float = field(default_factory=time.time)
     updated: float = field(default_factory=time.time)
-    status: str = "running"          # running | done | error
+    last_seen: float = field(default_factory=time.monotonic)
+    status: str = "queued"          # queued | running | done | error
+    cached: bool = False
     response: Optional[Dict[str, Any]] = None
     task: Optional[asyncio.Task] = None
 
     def to_dict(self) -> Dict[str, Any]:
-        return {
+        data: Dict[str, Any] = {
             "jobId": self.id,
             "sourceType": self.source_type,
             "status": self.status,
             "steps": self.progress.snapshot(),
-            "elapsedSec": round((self.updated if self.status != "running" else time.time()) - self.created, 2),
+            "elapsedSec": round((self.updated if self.status in ("done", "error") else time.time()) - self.created, 2),
             "response": self.response,
         }
+        position = cpu_gate.position(self.id)
+        if self.status == "queued" and position is not None:
+            data["queue"] = {"position": position, "estimatedWaitSec": cpu_gate.estimated_wait(position)}
+        if self.cached:
+            data["cached"] = True
+        return data
+
+
+class QueueFull(Exception):
+    pass
+
+
+Work = Callable[[Job], Awaitable[Dict[str, Any]]]
 
 
 class JobStore:
     def __init__(self) -> None:
         self._jobs: Dict[str, Job] = {}
+        self._inflight: Dict[str, str] = {}
 
     def _prune(self) -> None:
         cutoff = time.time() - JOB_TTL_SEC
-        for job_id in [j.id for j in self._jobs.values() if j.status != "running" and j.updated < cutoff]:
+        for job_id in [j.id for j in self._jobs.values() if j.status in ("done", "error") and j.updated < cutoff]:
             del self._jobs[job_id]
-        # Hard cap: drop the oldest finished jobs first.
         if len(self._jobs) > MAX_JOBS:
-            finished = sorted((j for j in self._jobs.values() if j.status != "running"), key=lambda j: j.updated)
+            finished = sorted((j for j in self._jobs.values() if j.status in ("done", "error")), key=lambda j: j.updated)
             for job in finished[: len(self._jobs) - MAX_JOBS]:
                 del self._jobs[job.id]
 
+    def pending_count(self) -> int:
+        return sum(1 for j in self._jobs.values() if j.status in ("queued", "running"))
+
     def running_count(self) -> int:
         return sum(1 for j in self._jobs.values() if j.status == "running")
 
-    def start(
-        self,
-        source_type: str,
-        steps: List[str],
-        work: Callable[[Progress], Awaitable[Dict[str, Any]]],
-    ) -> Job:
+    def start(self, source_type: str, steps: List[str], work: Work, key: Optional[str] = None) -> Job:
+        """Queue a job; the same file or link already in flight is shared, a recent one is served from cache."""
         self._prune()
-        job = Job(id=uuid.uuid4().hex, source_type=source_type, progress=Progress(steps))
+        if key and key in self._inflight:
+            shared = self._jobs.get(self._inflight[key])
+            if shared and shared.status in ("queued", "running"):
+                shared.last_seen = time.monotonic()
+                return shared
+
+        job = Job(id=uuid.uuid4().hex, source_type=source_type, progress=Progress(steps), key=key)
+        hit = result_cache.get(key)
+        if hit is not None:
+            job.status, job.cached, job.response = "done", True, hit
+            job.progress.skip_pending()
+            for step in steps:
+                job.progress.set(step, STEP_DONE)
+            self._jobs[job.id] = job
+            return job
+
+        if self.pending_count() >= MAX_PENDING:
+            raise QueueFull()
+
+        async def gone() -> bool:
+            return time.monotonic() - job.last_seen > ABANDON_SEC
+
+        job.progress.gone = gone
         self._jobs[job.id] = job
+        if key:
+            self._inflight[key] = job.id
 
         async def runner() -> None:
             try:
-                job.response = await work(job.progress)
-                job.status = "error" if job.response.get("errors") and not job.response.get("result") else "done"
+                job.response = await work(job)
+                failed = bool(job.response.get("errors")) and not job.response.get("result")
+                job.status = "error" if failed else "done"
+                result_cache.put(key, job.response)
+            except JobCancelled as stop:
+                job.progress.skip_pending()
+                job.response = {"result": None, "warnings": [], "errors": [stop.code]}
+                job.status = "error"
             except Exception:  # noqa: BLE001 — the job must always settle
+                job.progress.skip_pending()
                 job.response = {"result": None, "warnings": [], "errors": ["internal_error"]}
                 job.status = "error"
             finally:
                 job.updated = time.time()
+                if key and self._inflight.get(key) == job.id:
+                    del self._inflight[key]
+                job.task = None
 
         # Keep a reference: asyncio only holds weak references to tasks.
         job.task = asyncio.create_task(runner())
         return job
 
-    def get(self, job_id: str) -> Optional[Job]:
+    def get(self, job_id: str, touch: bool = True) -> Optional[Job]:
         self._prune()
-        return self._jobs.get(job_id)
+        job = self._jobs.get(job_id)
+        if job is not None and touch:
+            job.last_seen = time.monotonic()
+        return job
+
+    def cancel(self, job_id: str) -> Optional[Job]:
+        job = self._jobs.get(job_id)
+        if job is None:
+            return None
+        if job.status in ("queued", "running"):
+            job.progress.cancelled = True
+            # A job still waiting for the CPU never gets there.
+            if job.status == "queued" and job.task is not None:
+                job.task.cancel()
+        return job
 
 
 job_store = JobStore()