fix(api): persist scratch/local and bound heavy compute concurrency

Keep async job and chart-cache files across API recreates, freeze jyotish_api_server.py growth, and fail fast with 429 when rectification or high-rigor compute is saturated. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-02 06:12:23 +08:00
parent bf6989eccf
commit 124d3990b2
10 changed files with 530 additions and 3 deletions
@@ -0,0 +1,116 @@
+"""Fail-fast bounded concurrency for heavy Jyotish API compute endpoints.
+
+Request threads that run rectification scans or high-rigor workflows share one
+process-wide semaphore sized for the 2 vCPU production host. Saturated requests
+return immediately; they are not queued. Health checks and other light routes
+must not call this gate.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+
+DEFAULT_CONCURRENCY = 2
+DEFAULT_RETRY_AFTER_SECONDS = 2
+ENV_CONCURRENCY = "JYOTISH_HEAVY_COMPUTE_CONCURRENCY"
+ENV_RETRY_AFTER = "JYOTISH_HEAVY_COMPUTE_RETRY_AFTER_SECONDS"
+
+HEAVY_COMPUTE_PATHS = frozenset(
+    {
+        "/api/rectification/sensitivity_scan",
+        "/api/active_rectification_events",
+        "/api/active_rectification_events_v4",
+        "/api/rectification/v5/candidate-features",
+        "/api/rectification/v5/score",
+        "/api/rectification/v5/diagnostics",
+        "/api/rectification/v5/vedastro-validate",
+        "/api/dynamic_rectification_opportunities",
+        "/api/dynamic_rectification_score",
+        "/api/high_rigor_workflow",
+        "/api/consultation_workflow",
+        "/api/professional_reading",
+        "/api/vedastro/range_scan",
+        "/api/vedastro_gateway/run",
+        "/api/thematic_report",
+    }
+)
+
+_state_lock = threading.Lock()
+_semaphore: threading.BoundedSemaphore | None = None
+_retry_after_seconds = DEFAULT_RETRY_AFTER_SECONDS
+
+
+class HeavyComputeBusy(RuntimeError):
+    """No heavy-compute slot is free; callers must fail fast with HTTP 429."""
+
+    error_code = "ERR_COMPUTE_BUSY"
+
+    def __init__(self, retry_after_seconds: int) -> None:
+        super().__init__(
+            "Heavy compute capacity is saturated; retry after the Retry-After delay."
+        )
+        self.retry_after_seconds = max(int(retry_after_seconds), 1)
+
+
+def _parse_positive_int(raw: str | None, default: int) -> int:
+    try:
+        value = int(str(raw or "").strip())
+    except (TypeError, ValueError):
+        return default
+    return value if value >= 1 else default
+
+
+def is_heavy_compute_path(path: str) -> bool:
+    if path in HEAVY_COMPUTE_PATHS:
+        return True
+    return path.startswith("/api/vedastro_gateway/jobs/") and path.endswith("/run")
+
+
+def reset_heavy_compute_gate(
+    *,
+    limit: int | None = None,
+    retry_after_seconds: int | None = None,
+) -> None:
+    """Rebuild the process-wide semaphore. Tests must call this after env changes."""
+    global _semaphore, _retry_after_seconds
+    resolved_limit = (
+        limit
+        if limit is not None
+        else _parse_positive_int(os.environ.get(ENV_CONCURRENCY), DEFAULT_CONCURRENCY)
+    )
+    resolved_retry = (
+        retry_after_seconds
+        if retry_after_seconds is not None
+        else _parse_positive_int(os.environ.get(ENV_RETRY_AFTER), DEFAULT_RETRY_AFTER_SECONDS)
+    )
+    with _state_lock:
+        _retry_after_seconds = resolved_retry
+        _semaphore = threading.BoundedSemaphore(resolved_limit)
+
+
+def _ensure_locked() -> tuple[threading.BoundedSemaphore, int]:
+    global _semaphore, _retry_after_seconds
+    if _semaphore is None:
+        limit = _parse_positive_int(os.environ.get(ENV_CONCURRENCY), DEFAULT_CONCURRENCY)
+        _retry_after_seconds = _parse_positive_int(
+            os.environ.get(ENV_RETRY_AFTER), DEFAULT_RETRY_AFTER_SECONDS
+        )
+        _semaphore = threading.BoundedSemaphore(limit)
+    return _semaphore, _retry_after_seconds
+
+
+def acquire_heavy_compute_slot(path: str) -> threading.BoundedSemaphore | None:
+    """Acquire a slot for a gated path. Light paths return None. Fail-fast on saturation."""
+    if not is_heavy_compute_path(path):
+        return None
+    with _state_lock:
+        semaphore, retry_after = _ensure_locked()
+    if not semaphore.acquire(blocking=False):
+        raise HeavyComputeBusy(retry_after)
+    return semaphore
+
+
+def release_heavy_compute_slot(slot: threading.BoundedSemaphore | None) -> None:
+    if slot is not None:
+        slot.release()