Jesse_Chen
124d3990b2
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>
117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
"""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()
|