diff --git a/app/api/v1/endpoints/alpaca.py b/app/api/v1/endpoints/alpaca.py index 439494a..1e7e115 100644 --- a/app/api/v1/endpoints/alpaca.py +++ b/app/api/v1/endpoints/alpaca.py @@ -3,6 +3,7 @@ Alpaca Market Data endpoints — standalone price data via Alpaca API """ import asyncio +import gc from datetime import date, datetime, timezone, timedelta from typing import Optional from zoneinfo import ZoneInfo @@ -15,7 +16,8 @@ _MARKET_CLOSE_HOUR = 16 # 4:00 PM ET # Limit concurrent Alpaca intraday processing to prevent event-loop saturation # under bulk backfill workloads. Callers beyond this limit wait on the semaphore # (cheap asyncio wait) rather than flooding httpx connections and DB sessions. -_INTRADAY_SEMAPHORE = asyncio.Semaphore(3) +# Semaphore(5): BaseHTTPMiddleware removed → task count halved → safe to allow 5. +_INTRADAY_SEMAPHORE = asyncio.Semaphore(5) def _market_closed_for(d: date) -> bool: @@ -143,6 +145,11 @@ async def get_alpaca_intraday_multi( finally: await svc.client.close() + # Reclaim Alpaca HTTP buffers, intermediate bar dicts, and DB row objects + # before building the serialised response. Prevents Python heap from growing + # unboundedly across thousands of backfill requests in a long-running worker. + gc.collect() + bars = { ticker: [ { @@ -217,6 +224,8 @@ async def get_alpaca_intraday_today( finally: await svc.client.close() + gc.collect() + bars = { ticker: [ { diff --git a/app/middleware/error_logger.py b/app/middleware/error_logger.py index a8198a9..0790b1c 100644 --- a/app/middleware/error_logger.py +++ b/app/middleware/error_logger.py @@ -2,13 +2,19 @@ Error logging middleware for capturing and storing API errors. Request logs are buffered in an in-memory async queue and flushed in batch -every second (or when the buffer reaches 100 entries) by a background task +every second (or when the buffer reaches 500 entries) by a background task started at app startup via start_request_log_flusher(). This avoids one DB write per request at high throughput (1 M+ req/day) and eliminates DB lock contention as a contributor to p95 tail latency. Error logs (4xx/5xx) are still written synchronously because they are relatively rare and need accurate timing. + +Pure ASGI implementation — does NOT subclass BaseHTTPMiddleware. +BaseHTTPMiddleware spawns an extra asyncio task per request via call_next(), +doubling event-loop task counts under load and causing health-check timeouts. +A raw ASGI __call__ wraps the downstream send callable instead, staying in +the same task throughout. """ import asyncio @@ -17,15 +23,12 @@ import time import traceback import uuid from datetime import datetime, timedelta, timezone -from typing import Callable, Optional +from typing import Optional +from urllib.parse import parse_qs -from fastapi import Request, Response -from fastapi.responses import JSONResponse -from sqlalchemy.ext.asyncio import AsyncSession -from starlette.middleware.base import BaseHTTPMiddleware -from starlette.types import ASGIApp +from starlette.responses import JSONResponse +from starlette.types import ASGIApp, Receive, Scope, Send, Message -from app.core.database import get_db from app.models.error_log import ErrorLog from app.models.request_log import RequestLog import logging @@ -112,176 +115,149 @@ def start_request_log_flusher() -> None: logger.info("Request log flusher background task started") -class ErrorLoggingMiddleware(BaseHTTPMiddleware): - """Middleware to log all API requests and errors to database""" - - def __init__(self, app: ASGIApp): - super().__init__(app) - - async def dispatch(self, request: Request, call_next: Callable) -> Response: - """Process request and log any errors that occur""" - - # Generate unique request ID +class ErrorLoggingMiddleware: + """Pure ASGI middleware: logs all requests and 4xx/5xx errors to the database. + + Unlike a BaseHTTPMiddleware subclass, this implementation wraps the ASGI + `send` callable directly. Every request is handled in a single asyncio + task — no call_next() extra task, no event-loop task doubling. + """ + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + # Pass non-HTTP scopes (lifespan, websocket) straight through. + if scope["type"] != "http": + await self.app(scope, receive, send) + return + request_id = str(uuid.uuid4())[:8] - request.state.request_id = request_id - - # Track request start time start_time = time.time() - - # Store request details for potential error logging - request_info = self._extract_request_info_minimal(request) + request_info = self._extract_request_info(scope) + + # Mutable state shared with the send wrapper (nonlocal in closure). + status_code: list[int] = [200] + response_headers: list = [] + body_chunks: list[bytes] = [] + is_error: list[bool] = [False] + + async def send_wrapper(message: Message) -> None: + if message["type"] == "http.response.start": + status_code[0] = message["status"] + is_error[0] = status_code[0] >= 400 + # Inject X-Request-ID into the response headers. + headers = list(message.get("headers", [])) + headers.append((b"x-request-id", request_id.encode())) + response_headers.extend(headers) + await send({**message, "headers": headers}) + + elif message["type"] == "http.response.body": + if is_error[0]: + body_chunks.append(message.get("body", b"")) + await send(message) + + else: + await send(message) - logger.debug(f"Processing request {request_id}: {request.method} {request.url.path}") - try: - # Process the request - response = await call_next(request) - - # Calculate response time + await self.app(scope, receive, send_wrapper) + except Exception as exc: response_time_ms = (time.time() - start_time) * 1000 - - # Store response body for error cases - response_body = b"" - error_detail = None - - # Check if response indicates an error (4xx or 5xx) - if response.status_code >= 400: - logger.debug(f"Error response detected: {response.status_code} for request {request_id}") - - # Full extraction for error logging (body + headers) - request_info = await self._extract_request_info(request) - - # Try to capture response body for errors - # We need to consume the response body and recreate it - from starlette.responses import Response - - # Collect response body chunks - body_chunks = [] - async for chunk in response.body_iterator: - body_chunks.append(chunk) - response_body = b''.join(body_chunks) - - # Try to parse as JSON - try: - if response_body: - error_detail = json.loads(response_body.decode('utf-8')) - except Exception as e: - logger.warning(f"Could not parse error response body as JSON: {e}") - # Store raw text if not JSON - try: - error_detail = {"raw_response": response_body.decode('utf-8')} - except: - error_detail = {"raw_response": str(response_body)} - - # Log the error with response body - logger.debug(f"Logging error for request {request_id}") - await self._log_error( - request_id=request_id, - request_info=request_info, - status_code=response.status_code, - error_detail=error_detail, - response_time_ms=response_time_ms - ) - - # Recreate response with the same body - response = Response( - content=response_body, - status_code=response.status_code, - headers=dict(response.headers), - media_type=response.media_type - ) - - # Log all requests (not just errors) - # Add data source info to headers for successful responses - data_source = response.headers.get("X-Data-Source", None) - await self._log_request( - request_id=request_id, - request_info=request_info, - status_code=response.status_code, - response_time_ms=response_time_ms, - response_size=len(response_body) if response_body else None, - data_source=data_source - ) - - # Add request ID to response headers - response.headers["X-Request-ID"] = request_id - return response - - except Exception as e: - # Log unexpected errors - response_time_ms = (time.time() - start_time) * 1000 - - # Full extraction for error logging - request_info = await self._extract_request_info(request) - await self._log_error( request_id=request_id, request_info=request_info, status_code=500, error_type="INTERNAL_SERVER_ERROR", - error_message=str(e), + error_message=str(exc), stack_trace=traceback.format_exc(), - response_time_ms=response_time_ms + response_time_ms=response_time_ms, ) - - # Return error response - return JSONResponse( + error_response = JSONResponse( status_code=500, content={ "error_type": "INTERNAL_SERVER_ERROR", "message": "An unexpected error occurred", "request_id": request_id, - "timestamp": datetime.now(timezone.utc).isoformat() + "timestamp": datetime.now(timezone.utc).isoformat(), }, - headers={"X-Request-ID": request_id} + headers={"X-Request-ID": request_id}, + ) + await error_response(scope, receive, send) + return + + # --- Post-response logging (runs after all body chunks are sent) --- + response_time_ms = (time.time() - start_time) * 1000 + + if is_error[0]: + response_body = b"".join(body_chunks) + error_detail = None + try: + if response_body: + error_detail = json.loads(response_body.decode("utf-8")) + except Exception: + try: + error_detail = {"raw_response": response_body.decode("utf-8")} + except Exception: + error_detail = {"raw_response": str(response_body)} + + await self._log_error( + request_id=request_id, + request_info=request_info, + status_code=status_code[0], + error_detail=error_detail, + response_time_ms=response_time_ms, ) - - def _extract_request_info_minimal(self, request: Request) -> dict: - """Extract minimal request info (sync, no body read) for success-path logging.""" + + # Extract X-Data-Source from response headers if present. + data_source = None + for name, value in response_headers: + if name == b"x-data-source": + data_source = value.decode() + break + + response_size = sum(len(c) for c in body_chunks) if body_chunks else None + await self._log_request( + request_id=request_id, + request_info=request_info, + status_code=status_code[0], + response_time_ms=response_time_ms, + response_size=response_size, + data_source=data_source, + ) + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + def _extract_request_info(self, scope: Scope) -> dict: + """Extract minimal request info from the ASGI scope (sync, no body read).""" + headers_raw = dict(scope.get("headers", [])) + user_agent = headers_raw.get(b"user-agent", b"").decode("latin-1") or None + path = scope.get("path", "") + qs_bytes = scope.get("query_string", b"") + qs = qs_bytes.decode("latin-1") + full_url = f"{path}?{qs}" if qs else path + + query_params = None + if qs: + parsed = parse_qs(qs) + query_params = {k: v[0] if len(v) == 1 else v for k, v in parsed.items()} + + client = scope.get("client") + client_ip = client[0] if client else None + return { - "endpoint": str(request.url.path), - "method": request.method, - "path": str(request.url), - "query_params": dict(request.query_params) if request.query_params else None, + "endpoint": path, + "method": scope.get("method", ""), + "path": full_url, + "query_params": query_params, "request_body": None, "headers": None, - "user_agent": request.headers.get("user-agent"), - "client_ip": request.client.host if request.client else None, + "user_agent": user_agent, + "client_ip": client_ip, } - async def _extract_request_info(self, request: Request) -> dict: - """Extract full request information for error logging""" - - # Get request body if present - body = None - if request.method in ["POST", "PUT", "PATCH"]: - try: - body_bytes = await request.body() - if body_bytes: - body = json.loads(body_bytes.decode('utf-8')) - # Store body for later use in request processing - request._body = body_bytes - except Exception as e: - logger.warning(f"Could not parse request body: {e}") - - # Extract headers (remove sensitive ones) - headers = dict(request.headers) - sensitive_headers = ['authorization', 'api-key', 'x-api-key', 'cookie'] - for header in sensitive_headers: - if header in headers: - headers[header] = '***REDACTED***' - - return { - "endpoint": str(request.url.path), - "method": request.method, - "path": str(request.url), - "query_params": dict(request.query_params) if request.query_params else None, - "request_body": body, - "headers": headers, - "user_agent": headers.get("user-agent"), - "client_ip": request.client.host if request.client else None - } - async def _log_error( self, request_id: str, @@ -291,48 +267,37 @@ class ErrorLoggingMiddleware(BaseHTTPMiddleware): error_message: Optional[str] = None, error_detail: Optional[dict] = None, stack_trace: Optional[str] = None, - response_time_ms: Optional[float] = None - ): - """Log error to database""" - - # Skip logging errors for log deletion endpoints to avoid logging the deletion of logs + response_time_ms: Optional[float] = None, + ) -> None: + """Log error to database.""" endpoint = request_info["endpoint"] method = request_info["method"] - # Don't log errors for DELETE requests to log management endpoints - if (method == "DELETE" and - (endpoint.startswith("/api/v1/admin/requests/logs") or - endpoint.startswith("/api/v1/admin/errors/logs"))): + # Don't log errors for DELETE requests to log management endpoints. + if method == "DELETE" and ( + endpoint.startswith("/api/v1/admin/requests/logs") + or endpoint.startswith("/api/v1/admin/errors/logs") + ): return try: - # Get database session from app.core.database import AsyncSessionLocal async with AsyncSessionLocal() as db: - # Extract error information from detail if available if error_detail and isinstance(error_detail, dict): if not error_type: error_type = error_detail.get("error_type", "UNKNOWN_ERROR") if not error_message: error_message = error_detail.get("message", "Unknown error occurred") - - # Set defaults + if not error_type: - if status_code >= 500: - error_type = "INTERNAL_SERVER_ERROR" - elif status_code >= 400: - error_type = "CLIENT_ERROR" - else: - error_type = "UNKNOWN_ERROR" - + error_type = "INTERNAL_SERVER_ERROR" if status_code >= 500 else "CLIENT_ERROR" if not error_message: error_message = f"HTTP {status_code} error" - - # Create error log entry - error_log = ErrorLog( + + db.add(ErrorLog( request_id=request_id, - endpoint=request_info["endpoint"], - method=request_info["method"], + endpoint=endpoint, + method=method, path=request_info["path"], query_params=request_info["query_params"], request_body=request_info["request_body"], @@ -344,28 +309,22 @@ class ErrorLoggingMiddleware(BaseHTTPMiddleware): stack_trace=stack_trace, user_agent=request_info["user_agent"], client_ip=request_info["client_ip"], - response_time_ms=response_time_ms - ) - - db.add(error_log) + response_time_ms=response_time_ms, + )) await db.commit() - logger.error( - f"Error logged - Request ID: {request_id}, " - f"Endpoint: {request_info['endpoint']}, " - f"Status: {status_code}, " - f"Error: {error_type} - {error_message}" - ) - + logger.error( + f"Error logged - Request ID: {request_id}, " + f"Endpoint: {endpoint}, Status: {status_code}, " + f"Error: {error_type} - {error_message}" + ) except Exception as e: - # If we can't log to database, at least log to file logger.error(f"Failed to log error to database: {e}") logger.error( f"Original error - Request ID: {request_id}, " - f"Status: {status_code}, " - f"Error: {error_type} - {error_message}" + f"Status: {status_code}, Error: {error_type} - {error_message}" ) - + async def _log_request( self, request_id: str, @@ -373,28 +332,26 @@ class ErrorLoggingMiddleware(BaseHTTPMiddleware): status_code: int, response_time_ms: Optional[float] = None, response_size: Optional[int] = None, - data_source: Optional[str] = None - ): + data_source: Optional[str] = None, + ) -> None: """Enqueue a request log entry for async batch flushing. Non-blocking: the entry is placed in _REQUEST_LOG_QUEUE and written to the database in bulk by the background flusher, avoiding a synchronous DB round-trip on every request. """ - - # Skip logging for log deletion endpoints endpoint = request_info["endpoint"] method = request_info["method"] - if (method == "DELETE" and - (endpoint.startswith("/api/v1/admin/requests/logs") or - endpoint.startswith("/api/v1/admin/errors/logs"))): + if method == "DELETE" and ( + endpoint.startswith("/api/v1/admin/requests/logs") + or endpoint.startswith("/api/v1/admin/errors/logs") + ): return - # Augment headers with data source if provided - headers = request_info.get("headers") or {} + headers: dict = {} if data_source: - headers = {**headers, "X-Data-Source": data_source} + headers["X-Data-Source"] = data_source entry = { "request_id": request_id, @@ -403,7 +360,7 @@ class ErrorLoggingMiddleware(BaseHTTPMiddleware): "path": request_info["path"], "query_params": request_info["query_params"], "request_body": request_info["request_body"], - "headers": headers, + "headers": headers or None, "status_code": status_code, "response_size": response_size, "user_agent": request_info["user_agent"], @@ -414,7 +371,6 @@ class ErrorLoggingMiddleware(BaseHTTPMiddleware): try: _REQUEST_LOG_QUEUE.put_nowait(entry) except asyncio.QueueFull: - # Queue is at capacity — fall back to a direct write so we don't drop the log logger.warning("Request log queue full — writing synchronously") try: from app.core.database import AsyncSessionLocal @@ -422,4 +378,4 @@ class ErrorLoggingMiddleware(BaseHTTPMiddleware): db.add(RequestLog(**entry)) await db.commit() except Exception as e: - logger.error(f"Failed to write request log (fallback): {e}") \ No newline at end of file + logger.error(f"Failed to write request log (fallback): {e}") diff --git a/docker-compose.yml b/docker-compose.yml index 013d4b3..a5d3687 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -68,7 +68,7 @@ services: mem_limit: 2g memswap_limit: 2g restart: unless-stopped - command: ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "18000", "--reload", "--limit-concurrency", "20", "--limit-max-requests", "500"] + command: ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "18000", "--reload", "--limit-concurrency", "20"] # Frontend Application frontend: