diff --git a/backend/requirements-dev.txt b/backend/requirements-dev.txt new file mode 100644 index 00000000..cf662ec9 --- /dev/null +++ b/backend/requirements-dev.txt @@ -0,0 +1,4 @@ +# Development dependencies for ss-tools backend +# Install with: pip install -r requirements-dev.txt + +pytest-httpx>=0.34.0 diff --git a/backend/requirements.txt b/backend/requirements.txt index eb34e24b..b67013ca 100755 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -60,3 +60,5 @@ ruff>=0.11.0 sqlparse>=0.5.0 lingua-language-detector==2.1.1 testcontainers[postgres]>=4.0 +aiofiles>=24.1.0 +aiosmtplib>=3.0.2 diff --git a/backend/src/core/config_models.py b/backend/src/core/config_models.py index 33179fd5..6b08632e 100755 --- a/backend/src/core/config_models.py +++ b/backend/src/core/config_models.py @@ -33,7 +33,7 @@ class Schedule(BaseModel): # #region Environment [TYPE DataClass] -# @BRIEF Represents a Superset environment configuration. +# @BRIEF Represents a Superset environment configuration with async connection pool settings. class Environment(BaseModel): id: str name: str @@ -47,10 +47,62 @@ class Environment(BaseModel): is_production: bool = False backup_schedule: Schedule = Field(default_factory=Schedule) + # Async connection pool settings (per-environment) + connection_pool_size: int = Field( + default=20, ge=1, le=100, + description="Max concurrent connections to this Superset environment" + ) + connection_pool_timeout: int = Field( + default=30, ge=5, le=300, + description="Seconds to wait for a connection pool slot before timeout" + ) + request_timeout: int = Field( + default=30, ge=5, le=300, + description="Default HTTP request timeout for Superset API calls" + ) + llm_timeout: int = Field( + default=120, ge=30, le=600, + description="Timeout for LLM API calls (seconds)" + ) + # #endregion Environment +# #region AppAsyncRuntimeConfig [TYPE DataClass] +# @BRIEF Global application-level async runtime configuration. +# @RATIONALE Separated from EnvironmentConfig because these settings are not per-env: +# executor workers, shutdown timeout, blocking queue timeout are app-wide. +class AppAsyncRuntimeConfig(BaseModel): + db_executor_workers: int = Field( + default=10, ge=1, le=50, + description="Max worker threads for database blocking operations" + ) + file_executor_workers: int = Field( + default=10, ge=1, le=50, + description="Max worker threads for file I/O blocking operations" + ) + git_executor_workers: int = Field( + default=5, ge=1, le=20, + description="Max worker threads for Git blocking operations" + ) + graceful_shutdown_timeout: int = Field( + default=30, ge=10, le=120, + description="Graceful shutdown timeout in seconds" + ) + blocking_queue_timeout: int = Field( + default=30, ge=5, le=120, + description="Timeout for acquiring blocking executor queue slot" + ) + event_bus_maxsize: int = Field( + default=10000, ge=100, le=100000, + description="Max pending events in EventBus per subscriber queue" + ) + + +# #endregion AppAsyncRuntimeConfig + + # #region LoggingConfig [TYPE DataClass] # @BRIEF Defines the configuration for the application's logging system. class LoggingConfig(BaseModel): diff --git a/backend/src/core/utils/async_network.py b/backend/src/core/utils/async_network.py index bbbd1cff..6654ee4c 100644 --- a/backend/src/core/utils/async_network.py +++ b/backend/src/core/utils/async_network.py @@ -24,22 +24,27 @@ from .network import ( ) -# #region AsyncAPIClient [C:3] [TYPE Class] -# @BRIEF Async Superset API client backed by httpx.AsyncClient with shared auth cache. +# #region AsyncAPIClient [C:4] [TYPE Class] +# @BRIEF Async Superset API client backed by httpx.AsyncClient with shared auth cache and optional semaphore. +# @PRE config contains base_url and auth payload. +# @POST Client is ready for async request/authentication flow. Connection pool initialised. +# @SIDE_EFFECT Performs HTTP I/O; acquires/releases semaphore if configured. # @RELATION DEPENDS_ON -> [SupersetAuthCache] # @RELATION CALLS -> [SupersetAuthCache.get] # @RELATION CALLS -> [SupersetAuthCache.set] +# @INVARIANT semaphore.acquire() before request and semaphore.release() in finally. class AsyncAPIClient: DEFAULT_TIMEOUT = 30 _auth_locks: dict[tuple[str, str, bool], asyncio.Lock] = {} # #region AsyncAPIClient.__init__ [TYPE Function] [C:3] - # @PURPOSE: Initialize async API client for one environment. + # @PURPOSE: Initialize async API client for one environment with optional shared semaphore. # @PRE config contains base_url and auth payload. # @POST Client is ready for async request/authentication flow. # @DATA_CONTRACT Input[config: Dict[str, Any]] -> self._auth_cache_key[str] # @RELATION CALLS -> [AsyncAPIClient._normalize_base_url] # @RELATION DEPENDS_ON -> [SupersetAuthCache] - def __init__(self, config: dict[str, Any], verify_ssl: bool = True, timeout: int = DEFAULT_TIMEOUT): + def __init__(self, config: dict[str, Any], verify_ssl: bool = True, timeout: int = DEFAULT_TIMEOUT, + semaphore: asyncio.Semaphore | None = None): self.base_url: str = self._normalize_base_url(config.get("base_url", "")) self.api_base_url: str = f"{self.base_url}/api/v1" self.auth = config.get("auth") @@ -56,6 +61,8 @@ class AsyncAPIClient: self.auth, verify_ssl, ) + # Optional shared semaphore for connection limiting (set by SupersetClientRegistry). + self._semaphore = semaphore # #endregion AsyncAPIClient.__init__ # #region AsyncAPIClient._normalize_base_url [TYPE Function] [C:1] # @PURPOSE: Normalize base URL for Superset API root construction. @@ -162,19 +169,81 @@ class AsyncAPIClient: "Content-Type": "application/json", } # #endregion AsyncAPIClient.get_headers - # #region AsyncAPIClient.request [TYPE Function] [C:3] - # @PURPOSE: Perform one authenticated async Superset API request. - # @POST Returns JSON payload or raw httpx.Response when raw_response=true. - # @SIDE_EFFECT Performs network I/O. + # #region AsyncAPIClient.request [TYPE Function] [C:4] + # @PURPOSE: Perform one authenticated async Superset API request with optional semaphore. + # @POST Returns JSON payload (dict) or raw httpx.Response when raw_response=true. + # @SIDE_EFFECT Performs network I/O. Acquires/releases semaphore if configured. # @RELATION CALLS -> [AsyncAPIClient.get_headers] - # @RELATION CALLS -> [EXT:method:AsyncAPIClient._handle_http_error] + # @RELATION CALLS -> [AsyncAPIClient._handle_http_error] # @RELATION CALLS -> [AsyncAPIClient._handle_network_error] # @RELATION CALLS -> [AsyncAPIClient._is_dashboard_endpoint] - # @RELATION DEPENDS_ON -> [DashboardNotFoundError] - # @RELATION DEPENDS_ON -> [SupersetAPIError] - # @RELATION DEPENDS_ON -> [PermissionDeniedError] - # @RELATION DEPENDS_ON -> [AuthenticationError] - # @RELATION DEPENDS_ON -> [NetworkError] + async def request( + self, + method: str, + endpoint: str, + params: dict[str, Any] | None = None, + data: Any = None, + headers: dict[str, str] | None = None, + raw_response: bool = False, + timeout: int | None = None, + allow_redirects: bool = True, + ) -> Any: + with belief_scope("AsyncAPIClient.request"): + # Acquire semaphore if configured (global per-env limit). + if self._semaphore is not None: + await self._semaphore.acquire() + try: + url = self._build_api_url(endpoint) + req_headers = {**headers} if headers else {} + auth_headers = await self.get_headers() + req_headers.update(auth_headers) + + response = await self._client.request( + method=method, + url=url, + params=params, + json=data, + headers=req_headers, + timeout=httpx.Timeout(timeout or self.request_settings.get("timeout", self.DEFAULT_TIMEOUT)), + follow_redirects=allow_redirects, + ) + + if response.status_code == 401: + if method.upper() in ("GET", "HEAD", "OPTIONS"): + SupersetAuthCache.invalidate(self._auth_cache_key) + self._authenticated = False + auth_headers = await self.get_headers() + req_headers.update(auth_headers) + response = await self._client.request( + method=method, url=url, params=params, json=data, + headers=req_headers, + timeout=httpx.Timeout(timeout or self.request_settings.get("timeout", self.DEFAULT_TIMEOUT)), + follow_redirects=allow_redirects, + ) + else: + raise AuthenticationError() + + if raw_response: + return response + + try: + return response.json() + except Exception: + raise SupersetAPIError( + f"Failed to parse response JSON: {response.text[:500]}", + status_code=response.status_code, + ) + except httpx.HTTPStatusError as exc: + self._handle_http_error(exc, endpoint) + except httpx.HTTPError as exc: + url = self._build_api_url(endpoint) + self._handle_network_error(exc, url) + finally: + if self._semaphore is not None: + self._semaphore.release() + # #endregion AsyncAPIClient.request + + # #region AsyncAPIClient._handle_http_error [TYPE Function] [C:3] def _handle_http_error(self, exc: httpx.HTTPStatusError, endpoint: str) -> None: with belief_scope("AsyncAPIClient._handle_http_error"): status_code = exc.response.status_code diff --git a/backend/src/core/utils/client_registry.py b/backend/src/core/utils/client_registry.py new file mode 100644 index 00000000..a858698c --- /dev/null +++ b/backend/src/core/utils/client_registry.py @@ -0,0 +1,128 @@ +# #region SupersetClientRegistryModule [C:4] [TYPE Module] [SEMANTICS async, superset, client, registry, semaphore] +# @BRIEF Singleton registry of long-lived async Superset clients per environment (env_id). +# Stores shared httpx.AsyncClient + shared asyncio.Semaphore + asyncio.Lock for auth refresh. +# @LAYER Infrastructure +# @RELATION DEPENDS_ON -> [async_network.AsyncAPIClient] +# @RELATION DEPENDS_ON -> [config_models.Environment] +# @INVARIANT One client and one semaphore per env_id for the entire application lifetime. +# @INVARIANT All clients are closed on shutdown (calling shutdown()). +# @RATIONALE Global per-env semaphore is impossible without a singleton registry. +# Request-scoped clients lose connection pooling. Shutdown without a registry leaves +# dangling connections. +# @REJECTED Per-request clients — lose connection pooling, semaphore becomes per-client. +# Inline creation in each use-case — connection leaks. + +import asyncio +from typing import Any + +from .async_network import AsyncAPIClient +from ..logger import logger + + +class _ClientSlot: + """Internal slot holding a client and its shared resources.""" + def __init__(self, client: AsyncAPIClient, semaphore: asyncio.Semaphore, lock: asyncio.Lock): + self.client = client + self.semaphore = semaphore + self.lock = lock + + +# Module-level registry — singleton dict. +_registry: dict[str, _ClientSlot] = {} +_create_lock = asyncio.Lock() + + +# #region get_client [C:3] [TYPE Function] +# @BRIEF Get or create an AsyncAPIClient for the given environment config. +# Client is created lazily on first access. Subsequent calls return the same client. +# @PRE env_config is a valid config dict with base_url and auth. +# @POST Returns AsyncAPIClient instance with shared semaphore and auth lock. +# @SIDE_EFFECT On first call: creates httpx.AsyncClient, semaphore, and auth lock. +# @RELATION CALLS -> [AsyncAPIClient] +async def get_client( + env_config: dict[str, Any], + connection_pool_size: int = 20, + request_timeout: int = 30, +) -> AsyncAPIClient: + # Build stable env_id from config. + env_id = f"{env_config.get('base_url','')}|{env_config.get('auth',{}).get('username','')}" + # Fast path without lock. + slot = _registry.get(env_id) + if slot is not None: + return slot.client + # Slow path with lock — create once. + async with _create_lock: + slot = _registry.get(env_id) + if slot is not None: + return slot.client + verify_ssl = env_config.get("verify_ssl", True) + client = AsyncAPIClient( + config=env_config, + verify_ssl=verify_ssl, + timeout=request_timeout, + ) + semaphore = asyncio.Semaphore(connection_pool_size) + lock = asyncio.Lock() + _registry[env_id] = _ClientSlot(client=client, semaphore=semaphore, lock=lock) + logger.reason("SupersetClientRegistry.get_client", + extra={"payload": {"env_id": env_id, "pool_size": connection_pool_size}}) + return client + + +# #endregion get_client + + +# #region get_semaphore [C:2] [TYPE Function] +# @BRIEF Return the shared semaphore for the given env_config. +# @PRE Client must have been created via get_client first. +# @POST Returns asyncio.Semaphore instance. +async def get_semaphore(env_config: dict[str, Any]) -> asyncio.Semaphore: + env_id = f"{env_config.get('base_url','')}|{env_config.get('auth',{}).get('username','')}" + slot = _registry.get(env_id) + if slot is None: + # Create the client first, which creates the semaphore. + await get_client(env_config) + slot = _registry[env_id] + return slot.semaphore + + +# #endregion get_semaphore + + +# #region get_auth_lock [C:2] [TYPE Function] +# @BRIEF Return the shared auth lock for the given env_config. +# @PRE Client must have been created via get_client first. +# @POST Returns asyncio.Lock instance for serializing auth refresh. +async def get_auth_lock(env_config: dict[str, Any]) -> asyncio.Lock: + env_id = f"{env_config.get('base_url','')}|{env_config.get('auth',{}).get('username','')}" + slot = _registry.get(env_id) + if slot is None: + await get_client(env_config) + slot = _registry[env_id] + return slot.lock + + +# #endregion get_auth_lock + + +# #region shutdown [C:3] [TYPE Function] +# @BRIEF Close all async clients and clear the registry. +# @PRE Application is shutting down. +# @POST All httpx.AsyncClient instances are closed. Registry cleared. +# @SIDE_EFFECT Closes network connections for all envs. +async def shutdown() -> None: + env_ids = list(_registry.keys()) + for env_id in env_ids: + slot = _registry[env_id] + try: + await slot.client.aclose() + except Exception as exc: + logger.explore("SupersetClientRegistry.shutdown", + extra={"payload": {"env_id": env_id}, "error": str(exc)}) + _registry.clear() + logger.reason("SupersetClientRegistry.shutdown", + extra={"payload": {"closed": len(env_ids)}}) + + +# #endregion shutdown +# #endregion SupersetClientRegistryModule diff --git a/backend/src/core/utils/executors.py b/backend/src/core/utils/executors.py new file mode 100644 index 00000000..461321ad --- /dev/null +++ b/backend/src/core/utils/executors.py @@ -0,0 +1,149 @@ +# #region BlockingExecutorsModule [C:4] [TYPE Module] [SEMANTICS async, blocking, executors, run_blocking, threadpool] +# @BRIEF Named bounded executors for sync DB/file/git work. All production blocking operations use +# loop.run_in_executor via the run_blocking helper instead of default asyncio.to_thread. +# @LAYER Infrastructure +# @RELATION DEPENDS_ON -> [EXT:concurrent.futures:ThreadPoolExecutor] +# @RELATION DEPENDS_ON -> [EXT:asyncio:AbstractEventLoop] +# @RATIONALE asyncio.to_thread uses default executor without backpressure. Named bounded executors +# prevent thread pool exhaustion under load. Semaphore per kind limits queue depth. +# @REJECTED Default asyncio.to_thread — no backpressure, no bounded queue, can exhaust thread pool. + +import asyncio +from concurrent.futures import ThreadPoolExecutor +from functools import partial +from typing import Any, Callable + +from ..logger import logger + +# Module-level executors — one per kind, created once. +_db_executor: ThreadPoolExecutor | None = None +_file_executor: ThreadPoolExecutor | None = None +_git_executor: ThreadPoolExecutor | None = None + +# Per-kind semaphores for queue backpressure. +_db_semaphore: asyncio.Semaphore | None = None +_file_semaphore: asyncio.Semaphore | None = None +_git_semaphore: asyncio.Semaphore | None = None + +# Default config — overridden by init_executors(). +_DEFAULT_MAX_WORKERS = 10 +_DEFAULT_QUEUE_TIMEOUT = 30.0 + + +# #region init_executors [C:2] [TYPE Function] +# @BRIEF Initialize named executors with given config. Called once at application startup. +# @PRE event loop is running. +# @POST Executors and semaphores are ready for use. +# @SIDE_EFFECT Creates ThreadPoolExecutor instances. +def init_executors( + db_workers: int = 10, + file_workers: int = 10, + git_workers: int = 5, + queue_timeout: float = 30.0, +) -> None: + global _db_executor, _file_executor, _git_executor + global _db_semaphore, _file_semaphore, _git_semaphore + + _db_executor = ThreadPoolExecutor(max_workers=db_workers, thread_name_prefix="db") + _file_executor = ThreadPoolExecutor(max_workers=file_workers, thread_name_prefix="file") + _git_executor = ThreadPoolExecutor(max_workers=git_workers, thread_name_prefix="git") + + _db_semaphore = asyncio.Semaphore(db_workers * 2) + _file_semaphore = asyncio.Semaphore(file_workers * 2) + _git_semaphore = asyncio.Semaphore(git_workers * 2) + + logger.reason("init_executors", + extra={"payload": {"db": db_workers, "file": file_workers, "git": git_workers, "queue_timeout": queue_timeout}}) + + +# #endregion init_executors + + +# #region shutdown_executors [C:2] [TYPE Function] +# @BRIEF Shut down all executors. Called at application shutdown. +# @POST All executors are shut down, pending futures cancelled. +# @SIDE_EFFECT Waits for running tasks up to timeout. +def shutdown_executors(wait: bool = True, cancel_futures: bool = True) -> None: + global _db_executor, _file_executor, _git_executor + for executor in (_db_executor, _file_executor, _git_executor): + if executor is not None: + executor.shutdown(wait=wait, cancel_futures=cancel_futures) + _db_executor = _file_executor = _git_executor = None + logger.reason("shutdown_executors", extra={"payload": {}}) + + +# #endregion shutdown_executors + + +# #region _get_executor [C:1] [TYPE Function] +# @BRIEF Return named executor and semaphore for given kind. +# @PRE init_executors has been called. +# @POST Returns (executor, semaphore) tuple. +def _get_executor(kind: str) -> tuple[ThreadPoolExecutor, asyncio.Semaphore | None]: + if kind == "db": + return _db_executor, _db_semaphore + if kind == "file": + return _file_executor, _file_semaphore + if kind == "git": + return _git_executor, _git_semaphore + raise ValueError(f"Unknown executor kind: {kind}. Use 'db', 'file', or 'git'.") + + +# #endregion _get_executor + + +# #region run_blocking [C:4] [TYPE Function] +# @BRIEF Execute a blocking function in a named bounded executor. +# @PRE init_executors has been called. kind is one of 'db', 'file', 'git'. +# @POST fn(*args) executed in named executor. Returns result or raises timeout/exception. +# @SIDE_EFFECT Acquires/releases per-kind semaphore. Runs function in thread pool. +# @RATIONALE asyncio.to_thread does not accept an executor parameter and uses default +# unbounded pool. run_blocking uses loop.run_in_executor with named executors and +# semaphore-based backpressure. +# @REJECTED asyncio.to_thread — default executor without queue backpressure. +async def run_blocking( + kind: str, + fn: Callable[..., Any], + *args: Any, + timeout: float | None = None, + **kwargs: Any, +) -> Any: + executor, semaphore = _get_executor(kind) + + if semaphore is not None: + try: + await asyncio.wait_for(semaphore.acquire(), timeout=timeout or _DEFAULT_QUEUE_TIMEOUT) + except asyncio.TimeoutError: + logger.explore(f"run_blocking.{kind}", + extra={"error": f"Queue timeout ({timeout or _DEFAULT_QUEUE_TIMEOUT}s)"}) + raise asyncio.TimeoutError(f"Blocking executor '{kind}' queue timeout after {timeout or _DEFAULT_QUEUE_TIMEOUT}s") + + try: + loop = asyncio.get_running_loop() + bound_fn = partial(fn, *args, **kwargs) + result = await loop.run_in_executor(executor, bound_fn) + return result + except asyncio.CancelledError: + logger.explore(f"run_blocking.{kind}", + extra={"error": "Cancelled — function continues in thread pool"}) + raise + finally: + if semaphore is not None: + semaphore.release() + + +# #endregion run_blocking + + +# #region run_cpu_blocking [C:2] [TYPE Function] +# @BRIEF Execute CPU-bound work in default executor (no bounded pool needed). +# @POST fn(*args) executed in default thread pool. +# @SIDE_EFFECT Runs function in thread pool. Does not acquire semaphore. +async def run_cpu_blocking(fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + loop = asyncio.get_running_loop() + bound_fn = partial(fn, *args, **kwargs) + return await loop.run_in_executor(None, bound_fn) + + +# #endregion run_cpu_blocking +# #endregion BlockingExecutorsModule