From 3d92ff4e9bf26d54d7d6ed6f69a4b68935a4685f Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Sun, 1 Mar 2026 01:20:47 +0100 Subject: [PATCH 1/7] feat: misc issues fixed, also update saga/exec status --- backend/app/api/routes/admin/events.py | 12 +-- backend/app/api/routes/admin/executions.py | 1 + backend/app/api/routes/sse.py | 14 +--- backend/app/core/container.py | 2 +- backend/app/core/metrics/queue.py | 8 ++ backend/app/core/middlewares/__init__.py | 2 + backend/app/core/middlewares/csrf.py | 2 +- backend/app/core/middlewares/rate_limit.py | 27 ++++-- .../core/middlewares/request_size_limit.py | 37 +++++++- backend/app/core/security.py | 20 +++-- backend/app/core/utils.py | 48 ++++++----- backend/app/db/docs/resource.py | 4 +- .../app/db/repositories/event_repository.py | 14 +++- .../db/repositories/execution_repository.py | 84 +++++++++++++++++-- .../repositories/notification_repository.py | 31 ++++--- .../app/db/repositories/replay_repository.py | 9 +- .../app/db/repositories/saga_repository.py | 9 +- .../app/db/repositories/user_repository.py | 4 + backend/app/dlq/manager.py | 4 +- backend/app/dlq/models.py | 19 ++++- backend/app/domain/enums/__init__.py | 20 ++++- backend/app/domain/enums/execution.py | 26 ++++-- backend/app/domain/enums/replay.py | 21 ++++- backend/app/domain/enums/saga.py | 24 +++++- backend/app/domain/enums/storage.py | 7 ++ backend/app/events/core/producer.py | 11 ++- backend/app/events/handlers.py | 10 ++- backend/app/main.py | 24 ++++-- backend/app/services/auth_service.py | 23 ++++- .../services/event_replay/replay_service.py | 3 +- backend/app/services/execution_queue.py | 9 +- backend/app/services/execution_service.py | 72 ++-------------- backend/app/services/notification_service.py | 33 ++++---- .../app/services/pod_monitor/event_mapper.py | 11 ++- backend/app/services/rate_limit_service.py | 49 +++++++---- backend/app/services/saga/execution_saga.py | 53 ++++++------ .../app/services/saga/saga_orchestrator.py | 36 +++++++- backend/app/services/sse/sse_service.py | 12 +-- backend/app/settings.py | 4 +- .../saga/test_execution_saga_steps.py | 41 +++++---- docs/architecture/domain-exceptions.md | 1 + docs/architecture/middleware.md | 2 +- docs/architecture/overview.md | 2 +- docs/components/saved-scripts.md | 2 +- 44 files changed, 552 insertions(+), 295 deletions(-) diff --git a/backend/app/api/routes/admin/events.py b/backend/app/api/routes/admin/events.py index 9edf953e..5ee48fcd 100644 --- a/backend/app/api/routes/admin/events.py +++ b/backend/app/api/routes/admin/events.py @@ -9,6 +9,7 @@ from sse_starlette.sse import EventSourceResponse from app.api.dependencies import admin_user +from app.api.routes.common import SSEResponse from app.domain.enums import EventType, ExportFormat from app.domain.events import EventFilter as DomainEventFilter from app.domain.replay import ReplayFilter @@ -27,15 +28,6 @@ from app.services.admin import AdminEventsService from app.services.sse import SSEService - -class _SSEResponse(EventSourceResponse): - """Workaround: sse-starlette sets media_type only in __init__, not as a - class attribute. FastAPI reads the class attribute for OpenAPI generation, - so without this subclass every SSE endpoint shows application/json.""" - - media_type = "text/event-stream" - - router = APIRouter( prefix="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/admin/events", tags=["admin-events"], route_class=DishkaRoute, dependencies=[Depends(admin_user)] ) @@ -144,7 +136,7 @@ async def replay_events( @router.get( "/replay/{session_id}/status", - response_class=_SSEResponse, + response_class=SSEResponse, responses={ 200: {"model": EventReplayStatusResponse}, 404: {"model": ErrorResponse, "description": "Replay session not found"}, diff --git a/backend/app/api/routes/admin/executions.py b/backend/app/api/routes/admin/executions.py index eba0a421..02bf8526 100644 --- a/backend/app/api/routes/admin/executions.py +++ b/backend/app/api/routes/admin/executions.py @@ -20,6 +20,7 @@ prefix="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/admin/executions", tags=["admin-executions"], route_class=DishkaRoute, + dependencies=[Depends(admin_user)], ) diff --git a/backend/app/api/routes/sse.py b/backend/app/api/routes/sse.py index 9aab54ed..108557a2 100644 --- a/backend/app/api/routes/sse.py +++ b/backend/app/api/routes/sse.py @@ -6,26 +6,18 @@ from sse_starlette.sse import EventSourceResponse from app.api.dependencies import current_user +from app.api.routes.common import SSEResponse from app.domain.user import User from app.schemas_pydantic.notification import NotificationResponse from app.schemas_pydantic.sse import SSEExecutionEventSchema from app.services.sse import SSEService - -class _SSEResponse(EventSourceResponse): - """Workaround: sse-starlette sets media_type only in __init__, not as a - class attribute. FastAPI reads the class attribute for OpenAPI generation, - so without this subclass every SSE endpoint shows application/json.""" - - media_type = "text/event-stream" - - router = APIRouter(prefix="/events", tags=["sse"], route_class=DishkaRoute) @router.get( "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/notifications/stream", - response_class=_SSEResponse, + response_class=SSEResponse, responses={200: {"model": NotificationResponse}}, ) async def notification_stream( @@ -41,7 +33,7 @@ async def notification_stream( @router.get( "/executions/{execution_id}", - response_class=_SSEResponse, + response_class=SSEResponse, responses={200: {"model": SSEExecutionEventSchema}}, ) async def execution_events( diff --git a/backend/app/core/container.py b/backend/app/core/container.py index afbce68a..2ba39c62 100644 --- a/backend/app/core/container.py +++ b/backend/app/core/container.py @@ -35,7 +35,7 @@ def create_app_container(settings: Settings) -> AsyncContainer: Args: settings: Application settings (injected via from_context). - Note: init_beanie() must be called BEFORE this container is created. + Note: init_beanie() is called in the async lifespan AFTER this container is created. KafkaBroker is created by BrokerProvider and can be retrieved via container.get(KafkaBroker) after container creation. """ diff --git a/backend/app/core/metrics/queue.py b/backend/app/core/metrics/queue.py index 16d39109..2a1dc132 100644 --- a/backend/app/core/metrics/queue.py +++ b/backend/app/core/metrics/queue.py @@ -30,6 +30,11 @@ def _create_instruments(self) -> None: description="Time spent waiting in queue before scheduling", unit="s", ) + self._event_data_lost = self._meter.create_counter( + name="queue.event_data_lost", + description="Executions lost due to expired event data in Redis", + unit="1", + ) def record_enqueue(self) -> None: self._enqueue_total.add(1) @@ -45,3 +50,6 @@ def record_release(self) -> None: def record_wait_time(self, wait_seconds: float, priority: str) -> None: self._wait_time.record(wait_seconds, attributes={"priority": priority}) + + def record_event_data_lost(self) -> None: + self._event_data_lost.add(1) diff --git a/backend/app/core/middlewares/__init__.py b/backend/app/core/middlewares/__init__.py index 0ea15f3a..869869a7 100644 --- a/backend/app/core/middlewares/__init__.py +++ b/backend/app/core/middlewares/__init__.py @@ -3,6 +3,7 @@ from .metrics import MetricsMiddleware, create_system_metrics, setup_metrics from .rate_limit import RateLimitMiddleware from .request_size_limit import RequestSizeLimitMiddleware +from .security_headers import SecurityHeadersMiddleware __all__ = [ "CacheControlMiddleware", @@ -12,4 +13,5 @@ "create_system_metrics", "RequestSizeLimitMiddleware", "RateLimitMiddleware", + "SecurityHeadersMiddleware", ] diff --git a/backend/app/core/middlewares/csrf.py b/backend/app/core/middlewares/csrf.py index ad070b4d..c186fc8b 100644 --- a/backend/app/core/middlewares/csrf.py +++ b/backend/app/core/middlewares/csrf.py @@ -19,7 +19,7 @@ class CSRFMiddleware: Requests are skipped if: - Method is safe (GET, HEAD, OPTIONS) - - Path is an auth endpoint (login, register, logout) + - Path is an auth endpoint (login, register) - Path is not under /api/ - User is not authenticated (no access_token cookie) """ diff --git a/backend/app/core/middlewares/rate_limit.py b/backend/app/core/middlewares/rate_limit.py index c6862599..4f484d4c 100644 --- a/backend/app/core/middlewares/rate_limit.py +++ b/backend/app/core/middlewares/rate_limit.py @@ -7,7 +7,6 @@ from app.core.utils import get_client_ip from app.domain.rate_limit import RateLimitStatus -from app.domain.user import User from app.services.rate_limit_service import RateLimitService from app.settings import Settings @@ -102,10 +101,28 @@ async def send_wrapper(message: Message) -> None: await self.app(scope, receive, send_wrapper) # --8<-- [start:extract_user_id] - def _extract_user_id(self, request: Request) -> str: - user: User | None = request.state.__dict__.get("user") - if user: - return str(user.user_id) + @staticmethod + def _extract_user_id(request: Request) -> str: + """Extract user identifier for rate limiting. + + Reads the JWT payload from the access_token cookie without full + verification (that happens in route-level auth dependencies). This + is safe because the value is only used as a rate-limit bucket key. + Falls back to IP-based identification if no token is present or + the payload cannot be read. + """ + token = request.cookies.get("access_token") + if token: + import base64 + import json as _json + parts = token.split(".") + if len(parts) == 3: + # Pad the base64url payload segment + padded = parts[1] + "=" * (-len(parts[1]) % 4) + payload = _json.loads(base64.urlsafe_b64decode(padded)) + username = payload.get("sub") + if username: + return f"user:{username}" return f"ip:{get_client_ip(request)}" # --8<-- [end:extract_user_id] diff --git a/backend/app/core/middlewares/request_size_limit.py b/backend/app/core/middlewares/request_size_limit.py index 0093f7a2..a44c3ec0 100644 --- a/backend/app/core/middlewares/request_size_limit.py +++ b/backend/app/core/middlewares/request_size_limit.py @@ -1,10 +1,15 @@ from starlette.responses import JSONResponse -from starlette.types import ASGIApp, Receive, Scope, Send +from starlette.types import ASGIApp, Message, Receive, Scope, Send # --8<-- [start:RequestSizeLimitMiddleware] class RequestSizeLimitMiddleware: - """Middleware to limit request size, default 10MB""" + """Middleware to limit request size, default 10MB. + + Checks Content-Length header when present for an early reject, and wraps + the ASGI ``receive`` callable to count bytes as they stream — this + catches chunked-transfer requests that omit Content-Length. + """ def __init__(self, app: ASGIApp, max_size_mb: int = 10) -> None: self.app = app @@ -29,4 +34,30 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: await response(scope, receive, send) return - await self.app(scope, receive, send) + bytes_received = 0 + max_size = self.max_size_bytes + exceeded = False + + async def receive_wrapper() -> Message: + nonlocal bytes_received, exceeded + message = await receive() + if message["type"] == "http.request": + body = message.get("body", b"") + bytes_received += len(body) + if bytes_received > max_size: + exceeded = True + raise _RequestTooLarge() + return message + + try: + await self.app(scope, receive_wrapper, send) + except _RequestTooLarge: + response = JSONResponse( + status_code=413, + content={"detail": f"Request too large. Maximum size is {max_size / 1024 / 1024}MB"}, + ) + await response(scope, receive, send) + + +class _RequestTooLarge(Exception): + pass diff --git a/backend/app/core/security.py b/backend/app/core/security.py index 4901c310..93e5c4dc 100644 --- a/backend/app/core/security.py +++ b/backend/app/core/security.py @@ -42,16 +42,24 @@ def create_access_token(self, data: dict[str, Any], expires_delta: timedelta) -> return encoded_jwt # --8<-- [end:create_access_token] - def decode_token(self, token: str) -> str: - """Decode JWT and return the username (sub claim).""" + def decode_token(self, token: str, *, allow_expired: bool = False) -> str: + """Decode JWT and return the username (sub claim). + + Args: + token: The JWT token string. + allow_expired: If True, accept expired tokens (used for logout). + """ try: - payload = jwt.decode(token, self.settings.SECRET_KEY, algorithms=[self.settings.ALGORITHM]) + options = {"verify_exp": not allow_expired} + payload = jwt.decode( + token, self.settings.SECRET_KEY, algorithms=[self.settings.ALGORITHM], options=options, + ) username: str | None = payload.get("sub") if username is None: raise InvalidCredentialsError() - except jwt.PyJWTError as e: - raise InvalidCredentialsError() from e - return username + return username + except jwt.PyJWTError: + raise InvalidCredentialsError() from None def generate_csrf_token(self, session_id: str) -> str: """Generate a signed CSRF token bound to the given session (access_token). diff --git a/backend/app/core/utils.py b/backend/app/core/utils.py index 37f48058..e5c90fec 100644 --- a/backend/app/core/utils.py +++ b/backend/app/core/utils.py @@ -32,23 +32,33 @@ def __format__(self, format_spec: str) -> str: def get_client_ip(request: Request) -> str: + """Get client IP address from request. + + Uses the direct connection IP as the authoritative source. + X-Forwarded-For is only trusted when the direct connection comes from a + known local/proxy address, preventing attackers from spoofing their IP to + bypass rate limiting. """ - Safely get client IP address from request. - Handles both normal connections and proxy forwarded requests. - """ - # Check for proxy headers first (in order of preference) - forwarded_for = request.headers.get("x-forwarded-for") - if forwarded_for: - # X-Forwarded-For can contain multiple IPs, take the first one - return forwarded_for.split(",")[0].strip() - - real_ip = request.headers.get("x-real-ip") - if real_ip: - return real_ip - - # Fall back to direct client connection - if request.client: - return request.client.host - - # Should rarely happen, but handle the case where client is None - return "127.0.0.1" + direct_ip = request.client.host if request.client else "127.0.0.1" + + if _is_trusted_proxy(direct_ip): + forwarded_for = request.headers.get("x-forwarded-for") + if forwarded_for: + return forwarded_for.split(",")[0].strip() + + real_ip = request.headers.get("x-real-ip") + if real_ip: + return real_ip + + return direct_ip + + +def _is_trusted_proxy(ip: str) -> bool: + """Check if an IP belongs to a trusted proxy (loopback or Docker-internal ranges).""" + return ( + ip.startswith("127.") + or ip == "::1" + or ip.startswith("10.") + or ip.startswith("172.") + or ip.startswith("192.168.") + ) diff --git a/backend/app/db/docs/resource.py b/backend/app/db/docs/resource.py index ef16e814..68abbd2b 100644 --- a/backend/app/db/docs/resource.py +++ b/backend/app/db/docs/resource.py @@ -3,6 +3,8 @@ from beanie import Document, Indexed from pydantic import ConfigDict, Field +from app.domain.enums import AllocationStatus + class ResourceAllocationDocument(Document): """Resource allocation bookkeeping document used by saga steps. @@ -17,7 +19,7 @@ class ResourceAllocationDocument(Document): memory_request: str cpu_limit: str memory_limit: str - status: Indexed(str) = "active" # type: ignore[valid-type] # "active" | "released" + status: Indexed(str) = AllocationStatus.ACTIVE # type: ignore[valid-type] allocated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) released_at: datetime | None = None diff --git a/backend/app/db/repositories/event_repository.py b/backend/app/db/repositories/event_repository.py index 19f23270..3bac6e32 100644 --- a/backend/app/db/repositories/event_repository.py +++ b/backend/app/db/repositories/event_repository.py @@ -57,6 +57,12 @@ async def store_event(self, event: DomainEvent) -> str: self.logger.debug(f"Stored event {event.event_id} of type {event.event_type}") return event.event_id + async def mark_publish_failed(self, event_id: str) -> None: + """Mark an event as failed to publish to Kafka for later retry.""" + await EventDocument.find_one( + EventDocument.event_id == event_id, + ).update({"$set": {"publish_failed": True, "publish_failed_at": datetime.now(timezone.utc)}}) + async def get_event(self, event_id: str) -> DomainEvent | None: doc = await EventDocument.find_one(EventDocument.event_id == event_id) if not doc: @@ -257,8 +263,12 @@ async def delete_event_with_archival( deleted_at = datetime.now(timezone.utc) archive_fields = {"deleted_at": deleted_at, "deleted_by": deleted_by, "deletion_reason": deletion_reason} archived_doc = EventArchiveDocument.model_validate(doc).model_copy(update=archive_fields) - await archived_doc.insert() - await doc.delete() + + session = await EventDocument.get_motor_collection().database.client.start_session() + async with session.start_transaction(): + await archived_doc.insert(session=session) + await doc.delete(session=session) + return ArchivedEvent.model_validate(doc).model_copy(update=archive_fields) async def get_aggregate_replay_info(self, aggregate_id: str) -> EventReplayInfo | None: diff --git a/backend/app/db/repositories/execution_repository.py b/backend/app/db/repositories/execution_repository.py index 28caee27..91e71a49 100644 --- a/backend/app/db/repositories/execution_repository.py +++ b/backend/app/db/repositories/execution_repository.py @@ -4,9 +4,10 @@ import structlog from beanie.odm.enums import SortDirection +from beanie.operators import In from app.db.docs import ExecutionDocument -from app.domain.enums import QueuePriority +from app.domain.enums import EXECUTION_ACTIVE, QueuePriority from app.domain.events import ResourceUsageDomain from app.domain.execution import ( DomainExecution, @@ -46,13 +47,16 @@ async def get_execution(self, execution_id: str) -> DomainExecution | None: return self._to_domain(doc) async def write_terminal_result(self, result: ExecutionResultDomain) -> bool: - doc = await ExecutionDocument.find_one(ExecutionDocument.execution_id == result.execution_id) - if not doc: - self.logger.warning("No execution found", execution_id=result.execution_id) - return False - - await doc.set( - { + """Atomically write a terminal result, guarded by non-terminal status check. + + Uses find_one_and_update so a slower processor cannot overwrite a result + that was already written by a faster one. + """ + update_result = await ExecutionDocument.find_one( + ExecutionDocument.execution_id == result.execution_id, + In(ExecutionDocument.status, list(EXECUTION_ACTIVE)), + ).update( + {"$set": { "status": result.status, "exit_code": result.exit_code, "stdout": result.stdout, @@ -60,8 +64,13 @@ async def write_terminal_result(self, result: ExecutionResultDomain) -> bool: "resource_usage": dataclasses.asdict(result.resource_usage) if result.resource_usage else None, "error_type": result.error_type, "updated_at": datetime.now(timezone.utc), - } + }} ) + if not update_result or getattr(update_result, "modified_count", 0) == 0: + self.logger.warning( + "Execution not found or already in terminal state", execution_id=result.execution_id, + ) + return False return True async def get_executions( @@ -96,6 +105,63 @@ async def get_execution_result(self, execution_id: str) -> ExecutionResultDomain data["resource_usage"] = ResourceUsageDomain(**data["resource_usage"]) return ExecutionResultDomain(**data) + async def aggregate_stats(self, query: dict[str, Any]) -> dict[str, Any]: + """Compute execution statistics entirely in MongoDB via aggregation pipeline.""" + pipeline: list[dict[str, Any]] = [] + if query: + pipeline.append({"$match": query}) + + pipeline.append({ + "$facet": { + "by_status": [{"$group": {"_id": "$status", "count": {"$sum": 1}}}], + "by_language": [ + {"$group": { + "_id": {"$concat": ["$lang", "-", "$lang_version"]}, + "count": {"$sum": 1}, + }}, + ], + "totals": [{"$group": { + "_id": None, + "total": {"$sum": 1}, + "successful": {"$sum": {"$cond": [{"$eq": ["$status", "completed"]}, 1, 0]}}, + }}], + "avg_duration": [ + {"$match": { + "status": "completed", + "created_at": {"$ne": None}, + "updated_at": {"$ne": None}, + }}, + {"$group": { + "_id": None, + "avg_ms": {"$avg": { + "$multiply": [ + {"$divide": [{"$subtract": ["$updated_at", "$created_at"]}, 1]}, + ], + }}, + }}, + ], + }, + }) + + collection = ExecutionDocument.get_motor_collection() + results = await collection.aggregate(pipeline).to_list(length=1) + + if not results: + return {"total": 0, "by_status": {}, "by_language": {}, "average_duration_ms": 0, "success_rate": 0} + + facets = results[0] + totals = facets["totals"][0] if facets["totals"] else {"total": 0, "successful": 0} + total = totals["total"] + successful = totals["successful"] + + return { + "total": total, + "by_status": {item["_id"]: item["count"] for item in facets["by_status"]}, + "by_language": {item["_id"]: item["count"] for item in facets["by_language"]}, + "average_duration_ms": facets["avg_duration"][0]["avg_ms"] if facets["avg_duration"] else 0, + "success_rate": successful / total if total > 0 else 0, + } + async def delete_execution(self, execution_id: str) -> bool: doc = await ExecutionDocument.find_one(ExecutionDocument.execution_id == execution_id) if not doc: diff --git a/backend/app/db/repositories/notification_repository.py b/backend/app/db/repositories/notification_repository.py index 1631a00e..6365dbd3 100644 --- a/backend/app/db/repositories/notification_repository.py +++ b/backend/app/db/repositories/notification_repository.py @@ -137,18 +137,15 @@ async def find_due_notifications(self, limit: int = 50) -> list[DomainNotificati async def try_claim_pending(self, notification_id: str) -> bool: now = datetime.now(UTC) - doc = await NotificationDocument.find_one( + result = await NotificationDocument.find_one( NotificationDocument.notification_id == notification_id, NotificationDocument.status == NotificationStatus.PENDING, Or( NotificationDocument.scheduled_for == None, # noqa: E711 LTE(NotificationDocument.scheduled_for, now), ), - ) - if not doc: - return False - await doc.set({"status": NotificationStatus.SENDING, "sent_at": now}) - return True + ).update({"$set": {"status": NotificationStatus.SENDING, "sent_at": now}}) + return bool(result and getattr(result, "modified_count", 0) > 0) # Subscriptions async def get_subscription( @@ -187,17 +184,17 @@ async def upsert_subscription( return DomainNotificationSubscription(**doc.model_dump(include=_sub_fields)) async def get_all_subscriptions(self, user_id: str) -> list[DomainNotificationSubscription]: - subs: list[DomainNotificationSubscription] = [] - for channel in NotificationChannel: - doc = await NotificationSubscriptionDocument.find_one( - NotificationSubscriptionDocument.user_id == user_id, - NotificationSubscriptionDocument.channel == channel, - ) - if doc: - subs.append(DomainNotificationSubscription(**doc.model_dump(include=_sub_fields))) - else: - subs.append(DomainNotificationSubscription(user_id=user_id, channel=channel, enabled=True)) - return subs + docs = await NotificationSubscriptionDocument.find( + NotificationSubscriptionDocument.user_id == user_id, + ).to_list() + existing: dict[NotificationChannel, DomainNotificationSubscription] = { + doc.channel: DomainNotificationSubscription(**doc.model_dump(include=_sub_fields)) + for doc in docs + } + return [ + existing.get(channel, DomainNotificationSubscription(user_id=user_id, channel=channel, enabled=True)) + for channel in NotificationChannel + ] # User query operations async def get_users_by_roles(self, roles: list[UserRole]) -> list[str]: diff --git a/backend/app/db/repositories/replay_repository.py b/backend/app/db/repositories/replay_repository.py index 66219678..46ce1947 100644 --- a/backend/app/db/repositories/replay_repository.py +++ b/backend/app/db/repositories/replay_repository.py @@ -8,7 +8,7 @@ from app.db.docs import EventDocument, ReplaySessionDocument from app.domain.admin import ReplaySessionUpdate -from app.domain.enums import ReplayStatus +from app.domain.enums import REPLAY_TERMINAL, ReplayStatus from app.domain.replay import ReplayConfig, ReplayError, ReplayFilter, ReplaySessionState _replay_fields = set(ReplaySessionState.__dataclass_fields__) @@ -66,14 +66,9 @@ async def update_session_status(self, session_id: str, status: ReplayStatus) -> return True async def delete_old_sessions(self, cutoff_time: datetime) -> int: - terminal_statuses = [ - ReplayStatus.COMPLETED, - ReplayStatus.FAILED, - ReplayStatus.CANCELLED, - ] result = await ReplaySessionDocument.find( LT(ReplaySessionDocument.created_at, cutoff_time), - In(ReplaySessionDocument.status, terminal_statuses), + In(ReplaySessionDocument.status, list(REPLAY_TERMINAL)), ).delete() return result.deleted_count if result else 0 diff --git a/backend/app/db/repositories/saga_repository.py b/backend/app/db/repositories/saga_repository.py index c998288d..ea44376c 100644 --- a/backend/app/db/repositories/saga_repository.py +++ b/backend/app/db/repositories/saga_repository.py @@ -11,7 +11,7 @@ from monggregate import Pipeline, S from app.db.docs import ExecutionDocument, SagaDocument -from app.domain.enums import SagaState +from app.domain.enums import SAGA_ACTIVE, SagaState from app.domain.saga import ( Saga, SagaConcurrencyError, @@ -84,7 +84,7 @@ async def atomic_cancel_saga( """ doc = await SagaDocument.find_one( SagaDocument.saga_id == saga_id, - In(SagaDocument.state, [SagaState.RUNNING, SagaState.CREATED]), + In(SagaDocument.state, list(SAGA_ACTIVE)), ).update( Set({ # type: ignore[no-untyped-call] SagaDocument.state: SagaState.CANCELLED, @@ -167,8 +167,9 @@ async def list_sagas(self, saga_filter: SagaFilter, limit: int = 100, skip: int ) async def get_user_execution_ids(self, user_id: str) -> list[str]: - docs = await ExecutionDocument.find(ExecutionDocument.user_id == user_id).to_list() - return [doc.execution_id for doc in docs] + collection = ExecutionDocument.get_motor_collection() + result: list[str] = await collection.distinct("execution_id", {"user_id": user_id}) + return result async def find_timed_out_sagas( self, diff --git a/backend/app/db/repositories/user_repository.py b/backend/app/db/repositories/user_repository.py index ed9d7d7f..dfd83957 100644 --- a/backend/app/db/repositories/user_repository.py +++ b/backend/app/db/repositories/user_repository.py @@ -43,6 +43,10 @@ async def create_user(self, create_data: DomainUserCreate) -> User: raise ConflictError("User already exists") from e return User(**doc.model_dump(include=_user_fields)) + async def get_user_by_email(self, email: str) -> User | None: + doc = await UserDocument.find_one(UserDocument.email == email) + return User(**doc.model_dump(include=_user_fields)) if doc else None + async def get_user_by_id(self, user_id: str) -> User | None: doc = await UserDocument.find_one(UserDocument.user_id == user_id) return User(**doc.model_dump(include=_user_fields)) if doc else None diff --git a/backend/app/dlq/manager.py b/backend/app/dlq/manager.py index a027465e..139fc33e 100644 --- a/backend/app/dlq/manager.py +++ b/backend/app/dlq/manager.py @@ -226,7 +226,7 @@ async def retry_message_manually(self, event_id: str) -> bool: self.logger.error("Message not found in DLQ", event_id=event_id) return False - if message.status in {DLQMessageStatus.DISCARDED, DLQMessageStatus.RETRIED}: + if message.status.is_terminal: self.logger.info("Skipping manual retry", event_id=event_id, status=message.status) return False @@ -278,7 +278,7 @@ async def discard_message_manually(self, event_id: str, reason: str) -> bool: self.logger.error("Message not found in DLQ", event_id=event_id) return False - if message.status in {DLQMessageStatus.DISCARDED, DLQMessageStatus.RETRIED}: + if message.status.is_terminal: self.logger.info("Skipping manual discard", event_id=event_id, status=message.status) return False diff --git a/backend/app/dlq/models.py b/backend/app/dlq/models.py index 9bc613f9..f8928f17 100644 --- a/backend/app/dlq/models.py +++ b/backend/app/dlq/models.py @@ -10,10 +10,25 @@ class DLQMessageStatus(StringEnum): """Status of a message in the Dead Letter Queue.""" + _terminal: bool + PENDING = "pending" SCHEDULED = "scheduled" - RETRIED = "retried" - DISCARDED = "discarded" + RETRIED = ("retried", True) + DISCARDED = ("discarded", True) + + def __new__(cls, value: str, terminal: bool = False) -> "DLQMessageStatus": + obj = str.__new__(cls, value) + obj._value_ = value + obj._terminal = terminal + return obj + + @property + def is_terminal(self) -> bool: + return self._terminal + + +DLQ_TERMINAL = frozenset(s for s in DLQMessageStatus if s.is_terminal) class RetryStrategy(StringEnum): diff --git a/backend/app/domain/enums/__init__.py b/backend/app/domain/enums/__init__.py index 64ef713b..d6882b84 100644 --- a/backend/app/domain/enums/__init__.py +++ b/backend/app/domain/enums/__init__.py @@ -1,16 +1,22 @@ from app.domain.enums.auth import LoginMethod, SettingsType from app.domain.enums.common import Environment, ErrorType, ExportFormat, SortOrder, Theme from app.domain.enums.events import EventType -from app.domain.enums.execution import CancelStatus, ExecutionStatus, QueuePriority +from app.domain.enums.execution import ( + EXECUTION_ACTIVE, + EXECUTION_TERMINAL, + CancelStatus, + ExecutionStatus, + QueuePriority, +) from app.domain.enums.notification import ( NotificationChannel, NotificationSeverity, NotificationStatus, ) -from app.domain.enums.replay import ReplayStatus, ReplayTarget, ReplayType -from app.domain.enums.saga import SagaState +from app.domain.enums.replay import REPLAY_TERMINAL, ReplayStatus, ReplayTarget, ReplayType +from app.domain.enums.saga import SAGA_ACTIVE, SAGA_TERMINAL, SagaState from app.domain.enums.sse import SSEControlEvent -from app.domain.enums.storage import ExecutionErrorType, StorageType +from app.domain.enums.storage import AllocationStatus, ExecutionErrorType, StorageType from app.domain.enums.user import UserRole __all__ = [ @@ -27,6 +33,8 @@ "EventType", # Execution "CancelStatus", + "EXECUTION_ACTIVE", + "EXECUTION_TERMINAL", "ExecutionStatus", "QueuePriority", # Notification @@ -34,14 +42,18 @@ "NotificationSeverity", "NotificationStatus", # Replay + "REPLAY_TERMINAL", "ReplayStatus", "ReplayTarget", "ReplayType", # Saga + "SAGA_ACTIVE", + "SAGA_TERMINAL", "SagaState", # SSE "SSEControlEvent", # Storage + "AllocationStatus", "ExecutionErrorType", "StorageType", # User diff --git a/backend/app/domain/enums/execution.py b/backend/app/domain/enums/execution.py index 333f4c61..7a09cffc 100644 --- a/backend/app/domain/enums/execution.py +++ b/backend/app/domain/enums/execution.py @@ -14,14 +14,30 @@ class QueuePriority(StringEnum): class ExecutionStatus(StringEnum): """Status of an execution.""" + _terminal: bool + QUEUED = "queued" SCHEDULED = "scheduled" RUNNING = "running" - COMPLETED = "completed" - FAILED = "failed" - TIMEOUT = "timeout" - CANCELLED = "cancelled" - ERROR = "error" + COMPLETED = ("completed", True) + FAILED = ("failed", True) + TIMEOUT = ("timeout", True) + CANCELLED = ("cancelled", True) + ERROR = ("error", True) + + def __new__(cls, value: str, terminal: bool = False) -> "ExecutionStatus": + obj = str.__new__(cls, value) + obj._value_ = value + obj._terminal = terminal + return obj + + @property + def is_terminal(self) -> bool: + return self._terminal + + +EXECUTION_TERMINAL = frozenset(s for s in ExecutionStatus if s.is_terminal) +EXECUTION_ACTIVE = frozenset(s for s in ExecutionStatus if not s.is_terminal) class CancelStatus(StringEnum): diff --git a/backend/app/domain/enums/replay.py b/backend/app/domain/enums/replay.py index 10cfc170..d27efd17 100644 --- a/backend/app/domain/enums/replay.py +++ b/backend/app/domain/enums/replay.py @@ -11,14 +11,29 @@ class ReplayType(StringEnum): class ReplayStatus(StringEnum): # Unified replay lifecycle across admin + services + _terminal: bool + PREVIEW = "preview" # Dry-run preview state SCHEDULED = "scheduled" CREATED = "created" RUNNING = "running" PAUSED = "paused" - COMPLETED = "completed" - FAILED = "failed" - CANCELLED = "cancelled" + COMPLETED = ("completed", True) + FAILED = ("failed", True) + CANCELLED = ("cancelled", True) + + def __new__(cls, value: str, terminal: bool = False) -> "ReplayStatus": + obj = str.__new__(cls, value) + obj._value_ = value + obj._terminal = terminal + return obj + + @property + def is_terminal(self) -> bool: + return self._terminal + + +REPLAY_TERMINAL = frozenset(s for s in ReplayStatus if s.is_terminal) class ReplayTarget(StringEnum): diff --git a/backend/app/domain/enums/saga.py b/backend/app/domain/enums/saga.py index 7c563663..41c217f4 100644 --- a/backend/app/domain/enums/saga.py +++ b/backend/app/domain/enums/saga.py @@ -4,10 +4,26 @@ class SagaState(StringEnum): """Saga execution states.""" + _terminal: bool + CREATED = "created" RUNNING = "running" COMPENSATING = "compensating" - COMPLETED = "completed" - FAILED = "failed" - TIMEOUT = "timeout" - CANCELLED = "cancelled" + COMPLETED = ("completed", True) + FAILED = ("failed", True) + TIMEOUT = ("timeout", True) + CANCELLED = ("cancelled", True) + + def __new__(cls, value: str, terminal: bool = False) -> "SagaState": + obj = str.__new__(cls, value) + obj._value_ = value + obj._terminal = terminal + return obj + + @property + def is_terminal(self) -> bool: + return self._terminal + + +SAGA_TERMINAL = frozenset(s for s in SagaState if s.is_terminal) +SAGA_ACTIVE = frozenset(s for s in SagaState if not s.is_terminal) diff --git a/backend/app/domain/enums/storage.py b/backend/app/domain/enums/storage.py index 9f234a38..47117189 100644 --- a/backend/app/domain/enums/storage.py +++ b/backend/app/domain/enums/storage.py @@ -13,6 +13,13 @@ class ExecutionErrorType(StringEnum): PERMISSION_DENIED = "permission_denied" +class AllocationStatus(StringEnum): + """Status of a resource allocation.""" + + ACTIVE = "active" + RELEASED = "released" + + class StorageType(StringEnum): """Types of storage backends.""" diff --git a/backend/app/events/core/producer.py b/backend/app/events/core/producer.py index d4d82919..f5fb3bed 100644 --- a/backend/app/events/core/producer.py +++ b/backend/app/events/core/producer.py @@ -26,7 +26,11 @@ def __init__( self._event_metrics = event_metrics async def produce(self, event_to_produce: DomainEvent, key: str) -> None: - """Persist event to MongoDB, then publish to Kafka.""" + """Persist event to MongoDB, then publish to Kafka. + + On Kafka publish failure, the event is marked as failed-to-publish + in MongoDB before the exception propagates. + """ await self._event_repository.store_event(event_to_produce) topic = event_to_produce.event_type try: @@ -37,9 +41,10 @@ async def produce(self, event_to_produce: DomainEvent, key: str) -> None: ) self._event_metrics.record_kafka_message_produced(topic) - self.logger.debug(f"Event {event_to_produce.event_type} sent to topic: {topic}") + self.logger.debug("Event sent to topic", event_type=event_to_produce.event_type, topic=topic) except Exception as e: self._event_metrics.record_kafka_production_error(topic=topic, error_type=type(e).__name__) - self.logger.error(f"Failed to produce message: {e}") + self.logger.error("Failed to produce message", topic=topic, error=str(e)) + await self._event_repository.mark_publish_failed(event_to_produce.event_id) raise diff --git a/backend/app/events/handlers.py b/backend/app/events/handlers.py index 93e4d6f4..b864f0cb 100644 --- a/backend/app/events/handlers.py +++ b/backend/app/events/handlers.py @@ -162,8 +162,6 @@ async def on_execution_timeout( def register_saga_subscriber(broker: KafkaBroker) -> None: - # No with_idempotency — the saga state machine provides its own - # deduplication via status checks before each transition. @broker.subscriber( EventType.EXECUTION_REQUESTED, group_id="saga-orchestrator", @@ -172,10 +170,14 @@ def register_saga_subscriber(broker: KafkaBroker) -> None: async def on_execution_requested( body: ExecutionRequestedEvent, orchestrator: FromDishka[SagaOrchestrator], + idem: FromDishka[IdempotencyManager], + logger: FromDishka[structlog.stdlib.BoundLogger], event_metrics: FromDishka[EventMetrics], ) -> None: - await _track_consumed(event_metrics, body, "saga-orchestrator", - orchestrator.handle_execution_requested(body)) + coro = with_idempotency( + body, orchestrator.handle_execution_requested, idem, KeyStrategy.EVENT_BASED, 3600, logger, + ) + await _track_consumed(event_metrics, body, "saga-orchestrator", coro) @broker.subscriber( EventType.EXECUTION_COMPLETED, diff --git a/backend/app/main.py b/backend/app/main.py index 40df51af..e6f2c810 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -39,6 +39,7 @@ MetricsMiddleware, RateLimitMiddleware, RequestSizeLimitMiddleware, + SecurityHeadersMiddleware, setup_metrics, ) from app.settings import Settings @@ -58,13 +59,16 @@ def create_app(settings: Settings | None = None) -> FastAPI: settings = settings or Settings() logger = setup_logger(settings.LOG_LEVEL) - # Disable OpenAPI/Docs in production for security; health endpoints provide readiness + openapi_url = "/openapi.json" if settings.DEVELOPMENT_MODE else None + docs_url = "/docs" if settings.DEVELOPMENT_MODE else None + redoc_url = "/redoc" if settings.DEVELOPMENT_MODE else None + app = FastAPI( title=settings.PROJECT_NAME, lifespan=lifespan, - openapi_url=None, - docs_url=None, - redoc_url=None, + openapi_url=openapi_url, + docs_url=docs_url, + redoc_url=redoc_url, ) # Store settings on app state for lifespan access @@ -81,13 +85,14 @@ def create_app(settings: Settings | None = None) -> FastAPI: app.add_middleware(RateLimitMiddleware, settings=settings) app.add_middleware(CSRFMiddleware) app.add_middleware(RequestSizeLimitMiddleware) + app.add_middleware(SecurityHeadersMiddleware) app.add_middleware(CacheControlMiddleware) app.add_middleware( CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_credentials=True, - allow_methods=["GET", "POST", "PUT", "DELETE"], + allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"], allow_headers=[ "Authorization", "Content-Type", @@ -96,7 +101,14 @@ def create_app(settings: Settings | None = None) -> FastAPI: "X-Requested-With", "X-CSRF-Token", ], - expose_headers=["Content-Length", "Content-Range"], + expose_headers=[ + "Content-Length", + "Content-Range", + "X-RateLimit-Limit", + "X-RateLimit-Remaining", + "X-RateLimit-Reset", + "Retry-After", + ], ) logger.info("CORS middleware configured", origins=settings.CORS_ORIGINS) diff --git a/backend/app/services/auth_service.py b/backend/app/services/auth_service.py index 713c72cd..0d1dbe62 100644 --- a/backend/app/services/auth_service.py +++ b/backend/app/services/auth_service.py @@ -76,6 +76,9 @@ async def get_current_user(self, request: Request) -> User: if user is None: raise InvalidCredentialsError() + if not user.is_active: + raise InvalidCredentialsError() + return user async def get_admin(self, request: Request) -> User: @@ -84,7 +87,7 @@ async def get_admin(self, request: Request) -> User: "/admin", request.method, user.role == UserRole.ADMIN, user_role=user.role, ) if user.role != UserRole.ADMIN: - self.logger.warning(f"Admin access denied for user: {user.username} (role: {user.role})") + self.logger.warning("Admin access denied", username=user.username, role=user.role) raise AdminAccessRequiredError(user.username) return user @@ -97,7 +100,8 @@ async def _fail_login( user_id: str = "", ) -> NoReturn: self.logger.warning( - f"Login failed - {reason}", + "Login failed", + reason=reason, username=username, client_ip=ip_address, user_agent=user_agent, @@ -137,6 +141,9 @@ async def login( if not self.security_service.verify_password(password, user.hashed_password): await self._fail_login(username, "invalid_password", ip_address, user_agent, user_id=user.user_id) + if not user.is_active: + await self._fail_login(username, "account_deactivated", ip_address, user_agent, user_id=user.user_id) + await self._lockout.clear_attempts(username) effective = await self._runtime_settings.get_effective_settings() @@ -199,6 +206,16 @@ async def register( ) raise ConflictError("Username already registered") + existing_email = await self.user_repo.get_user_by_email(email) + if existing_email: + self.logger.warning( + "Registration failed - email taken", + username=username, + client_ip=ip_address, + user_agent=user_agent, + ) + raise ConflictError("Email already registered") + hashed_password = self.security_service.get_password_hash(password) create_data = DomainUserCreate( username=username, @@ -232,7 +249,7 @@ async def register( async def publish_logout_event(self, token: str | None) -> None: if not token: return - username = self.security_service.decode_token(token) + username = self.security_service.decode_token(token, allow_expired=True) user = await self.user_repo.get_user(username) if not user: return diff --git a/backend/app/services/event_replay/replay_service.py b/backend/app/services/event_replay/replay_service.py index 0cdf7344..34508f1c 100644 --- a/backend/app/services/event_replay/replay_service.py +++ b/backend/app/services/event_replay/replay_service.py @@ -171,10 +171,9 @@ async def cleanup_old_sessions(self, older_than_hours: int = 24) -> CleanupResul cutoff_time = datetime.now(timezone.utc) - timedelta(hours=older_than_hours) removed_memory = 0 - completed_statuses = {ReplayStatus.COMPLETED, ReplayStatus.FAILED, ReplayStatus.CANCELLED} for session_id in list(self._sessions.keys()): session = self._sessions[session_id] - if session.status in completed_statuses and session.created_at < cutoff_time: + if session.status.is_terminal and session.created_at < cutoff_time: del self._sessions[session_id] removed_memory += 1 diff --git a/backend/app/services/execution_queue.py b/backend/app/services/execution_queue.py index 3d51c1d2..d103aa22 100644 --- a/backend/app/services/execution_queue.py +++ b/backend/app/services/execution_queue.py @@ -35,7 +35,6 @@ def _pending_key(priority: QueuePriority) -> str: _UPDATE_PRIORITY_LUA = """ local new_key = KEYS[1] local exec_id = ARGV[1] -local new_score = tonumber(ARGV[2]) for i = 2, #KEYS do local score = redis.call('ZSCORE', KEYS[i], exec_id) @@ -132,8 +131,12 @@ async def try_schedule(self, max_active: int) -> tuple[str, ExecutionRequestedEv event_json = await self._redis.get(_event_key(execution_id)) if event_json is None: - self._logger.warning("Event data missing for scheduled execution", execution_id=execution_id) + self._logger.error( + "Event data expired/missing for scheduled execution — execution lost", + execution_id=execution_id, + ) await self._redis.srem(_ACTIVE_KEY, execution_id) # type: ignore[misc] + self._metrics.record_event_data_lost() return None event_str = event_json if isinstance(event_json, str) else event_json.decode() @@ -159,7 +162,7 @@ async def update_priority(self, execution_id: str, new_priority: QueuePriority) script = await self._get_update_priority_script() result = await script( keys=[_pending_key(new_priority), *_PENDING_KEYS], - args=[execution_id, time.time()], + args=[execution_id], ) if not result: return False diff --git a/backend/app/services/execution_service.py b/backend/app/services/execution_service.py index 04410593..4b8178b2 100644 --- a/backend/app/services/execution_service.py +++ b/backend/app/services/execution_service.py @@ -213,14 +213,7 @@ async def cancel_execution( Raises: ExecutionTerminalError: If execution is in a terminal state. """ - terminal_states = { - ExecutionStatus.COMPLETED, - ExecutionStatus.FAILED, - ExecutionStatus.TIMEOUT, - ExecutionStatus.ERROR, - } - - if current_status in terminal_states: + if current_status.is_terminal: raise ExecutionTerminalError(execution_id, current_status) if current_status == ExecutionStatus.CANCELLED: @@ -521,12 +514,10 @@ async def delete_execution(self, execution_id: str, user_id: str) -> bool: return True async def _publish_deletion_event(self, execution_id: str, user_id: str) -> None: - """ - Publish execution deletion/cancellation event. + """Publish cancellation event for a deleted execution. - Args: - execution_id: UUID of deleted execution. - user_id: ID of user who deleted it. + Uses ExecutionCancelledEvent because no dedicated deletion event type + exists yet — the saga orchestrator treats both the same way. """ metadata = self._create_event_metadata(user_id=user_id) @@ -540,7 +531,7 @@ async def _publish_deletion_event(self, execution_id: str, user_id: str) -> None await self.producer.produce(event_to_produce=event, key=execution_id) self.logger.info( - "Published cancellation event", + "Published deletion event (as cancellation)", execution_id=execution_id, event_id=event.event_id, ) @@ -559,14 +550,7 @@ async def get_execution_stats( Dictionary containing execution statistics. """ query = self._build_stats_query(user_id, time_range) - - # Get executions for stats - executions = await self.execution_repo.get_executions( - query=query, - limit=1000, # Reasonable limit for stats - ) - - return self._calculate_stats(executions) + return await self.execution_repo.aggregate_stats(query) def _build_stats_query( self, user_id: str | None, time_range: tuple[datetime | None, datetime | None] @@ -597,47 +581,3 @@ def _build_stats_query( return query - def _calculate_stats(self, executions: list[DomainExecution]) -> dict[str, Any]: - """ - Calculate statistics from executions. - - Args: - executions: List of executions to analyze. - - Returns: - Statistics dictionary. - """ - stats: dict[str, Any] = { - "total": len(executions), - "by_status": {}, - "by_language": {}, - "average_duration_ms": 0, - "success_rate": 0, - } - - total_duration = 0.0 - successful = 0 - - for execution in executions: - # Count by status - status = execution.status - stats["by_status"][status] = stats["by_status"].get(status, 0) + 1 - - # Count by language - lang_key = f"{execution.lang}-{execution.lang_version}" - stats["by_language"][lang_key] = stats["by_language"].get(lang_key, 0) + 1 - - # Track success and duration - if status == ExecutionStatus.COMPLETED: - successful += 1 - if execution.created_at and execution.updated_at: - duration = (execution.updated_at - execution.created_at).total_seconds() * 1000 - total_duration += duration - - # Calculate averages - if stats["total"] > 0: - stats["success_rate"] = successful / stats["total"] - if successful > 0: - stats["average_duration_ms"] = total_duration / successful - - return stats diff --git a/backend/app/services/notification_service.py b/backend/app/services/notification_service.py index 518cb2b0..cd9063cb 100644 --- a/backend/app/services/notification_service.py +++ b/backend/app/services/notification_service.py @@ -111,6 +111,7 @@ def __init__( self.settings = settings self.sse_bus = sse_bus self.logger = logger + self._http_client = httpx.AsyncClient(timeout=30.0) self._throttle_cache = ThrottleCache() @@ -363,15 +364,14 @@ async def _send_webhook( "notification.channel": "webhook", "notification.webhook_host": safe_host, }) - async with httpx.AsyncClient() as client: - response = await client.post(webhook_url, json=payload, headers=headers, timeout=30.0) - response.raise_for_status() - self.logger.debug( - "Webhook delivered successfully", - notification_id=str(notification.notification_id), - status_code=response.status_code, - response_time_ms=int(response.elapsed.total_seconds() * 1000), - ) + response = await self._http_client.post(webhook_url, json=payload, headers=headers) + response.raise_for_status() + self.logger.debug( + "Webhook delivered successfully", + notification_id=str(notification.notification_id), + status_code=response.status_code, + response_time_ms=int(response.elapsed.total_seconds() * 1000), + ) async def _send_slack(self, notification: DomainNotification, subscription: DomainNotificationSubscription) -> None: """Send Slack notification.""" @@ -411,14 +411,13 @@ async def _send_slack(self, notification: DomainNotification, subscription: Doma "notification.id": str(notification.notification_id), "notification.channel": "slack", }) - async with httpx.AsyncClient() as client: - response = await client.post(subscription.slack_webhook, json=slack_message, timeout=30.0) - response.raise_for_status() - self.logger.debug( - "Slack notification delivered successfully", - notification_id=str(notification.notification_id), - status_code=response.status_code, - ) + response = await self._http_client.post(subscription.slack_webhook, json=slack_message) + response.raise_for_status() + self.logger.debug( + "Slack notification delivered successfully", + notification_id=str(notification.notification_id), + status_code=response.status_code, + ) def _get_slack_color(self, priority: NotificationSeverity) -> str: """Get Slack color based on severity.""" diff --git a/backend/app/services/pod_monitor/event_mapper.py b/backend/app/services/pod_monitor/event_mapper.py index 34a53c39..2983fe0d 100644 --- a/backend/app/services/pod_monitor/event_mapper.py +++ b/backend/app/services/pod_monitor/event_mapper.py @@ -1,4 +1,4 @@ -import ast +import json from collections.abc import Awaitable, Callable from dataclasses import dataclass from uuid import uuid4 @@ -59,6 +59,8 @@ class PodLogs: class PodEventMapper: """Maps Kubernetes pod objects to application events""" + _MAX_CACHE_SIZE = 10_000 + def __init__(self, logger: structlog.stdlib.BoundLogger, k8s_api: k8s_client.CoreV1Api | None = None) -> None: self.logger = logger self._event_cache: dict[str, PodPhase] = {} @@ -193,6 +195,11 @@ def _is_duplicate(self, pod_name: str, phase: PodPhase) -> bool: """Check if this is a duplicate event""" if self._event_cache.get(pod_name) == phase: return True + if len(self._event_cache) >= self._MAX_CACHE_SIZE: + # Evict oldest half to amortise cleanup cost + keys = list(self._event_cache)[:self._MAX_CACHE_SIZE // 2] + for k in keys: + del self._event_cache[k] self._event_cache[pod_name] = phase return False @@ -500,7 +507,7 @@ def _try_parse_json(self, text: str) -> PodLogs | None: if not (text.startswith("{") and text.endswith("}")): return None - data = ast.literal_eval(text) + data = json.loads(text) return PodLogs( stdout=data.get("stdout", ""), stderr=data.get("stderr", ""), diff --git a/backend/app/services/rate_limit_service.py b/backend/app/services/rate_limit_service.py index 0fddf07d..897fe7c7 100644 --- a/backend/app/services/rate_limit_service.py +++ b/backend/app/services/rate_limit_service.py @@ -164,6 +164,35 @@ async def _check_sliding_window( algorithm=RateLimitAlgorithm.SLIDING_WINDOW, ) + _TOKEN_BUCKET_LUA = """ +local key = KEYS[1] +local max_tokens = tonumber(ARGV[1]) +local refill_rate = tonumber(ARGV[2]) +local now = tonumber(ARGV[3]) +local ttl = tonumber(ARGV[4]) + +local tokens = max_tokens +local last_refill = now + +local bucket_data = redis.call('GET', key) +if bucket_data then + local bucket = cjson.decode(bucket_data) + tokens = bucket['tokens'] + last_refill = bucket['last_refill'] + local time_passed = now - last_refill + tokens = math.min(max_tokens, tokens + time_passed * refill_rate) +end + +local allowed = 0 +if tokens >= 1 then + tokens = tokens - 1 + allowed = 1 +end + +redis.call('SETEX', key, ttl, cjson.encode({tokens=tokens, last_refill=now})) +return {allowed, tostring(tokens)} +""" + async def _check_token_bucket( self, user_id: str, endpoint: str, limit: int, window_seconds: int, burst_multiplier: float, rule: RateLimitRule ) -> RateLimitStatus: @@ -175,22 +204,14 @@ async def _check_token_bucket( await self._register_user_key(user_id, key) # --8<-- [start:check_token_bucket] - bucket_data = await self.redis.get(key) - if bucket_data: - bucket = json.loads(bucket_data) - tokens = bucket["tokens"] - last_refill = bucket["last_refill"] - time_passed = now - last_refill - tokens_to_add = time_passed * refill_rate - tokens = min(max_tokens, tokens + tokens_to_add) - else: - tokens = max_tokens - - if tokens >= 1: - tokens -= 1 - await self.redis.setex(key, window_seconds * 2, json.dumps({"tokens": tokens, "last_refill": now})) + result = await self.redis.eval( # type: ignore[misc] + self._TOKEN_BUCKET_LUA, 1, key, max_tokens, refill_rate, now, window_seconds * 2, + ) + allowed = bool(result[0]) + tokens = float(result[1]) # --8<-- [end:check_token_bucket] + if allowed: return RateLimitStatus( allowed=True, limit=limit, diff --git a/backend/app/services/saga/execution_saga.py b/backend/app/services/saga/execution_saga.py index 1401fb79..dc59b8df 100644 --- a/backend/app/services/saga/execution_saga.py +++ b/backend/app/services/saga/execution_saga.py @@ -10,17 +10,16 @@ from .saga_step import CompensationStep, SagaContext, SagaStep -logger = structlog.get_logger(__name__) - class ValidateExecutionStep(SagaStep[ExecutionRequestedEvent]): """Validate execution request.""" - def __init__(self) -> None: + def __init__(self, logger: structlog.stdlib.BoundLogger) -> None: super().__init__("validate_execution") + self.logger = logger async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: - logger.info(f"Validating execution {event.execution_id}") + self.logger.info("Validating execution", execution_id=event.execution_id) context.set("execution_id", event.execution_id) context.set("language", event.language) @@ -43,13 +42,14 @@ def get_compensation(self) -> CompensationStep | None: class AllocateResourcesStep(SagaStep[ExecutionRequestedEvent]): """Allocate resources for execution.""" - def __init__(self, alloc_repo: ResourceAllocationRepository) -> None: + def __init__(self, alloc_repo: ResourceAllocationRepository, logger: structlog.stdlib.BoundLogger) -> None: super().__init__("allocate_resources") self.alloc_repo = alloc_repo + self.logger = logger async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: execution_id = context.get("execution_id") - logger.info(f"Allocating resources for execution {execution_id}") + self.logger.info("Allocating resources for execution", execution_id=execution_id) active_count = await self.alloc_repo.count_active(event.language) if active_count >= 100: @@ -72,29 +72,31 @@ async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> return True def get_compensation(self) -> CompensationStep | None: - return ReleaseResourcesCompensation(alloc_repo=self.alloc_repo) + return ReleaseResourcesCompensation(alloc_repo=self.alloc_repo, logger=self.logger) class CreatePodStep(SagaStep[ExecutionRequestedEvent]): """Create Kubernetes pod.""" - def __init__(self, producer: UnifiedProducer, publish_commands: bool) -> None: + def __init__(self, producer: UnifiedProducer, publish_commands: bool, logger: structlog.stdlib.BoundLogger) -> None: super().__init__("create_pod") self.producer = producer self.publish_commands = publish_commands + self.logger = logger async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: execution_id = context.get("execution_id") if not self.publish_commands: - logger.info( - f"Skipping CreatePodCommandEvent publish for execution {execution_id} " - f"because publish_commands flag is disabled" + self.logger.info( + "Skipping CreatePodCommandEvent publish", + execution_id=execution_id, + reason="publish_commands disabled", ) context.set("pod_creation_triggered", False) return True - logger.info(f"Publishing CreatePodCommandEvent for execution {execution_id}") + self.logger.info("Publishing CreatePodCommandEvent", execution_id=execution_id) create_pod_cmd = CreatePodCommandEvent( saga_id=context.saga_id, @@ -121,30 +123,28 @@ async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> await self.producer.produce(event_to_produce=create_pod_cmd, key=execution_id) context.set("pod_creation_triggered", True) - logger.info(f"CreatePodCommandEvent published for execution {execution_id}") + self.logger.info("CreatePodCommandEvent published", execution_id=execution_id) return True def get_compensation(self) -> CompensationStep | None: - return DeletePodCompensation(producer=self.producer) - - -# Compensation Steps + return DeletePodCompensation(producer=self.producer, logger=self.logger) class ReleaseResourcesCompensation(CompensationStep): """Release allocated resources.""" - def __init__(self, alloc_repo: ResourceAllocationRepository) -> None: + def __init__(self, alloc_repo: ResourceAllocationRepository, logger: structlog.stdlib.BoundLogger) -> None: super().__init__("release_resources") self.alloc_repo = alloc_repo + self.logger = logger async def compensate(self, context: SagaContext) -> bool: allocation_id = context.get("allocation_id") if not allocation_id: return True - logger.info(f"Releasing resources for allocation {allocation_id}") + self.logger.info("Releasing resources", allocation_id=allocation_id) await self.alloc_repo.release_allocation(allocation_id) return True @@ -153,16 +153,17 @@ async def compensate(self, context: SagaContext) -> bool: class DeletePodCompensation(CompensationStep): """Delete created pod.""" - def __init__(self, producer: UnifiedProducer) -> None: + def __init__(self, producer: UnifiedProducer, logger: structlog.stdlib.BoundLogger) -> None: super().__init__("delete_pod") self.producer = producer + self.logger = logger async def compensate(self, context: SagaContext) -> bool: execution_id = context.get("execution_id") if not execution_id or not context.get("pod_creation_triggered"): return True - logger.info(f"Publishing DeletePodCommandEvent for execution {execution_id}") + self.logger.info("Publishing DeletePodCommandEvent", execution_id=execution_id) delete_pod_cmd = DeletePodCommandEvent( saga_id=context.saga_id, @@ -177,7 +178,7 @@ async def compensate(self, context: SagaContext) -> bool: await self.producer.produce(event_to_produce=delete_pod_cmd, key=execution_id) - logger.info(f"DeletePodCommandEvent published for {execution_id}") + self.logger.info("DeletePodCommandEvent published", execution_id=execution_id) return True @@ -193,14 +194,16 @@ def bind_dependencies( producer: UnifiedProducer, alloc_repo: ResourceAllocationRepository, publish_commands: bool, + logger: structlog.stdlib.BoundLogger, ) -> None: self._producer = producer self._alloc_repo = alloc_repo self._publish_commands = publish_commands + self._logger = logger def get_steps(self) -> list[SagaStep[Any]]: return [ - ValidateExecutionStep(), - AllocateResourcesStep(alloc_repo=self._alloc_repo), - CreatePodStep(producer=self._producer, publish_commands=self._publish_commands), + ValidateExecutionStep(logger=self._logger), + AllocateResourcesStep(alloc_repo=self._alloc_repo, logger=self._logger), + CreatePodStep(producer=self._producer, publish_commands=self._publish_commands, logger=self._logger), ] diff --git a/backend/app/services/saga/saga_orchestrator.py b/backend/app/services/saga/saga_orchestrator.py index b208d9b4..a79af3f0 100644 --- a/backend/app/services/saga/saga_orchestrator.py +++ b/backend/app/services/saga/saga_orchestrator.py @@ -106,7 +106,7 @@ async def _resolve_completion( saga = await self._repo.get_saga_by_execution_and_name(execution_id, _SAGA_NAME) if not saga: self.logger.debug("No execution_saga found for execution", execution_id=execution_id) - elif saga.state not in (SagaState.RUNNING, SagaState.CREATED): + elif saga.state.is_terminal: self.logger.debug("Saga already in terminal state", saga_id=saga.saga_id, state=saga.state) else: self.logger.info("Marking saga terminal state", saga_id=saga.saga_id, state=state) @@ -117,6 +117,8 @@ async def _resolve_completion( await self._queue.release(execution_id) await self.try_schedule_from_queue() + _MAX_SAGA_START_RETRIES = 3 + async def try_schedule_from_queue(self) -> None: """Try to schedule pending executions from the queue.""" settings = await self._runtime_settings.get_effective_settings() @@ -128,11 +130,26 @@ async def try_schedule_from_queue(self) -> None: try: await self._start_saga(event) except Exception: + retry_count = getattr(event, "_retry_count", 0) + 1 self.logger.error( - "Failed to start saga, re-enqueueing execution", execution_id=execution_id, exc_info=True, + "Failed to start saga", + execution_id=execution_id, + retry_count=retry_count, + exc_info=True, ) await self._queue.release(execution_id) - await self._queue.enqueue(event) + if retry_count >= self._MAX_SAGA_START_RETRIES: + self.logger.error( + "Max saga start retries exceeded, dropping execution", + execution_id=execution_id, + ) + await self._resolve_completion( + execution_id, SagaState.FAILED, + f"Failed to start saga after {retry_count} attempts", + ) + else: + event._retry_count = retry_count # type: ignore[attr-defined] + await self._queue.enqueue(event) break async def _start_saga(self, trigger_event: ExecutionRequestedEvent) -> str: @@ -172,6 +189,7 @@ def _create_saga_instance(self) -> ExecutionSaga: producer=self._producer, alloc_repo=self._alloc_repo, publish_commands=self.config.publish_commands, + logger=self.logger, ) return saga @@ -190,7 +208,7 @@ async def _execute_saga( for step in steps: saved = await self._repo.save_saga(instance.saga_id, current_step=step.name) - if saved.state not in (SagaState.RUNNING, SagaState.CREATED): + if saved.state.is_terminal: self.logger.info( "Saga no longer active, stopping", saga_id=instance.saga_id, state=saved.state ) @@ -259,6 +277,7 @@ async def _compensate_saga(self, saga_id: str, context: SagaContext) -> None: await self._repo.save_saga(saga_id, state=SagaState.COMPENSATING) compensated: list[str] = list(saga.compensated_steps) + failed_compensations: list[str] = [] for compensation in reversed(context.compensations): try: self.logger.info("Executing compensation", compensation=compensation.name, saga_id=saga_id) @@ -268,13 +287,22 @@ async def _compensate_saga(self, saga_id: str, context: SagaContext) -> None: if success: compensated.append(compensation.name) else: + failed_compensations.append(compensation.name) self.logger.error("Compensation failed", compensation=compensation.name, saga_id=saga_id) except Exception: + failed_compensations.append(compensation.name) self.logger.error( "Error in compensation", compensation=compensation.name, saga_id=saga_id, exc_info=True ) + if failed_compensations: + self.logger.error( + "Partial compensation — some steps could not be undone", + saga_id=saga_id, + failed_compensations=failed_compensations, + ) + if was_cancelled: await self._repo.save_saga(saga_id, compensated_steps=compensated) self.logger.info("Saga compensation completed after cancellation", saga_id=saga_id) diff --git a/backend/app/services/sse/sse_service.py b/backend/app/services/sse/sse_service.py index 0058d597..edb7a9ca 100644 --- a/backend/app/services/sse/sse_service.py +++ b/backend/app/services/sse/sse_service.py @@ -6,7 +6,7 @@ from pydantic import TypeAdapter from app.db.repositories import ExecutionRepository -from app.domain.enums import EventType, ReplayStatus, SSEControlEvent, UserRole +from app.domain.enums import EventType, SSEControlEvent, UserRole from app.domain.exceptions import ForbiddenError from app.domain.execution import ExecutionNotFoundError from app.domain.execution.models import DomainExecution @@ -24,12 +24,6 @@ EventType.RESULT_FAILED, }) -_TERMINAL_REPLAY_STATUSES: frozenset[ReplayStatus] = frozenset({ - ReplayStatus.COMPLETED, - ReplayStatus.FAILED, - ReplayStatus.CANCELLED, -}) - class SSEService: """SSE service — transforms bus events and DB state into SSE wire format.""" @@ -97,10 +91,10 @@ async def _replay_pipeline( ) -> AsyncGenerator[dict[str, Any], None]: session_id = initial_status.session_id yield {"data": _replay_adapter.dump_json(initial_status).decode()} - if initial_status.status in _TERMINAL_REPLAY_STATUSES: + if initial_status.status.is_terminal: return async for status in self._bus.listen_replay(session_id): self._logger.info("SSE replay event", session_id=session_id, status=status.status) yield {"data": _replay_adapter.dump_json(status).decode()} - if status.status in _TERMINAL_REPLAY_STATUSES: + if status.status.is_terminal: return diff --git a/backend/app/settings.py b/backend/app/settings.py index 70425bd1..61dbcaca 100644 --- a/backend/app/settings.py +++ b/backend/app/settings.py @@ -149,7 +149,7 @@ def __init__( # Service metadata SERVICE_NAME: str = "integr8scode-backend" SERVICE_VERSION: str = "1.0.0" - ENVIRONMENT: str = "production" # deployment environment (production, staging, development) + ENVIRONMENT: str = "development" # deployment environment (production, staging, development) HOSTNAME: str = "unknown" # container hostname, set via TOML or override # OpenTelemetry metrics export endpoint @@ -178,4 +178,4 @@ def __init__( ]) # Logging configuration - LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "DEBUG" + LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO" diff --git a/backend/tests/unit/services/saga/test_execution_saga_steps.py b/backend/tests/unit/services/saga/test_execution_saga_steps.py index d690484c..b56b6919 100644 --- a/backend/tests/unit/services/saga/test_execution_saga_steps.py +++ b/backend/tests/unit/services/saga/test_execution_saga_steps.py @@ -1,3 +1,4 @@ +import structlog import pytest from app.db.repositories import ResourceAllocationRepository from app.domain.events import DomainEvent, ExecutionRequestedEvent @@ -17,6 +18,8 @@ pytestmark = pytest.mark.unit +_logger = structlog.get_logger() + def _req(timeout: int = 30, script: str = "print('x')") -> ExecutionRequestedEvent: return make_execution_requested_event(execution_id="e1", script=script, timeout_seconds=timeout) @@ -25,19 +28,19 @@ def _req(timeout: int = 30, script: str = "print('x')") -> ExecutionRequestedEve @pytest.mark.asyncio async def test_validate_execution_step_success_and_failures() -> None: ctx = SagaContext("s1", "e1") - ok = await ValidateExecutionStep().execute(ctx, _req()) + ok = await ValidateExecutionStep(logger=_logger).execute(ctx, _req()) assert ok is True and ctx.get("execution_id") == "e1" # Timeout too large → raises ctx2 = SagaContext("s1", "e1") with pytest.raises(ValueError, match="Timeout exceeds maximum"): - await ValidateExecutionStep().execute(ctx2, _req(timeout=301)) + await ValidateExecutionStep(logger=_logger).execute(ctx2, _req(timeout=301)) # Script too big → raises ctx3 = SagaContext("s1", "e1") big = "x" * (1024 * 1024 + 1) with pytest.raises(ValueError, match="Script size exceeds limit"): - await ValidateExecutionStep().execute(ctx3, _req(script=big)) + await ValidateExecutionStep(logger=_logger).execute(ctx3, _req(script=big)) class _FakeAllocRepo(ResourceAllocationRepository): @@ -71,14 +74,16 @@ async def release_allocation(self, allocation_id: str) -> bool: async def test_allocate_resources_step_paths() -> None: ctx = SagaContext("s1", "e1") ctx.set("execution_id", "e1") - ok = await AllocateResourcesStep(alloc_repo=_FakeAllocRepo(active=0, alloc_id="alloc-1")).execute(ctx, _req()) + ok = await AllocateResourcesStep( + alloc_repo=_FakeAllocRepo(active=0, alloc_id="alloc-1"), logger=_logger, + ).execute(ctx, _req()) assert ok is True and ctx.get("resources_allocated") is True and ctx.get("allocation_id") == "alloc-1" # Limit exceeded → raises ctx2 = SagaContext("s2", "e2") ctx2.set("execution_id", "e2") with pytest.raises(ValueError, match="Resource limit exceeded"): - await AllocateResourcesStep(alloc_repo=_FakeAllocRepo(active=100)).execute(ctx2, _req()) + await AllocateResourcesStep(alloc_repo=_FakeAllocRepo(active=100), logger=_logger).execute(ctx2, _req()) class _FakeProducer(UnifiedProducer): @@ -99,19 +104,19 @@ async def test_create_pod_step_publish_flag_and_compensation() -> None: # Skip publish path ctx = SagaContext("s1", "e1") ctx.set("execution_id", "e1") - s1 = CreatePodStep(producer=prod, publish_commands=False) + s1 = CreatePodStep(producer=prod, publish_commands=False, logger=_logger) ok1 = await s1.execute(ctx, _req()) assert ok1 is True and ctx.get("pod_creation_triggered") is False # Publish path succeeds ctx2 = SagaContext("s2", "e2") ctx2.set("execution_id", "e2") - s2 = CreatePodStep(producer=prod, publish_commands=True) + s2 = CreatePodStep(producer=prod, publish_commands=True, logger=_logger) ok2 = await s2.execute(ctx2, _req()) assert ok2 is True and ctx2.get("pod_creation_triggered") is True and prod.events # DeletePod compensation triggers only when flagged and producer exists - comp = DeletePodCompensation(producer=prod) + comp = DeletePodCompensation(producer=prod, logger=_logger) ctx2.set("pod_creation_triggered", True) assert await comp.compensate(ctx2) is True @@ -119,14 +124,14 @@ async def test_create_pod_step_publish_flag_and_compensation() -> None: @pytest.mark.asyncio async def test_release_resources_compensation() -> None: repo = _FakeAllocRepo() - comp = ReleaseResourcesCompensation(alloc_repo=repo) + comp = ReleaseResourcesCompensation(alloc_repo=repo, logger=_logger) ctx = SagaContext("s1", "e1") ctx.set("allocation_id", "alloc-1") assert await comp.compensate(ctx) is True and repo.released == ["alloc-1"] # Missing allocation_id -> True short-circuit ctx2 = SagaContext("sX", "eX") - assert await ReleaseResourcesCompensation(alloc_repo=repo).compensate(ctx2) is True + assert await ReleaseResourcesCompensation(alloc_repo=repo, logger=_logger).compensate(ctx2) is True @pytest.mark.asyncio @@ -136,19 +141,23 @@ async def test_delete_pod_compensation_variants() -> None: # Not triggered -> True early ctx = SagaContext("s", "e") ctx.set("pod_creation_triggered", False) - assert await DeletePodCompensation(producer=prod).compensate(ctx) is True + assert await DeletePodCompensation(producer=prod, logger=_logger).compensate(ctx) is True # Triggered -> publishes delete command ctx2 = SagaContext("s2", "e2") ctx2.set("pod_creation_triggered", True) ctx2.set("execution_id", "e2") - assert await DeletePodCompensation(producer=prod).compensate(ctx2) is True + assert await DeletePodCompensation(producer=prod, logger=_logger).compensate(ctx2) is True assert len(prod.events) == 1 # get_compensation return types - assert ValidateExecutionStep().get_compensation() is None - assert isinstance(AllocateResourcesStep(_FakeAllocRepo()).get_compensation(), ReleaseResourcesCompensation) - assert isinstance(CreatePodStep(prod, publish_commands=False).get_compensation(), DeletePodCompensation) + assert ValidateExecutionStep(logger=_logger).get_compensation() is None + assert isinstance( + AllocateResourcesStep(_FakeAllocRepo(), logger=_logger).get_compensation(), ReleaseResourcesCompensation, + ) + assert isinstance( + CreatePodStep(prod, publish_commands=False, logger=_logger).get_compensation(), DeletePodCompensation, + ) def test_execution_saga_bind_and_get_steps_sets_flags_and_types() -> None: @@ -161,7 +170,7 @@ def __init__(self) -> None: pass s = ExecutionSaga() - s.bind_dependencies(producer=DummyProd(), alloc_repo=DummyAlloc(), publish_commands=True) + s.bind_dependencies(producer=DummyProd(), alloc_repo=DummyAlloc(), publish_commands=True, logger=_logger) steps = s.get_steps() assert len(steps) == 3 cps = [st for st in steps if isinstance(st, CreatePodStep)][0] diff --git a/docs/architecture/domain-exceptions.md b/docs/architecture/domain-exceptions.md index e82a821c..2abe1acb 100644 --- a/docs/architecture/domain-exceptions.md +++ b/docs/architecture/domain-exceptions.md @@ -30,6 +30,7 @@ HTTP status codes: | `ForbiddenError` | 403 | Authenticated but not allowed | | `InvalidStateError` | 400 | Operation invalid for current state | | `InfrastructureError` | 500 | External system failure | +| `AccountLockedError` | 423 | Account temporarily locked | Each domain module defines specific exceptions that inherit from these bases. The hierarchy looks like this: diff --git a/docs/architecture/middleware.md b/docs/architecture/middleware.md index 785f2417..7ecc9bae 100644 --- a/docs/architecture/middleware.md +++ b/docs/architecture/middleware.md @@ -58,7 +58,7 @@ Adds appropriate `Cache-Control` headers to GET responses based on endpoint patt |-----------------------------|-------------------|------------| | `/api/v1/k8s-limits` | public | 5 minutes | | `/api/v1/example-scripts` | public | 10 minutes | -| `/api/v1/auth/verify-token` | private, no-cache | - | +| `/api/v1/auth/me` | private, no-cache | - | | `/api/v1/notifications` | private, no-cache | - | Public endpoints also get a `Vary: Accept-Encoding` header for proper proxy caching. Cache headers are only added to diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 699138b7..31eb5d83 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -1,6 +1,6 @@ # Architecture overview -Integr8sCode lets users submit Python scripts through a Svelte SPA. The FastAPI backend validates requests, publishes events to Kafka, and streams results back via SSE. Kafka workers coordinate isolated Kubernetes pods that run the code, collect results into MongoDB, and push them through Redis pub/sub so any API instance can deliver them over SSE. +Integr8sCode lets users submit code in multiple languages (Python, JavaScript, TypeScript, Ruby, Go) through a Svelte SPA. The FastAPI backend validates requests, publishes events to Kafka, and streams results back via SSE. Kafka workers coordinate isolated Kubernetes pods that run the code, collect results into MongoDB, and push them through Redis pub/sub so any API instance can deliver them over SSE.
diff --git a/docs/components/saved-scripts.md b/docs/components/saved-scripts.md index 62d7cd97..7f74b22c 100644 --- a/docs/components/saved-scripts.md +++ b/docs/components/saved-scripts.md @@ -28,7 +28,7 @@ All operations log the user ID, script ID, and relevant metadata for auditing. ## Storage -Scripts are stored in the `saved_scripts` MongoDB collection with a compound index on `(user_id, script_id)` for efficient per-user queries. +Scripts are stored in the `saved_scripts` MongoDB collection with individual indexes on `script_id` (unique) and `user_id` for efficient per-user queries. The repository enforces user isolation—queries always filter by `user_id` to prevent cross-user access. From c1b2656649aa8faa8f93e92ddc6adc5b56550f43 Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Sun, 1 Mar 2026 01:21:04 +0100 Subject: [PATCH 2/7] feat: misc issues fixed, also update saga/exec status --- backend/app/api/routes/common.py | 9 ++++++ .../app/core/middlewares/security_headers.py | 31 +++++++++++++++++++ 2 files changed, 40 insertions(+) create mode 100644 backend/app/api/routes/common.py create mode 100644 backend/app/core/middlewares/security_headers.py diff --git a/backend/app/api/routes/common.py b/backend/app/api/routes/common.py new file mode 100644 index 00000000..a7e7dbde --- /dev/null +++ b/backend/app/api/routes/common.py @@ -0,0 +1,9 @@ +from sse_starlette.sse import EventSourceResponse + + +class SSEResponse(EventSourceResponse): + """Workaround: sse-starlette sets media_type only in __init__, not as a + class attribute. FastAPI reads the class attribute for OpenAPI generation, + so without this subclass every SSE endpoint shows application/json.""" + + media_type = "text/event-stream" diff --git a/backend/app/core/middlewares/security_headers.py b/backend/app/core/middlewares/security_headers.py new file mode 100644 index 00000000..a07f4795 --- /dev/null +++ b/backend/app/core/middlewares/security_headers.py @@ -0,0 +1,31 @@ +from starlette.datastructures import MutableHeaders +from starlette.types import ASGIApp, Message, Receive, Scope, Send + + +class SecurityHeadersMiddleware: + """Add standard security response headers to every HTTP response.""" + + _HEADERS = { + "X-Content-Type-Options": "nosniff", + "X-Frame-Options": "DENY", + "Referrer-Policy": "strict-origin-when-cross-origin", + "Strict-Transport-Security": "max-age=63072000; includeSubDomains", + "Content-Security-Policy": "default-src 'self'; frame-ancestors 'none'", + } + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + async def send_wrapper(message: Message) -> None: + if message["type"] == "http.response.start": + headers = MutableHeaders(scope=message) + for name, value in self._HEADERS.items(): + headers[name] = value + await send(message) + + await self.app(scope, receive, send_wrapper) From 71c37a8a37da7b448c8ac30328a8925a811d5a41 Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Sun, 1 Mar 2026 02:18:57 +0100 Subject: [PATCH 3/7] feat: better msg parsing from pods: using NAME+BYTESIZE instead of JSON. Also simplified logging in event mapper --- backend/app/scripts/entrypoint.sh | 48 ++-- .../app/services/k8s_worker/pod_builder.py | 1 + .../app/services/pod_monitor/event_mapper.py | 96 ++++--- .../services/pod_monitor/test_event_mapper.py | 238 ++++++++++++++---- 4 files changed, 251 insertions(+), 132 deletions(-) diff --git a/backend/app/scripts/entrypoint.sh b/backend/app/scripts/entrypoint.sh index 99cffc17..31c640f8 100644 --- a/backend/app/scripts/entrypoint.sh +++ b/backend/app/scripts/entrypoint.sh @@ -2,23 +2,16 @@ # # Strict, portable POSIX-sh entrypoint that runs an arbitrary # command, captures its output, exit-code and coarse resource -# usage, then prints a single-line JSON blob to stdout. - -# Very small, POSIX-compliant JSON string escaper -json_escape() { - sed -e ':a;N;$!ba' \ - -e 's/\\/\\\\/g' \ - -e 's/"/\\"/g' \ - -e 's/\n/\\n/g' \ - -e 's/\t/\\t/g' \ - -e 's/\r/\\r/g' -} +# usage, then writes metrics to /dev/termination-log and +# length-prefixed stdout/stderr to container stdout. # ---------- argument check -------------------------------------------------- if [ "$#" -eq 0 ]; then - printf '{"exit_code":127,"resource_usage":null,"stdout":"","stderr":"Entrypoint Error: No command provided."}' - exit 0 + printf 'cpu_jiffies=0\nclk_tck=100\npeak_memory_kb=0\nwall_seconds=0\n' > /dev/termination-log + ERR_MSG="Entrypoint Error: No command provided." + printf 'STDOUT 0\nSTDERR %d\n%s' "${#ERR_MSG}" "$ERR_MSG" + exit 127 fi # ---------- temp files & timing -------------------------------------------- @@ -68,16 +61,19 @@ EXIT_CODE=$? END_TIME="$(date +%s.%N)" ELAPSED_S=$(printf '%s\n' "$END_TIME $START_TIME" | awk '{printf "%.6f",$1-$2}') -# ---------- emit single-line JSON ------------------------------------------ - -# Build JSON on a single line -# Note: CLK_TCK included for CPU time conversion (cpu_seconds = jiffies / clk_tck) -# Security: CLK_TCK is obtainable by any user via getconf, not sensitive -printf '{"exit_code":%d,"resource_usage":{"execution_time_wall_seconds":%s,"cpu_time_jiffies":%d,"clk_tck_hertz":%d,"peak_memory_kb":%d},"stdout":"%s","stderr":"%s"}' \ - "${EXIT_CODE:-1}" \ - "${ELAPSED_S:-0}" \ - "${JIFS:-0}" \ - "${CLK_TCK:-100}" \ - "${PEAK_KB:-0}" \ - "$(cat "$OUT" | json_escape)" \ - "$(cat "$ERR" | json_escape)" +# ---------- write resource metrics to termination log ---------------------- + +printf 'cpu_jiffies=%d\nclk_tck=%d\npeak_memory_kb=%d\nwall_seconds=%s\n' \ + "${JIFS:-0}" "${CLK_TCK:-100}" "${PEAK_KB:-0}" "${ELAPSED_S:-0}" \ + > /dev/termination-log + +# ---------- write length-prefixed stdout/stderr ---------------------------- + +STDOUT_BYTES=$(wc -c < "$OUT") +STDERR_BYTES=$(wc -c < "$ERR") +printf 'STDOUT %d\n' "$STDOUT_BYTES" +cat "$OUT" +printf 'STDERR %d\n' "$STDERR_BYTES" +cat "$ERR" + +exit 0 diff --git a/backend/app/services/k8s_worker/pod_builder.py b/backend/app/services/k8s_worker/pod_builder.py index 3f621add..1579fdee 100644 --- a/backend/app/services/k8s_worker/pod_builder.py +++ b/backend/app/services/k8s_worker/pod_builder.py @@ -77,6 +77,7 @@ def _build_container(self, command: CreatePodCommandEvent) -> k8s_client.V1Conta k8s_client.V1EnvVar(name="EXECUTION_ID", value=execution_id), k8s_client.V1EnvVar(name="OUTPUT_PATH", value="/output"), ], + termination_message_policy="FallbackToLogsOnError", ) # SECURITY: Always enforce strict security context diff --git a/backend/app/services/pod_monitor/event_mapper.py b/backend/app/services/pod_monitor/event_mapper.py index 2983fe0d..f73bb1d2 100644 --- a/backend/app/services/pod_monitor/event_mapper.py +++ b/backend/app/services/pod_monitor/event_mapper.py @@ -1,4 +1,3 @@ -import json from collections.abc import Awaitable, Callable from dataclasses import dataclass from uuid import uuid4 @@ -457,72 +456,65 @@ def _analyze_failure(self, pod: k8s_client.V1Pod) -> FailureInfo: async def _extract_logs(self, pod: k8s_client.V1Pod) -> PodLogs | None: """Extract and parse pod logs. Returns None if extraction fails.""" - # Without k8s API or metadata, can't fetch logs if not self._k8s_api or not pod.metadata: return None - # Check if any container terminated - has_terminated = any( - status.state and status.state.terminated for status in (pod.status.container_statuses if pod.status else []) - ) - - if not has_terminated: - self.logger.debug(f"Pod {pod.metadata.name} has no terminated containers") + container = self._get_main_container(pod) + if not container or not container.state or not container.state.terminated: return None + terminated = container.state.terminated + + meta = self._parse_termination_message(terminated.message or "") + try: logs = await self._k8s_api.read_namespaced_pod_log( - name=pod.metadata.name, namespace=pod.metadata.namespace or "integr8scode", tail_lines=10000 + name=pod.metadata.name, + namespace=pod.metadata.namespace or "integr8scode", + tail_lines=10000, ) - - if not logs: - return None - - # Try to parse executor JSON - return self._parse_executor_output(logs) - - except Exception as e: - self._log_extraction_error(pod.metadata.name, str(e)) + except Exception: + self.logger.warning("Failed to fetch pod logs", pod_name=pod.metadata.name, exc_info=True) return None - def _parse_executor_output(self, logs: str) -> PodLogs | None: - """Parse executor JSON output from logs. Returns None if parsing fails.""" - logs_stripped = logs.strip() + if not logs: + return None - # Try full output as JSON - if result := self._try_parse_json(logs_stripped): - return result + stdout, stderr = self._parse_framed_output(logs) - # Try line by line - for line in logs_stripped.split("\n"): - if result := self._try_parse_json(line.strip()): - return result + return PodLogs( + exit_code=terminated.exit_code or 0, + stdout=stdout, + stderr=stderr, + resource_usage=ResourceUsageDomain( + execution_time_wall_seconds=float(meta.get("wall_seconds", "0")), + cpu_time_jiffies=int(meta.get("cpu_jiffies", "0")), + clk_tck_hertz=int(meta.get("clk_tck", "100")), + peak_memory_kb=int(meta.get("peak_memory_kb", "0")), + ), + ) - # No valid executor JSON found - self.logger.warning("Logs do not contain valid executor JSON") - return None + @staticmethod + def _parse_termination_message(raw: str) -> dict[str, str]: + """Parse key=value metadata from K8s termination message.""" + return dict(line.split("=", 1) for line in raw.strip().splitlines() if "=" in line) - def _try_parse_json(self, text: str) -> PodLogs | None: - """Try to parse text as executor JSON output""" - if not (text.startswith("{") and text.endswith("}")): - return None + @staticmethod + def _parse_framed_output(logs: str) -> tuple[str, str]: + """Extract stdout/stderr from length-prefixed framed output.""" + try: + idx = logs.index("STDOUT ") + 7 + nl = logs.index("\n", idx) + stdout_len = int(logs[idx:nl]) + stdout = logs[nl + 1 : nl + 1 + stdout_len] - data = json.loads(text) - return PodLogs( - stdout=data.get("stdout", ""), - stderr=data.get("stderr", ""), - exit_code=data.get("exit_code", 0), - resource_usage=ResourceUsageDomain(**data.get("resource_usage", {})), - ) + idx = logs.index("STDERR ", nl + 1 + stdout_len) + 7 + nl = logs.index("\n", idx) + stderr_len = int(logs[idx:nl]) + stderr = logs[nl + 1 : nl + 1 + stderr_len] + except (ValueError, IndexError): + return "", "" - def _log_extraction_error(self, pod_name: str, error: str) -> None: - """Log extraction errors with appropriate level""" - error_lower = error.lower() + return stdout, stderr - if "404" in error or "not found" in error_lower: - self.logger.debug(f"Pod {pod_name} logs not found - pod may have been deleted") - elif "400" in error: - self.logger.debug(f"Pod {pod_name} logs not available - container may still be creating") - else: - self.logger.warning(f"Failed to extract logs from pod {pod_name}: {error}") diff --git a/backend/tests/unit/services/pod_monitor/test_event_mapper.py b/backend/tests/unit/services/pod_monitor/test_event_mapper.py index 7d5d586b..aed333c9 100644 --- a/backend/tests/unit/services/pod_monitor/test_event_mapper.py +++ b/backend/tests/unit/services/pod_monitor/test_event_mapper.py @@ -1,10 +1,7 @@ -import json -import structlog from unittest.mock import AsyncMock, MagicMock import pytest -from kubernetes_asyncio.client import V1Pod, V1PodCondition - +import structlog from app.domain.enums import EventType, ExecutionErrorType from app.domain.events import ( EventMetadata, @@ -14,12 +11,22 @@ PodRunningEvent, ) from app.services.pod_monitor import PodContext, PodEventMapper, WatchEventType +from kubernetes_asyncio.client import V1Pod, V1PodCondition + from tests.unit.conftest import make_container_status, make_pod pytestmark = pytest.mark.unit _test_logger = structlog.get_logger("test.services.pod_monitor.event_mapper") +_TERM_MSG = "cpu_jiffies=100\nclk_tck=100\npeak_memory_kb=1024\nwall_seconds=0.5\n" +_TERM_MSG_EMPTY = "cpu_jiffies=0\nclk_tck=100\npeak_memory_kb=0\nwall_seconds=0\n" + + +def _framed(stdout: str = "", stderr: str = "") -> str: + """Build length-prefixed framed output string.""" + return f"STDOUT {len(stdout)}\n{stdout}STDERR {len(stderr)}\n{stderr}" + def _ctx(pod: V1Pod, event_type: WatchEventType = WatchEventType.ADDED) -> PodContext: return PodContext( @@ -31,7 +38,7 @@ def _ctx(pod: V1Pod, event_type: WatchEventType = WatchEventType.ADDED) -> PodCo ) -def _make_mock_api(logs: str = "{}") -> MagicMock: +def _make_mock_api(logs: str = "") -> MagicMock: mock = MagicMock() mock.read_namespaced_pod_log = AsyncMock(return_value=logs) return mock @@ -39,18 +46,8 @@ def _make_mock_api(logs: str = "{}") -> MagicMock: @pytest.mark.asyncio async def test_pending_running_and_succeeded_mapping() -> None: - logs_json = json.dumps({ - "stdout": "ok", - "stderr": "", - "exit_code": 0, - "resource_usage": { - "execution_time_wall_seconds": 0, - "cpu_time_jiffies": 0, - "clk_tck_hertz": 0, - "peak_memory_kb": 0, - }, - }) - pem = PodEventMapper(k8s_api=_make_mock_api(logs_json), logger=_test_logger) + logs = _framed("ok", "") + pem = PodEventMapper(k8s_api=_make_mock_api(logs), logger=_test_logger) # Pending -> scheduled pend = make_pod( @@ -81,30 +78,33 @@ async def test_pending_running_and_succeeded_mapping() -> None: "terminated" in s.state for s in pr.container_statuses ) - # Succeeded -> completed + # Succeeded -> completed (with termination message for resource metrics) suc = make_pod( name="p", phase="Succeeded", labels={"execution-id": "e1"}, - container_statuses=[make_container_status(terminated_exit_code=0)], + container_statuses=[make_container_status(terminated_exit_code=0, terminated_message=_TERM_MSG)], ) evts = await pem.map_pod_event(suc, WatchEventType.MODIFIED) comp = [e for e in evts if e.event_type == EventType.EXECUTION_COMPLETED][0] assert isinstance(comp, ExecutionCompletedEvent) assert comp.exit_code == 0 and comp.stdout == "ok" + assert comp.resource_usage is not None + assert comp.resource_usage.cpu_time_jiffies == 100 + assert comp.resource_usage.peak_memory_kb == 1024 @pytest.mark.asyncio async def test_failed_timeout_and_deleted() -> None: - valid_logs = json.dumps({"stdout": "", "stderr": "", "exit_code": 137, "resource_usage": {}}) - pem = PodEventMapper(k8s_api=_make_mock_api(valid_logs), logger=_test_logger) + logs = _framed("", "") + pem = PodEventMapper(k8s_api=_make_mock_api(logs), logger=_test_logger) # Timeout via DeadlineExceeded pod_to = make_pod( name="p", phase="Failed", labels={"execution-id": "e1"}, - container_statuses=[make_container_status(terminated_exit_code=137)], + container_statuses=[make_container_status(terminated_exit_code=137, terminated_message=_TERM_MSG_EMPTY)], reason="DeadlineExceeded", active_deadline_seconds=5, ) @@ -113,13 +113,17 @@ async def test_failed_timeout_and_deleted() -> None: assert ev.event_type == EventType.EXECUTION_TIMEOUT and ev.timeout_seconds == 5 # DeadlineExceeded with clean container exit should be treated as completed - valid_logs_done = json.dumps({"stdout": "ok", "stderr": "", "exit_code": 0, "resource_usage": {}}) - pem_done = PodEventMapper(k8s_api=_make_mock_api(valid_logs_done), logger=_test_logger) + logs_done = _framed("ok", "") + pem_done = PodEventMapper(k8s_api=_make_mock_api(logs_done), logger=_test_logger) pod_to_done = make_pod( name="p0", phase="Failed", labels={"execution-id": "e0"}, - container_statuses=[make_container_status(terminated_exit_code=0, terminated_reason="Completed")], + container_statuses=[ + make_container_status( + terminated_exit_code=0, terminated_reason="Completed", terminated_message=_TERM_MSG + ) + ], reason="DeadlineExceeded", active_deadline_seconds=5, ) @@ -140,13 +144,17 @@ async def test_failed_timeout_and_deleted() -> None: assert evf.event_type == EventType.EXECUTION_FAILED and evf.error_type in {ExecutionErrorType.SCRIPT_ERROR} # Deleted with exit code 0 returns completed - valid_logs_0 = json.dumps({"stdout": "", "stderr": "", "exit_code": 0, "resource_usage": {}}) - pem_completed = PodEventMapper(k8s_api=_make_mock_api(valid_logs_0), logger=_test_logger) + logs_0 = _framed("", "") + pem_completed = PodEventMapper(k8s_api=_make_mock_api(logs_0), logger=_test_logger) pod_del = make_pod( name="p3", phase="Failed", labels={"execution-id": "e3"}, - container_statuses=[make_container_status(terminated_exit_code=0, terminated_reason="Completed")], + container_statuses=[ + make_container_status( + terminated_exit_code=0, terminated_reason="Completed", terminated_message=_TERM_MSG_EMPTY + ) + ], ) evd = (await pem_completed.map_pod_event(pod_del, WatchEventType.DELETED))[0] assert evd.event_type == EventType.EXECUTION_COMPLETED @@ -201,35 +209,116 @@ async def test_scheduled_requires_condition() -> None: @pytest.mark.asyncio -async def test_parse_and_log_paths_and_analyze_failure_variants(caplog: pytest.LogCaptureFixture) -> None: - line_json = '{"stdout":"x","stderr":"","exit_code":3,"resource_usage":{}}' - pem = PodEventMapper(k8s_api=_make_mock_api("junk\n" + line_json), logger=_test_logger) +async def test_extract_logs_with_framed_output() -> None: + """Test log extraction with length-prefixed framing and termination message.""" + logs = _framed("hello world", "some warning") + pem = PodEventMapper(k8s_api=_make_mock_api(logs), logger=_test_logger) + pod = make_pod( + name="p", + phase="Succeeded", + container_statuses=[make_container_status(terminated_exit_code=0, terminated_message=_TERM_MSG)], + ) + result = await pem._extract_logs(pod) + assert result is not None + assert result.stdout == "hello world" + assert result.stderr == "some warning" + assert result.exit_code == 0 + assert result.resource_usage.cpu_time_jiffies == 100 + assert result.resource_usage.clk_tck_hertz == 100 + assert result.resource_usage.peak_memory_kb == 1024 + assert result.resource_usage.execution_time_wall_seconds == 0.5 + + +@pytest.mark.asyncio +async def test_extract_logs_empty_stdout_stderr() -> None: + """Test extraction when both stdout and stderr are empty.""" + logs = _framed("", "") + pem = PodEventMapper(k8s_api=_make_mock_api(logs), logger=_test_logger) + pod = make_pod( + name="p", + phase="Succeeded", + container_statuses=[make_container_status(terminated_exit_code=0, terminated_message=_TERM_MSG_EMPTY)], + ) + result = await pem._extract_logs(pod) + assert result is not None + assert result.stdout == "" + assert result.stderr == "" + + +@pytest.mark.asyncio +async def test_extract_logs_large_stdout() -> None: + """Test extraction with large stdout content.""" + large_content = "x" * 50_000 + logs = _framed(large_content, "err") + pem = PodEventMapper(k8s_api=_make_mock_api(logs), logger=_test_logger) + pod = make_pod( + name="p", + phase="Succeeded", + container_statuses=[make_container_status(terminated_exit_code=0, terminated_message=_TERM_MSG)], + ) + result = await pem._extract_logs(pod) + assert result is not None + assert result.stdout == large_content + assert result.stderr == "err" + + +@pytest.mark.asyncio +async def test_extract_logs_missing_termination_message() -> None: + """Test extraction when termination message is absent — defaults to zero values.""" + logs = _framed("out", "") + pem = PodEventMapper(k8s_api=_make_mock_api(logs), logger=_test_logger) + pod = make_pod( + name="p", + phase="Succeeded", + container_statuses=[make_container_status(terminated_exit_code=0)], + ) + result = await pem._extract_logs(pod) + assert result is not None + assert result.stdout == "out" + assert result.resource_usage.cpu_time_jiffies == 0 + assert result.resource_usage.clk_tck_hertz == 100 + assert result.resource_usage.peak_memory_kb == 0 + + +@pytest.mark.asyncio +async def test_extract_logs_malformed_framing_returns_empty() -> None: + """Test that malformed log framing returns empty strings.""" + pem = PodEventMapper(k8s_api=_make_mock_api("garbage data"), logger=_test_logger) + pod = make_pod( + name="p", + phase="Succeeded", + container_statuses=[make_container_status(terminated_exit_code=0, terminated_message=_TERM_MSG)], + ) + result = await pem._extract_logs(pod) + assert result is not None + assert result.stdout == "" + assert result.stderr == "" + + +@pytest.mark.asyncio +async def test_extract_logs_error_paths() -> None: + """Test error paths: no API, API exceptions.""" pod = make_pod( name="p", phase="Succeeded", container_statuses=[make_container_status(terminated_exit_code=0)], ) - logs = await pem._extract_logs(pod) - assert logs is not None - assert logs.exit_code == 3 and logs.stdout == "x" # no api -> returns None - pem2 = PodEventMapper(k8s_api=None, logger=_test_logger) - assert await pem2._extract_logs(pod) is None - - # exceptions -> all return None - mock_404 = MagicMock() - mock_404.read_namespaced_pod_log = AsyncMock(side_effect=Exception("404 Not Found")) - mock_400 = MagicMock() - mock_400.read_namespaced_pod_log = AsyncMock(side_effect=Exception("400 Bad Request")) - mock_gen = MagicMock() - mock_gen.read_namespaced_pod_log = AsyncMock(side_effect=Exception("boom")) - - assert await PodEventMapper(k8s_api=mock_404, logger=_test_logger)._extract_logs(pod) is None - assert await PodEventMapper(k8s_api=mock_400, logger=_test_logger)._extract_logs(pod) is None - assert await PodEventMapper(k8s_api=mock_gen, logger=_test_logger)._extract_logs(pod) is None - - # _analyze_failure: Evicted + pem_no_api = PodEventMapper(k8s_api=None, logger=_test_logger) + assert await pem_no_api._extract_logs(pod) is None + + # API exception -> returns None + mock_err = MagicMock() + mock_err.read_namespaced_pod_log = AsyncMock(side_effect=Exception("boom")) + assert await PodEventMapper(k8s_api=mock_err, logger=_test_logger)._extract_logs(pod) is None + + +def test_analyze_failure_variants() -> None: + """Test _analyze_failure with various pod failure scenarios.""" + pem = PodEventMapper(k8s_api=_make_mock_api(""), logger=_test_logger) + + # Evicted pod_e = make_pod(name="p", phase="Failed", reason="Evicted") assert pem._analyze_failure(pod_e).error_type == ExecutionErrorType.RESOURCE_LIMIT @@ -262,16 +351,16 @@ async def test_parse_and_log_paths_and_analyze_failure_variants(caplog: pytest.L @pytest.mark.asyncio async def test_all_containers_succeeded_and_cache_behavior() -> None: - valid_logs = json.dumps({"stdout": "", "stderr": "", "exit_code": 0, "resource_usage": {}}) - pem = PodEventMapper(k8s_api=_make_mock_api(valid_logs), logger=_test_logger) + logs = _framed("", "") + pem = PodEventMapper(k8s_api=_make_mock_api(logs), logger=_test_logger) pod = make_pod( name="p", phase="Failed", labels={"execution-id": "e1"}, container_statuses=[ - make_container_status(terminated_exit_code=0), - make_container_status(terminated_exit_code=0), + make_container_status(terminated_exit_code=0, terminated_message=_TERM_MSG_EMPTY), + make_container_status(terminated_exit_code=0, terminated_message=_TERM_MSG_EMPTY), ], ) # When all succeeded, failed mapping returns completed instead of failed @@ -284,3 +373,44 @@ async def test_all_containers_succeeded_and_cache_behavior() -> None: b = await pem.map_pod_event(p2, WatchEventType.MODIFIED) assert a == [] or all(x.event_type for x in a) assert b == [] or all(x.event_type for x in b) + + +def test_parse_termination_message() -> None: + """Test _parse_termination_message with various inputs.""" + parse = PodEventMapper._parse_termination_message + + # Normal message + result = parse("cpu_jiffies=100\nclk_tck=100\npeak_memory_kb=1024\nwall_seconds=0.5\n") + assert result == {"cpu_jiffies": "100", "clk_tck": "100", "peak_memory_kb": "1024", "wall_seconds": "0.5"} + + # Empty string + assert parse("") == {} + + # Lines without = are skipped + assert parse("no-equals\ncpu_jiffies=50\n") == {"cpu_jiffies": "50"} + + # Value containing = sign + assert parse("key=val=ue\n") == {"key": "val=ue"} + + +def test_parse_framed_output() -> None: + """Test _parse_framed_output with various inputs.""" + parse = PodEventMapper._parse_framed_output + + # Normal case + assert parse("STDOUT 5\nhelloSTDERR 3\nerr") == ("hello", "err") + + # Empty both + assert parse("STDOUT 0\nSTDERR 0\n") == ("", "") + + # Content with newlines + content = "line1\nline2\nline3" + assert parse(f"STDOUT {len(content)}\n{content}STDERR 0\n") == (content, "") + + # Malformed input + assert parse("garbage") == ("", "") + assert parse("") == ("", "") + + # Content containing STDOUT/STDERR markers (length-prefix makes this safe) + tricky = "STDOUT 5\nfake" + assert parse(f"STDOUT {len(tricky)}\n{tricky}STDERR 0\n") == (tricky, "") From 6a67412ebfce83d0fd1ad5daafa5d05431ad5124 Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Sun, 1 Mar 2026 02:35:47 +0100 Subject: [PATCH 4/7] feat: issues fixing (passing exit code instead of 0, previous issues, ..) --- backend/app/db/repositories/event_repository.py | 4 ++-- backend/app/db/repositories/execution_repository.py | 5 +++-- backend/app/db/repositories/saga_repository.py | 2 +- backend/app/scripts/entrypoint.sh | 2 +- backend/app/services/rate_limit_service.py | 12 +++++++----- .../services/rate_limit/test_rate_limit_service.py | 9 ++++----- backend/tests/e2e/test_auth_routes.py | 2 +- 7 files changed, 19 insertions(+), 17 deletions(-) diff --git a/backend/app/db/repositories/event_repository.py b/backend/app/db/repositories/event_repository.py index 3bac6e32..6dab1cb5 100644 --- a/backend/app/db/repositories/event_repository.py +++ b/backend/app/db/repositories/event_repository.py @@ -264,8 +264,8 @@ async def delete_event_with_archival( archive_fields = {"deleted_at": deleted_at, "deleted_by": deleted_by, "deletion_reason": deletion_reason} archived_doc = EventArchiveDocument.model_validate(doc).model_copy(update=archive_fields) - session = await EventDocument.get_motor_collection().database.client.start_session() - async with session.start_transaction(): + session = EventDocument.get_pymongo_collection().database.client.start_session() + async with await session.start_transaction(): await archived_doc.insert(session=session) await doc.delete(session=session) diff --git a/backend/app/db/repositories/execution_repository.py b/backend/app/db/repositories/execution_repository.py index 91e71a49..05c83390 100644 --- a/backend/app/db/repositories/execution_repository.py +++ b/backend/app/db/repositories/execution_repository.py @@ -143,8 +143,9 @@ async def aggregate_stats(self, query: dict[str, Any]) -> dict[str, Any]: }, }) - collection = ExecutionDocument.get_motor_collection() - results = await collection.aggregate(pipeline).to_list(length=1) + collection = ExecutionDocument.get_pymongo_collection() + cursor = await collection.aggregate(pipeline) + results = await cursor.to_list(length=1) if not results: return {"total": 0, "by_status": {}, "by_language": {}, "average_duration_ms": 0, "success_rate": 0} diff --git a/backend/app/db/repositories/saga_repository.py b/backend/app/db/repositories/saga_repository.py index ea44376c..c812381d 100644 --- a/backend/app/db/repositories/saga_repository.py +++ b/backend/app/db/repositories/saga_repository.py @@ -167,7 +167,7 @@ async def list_sagas(self, saga_filter: SagaFilter, limit: int = 100, skip: int ) async def get_user_execution_ids(self, user_id: str) -> list[str]: - collection = ExecutionDocument.get_motor_collection() + collection = ExecutionDocument.get_pymongo_collection() result: list[str] = await collection.distinct("execution_id", {"user_id": user_id}) return result diff --git a/backend/app/scripts/entrypoint.sh b/backend/app/scripts/entrypoint.sh index 31c640f8..6e1dae6e 100644 --- a/backend/app/scripts/entrypoint.sh +++ b/backend/app/scripts/entrypoint.sh @@ -76,4 +76,4 @@ cat "$OUT" printf 'STDERR %d\n' "$STDERR_BYTES" cat "$ERR" -exit 0 +exit "$EXIT_CODE" diff --git a/backend/app/services/rate_limit_service.py b/backend/app/services/rate_limit_service.py index 897fe7c7..a0ccb9c4 100644 --- a/backend/app/services/rate_limit_service.py +++ b/backend/app/services/rate_limit_service.py @@ -176,11 +176,13 @@ async def _check_sliding_window( local bucket_data = redis.call('GET', key) if bucket_data then - local bucket = cjson.decode(bucket_data) - tokens = bucket['tokens'] - last_refill = bucket['last_refill'] - local time_passed = now - last_refill - tokens = math.min(max_tokens, tokens + time_passed * refill_rate) + local ok, bucket = pcall(cjson.decode, bucket_data) + if ok and bucket['tokens'] and bucket['last_refill'] then + tokens = bucket['tokens'] + last_refill = bucket['last_refill'] + local time_passed = now - last_refill + tokens = math.min(max_tokens, tokens + time_passed * refill_rate) + end end local allowed = 0 diff --git a/backend/tests/e2e/services/rate_limit/test_rate_limit_service.py b/backend/tests/e2e/services/rate_limit/test_rate_limit_service.py index 2452e845..283ea933 100644 --- a/backend/tests/e2e/services/rate_limit/test_rate_limit_service.py +++ b/backend/tests/e2e/services/rate_limit/test_rate_limit_service.py @@ -1,5 +1,4 @@ import asyncio -import json from collections.abc import Awaitable from typing import Any, cast from uuid import uuid4 @@ -231,10 +230,10 @@ async def test_token_bucket_invalid_data(scope: AsyncContainer) -> None: algorithm=RateLimitAlgorithm.TOKEN_BUCKET, ) - with pytest.raises(json.JSONDecodeError): - await svc._check_token_bucket( - "user", "/api", int(rule.requests), rule.window_seconds, rule.burst_multiplier or 1.0, rule - ) + status = await svc._check_token_bucket( + "user", "/api", int(rule.requests), rule.window_seconds, rule.burst_multiplier or 1.0, rule + ) + assert status.allowed is True @pytest.mark.asyncio diff --git a/backend/tests/e2e/test_auth_routes.py b/backend/tests/e2e/test_auth_routes.py index 00545c47..f72b9774 100644 --- a/backend/tests/e2e/test_auth_routes.py +++ b/backend/tests/e2e/test_auth_routes.py @@ -135,7 +135,7 @@ async def test_register_duplicate_email( ) assert response.status_code == 409 - assert response.json()["detail"] == "User already exists" + assert response.json()["detail"] == "Email already registered" @pytest.mark.asyncio async def test_register_invalid_email_format(self, client: AsyncClient) -> None: From 310e0d11c4b36dd9916b6bb110709c8346d51c40 Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Sun, 1 Mar 2026 02:55:37 +0100 Subject: [PATCH 5/7] fix: detected issues --- .../app/db/repositories/notification_repository.py | 2 +- backend/app/scripts/entrypoint.sh | 8 ++++---- backend/app/services/pod_monitor/event_mapper.py | 2 +- .../unit/services/pod_monitor/test_event_mapper.py | 14 ++++++++------ 4 files changed, 14 insertions(+), 12 deletions(-) diff --git a/backend/app/db/repositories/notification_repository.py b/backend/app/db/repositories/notification_repository.py index 6365dbd3..c8016dba 100644 --- a/backend/app/db/repositories/notification_repository.py +++ b/backend/app/db/repositories/notification_repository.py @@ -55,7 +55,7 @@ async def mark_as_read(self, notification_id: str, user_id: str) -> bool: async def mark_all_as_read(self, user_id: str) -> int: result = await NotificationDocument.find( NotificationDocument.user_id == user_id, - NotificationDocument.status == NotificationStatus.DELIVERED, + NotificationDocument.status != NotificationStatus.READ, ).update_many({"$set": {"status": NotificationStatus.READ, "read_at": datetime.now(UTC)}}) return result.modified_count if result and hasattr(result, "modified_count") else 0 diff --git a/backend/app/scripts/entrypoint.sh b/backend/app/scripts/entrypoint.sh index 6e1dae6e..a5796be5 100644 --- a/backend/app/scripts/entrypoint.sh +++ b/backend/app/scripts/entrypoint.sh @@ -8,7 +8,7 @@ # ---------- argument check -------------------------------------------------- if [ "$#" -eq 0 ]; then - printf 'cpu_jiffies=0\nclk_tck=100\npeak_memory_kb=0\nwall_seconds=0\n' > /dev/termination-log + printf 'exit_code=127\ncpu_jiffies=0\nclk_tck=100\npeak_memory_kb=0\nwall_seconds=0\n' > /dev/termination-log ERR_MSG="Entrypoint Error: No command provided." printf 'STDOUT 0\nSTDERR %d\n%s' "${#ERR_MSG}" "$ERR_MSG" exit 127 @@ -63,8 +63,8 @@ ELAPSED_S=$(printf '%s\n' "$END_TIME $START_TIME" | awk '{printf "%.6f",$1-$2}') # ---------- write resource metrics to termination log ---------------------- -printf 'cpu_jiffies=%d\nclk_tck=%d\npeak_memory_kb=%d\nwall_seconds=%s\n' \ - "${JIFS:-0}" "${CLK_TCK:-100}" "${PEAK_KB:-0}" "${ELAPSED_S:-0}" \ +printf 'exit_code=%d\ncpu_jiffies=%d\nclk_tck=%d\npeak_memory_kb=%d\nwall_seconds=%s\n' \ + "${EXIT_CODE:-1}" "${JIFS:-0}" "${CLK_TCK:-100}" "${PEAK_KB:-0}" "${ELAPSED_S:-0}" \ > /dev/termination-log # ---------- write length-prefixed stdout/stderr ---------------------------- @@ -76,4 +76,4 @@ cat "$OUT" printf 'STDERR %d\n' "$STDERR_BYTES" cat "$ERR" -exit "$EXIT_CODE" +exit 0 diff --git a/backend/app/services/pod_monitor/event_mapper.py b/backend/app/services/pod_monitor/event_mapper.py index f73bb1d2..7e308a9f 100644 --- a/backend/app/services/pod_monitor/event_mapper.py +++ b/backend/app/services/pod_monitor/event_mapper.py @@ -483,7 +483,7 @@ async def _extract_logs(self, pod: k8s_client.V1Pod) -> PodLogs | None: stdout, stderr = self._parse_framed_output(logs) return PodLogs( - exit_code=terminated.exit_code or 0, + exit_code=int(meta.get("exit_code", str(terminated.exit_code or 0))), stdout=stdout, stderr=stderr, resource_usage=ResourceUsageDomain( diff --git a/backend/tests/unit/services/pod_monitor/test_event_mapper.py b/backend/tests/unit/services/pod_monitor/test_event_mapper.py index aed333c9..6e5fdb97 100644 --- a/backend/tests/unit/services/pod_monitor/test_event_mapper.py +++ b/backend/tests/unit/services/pod_monitor/test_event_mapper.py @@ -19,8 +19,8 @@ _test_logger = structlog.get_logger("test.services.pod_monitor.event_mapper") -_TERM_MSG = "cpu_jiffies=100\nclk_tck=100\npeak_memory_kb=1024\nwall_seconds=0.5\n" -_TERM_MSG_EMPTY = "cpu_jiffies=0\nclk_tck=100\npeak_memory_kb=0\nwall_seconds=0\n" +_TERM_MSG = "exit_code=0\ncpu_jiffies=100\nclk_tck=100\npeak_memory_kb=1024\nwall_seconds=0.5\n" +_TERM_MSG_EMPTY = "exit_code=0\ncpu_jiffies=0\nclk_tck=100\npeak_memory_kb=0\nwall_seconds=0\n" def _framed(stdout: str = "", stderr: str = "") -> str: @@ -99,12 +99,12 @@ async def test_failed_timeout_and_deleted() -> None: logs = _framed("", "") pem = PodEventMapper(k8s_api=_make_mock_api(logs), logger=_test_logger) - # Timeout via DeadlineExceeded + # Timeout via DeadlineExceeded (entrypoint killed, no termination message) pod_to = make_pod( name="p", phase="Failed", labels={"execution-id": "e1"}, - container_statuses=[make_container_status(terminated_exit_code=137, terminated_message=_TERM_MSG_EMPTY)], + container_statuses=[make_container_status(terminated_exit_code=137)], reason="DeadlineExceeded", active_deadline_seconds=5, ) @@ -380,8 +380,10 @@ def test_parse_termination_message() -> None: parse = PodEventMapper._parse_termination_message # Normal message - result = parse("cpu_jiffies=100\nclk_tck=100\npeak_memory_kb=1024\nwall_seconds=0.5\n") - assert result == {"cpu_jiffies": "100", "clk_tck": "100", "peak_memory_kb": "1024", "wall_seconds": "0.5"} + result = parse("exit_code=0\ncpu_jiffies=100\nclk_tck=100\npeak_memory_kb=1024\nwall_seconds=0.5\n") + assert result == { + "exit_code": "0", "cpu_jiffies": "100", "clk_tck": "100", "peak_memory_kb": "1024", "wall_seconds": "0.5", + } # Empty string assert parse("") == {} From 30b1a55b1030c120d3b6eab1f5fd4bae36b8f859 Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Sun, 1 Mar 2026 13:05:55 +0100 Subject: [PATCH 6/7] fix: detected issues --- backend/app/core/middlewares/rate_limit.py | 22 +--- backend/app/core/utils.py | 15 ++- .../app/db/repositories/event_repository.py | 8 +- .../db/repositories/execution_repository.py | 15 +-- .../repositories/notification_repository.py | 1 + backend/app/domain/user/__init__.py | 2 + backend/app/domain/user/exceptions.py | 8 ++ backend/app/services/auth_service.py | 9 +- backend/app/services/execution_queue.py | 12 ++ backend/app/services/execution_service.py | 6 +- backend/app/services/notification_service.py | 12 +- .../app/services/pod_monitor/event_mapper.py | 5 +- .../app/services/saga/saga_orchestrator.py | 3 +- .../repositories/test_execution_repository.py | 109 +++++++++++++++++- .../services/saga/test_saga_orchestrator.py | 54 +++++++++ .../unit/services/test_execution_queue.py | 32 ++++- docs/architecture/domain-exceptions.md | 11 +- 17 files changed, 258 insertions(+), 66 deletions(-) diff --git a/backend/app/core/middlewares/rate_limit.py b/backend/app/core/middlewares/rate_limit.py index 4f484d4c..d1349b17 100644 --- a/backend/app/core/middlewares/rate_limit.py +++ b/backend/app/core/middlewares/rate_limit.py @@ -103,26 +103,12 @@ async def send_wrapper(message: Message) -> None: # --8<-- [start:extract_user_id] @staticmethod def _extract_user_id(request: Request) -> str: - """Extract user identifier for rate limiting. + """Extract rate-limit bucket key from client IP. - Reads the JWT payload from the access_token cookie without full - verification (that happens in route-level auth dependencies). This - is safe because the value is only used as a rate-limit bucket key. - Falls back to IP-based identification if no token is present or - the payload cannot be read. + Middleware runs before route-level auth, so no verified identity is + available here. Using unverified JWT claims would let an attacker + craft arbitrary bucket keys to bypass IP-based limits. """ - token = request.cookies.get("access_token") - if token: - import base64 - import json as _json - parts = token.split(".") - if len(parts) == 3: - # Pad the base64url payload segment - padded = parts[1] + "=" * (-len(parts[1]) % 4) - payload = _json.loads(base64.urlsafe_b64decode(padded)) - username = payload.get("sub") - if username: - return f"user:{username}" return f"ip:{get_client_ip(request)}" # --8<-- [end:extract_user_id] diff --git a/backend/app/core/utils.py b/backend/app/core/utils.py index e5c90fec..bad8343a 100644 --- a/backend/app/core/utils.py +++ b/backend/app/core/utils.py @@ -1,4 +1,5 @@ from enum import StrEnum +from ipaddress import ip_address from fastapi import Request @@ -54,11 +55,9 @@ def get_client_ip(request: Request) -> str: def _is_trusted_proxy(ip: str) -> bool: - """Check if an IP belongs to a trusted proxy (loopback or Docker-internal ranges).""" - return ( - ip.startswith("127.") - or ip == "::1" - or ip.startswith("10.") - or ip.startswith("172.") - or ip.startswith("192.168.") - ) + """Check if an IP belongs to a trusted proxy (loopback or RFC 1918 private ranges).""" + try: + addr = ip_address(ip) + except ValueError: + return False + return addr.is_loopback or addr.is_private diff --git a/backend/app/db/repositories/event_repository.py b/backend/app/db/repositories/event_repository.py index 6dab1cb5..f04043cc 100644 --- a/backend/app/db/repositories/event_repository.py +++ b/backend/app/db/repositories/event_repository.py @@ -264,10 +264,10 @@ async def delete_event_with_archival( archive_fields = {"deleted_at": deleted_at, "deleted_by": deleted_by, "deletion_reason": deletion_reason} archived_doc = EventArchiveDocument.model_validate(doc).model_copy(update=archive_fields) - session = EventDocument.get_pymongo_collection().database.client.start_session() - async with await session.start_transaction(): - await archived_doc.insert(session=session) - await doc.delete(session=session) + async with EventDocument.get_pymongo_collection().database.client.start_session() as session: + async with await session.start_transaction(): + await archived_doc.insert(session=session) + await doc.delete(session=session) return ArchivedEvent.model_validate(doc).model_copy(update=archive_fields) diff --git a/backend/app/db/repositories/execution_repository.py b/backend/app/db/repositories/execution_repository.py index 05c83390..ece33df0 100644 --- a/backend/app/db/repositories/execution_repository.py +++ b/backend/app/db/repositories/execution_repository.py @@ -7,7 +7,7 @@ from beanie.operators import In from app.db.docs import ExecutionDocument -from app.domain.enums import EXECUTION_ACTIVE, QueuePriority +from app.domain.enums import EXECUTION_ACTIVE, ExecutionStatus, QueuePriority from app.domain.events import ResourceUsageDomain from app.domain.execution import ( DomainExecution, @@ -49,7 +49,8 @@ async def get_execution(self, execution_id: str) -> DomainExecution | None: async def write_terminal_result(self, result: ExecutionResultDomain) -> bool: """Atomically write a terminal result, guarded by non-terminal status check. - Uses find_one_and_update so a slower processor cannot overwrite a result + Performs a conditional update that only applies when the current status is + still in EXECUTION_ACTIVE, so a slower processor cannot overwrite a result that was already written by a faster one. """ update_result = await ExecutionDocument.find_one( @@ -123,21 +124,17 @@ async def aggregate_stats(self, query: dict[str, Any]) -> dict[str, Any]: "totals": [{"$group": { "_id": None, "total": {"$sum": 1}, - "successful": {"$sum": {"$cond": [{"$eq": ["$status", "completed"]}, 1, 0]}}, + "successful": {"$sum": {"$cond": [{"$eq": ["$status", ExecutionStatus.COMPLETED]}, 1, 0]}}, }}], "avg_duration": [ {"$match": { - "status": "completed", + "status": ExecutionStatus.COMPLETED, "created_at": {"$ne": None}, "updated_at": {"$ne": None}, }}, {"$group": { "_id": None, - "avg_ms": {"$avg": { - "$multiply": [ - {"$divide": [{"$subtract": ["$updated_at", "$created_at"]}, 1]}, - ], - }}, + "avg_ms": {"$avg": {"$subtract": ["$updated_at", "$created_at"]}}, }}, ], }, diff --git a/backend/app/db/repositories/notification_repository.py b/backend/app/db/repositories/notification_repository.py index c8016dba..4a33a067 100644 --- a/backend/app/db/repositories/notification_repository.py +++ b/backend/app/db/repositories/notification_repository.py @@ -56,6 +56,7 @@ async def mark_all_as_read(self, user_id: str) -> int: result = await NotificationDocument.find( NotificationDocument.user_id == user_id, NotificationDocument.status != NotificationStatus.READ, + NotificationDocument.status != NotificationStatus.CLICKED, ).update_many({"$set": {"status": NotificationStatus.READ, "read_at": datetime.now(UTC)}}) return result.modified_count if result and hasattr(result, "modified_count") else 0 diff --git a/backend/app/domain/user/__init__.py b/backend/app/domain/user/__init__.py index 4f5957f5..05ce83b7 100644 --- a/backend/app/domain/user/__init__.py +++ b/backend/app/domain/user/__init__.py @@ -1,6 +1,7 @@ from app.domain.enums import UserRole from .exceptions import ( + AccountDeactivatedError, AdminAccessRequiredError, AuthenticationRequiredError, CSRFValidationError, @@ -31,6 +32,7 @@ ) __all__ = [ + "AccountDeactivatedError", "AdminAccessRequiredError", "AuthenticationRequiredError", "CachedSettings", diff --git a/backend/app/domain/user/exceptions.py b/backend/app/domain/user/exceptions.py index 8b87c7a3..b5a8f57e 100644 --- a/backend/app/domain/user/exceptions.py +++ b/backend/app/domain/user/exceptions.py @@ -38,6 +38,14 @@ def __init__(self, username: str | None = None) -> None: super().__init__(msg) +class AccountDeactivatedError(ForbiddenError): + """Raised when a deactivated account attempts to log in.""" + + def __init__(self, username: str) -> None: + self.username = username + super().__init__(f"Account '{username}' is deactivated") + + class UserNotFoundError(NotFoundError): """Raised when a user is not found.""" diff --git a/backend/app/services/auth_service.py b/backend/app/services/auth_service.py index 0d1dbe62..5216d106 100644 --- a/backend/app/services/auth_service.py +++ b/backend/app/services/auth_service.py @@ -17,6 +17,7 @@ ) from app.domain.exceptions import AccountLockedError, ConflictError, ValidationError from app.domain.user import ( + AccountDeactivatedError, AdminAccessRequiredError, AuthenticationRequiredError, DomainUserCreate, @@ -142,7 +143,13 @@ async def login( await self._fail_login(username, "invalid_password", ip_address, user_agent, user_id=user.user_id) if not user.is_active: - await self._fail_login(username, "account_deactivated", ip_address, user_agent, user_id=user.user_id) + self.logger.warning( + "Login rejected: account deactivated", + username=username, + client_ip=ip_address, + user_agent=user_agent, + ) + raise AccountDeactivatedError(username) await self._lockout.clear_attempts(username) diff --git a/backend/app/services/execution_queue.py b/backend/app/services/execution_queue.py index d103aa22..ee5de60f 100644 --- a/backend/app/services/execution_queue.py +++ b/backend/app/services/execution_queue.py @@ -28,6 +28,10 @@ def _event_key(execution_id: str) -> str: return f"exec_queue:event:{execution_id}" +def _retry_key(execution_id: str) -> str: + return f"exec_queue:retries:{execution_id}" + + def _pending_key(priority: QueuePriority) -> str: return f"{_PENDING_PREFIX}{priority}" @@ -150,6 +154,13 @@ async def try_schedule(self, max_active: int) -> tuple[str, ExecutionRequestedEv ) return execution_id, event + async def increment_retry_count(self, execution_id: str) -> int: + """Atomically increment and return the retry count for an execution.""" + key = _retry_key(execution_id) + count: int = await self._redis.incr(key) + await self._redis.expire(key, _EVENT_TTL) + return count + async def release(self, execution_id: str) -> None: pipe = self._redis.pipeline(transaction=True) pipe.srem(_ACTIVE_KEY, execution_id) @@ -176,6 +187,7 @@ async def remove(self, execution_id: str) -> bool: for key in _PENDING_KEYS: pipe.zrem(key, execution_id) pipe.delete(_event_key(execution_id)) + pipe.delete(_retry_key(execution_id)) pipe.srem(_ACTIVE_KEY, execution_id) results = await pipe.execute() removed = any(results[: len(_PENDING_KEYS)]) or bool(results[-1]) diff --git a/backend/app/services/execution_service.py b/backend/app/services/execution_service.py index 4b8178b2..7f021498 100644 --- a/backend/app/services/execution_service.py +++ b/backend/app/services/execution_service.py @@ -213,9 +213,6 @@ async def cancel_execution( Raises: ExecutionTerminalError: If execution is in a terminal state. """ - if current_status.is_terminal: - raise ExecutionTerminalError(execution_id, current_status) - if current_status == ExecutionStatus.CANCELLED: return CancelResult( execution_id=execution_id, @@ -224,6 +221,9 @@ async def cancel_execution( event_id=None, ) + if current_status.is_terminal: + raise ExecutionTerminalError(execution_id, current_status) + metadata = self._create_event_metadata(user_id=user_id) event = ExecutionCancelledEvent( execution_id=execution_id, diff --git a/backend/app/services/notification_service.py b/backend/app/services/notification_service.py index cd9063cb..069f7dbc 100644 --- a/backend/app/services/notification_service.py +++ b/backend/app/services/notification_service.py @@ -111,8 +111,6 @@ def __init__( self.settings = settings self.sse_bus = sse_bus self.logger = logger - self._http_client = httpx.AsyncClient(timeout=30.0) - self._throttle_cache = ThrottleCache() # --8<-- [start:channel_handlers] @@ -364,8 +362,9 @@ async def _send_webhook( "notification.channel": "webhook", "notification.webhook_host": safe_host, }) - response = await self._http_client.post(webhook_url, json=payload, headers=headers) - response.raise_for_status() + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post(webhook_url, json=payload, headers=headers) + response.raise_for_status() self.logger.debug( "Webhook delivered successfully", notification_id=str(notification.notification_id), @@ -411,8 +410,9 @@ async def _send_slack(self, notification: DomainNotification, subscription: Doma "notification.id": str(notification.notification_id), "notification.channel": "slack", }) - response = await self._http_client.post(subscription.slack_webhook, json=slack_message) - response.raise_for_status() + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post(subscription.slack_webhook, json=slack_message) + response.raise_for_status() self.logger.debug( "Slack notification delivered successfully", notification_id=str(notification.notification_id), diff --git a/backend/app/services/pod_monitor/event_mapper.py b/backend/app/services/pod_monitor/event_mapper.py index 7e308a9f..74b98039 100644 --- a/backend/app/services/pod_monitor/event_mapper.py +++ b/backend/app/services/pod_monitor/event_mapper.py @@ -477,10 +477,7 @@ async def _extract_logs(self, pod: k8s_client.V1Pod) -> PodLogs | None: self.logger.warning("Failed to fetch pod logs", pod_name=pod.metadata.name, exc_info=True) return None - if not logs: - return None - - stdout, stderr = self._parse_framed_output(logs) + stdout, stderr = self._parse_framed_output(logs or "") return PodLogs( exit_code=int(meta.get("exit_code", str(terminated.exit_code or 0))), diff --git a/backend/app/services/saga/saga_orchestrator.py b/backend/app/services/saga/saga_orchestrator.py index a79af3f0..62dfa5d2 100644 --- a/backend/app/services/saga/saga_orchestrator.py +++ b/backend/app/services/saga/saga_orchestrator.py @@ -130,7 +130,7 @@ async def try_schedule_from_queue(self) -> None: try: await self._start_saga(event) except Exception: - retry_count = getattr(event, "_retry_count", 0) + 1 + retry_count = await self._queue.increment_retry_count(execution_id) self.logger.error( "Failed to start saga", execution_id=execution_id, @@ -148,7 +148,6 @@ async def try_schedule_from_queue(self) -> None: f"Failed to start saga after {retry_count} attempts", ) else: - event._retry_count = retry_count # type: ignore[attr-defined] await self._queue.enqueue(event) break diff --git a/backend/tests/e2e/db/repositories/test_execution_repository.py b/backend/tests/e2e/db/repositories/test_execution_repository.py index c1396020..aca51e60 100644 --- a/backend/tests/e2e/db/repositories/test_execution_repository.py +++ b/backend/tests/e2e/db/repositories/test_execution_repository.py @@ -1,10 +1,10 @@ -import structlog from uuid import uuid4 import pytest +import structlog from app.db.repositories import ExecutionRepository -from app.domain.enums import ExecutionStatus -from app.domain.execution import DomainExecutionCreate +from app.domain.enums import ExecutionErrorType, ExecutionStatus +from app.domain.execution import DomainExecutionCreate, ExecutionResultDomain _test_logger = structlog.get_logger("test.db.repositories.execution_repository") @@ -37,3 +37,106 @@ async def test_execution_crud_and_query() -> None: # Delete assert await repo.delete_execution(created.execution_id) is True assert await repo.get_execution(created.execution_id) is None + + +async def _create_and_complete( + repo: ExecutionRepository, + user_id: str, + *, + status: ExecutionStatus = ExecutionStatus.COMPLETED, + lang: str = "python", + lang_version: str = "3.11", +) -> str: + """Helper: create an execution then transition it to a terminal status.""" + created = await repo.create_execution(DomainExecutionCreate( + script="x = 1", lang=lang, lang_version=lang_version, user_id=user_id, + )) + await repo.write_terminal_result(ExecutionResultDomain( + execution_id=created.execution_id, + status=status, + exit_code=0 if status == ExecutionStatus.COMPLETED else 1, + stdout="ok", + stderr="", + error_type=ExecutionErrorType.SCRIPT_ERROR if status == ExecutionStatus.FAILED else None, + )) + return created.execution_id + + +@pytest.mark.asyncio +async def test_aggregate_stats_empty() -> None: + """Empty result set returns zero-value stats.""" + repo = ExecutionRepository(logger=_test_logger) + bogus_user = str(uuid4()) + + stats = await repo.aggregate_stats({"user_id": bogus_user}) + + assert stats["total"] == 0 + assert stats["by_status"] == {} + assert stats["by_language"] == {} + assert stats["average_duration_ms"] == 0 + assert stats["success_rate"] == 0 + + +@pytest.mark.asyncio +async def test_aggregate_stats_completed_and_failed() -> None: + """Stats reflect completed vs failed counts, success rate, and duration.""" + repo = ExecutionRepository(logger=_test_logger) + user_id = str(uuid4()) + exec_ids: list[str] = [] + + exec_ids.append(await _create_and_complete(repo, user_id, status=ExecutionStatus.COMPLETED)) + exec_ids.append(await _create_and_complete(repo, user_id, status=ExecutionStatus.COMPLETED)) + exec_ids.append(await _create_and_complete(repo, user_id, status=ExecutionStatus.FAILED)) + + try: + stats = await repo.aggregate_stats({"user_id": user_id}) + + assert stats["total"] == 3 + assert stats["by_status"][ExecutionStatus.COMPLETED] == 2 + assert stats["by_status"][ExecutionStatus.FAILED] == 1 + assert stats["success_rate"] == pytest.approx(2 / 3) + assert stats["average_duration_ms"] >= 0 + assert "python-3.11" in stats["by_language"] + assert stats["by_language"]["python-3.11"] == 3 + finally: + for eid in exec_ids: + await repo.delete_execution(eid) + + +@pytest.mark.asyncio +async def test_aggregate_stats_no_completed() -> None: + """When nothing completed, avg_duration is 0 and success_rate is 0.""" + repo = ExecutionRepository(logger=_test_logger) + user_id = str(uuid4()) + + eid = await _create_and_complete(repo, user_id, status=ExecutionStatus.FAILED) + + try: + stats = await repo.aggregate_stats({"user_id": user_id}) + + assert stats["total"] == 1 + assert stats["average_duration_ms"] == 0 + assert stats["success_rate"] == 0.0 + finally: + await repo.delete_execution(eid) + + +@pytest.mark.asyncio +async def test_aggregate_stats_multiple_languages() -> None: + """by_language groups by lang-lang_version correctly.""" + repo = ExecutionRepository(logger=_test_logger) + user_id = str(uuid4()) + exec_ids: list[str] = [] + + exec_ids.append(await _create_and_complete(repo, user_id, lang="python", lang_version="3.11")) + exec_ids.append(await _create_and_complete(repo, user_id, lang="python", lang_version="3.12")) + exec_ids.append(await _create_and_complete(repo, user_id, lang="python", lang_version="3.12")) + + try: + stats = await repo.aggregate_stats({"user_id": user_id}) + + assert stats["by_language"]["python-3.11"] == 1 + assert stats["by_language"]["python-3.12"] == 2 + finally: + for eid in exec_ids: + await repo.delete_execution(eid) diff --git a/backend/tests/unit/services/saga/test_saga_orchestrator.py b/backend/tests/unit/services/saga/test_saga_orchestrator.py index ceabace9..18a3db7f 100644 --- a/backend/tests/unit/services/saga/test_saga_orchestrator.py +++ b/backend/tests/unit/services/saga/test_saga_orchestrator.py @@ -113,6 +113,7 @@ def __init__(self) -> None: self.enqueued: list[ExecutionRequestedEvent] = [] self._pending: list[tuple[str, ExecutionRequestedEvent]] = [] self.released: list[str] = [] + self._retry_counts: dict[str, int] = {} async def enqueue(self, event: ExecutionRequestedEvent) -> int: self.enqueued.append(event) @@ -131,6 +132,10 @@ async def remove(self, execution_id: str) -> bool: self._pending = [(eid, ev) for eid, ev in self._pending if eid != execution_id] return True + async def increment_retry_count(self, execution_id: str) -> int: + self._retry_counts[execution_id] = self._retry_counts.get(execution_id, 0) + 1 + return self._retry_counts[execution_id] + async def update_priority(self, execution_id: str, new_priority: Any) -> bool: return True @@ -318,3 +323,52 @@ async def test_resolve_completion_releases_slot_when_no_saga_found() -> None: # Slot released even though no saga was found assert "orphan" in fake_queue.released + + +@pytest.mark.asyncio +async def test_max_retries_exceeded_fails_execution() -> None: + """After _MAX_SAGA_START_RETRIES failures, the execution is dropped with FAILED state.""" + fake_repo = _FakeRepo() + fake_queue = _FakeQueue() + fake_repo.fail_on_create = True + orch = _orch(repo=fake_repo, queue=fake_queue) + + event = make_execution_requested_event(execution_id="e1") + + # Simulate repeated failures by pre-setting retry count to threshold - 1 + fake_queue._retry_counts["e1"] = orch._MAX_SAGA_START_RETRIES - 1 + + await orch.handle_execution_requested(event) + + # Retry count reached max — execution should NOT be re-enqueued a second time + # (only the initial enqueue from handle_execution_requested) + assert len(fake_queue.enqueued) == 1 + # Slot was released + assert "e1" in fake_queue.released + + +@pytest.mark.asyncio +async def test_retry_count_increments_across_failures() -> None: + """Each saga start failure increments the retry count via the queue service.""" + fake_repo = _FakeRepo() + fake_queue = _FakeQueue() + fake_repo.fail_on_create = True + orch = _orch(repo=fake_repo, queue=fake_queue) + + event = make_execution_requested_event(execution_id="e1") + await orch.handle_execution_requested(event) + + # First failure: retry count = 1, event re-enqueued + assert fake_queue._retry_counts["e1"] == 1 + assert len(fake_queue.enqueued) == 2 # initial + re-enqueue + + # Trigger second attempt (fake queue has the re-enqueued event) + await orch.try_schedule_from_queue() + assert fake_queue._retry_counts["e1"] == 2 + assert len(fake_queue.enqueued) == 3 # another re-enqueue + + # Trigger third attempt — should exceed max retries (3 >= 3) + await orch.try_schedule_from_queue() + assert fake_queue._retry_counts["e1"] == 3 + # No more re-enqueues after max retries + assert len(fake_queue.enqueued) == 3 diff --git a/backend/tests/unit/services/test_execution_queue.py b/backend/tests/unit/services/test_execution_queue.py index 9e36d874..79fde15d 100644 --- a/backend/tests/unit/services/test_execution_queue.py +++ b/backend/tests/unit/services/test_execution_queue.py @@ -88,6 +88,20 @@ async def test_try_schedule_returns_event(queue_service: ExecutionQueueService, assert len(keys) == 1 + len(PRIORITY_ORDER) +@pytest.mark.asyncio +async def test_try_schedule_event_data_expired(queue_service: ExecutionQueueService, mock_redis: AsyncMock) -> None: + """When the event JSON has expired from Redis, try_schedule cleans the active set and returns None.""" + script = AsyncMock(return_value=[b"e-gone", str(time.time()).encode(), 2]) + mock_redis.register_script = MagicMock(return_value=script) + mock_redis.get = AsyncMock(return_value=None) + queue_service._schedule_script = None + + result = await queue_service.try_schedule(5) + + assert result is None + mock_redis.srem.assert_awaited_once_with(_ACTIVE_KEY, "e-gone") + + @pytest.mark.asyncio async def test_release(queue_service: ExecutionQueueService, mock_redis: AsyncMock) -> None: await queue_service.release("e1") @@ -116,8 +130,8 @@ async def test_update_priority_missing(queue_service: ExecutionQueueService, moc @pytest.mark.asyncio async def test_remove(queue_service: ExecutionQueueService, mock_redis: AsyncMock) -> None: pipe = MagicMock() - # 5 ZREM results + 1 DELETE + 1 SREM - pipe.execute = AsyncMock(return_value=[0, 0, 1, 0, 0, 1, 0]) + # 5 ZREM results + 1 DELETE (event) + 1 DELETE (retry) + 1 SREM + pipe.execute = AsyncMock(return_value=[0, 0, 1, 0, 0, 1, 0, 0]) mock_redis.pipeline.return_value = pipe result = await queue_service.remove("e1") assert result is True @@ -126,7 +140,7 @@ async def test_remove(queue_service: ExecutionQueueService, mock_redis: AsyncMoc @pytest.mark.asyncio async def test_remove_not_found(queue_service: ExecutionQueueService, mock_redis: AsyncMock) -> None: pipe = MagicMock() - pipe.execute = AsyncMock(return_value=[0, 0, 0, 0, 0, 0, 0]) + pipe.execute = AsyncMock(return_value=[0, 0, 0, 0, 0, 0, 0, 0]) mock_redis.pipeline.return_value = pipe result = await queue_service.remove("e1") assert result is False @@ -155,3 +169,15 @@ async def test_get_pending_by_priority(queue_service: ExecutionQueueService, moc assert counts[QueuePriority.BACKGROUND] == 1 assert QueuePriority.CRITICAL not in counts assert QueuePriority.LOW not in counts + + +@pytest.mark.asyncio +async def test_increment_retry_count(queue_service: ExecutionQueueService, mock_redis: AsyncMock) -> None: + mock_redis.incr = AsyncMock(return_value=1) + mock_redis.expire = AsyncMock(return_value=True) + + count = await queue_service.increment_retry_count("e1") + + assert count == 1 + mock_redis.incr.assert_awaited_once_with("exec_queue:retries:e1") + mock_redis.expire.assert_awaited_once() diff --git a/docs/architecture/domain-exceptions.md b/docs/architecture/domain-exceptions.md index 2abe1acb..4a1f8319 100644 --- a/docs/architecture/domain-exceptions.md +++ b/docs/architecture/domain-exceptions.md @@ -60,10 +60,11 @@ DomainError │ └── CSRFValidationError ├── InvalidStateError │ └── SagaInvalidStateError -└── InfrastructureError - ├── EventPublishError - ├── SagaTimeoutError - └── ReplayOperationError +├── InfrastructureError +│ ├── EventPublishError +│ ├── SagaTimeoutError +│ └── ReplayOperationError +└── AccountLockedError ``` ## Exception locations @@ -72,7 +73,7 @@ Domain exceptions live in their respective domain modules: | Module | File | Exceptions | |--------------|-----------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------| -| Base | `app/domain/exceptions.py` | `DomainError`, `NotFoundError`, `ValidationError`, etc. | +| Base | `app/domain/exceptions.py` | `DomainError`, `NotFoundError`, `ValidationError`, `AccountLockedError`, etc. | | Execution | `app/domain/execution/exceptions.py` | `ExecutionNotFoundError`, `RuntimeNotSupportedError`, `EventPublishError` | | Saga | `app/domain/saga/exceptions.py` | `SagaNotFoundError`, `SagaAccessDeniedError`, `SagaInvalidStateError`, `SagaTimeoutError`, `SagaConcurrencyError` | | Notification | `app/domain/notification/exceptions.py` | `NotificationNotFoundError`, `NotificationThrottledError`, `NotificationValidationError` | From 5d6b62d73c073453b65bd34c27ca748d027348d9 Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Sun, 1 Mar 2026 13:19:47 +0100 Subject: [PATCH 7/7] fix: detected issues --- backend/tests/e2e/db/repositories/test_execution_repository.py | 3 ++- backend/tests/unit/services/saga/test_saga_orchestrator.py | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/backend/tests/e2e/db/repositories/test_execution_repository.py b/backend/tests/e2e/db/repositories/test_execution_repository.py index aca51e60..6fc24f27 100644 --- a/backend/tests/e2e/db/repositories/test_execution_repository.py +++ b/backend/tests/e2e/db/repositories/test_execution_repository.py @@ -51,7 +51,7 @@ async def _create_and_complete( created = await repo.create_execution(DomainExecutionCreate( script="x = 1", lang=lang, lang_version=lang_version, user_id=user_id, )) - await repo.write_terminal_result(ExecutionResultDomain( + wrote = await repo.write_terminal_result(ExecutionResultDomain( execution_id=created.execution_id, status=status, exit_code=0 if status == ExecutionStatus.COMPLETED else 1, @@ -59,6 +59,7 @@ async def _create_and_complete( stderr="", error_type=ExecutionErrorType.SCRIPT_ERROR if status == ExecutionStatus.FAILED else None, )) + assert wrote, f"write_terminal_result failed for {created.execution_id}" return created.execution_id diff --git a/backend/tests/unit/services/saga/test_saga_orchestrator.py b/backend/tests/unit/services/saga/test_saga_orchestrator.py index 18a3db7f..7966f2a2 100644 --- a/backend/tests/unit/services/saga/test_saga_orchestrator.py +++ b/backend/tests/unit/services/saga/test_saga_orchestrator.py @@ -327,7 +327,7 @@ async def test_resolve_completion_releases_slot_when_no_saga_found() -> None: @pytest.mark.asyncio async def test_max_retries_exceeded_fails_execution() -> None: - """After _MAX_SAGA_START_RETRIES failures, the execution is dropped with FAILED state.""" + """After _MAX_SAGA_START_RETRIES failures, the execution is not re-enqueued and the slot is released.""" fake_repo = _FakeRepo() fake_queue = _FakeQueue() fake_repo.fail_on_create = True