diff --git a/backend/app/core/providers.py b/backend/app/core/providers.py index a4666420..68b72bf8 100644 --- a/backend/app/core/providers.py +++ b/backend/app/core/providers.py @@ -46,6 +46,7 @@ from app.db.repositories.resource_allocation_repository import ResourceAllocationRepository from app.db.repositories.user_settings_repository import UserSettingsRepository from app.dlq.manager import DLQManager +from app.domain.enums.events import EventType from app.domain.enums.kafka import CONSUMER_GROUP_SUBSCRIPTIONS, GroupId from app.domain.idempotency import KeyStrategy from app.domain.saga.models import SagaConfig @@ -72,7 +73,7 @@ from app.services.pod_monitor.monitor import PodMonitor from app.services.rate_limit_service import RateLimitService from app.services.result_processor.resource_cleaner import ResourceCleaner -from app.services.saga import SagaOrchestrator, create_saga_orchestrator +from app.services.saga import SagaOrchestrator from app.services.saga.saga_service import SagaService from app.services.saved_script_service import SavedScriptService from app.services.sse.redis_bus import SSERedisBus @@ -587,32 +588,6 @@ def _create_default_saga_config() -> SagaConfig: ) -# Standalone factory functions for lifecycle-managed services (eliminates duplication) -async def _provide_saga_orchestrator( - saga_repository: SagaRepository, - kafka_producer: UnifiedProducer, - schema_registry: SchemaRegistryManager, - settings: Settings, - event_store: EventStore, - resource_allocation_repository: ResourceAllocationRepository, - logger: logging.Logger, - event_metrics: EventMetrics, -) -> AsyncIterator[SagaOrchestrator]: - """Shared factory for SagaOrchestrator with lifecycle management.""" - async with create_saga_orchestrator( - saga_repository=saga_repository, - producer=kafka_producer, - schema_registry_manager=schema_registry, - settings=settings, - event_store=event_store, - resource_allocation_repository=resource_allocation_repository, - config=_create_default_saga_config(), - logger=logger, - event_metrics=event_metrics, - ) as orchestrator: - yield orchestrator - - class BusinessServicesProvider(Provider): scope = Scope.REQUEST @@ -864,9 +839,69 @@ async def get_pod_monitor( class SagaOrchestratorProvider(Provider): scope = Scope.APP - def __init__(self) -> None: - super().__init__() - self.provide(_provide_saga_orchestrator) + @provide + async def get_saga_orchestrator( + self, + saga_repository: SagaRepository, + kafka_producer: UnifiedProducer, + schema_registry: SchemaRegistryManager, + settings: Settings, + resource_allocation_repository: ResourceAllocationRepository, + logger: logging.Logger, + event_metrics: EventMetrics, + ) -> AsyncIterator[SagaOrchestrator]: + orchestrator = SagaOrchestrator( + config=_create_default_saga_config(), + saga_repository=saga_repository, + producer=kafka_producer, + resource_allocation_repository=resource_allocation_repository, + logger=logger, + ) + + dispatcher = EventDispatcher(logger=logger) + dispatcher.register_handler(EventType.EXECUTION_REQUESTED, orchestrator.handle_execution_requested) + dispatcher.register_handler(EventType.EXECUTION_COMPLETED, orchestrator.handle_execution_completed) + dispatcher.register_handler(EventType.EXECUTION_FAILED, orchestrator.handle_execution_failed) + dispatcher.register_handler(EventType.EXECUTION_TIMEOUT, orchestrator.handle_execution_timeout) + + consumer_config = ConsumerConfig( + bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS, + group_id=GroupId.SAGA_ORCHESTRATOR, + enable_auto_commit=False, + session_timeout_ms=settings.KAFKA_SESSION_TIMEOUT_MS, + heartbeat_interval_ms=settings.KAFKA_HEARTBEAT_INTERVAL_MS, + max_poll_interval_ms=settings.KAFKA_MAX_POLL_INTERVAL_MS, + request_timeout_ms=settings.KAFKA_REQUEST_TIMEOUT_MS, + ) + + consumer = UnifiedConsumer( + consumer_config, + event_dispatcher=dispatcher, + schema_registry=schema_registry, + settings=settings, + logger=logger, + event_metrics=event_metrics, + ) + + await consumer.start(list(CONSUMER_GROUP_SUBSCRIPTIONS[GroupId.SAGA_ORCHESTRATOR])) + + async def timeout_loop() -> None: + while True: + await asyncio.sleep(30) + try: + await orchestrator.check_timeouts() + except Exception as exc: + logger.error(f"Error checking saga timeouts: {exc}") + + timeout_task = asyncio.create_task(timeout_loop()) + logger.info("Saga orchestrator consumer and timeout checker started") + + try: + yield orchestrator + finally: + timeout_task.cancel() + await consumer.stop() + logger.info("Saga orchestrator stopped") class EventReplayProvider(Provider): diff --git a/backend/app/db/docs/saga.py b/backend/app/db/docs/saga.py index 87b7a62a..c762d6c8 100644 --- a/backend/app/db/docs/saga.py +++ b/backend/app/db/docs/saga.py @@ -37,4 +37,9 @@ class Settings: indexes = [ IndexModel([("state", ASCENDING)], name="idx_saga_state"), IndexModel([("state", ASCENDING), ("created_at", ASCENDING)], name="idx_saga_state_created"), + IndexModel( + [("execution_id", ASCENDING), ("saga_name", ASCENDING)], + unique=True, + name="idx_saga_execution_name_unique", + ), ] diff --git a/backend/app/db/repositories/saga_repository.py b/backend/app/db/repositories/saga_repository.py index 4adb7299..a1d3f3e8 100644 --- a/backend/app/db/repositories/saga_repository.py +++ b/backend/app/db/repositories/saga_repository.py @@ -3,6 +3,7 @@ from beanie.odm.enums import SortDirection from beanie.odm.operators.find import BaseFindOperator +from beanie.odm.queries.update import UpdateResponse from beanie.operators import GT, LT, NE, Eq, In from monggregate import Pipeline, S @@ -41,6 +42,30 @@ async def upsert_saga(self, saga: Saga) -> bool: await doc.save() return existing is not None + async def get_or_create_saga(self, saga: Saga) -> tuple[Saga, bool]: + """Atomically get or create a saga by (execution_id, saga_name). + + Uses MongoDB findOneAndUpdate with $setOnInsert + upsert in a single + atomic round-trip. Returns (saga, created). + """ + insert_doc = SagaDocument(**saga.model_dump()) + insert_data = insert_doc.model_dump() + insert_data.pop("id", None) + insert_data.pop("revision_id", None) + + doc = await SagaDocument.find_one( + SagaDocument.execution_id == saga.execution_id, + SagaDocument.saga_name == saga.saga_name, + ).upsert( + {"$setOnInsert": insert_data}, + on_insert=insert_doc, + response_type=UpdateResponse.NEW_DOCUMENT, + upsert=True, + ) + assert doc is not None + created = doc.saga_id == saga.saga_id + return Saga.model_validate(doc, from_attributes=True), created + async def get_saga_by_execution_and_name(self, execution_id: str, saga_name: str) -> Saga | None: doc = await SagaDocument.find_one( SagaDocument.execution_id == execution_id, diff --git a/backend/app/domain/enums/kafka.py b/backend/app/domain/enums/kafka.py index 1f3892eb..86b24164 100644 --- a/backend/app/domain/enums/kafka.py +++ b/backend/app/domain/enums/kafka.py @@ -125,7 +125,12 @@ class GroupId(StringEnum): EventType.EXECUTION_FAILED, EventType.EXECUTION_TIMEOUT, }, - GroupId.SAGA_ORCHESTRATOR: set(), + GroupId.SAGA_ORCHESTRATOR: { + EventType.EXECUTION_REQUESTED, + EventType.EXECUTION_COMPLETED, + EventType.EXECUTION_FAILED, + EventType.EXECUTION_TIMEOUT, + }, GroupId.WEBSOCKET_GATEWAY: { EventType.EXECUTION_REQUESTED, EventType.EXECUTION_STARTED, diff --git a/backend/app/services/saga/__init__.py b/backend/app/services/saga/__init__.py index e89535ae..de5ad07c 100644 --- a/backend/app/services/saga/__init__.py +++ b/backend/app/services/saga/__init__.py @@ -1,18 +1,14 @@ from app.domain.enums.saga import SagaState from app.domain.saga.models import SagaConfig, SagaInstance -from app.services.saga.base_saga import BaseSaga from app.services.saga.execution_saga import ( AllocateResourcesStep, CreatePodStep, DeletePodCompensation, ExecutionSaga, - MonitorExecutionStep, - QueueExecutionStep, ReleaseResourcesCompensation, - RemoveFromQueueCompensation, ValidateExecutionStep, ) -from app.services.saga.saga_orchestrator import SagaOrchestrator, create_saga_orchestrator +from app.services.saga.saga_orchestrator import SagaOrchestrator from app.services.saga.saga_step import CompensationStep, SagaContext, SagaStep __all__ = [ @@ -23,16 +19,11 @@ "SagaContext", "SagaStep", "CompensationStep", - "BaseSaga", "ExecutionSaga", # Steps and compensations (execution saga) "ValidateExecutionStep", "AllocateResourcesStep", - "QueueExecutionStep", "CreatePodStep", - "MonitorExecutionStep", "ReleaseResourcesCompensation", - "RemoveFromQueueCompensation", "DeletePodCompensation", - "create_saga_orchestrator", ] diff --git a/backend/app/services/saga/base_saga.py b/backend/app/services/saga/base_saga.py deleted file mode 100644 index 6e64a17a..00000000 --- a/backend/app/services/saga/base_saga.py +++ /dev/null @@ -1,52 +0,0 @@ -from abc import ABC, abstractmethod -from typing import Any - -from app.domain.enums.events import EventType -from app.services.saga.saga_step import SagaStep - - -class BaseSaga(ABC): - """Base class for saga implementations. - - All saga implementations should inherit from this class and implement - the required abstract methods to define their workflow. - """ - - @classmethod - @abstractmethod - def get_name(cls) -> str: - """Get the unique name of this saga. - - Returns: - String identifier for this saga type - """ - pass - - @classmethod - @abstractmethod - def get_trigger_events(cls) -> list[EventType]: - """Get event types that trigger this saga. - - Returns: - List of event types that should start this saga - """ - pass - - @abstractmethod - def get_steps(self) -> list[SagaStep[Any]]: - """Get saga steps in execution order. - - Returns: - Ordered list of steps to execute for this saga - """ - pass - - # Optional DI hook: concrete sagas may override to capture runtime deps - def bind_dependencies(self, **_: object) -> None: - """Inject runtime dependencies into the saga instance. - - Default implementation is a no-op; concrete sagas can override to - accept named dependencies (e.g., producer, repositories) and store them - for step construction. This avoids passing opaque context for DI. - """ - return None diff --git a/backend/app/services/saga/execution_saga.py b/backend/app/services/saga/execution_saga.py index 1f426ff4..62dec39b 100644 --- a/backend/app/services/saga/execution_saga.py +++ b/backend/app/services/saga/execution_saga.py @@ -2,364 +2,203 @@ from typing import Any from app.db.repositories.resource_allocation_repository import ResourceAllocationRepository -from app.domain.enums.events import EventType from app.domain.events.typed import CreatePodCommandEvent, DeletePodCommandEvent, EventMetadata, ExecutionRequestedEvent from app.domain.saga import DomainResourceAllocationCreate from app.events.core import UnifiedProducer -from .base_saga import BaseSaga from .saga_step import CompensationStep, SagaContext, SagaStep logger = logging.getLogger(__name__) class ValidateExecutionStep(SagaStep[ExecutionRequestedEvent]): - """Validate execution request""" + """Validate execution request.""" def __init__(self) -> None: super().__init__("validate_execution") async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: - """Validate execution parameters""" - try: - logger.info(f"Validating execution {event.execution_id}") + logger.info(f"Validating execution {event.execution_id}") - # Store execution details in context - context.set("execution_id", event.execution_id) - context.set("language", event.language) - context.set("language_version", event.language_version) - context.set("script", event.script) - context.set("timeout_seconds", event.timeout_seconds) + context.set("execution_id", event.execution_id) + context.set("language", event.language) + context.set("language_version", event.language_version) + context.set("script", event.script) + context.set("timeout_seconds", event.timeout_seconds) - # Validate script size - if len(event.script) > 1024 * 1024: # 1MB limit - raise ValueError("Script size exceeds limit") + if len(event.script) > 1024 * 1024: + raise ValueError("Script size exceeds limit") - # Validate timeout - if event.timeout_seconds is not None and event.timeout_seconds > 300: # 5 minutes max - raise ValueError("Timeout exceeds maximum allowed") + if event.timeout_seconds is not None and event.timeout_seconds > 300: + raise ValueError("Timeout exceeds maximum allowed") - # Additional validations can be added here - - return True - - except Exception as e: - logger.error(f"Validation failed: {e}") - context.set_error(e) - return False + return True def get_compensation(self) -> CompensationStep | None: - """No compensation needed for validation""" return None class AllocateResourcesStep(SagaStep[ExecutionRequestedEvent]): - """Allocate resources for execution""" + """Allocate resources for execution.""" - def __init__(self, alloc_repo: ResourceAllocationRepository | None = None) -> None: + def __init__(self, alloc_repo: ResourceAllocationRepository) -> None: super().__init__("allocate_resources") - self.alloc_repo: ResourceAllocationRepository | None = alloc_repo - - async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: - """Allocate computational resources""" - try: - if self.alloc_repo is None: - raise RuntimeError("ResourceAllocationRepository dependency not injected") - - execution_id = context.get("execution_id") - logger.info(f"Allocating resources for execution {execution_id}") - - # Check resource availability - # Count current allocations - active_count = await self.alloc_repo.count_active(event.language) - - # Simple resource limit check (e.g., max 100 concurrent per language) - if active_count >= 100: - raise ValueError("Resource limit exceeded") - - # Create allocation record via repository - allocation = await self.alloc_repo.create_allocation( - DomainResourceAllocationCreate( - execution_id=execution_id, - language=event.language, - cpu_request=event.cpu_request, - memory_request=event.memory_request, - cpu_limit=event.cpu_limit, - memory_limit=event.memory_limit, - ) - ) - - context.set("allocation_id", allocation.allocation_id) - context.set("resources_allocated", True) - - return True - - except Exception as e: - logger.error(f"Resource allocation failed: {e}") - context.set_error(e) - return False - - def get_compensation(self) -> CompensationStep | None: - """Return compensation to release resources""" - return ReleaseResourcesCompensation(alloc_repo=self.alloc_repo) - - -class QueueExecutionStep(SagaStep[ExecutionRequestedEvent]): - """Queue execution for processing""" - - def __init__(self) -> None: - super().__init__("queue_execution") + self.alloc_repo = alloc_repo async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: - """Queue execution""" - try: - execution_id = context.get("execution_id") - logger.info(f"Queueing execution {execution_id}") + execution_id = context.get("execution_id") + logger.info(f"Allocating resources for execution {execution_id}") - # Since we removed execution.queued, we'll mark the context directly - # The execution is already requested, so we just track that it's ready - context.set("queued", True) - logger.info(f"Execution {execution_id} ready for processing") + active_count = await self.alloc_repo.count_active(event.language) + if active_count >= 100: + raise ValueError("Resource limit exceeded") - return True - - except Exception as e: - logger.error(f"Queue execution failed: {e}") - context.set_error(e) - return False - - def get_compensation(self) -> CompensationStep | None: - """Return compensation to remove from queue""" - return RemoveFromQueueCompensation() - - -class CreatePodStep(SagaStep[ExecutionRequestedEvent]): - """Create Kubernetes pod""" - - def __init__(self, producer: UnifiedProducer | None = None, publish_commands: bool | None = None) -> None: - super().__init__("create_pod") - self.producer: UnifiedProducer | None = producer - self.publish_commands: bool | None = publish_commands - - async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: - """Trigger pod creation by publishing CreatePodCommandEvent""" - try: - execution_id = context.get("execution_id") - saga_id = context.saga_id - logger.info(f"Publishing CreatePodCommandEvent for execution {execution_id}") - - # Allow deployments where coordinator publishes commands to avoid duplicates - publish_commands: bool = bool(self.publish_commands) - if not publish_commands: - logger.info( - f"Skipping CreatePodCommandEvent publish for execution {execution_id} " - f"because publish_commands flag is disabled" - ) - context.set("pod_creation_triggered", False) - return True - - # Create the command event for K8sWorker - create_pod_cmd = CreatePodCommandEvent( - saga_id=saga_id, + allocation = await self.alloc_repo.create_allocation( + DomainResourceAllocationCreate( execution_id=execution_id, - script=event.script, language=event.language, - language_version=event.language_version, - runtime_image=event.runtime_image, - runtime_command=event.runtime_command, - runtime_filename=event.runtime_filename, - timeout_seconds=event.timeout_seconds, - cpu_limit=event.cpu_limit, - memory_limit=event.memory_limit, cpu_request=event.cpu_request, memory_request=event.memory_request, - priority=event.priority, - metadata=EventMetadata( - service_name="saga-orchestrator", - service_version="1.0.0", - user_id=event.metadata.user_id or "system", - ), + cpu_limit=event.cpu_limit, + memory_limit=event.memory_limit, ) + ) - # Publish command to saga_commands topic - if not self.producer: - raise RuntimeError("Producer dependency not injected") - 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}") + context.set("allocation_id", allocation.allocation_id) + context.set("resources_allocated", True) - return True - - except Exception as e: - logger.error(f"Pod creation trigger failed: {e}") - context.set_error(e) - return False + return True def get_compensation(self) -> CompensationStep | None: - """Return compensation to delete pod""" - return DeletePodCompensation(producer=self.producer) + return ReleaseResourcesCompensation(alloc_repo=self.alloc_repo) -class MonitorExecutionStep(SagaStep[ExecutionRequestedEvent]): - """Monitor execution progress""" +class CreatePodStep(SagaStep[ExecutionRequestedEvent]): + """Create Kubernetes pod.""" - def __init__(self) -> None: - super().__init__("monitor_execution") + def __init__(self, producer: UnifiedProducer, publish_commands: bool) -> None: + super().__init__("create_pod") + self.producer = producer + self.publish_commands = publish_commands async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: - """Set up execution monitoring""" - try: - execution_id = context.get("execution_id") - logger.info(f"Setting up monitoring for execution {execution_id}") - - # Monitoring is handled by PodMonitor service - # This step ensures the saga waits for completion - - context.set("monitoring_active", True) + 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" + ) + context.set("pod_creation_triggered", False) return True - except Exception as e: - logger.error(f"Monitor setup failed: {e}") - context.set_error(e) - return False + logger.info(f"Publishing CreatePodCommandEvent for execution {execution_id}") + + create_pod_cmd = CreatePodCommandEvent( + saga_id=context.saga_id, + execution_id=execution_id, + script=event.script, + language=event.language, + language_version=event.language_version, + runtime_image=event.runtime_image, + runtime_command=event.runtime_command, + runtime_filename=event.runtime_filename, + timeout_seconds=event.timeout_seconds, + cpu_limit=event.cpu_limit, + memory_limit=event.memory_limit, + cpu_request=event.cpu_request, + memory_request=event.memory_request, + priority=event.priority, + metadata=EventMetadata( + service_name="saga-orchestrator", + service_version="1.0.0", + user_id=event.metadata.user_id or "system", + ), + ) + + 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}") + + return True def get_compensation(self) -> CompensationStep | None: - """No compensation needed for monitoring""" - return None + return DeletePodCompensation(producer=self.producer) # Compensation Steps class ReleaseResourcesCompensation(CompensationStep): - """Release allocated resources""" + """Release allocated resources.""" - def __init__(self, alloc_repo: ResourceAllocationRepository | None = None) -> None: + def __init__(self, alloc_repo: ResourceAllocationRepository) -> None: super().__init__("release_resources") - self.alloc_repo: ResourceAllocationRepository | None = alloc_repo + self.alloc_repo = alloc_repo async def compensate(self, context: SagaContext) -> bool: - """Release allocated resources""" - try: - if self.alloc_repo is None: - raise RuntimeError("ResourceAllocationRepository dependency not injected") - - allocation_id = context.get("allocation_id") - if not allocation_id: - return True - - logger.info(f"Releasing resources for allocation {allocation_id}") - - await self.alloc_repo.release_allocation(allocation_id) - + allocation_id = context.get("allocation_id") + if not allocation_id: return True - except Exception as e: - logger.error(f"Failed to release resources: {e}") - return False - + logger.info(f"Releasing resources for allocation {allocation_id}") + await self.alloc_repo.release_allocation(allocation_id) -class RemoveFromQueueCompensation(CompensationStep): - """Remove execution from queue""" - - def __init__(self, producer: UnifiedProducer | None = None) -> None: - super().__init__("remove_from_queue") - self.producer: UnifiedProducer | None = producer - - async def compensate(self, context: SagaContext) -> bool: - """Remove from execution queue""" - try: - execution_id = context.get("execution_id") - if not execution_id or not context.get("queued"): - return True - - logger.info(f"Removing execution {execution_id} from queue") - - # In a real implementation, this would remove from the actual queue - # For now, we'll publish a cancellation event - - return True - - except Exception as e: - logger.error(f"Failed to remove from queue: {e}") - return False + return True class DeletePodCompensation(CompensationStep): - """Delete created pod""" + """Delete created pod.""" - def __init__(self, producer: UnifiedProducer | None = None) -> None: + def __init__(self, producer: UnifiedProducer) -> None: super().__init__("delete_pod") - self.producer: UnifiedProducer | None = producer + self.producer = producer async def compensate(self, context: SagaContext) -> bool: - """Delete Kubernetes pod""" - try: - execution_id = context.get("execution_id") - if not execution_id or not context.get("pod_creation_triggered"): - return True - - saga_id = context.saga_id - logger.info(f"Publishing DeletePodCommandEvent for execution {execution_id}") - - if not self.producer: - raise RuntimeError("Producer dependency not injected") + execution_id = context.get("execution_id") + if not execution_id or not context.get("pod_creation_triggered"): + return True - # Publish DeletePodCommandEvent for K8sWorker - delete_pod_cmd = DeletePodCommandEvent( - saga_id=saga_id, - execution_id=execution_id, - reason="Saga compensation due to failure", - metadata=EventMetadata( - service_name="saga-orchestrator", service_version="1.0.0", user_id=context.get("user_id", "system") - ), - ) + logger.info(f"Publishing DeletePodCommandEvent for execution {execution_id}") - await self.producer.produce(event_to_produce=delete_pod_cmd, key=execution_id) + delete_pod_cmd = DeletePodCommandEvent( + saga_id=context.saga_id, + execution_id=execution_id, + reason="Saga compensation due to failure", + metadata=EventMetadata( + service_name="saga-orchestrator", + service_version="1.0.0", + user_id=context.get("user_id", "system"), + ), + ) - logger.info(f"DeletePodCommandEvent published for {execution_id}") - return True + await self.producer.produce(event_to_produce=delete_pod_cmd, key=execution_id) - except Exception as e: - logger.error(f"Failed to trigger pod deletion: {e}") - return False + logger.info(f"DeletePodCommandEvent published for {execution_id}") + return True -class ExecutionSaga(BaseSaga): - """Saga for managing execution lifecycle""" +class ExecutionSaga: + """Saga for managing execution lifecycle.""" @classmethod def get_name(cls) -> str: - """Get saga name""" return "execution_saga" - @classmethod - def get_trigger_events(cls) -> list[EventType]: - """Get events that trigger this saga""" - return [EventType.EXECUTION_REQUESTED] + def bind_dependencies( + self, + producer: UnifiedProducer, + alloc_repo: ResourceAllocationRepository, + publish_commands: bool, + ) -> None: + self._producer = producer + self._alloc_repo = alloc_repo + self._publish_commands = publish_commands def get_steps(self) -> list[SagaStep[Any]]: - """Get saga steps in order""" - alloc_repo = getattr(self, "_alloc_repo", None) - producer = getattr(self, "_producer", None) - publish_commands = bool(getattr(self, "_publish_commands", False)) return [ ValidateExecutionStep(), - AllocateResourcesStep(alloc_repo=alloc_repo), - QueueExecutionStep(), - CreatePodStep(producer=producer, publish_commands=publish_commands), - MonitorExecutionStep(), + AllocateResourcesStep(alloc_repo=self._alloc_repo), + CreatePodStep(producer=self._producer, publish_commands=self._publish_commands), ] - - def bind_dependencies(self, **kwargs: object) -> None: - producer = kwargs.get("producer") - alloc_repo = kwargs.get("alloc_repo") - publish_commands = kwargs.get("publish_commands") - if isinstance(producer, UnifiedProducer): - self._producer = producer - if isinstance(alloc_repo, ResourceAllocationRepository): - self._alloc_repo = alloc_repo - if isinstance(publish_commands, bool): - self._publish_commands = publish_commands diff --git a/backend/app/services/saga/saga_orchestrator.py b/backend/app/services/saga/saga_orchestrator.py index d1f3d9b8..04849d96 100644 --- a/backend/app/services/saga/saga_orchestrator.py +++ b/backend/app/services/saga/saga_orchestrator.py @@ -5,309 +5,147 @@ from opentelemetry.trace import SpanKind -from app.core.lifecycle import LifecycleEnabled -from app.core.metrics import EventMetrics from app.core.tracing import EventAttributes from app.core.tracing.utils import get_tracer from app.db.repositories.resource_allocation_repository import ResourceAllocationRepository from app.db.repositories.saga_repository import SagaRepository -from app.domain.enums.events import EventType from app.domain.enums.saga import SagaState -from app.domain.events.typed import DomainEvent, EventMetadata, SagaCancelledEvent +from app.domain.events.typed import ( + DomainEvent, + EventMetadata, + ExecutionCompletedEvent, + ExecutionFailedEvent, + ExecutionRequestedEvent, + ExecutionTimeoutEvent, + SagaCancelledEvent, +) from app.domain.saga.models import Saga, SagaConfig -from app.events.core import ConsumerConfig, EventDispatcher, UnifiedConsumer, UnifiedProducer -from app.events.event_store import EventStore -from app.events.schema.schema_registry import SchemaRegistryManager -from app.infrastructure.kafka.mappings import get_topic_for_event -from app.settings import Settings +from app.events.core import UnifiedProducer -from .base_saga import BaseSaga from .execution_saga import ExecutionSaga from .saga_step import SagaContext +_SAGA_NAME = ExecutionSaga.get_name() -class SagaOrchestrator(LifecycleEnabled): - """Orchestrates saga execution and compensation""" + +class SagaOrchestrator: + """Orchestrates saga execution and compensation.""" def __init__( self, config: SagaConfig, saga_repository: SagaRepository, producer: UnifiedProducer, - schema_registry_manager: SchemaRegistryManager, - settings: Settings, - event_store: EventStore, resource_allocation_repository: ResourceAllocationRepository, logger: logging.Logger, - event_metrics: EventMetrics, ): - super().__init__() self.config = config - self._sagas: dict[str, type[BaseSaga]] = {} - self._running_instances: dict[str, Saga] = {} - self._consumer: UnifiedConsumer | None = None self._producer = producer - self._schema_registry_manager = schema_registry_manager - self._settings = settings - self._event_store = event_store self._repo: SagaRepository = saga_repository self._alloc_repo: ResourceAllocationRepository = resource_allocation_repository - self._tasks: list[asyncio.Task[None]] = [] self.logger = logger - self._event_metrics = event_metrics - - def register_saga(self, saga_class: type[BaseSaga]) -> None: - self._sagas[saga_class.get_name()] = saga_class - self.logger.info(f"Registered saga: {saga_class.get_name()}") - - def _register_default_sagas(self) -> None: - self.register_saga(ExecutionSaga) - self.logger.info("Registered default sagas") - - async def _on_start(self) -> None: - """Start the saga orchestrator.""" - self.logger.info(f"Starting saga orchestrator: {self.config.name}") - - self._register_default_sagas() - - await self._start_consumer() - - timeout_task = asyncio.create_task(self._check_timeouts()) - self._tasks.append(timeout_task) - - self.logger.info("Saga orchestrator started") - - async def _on_stop(self) -> None: - """Stop the saga orchestrator.""" - self.logger.info("Stopping saga orchestrator...") - if self._consumer: - await self._consumer.stop() - - for task in self._tasks: - if not task.done(): - task.cancel() - - if self._tasks: - await asyncio.gather(*self._tasks, return_exceptions=True) - - self.logger.info("Saga orchestrator stopped") - - async def _start_consumer(self) -> None: - self.logger.info(f"Registered sagas: {list(self._sagas.keys())}") - topics = set() - event_types_to_register = set() - - for saga_class in self._sagas.values(): - trigger_event_types = saga_class.get_trigger_events() - self.logger.info(f"Saga {saga_class.get_name()} triggers on event types: {trigger_event_types}") - - # Convert event types to topics for subscription - for event_type in trigger_event_types: - topic = get_topic_for_event(event_type) - topics.add(topic) - event_types_to_register.add(event_type) - self.logger.debug(f"Event type {event_type} maps to topic {topic}") - - # Also register handlers for completion events so execution sagas can complete - completion_event_types = { - EventType.EXECUTION_COMPLETED, - EventType.EXECUTION_FAILED, - EventType.EXECUTION_TIMEOUT, - } - for event_type in completion_event_types: - topic = get_topic_for_event(event_type) - topics.add(topic) - event_types_to_register.add(event_type) - self.logger.debug(f"Completion event type {event_type} maps to topic {topic}") - - if not topics: - self.logger.warning("No trigger events found in registered sagas") - return - - consumer_config = ConsumerConfig( - bootstrap_servers=self._settings.KAFKA_BOOTSTRAP_SERVERS, - group_id=f"saga-{self.config.name}", - enable_auto_commit=False, - session_timeout_ms=self._settings.KAFKA_SESSION_TIMEOUT_MS, - heartbeat_interval_ms=self._settings.KAFKA_HEARTBEAT_INTERVAL_MS, - max_poll_interval_ms=self._settings.KAFKA_MAX_POLL_INTERVAL_MS, - request_timeout_ms=self._settings.KAFKA_REQUEST_TIMEOUT_MS, + async def handle_execution_requested(self, event: DomainEvent) -> None: + """Handle EXECUTION_REQUESTED — starts a new saga.""" + if not isinstance(event, ExecutionRequestedEvent): + raise TypeError(f"Expected ExecutionRequestedEvent, got {type(event).__name__}") + await self._start_saga(event) + + async def handle_execution_completed(self, event: DomainEvent) -> None: + """Handle EXECUTION_COMPLETED — marks saga as completed.""" + if not isinstance(event, ExecutionCompletedEvent): + raise TypeError(f"Expected ExecutionCompletedEvent, got {type(event).__name__}") + await self._resolve_completion(event.execution_id, SagaState.COMPLETED) + + async def handle_execution_failed(self, event: DomainEvent) -> None: + """Handle EXECUTION_FAILED — marks saga as failed.""" + if not isinstance(event, ExecutionFailedEvent): + raise TypeError(f"Expected ExecutionFailedEvent, got {type(event).__name__}") + await self._resolve_completion( + event.execution_id, SagaState.FAILED, event.error_message or f"Execution {event.event_type}" ) - dispatcher = EventDispatcher(logger=self.logger) - for event_type in event_types_to_register: - dispatcher.register_handler(event_type, self._handle_event) - self.logger.info(f"Registered handler for event type: {event_type}") - - self._consumer = UnifiedConsumer( - config=consumer_config, - event_dispatcher=dispatcher, - schema_registry=self._schema_registry_manager, - settings=self._settings, - logger=self.logger, - event_metrics=self._event_metrics, + async def handle_execution_timeout(self, event: DomainEvent) -> None: + """Handle EXECUTION_TIMEOUT — marks saga as timed out.""" + if not isinstance(event, ExecutionTimeoutEvent): + raise TypeError(f"Expected ExecutionTimeoutEvent, got {type(event).__name__}") + await self._resolve_completion( + event.execution_id, SagaState.TIMEOUT, f"Execution timed out after {event.timeout_seconds} seconds" ) - await self._consumer.start(list(topics)) - - self.logger.info(f"Saga consumer started for topics: {topics}") - - async def _handle_event(self, event: DomainEvent) -> None: - """Handle incoming event""" - self.logger.info(f"Saga orchestrator handling event: type={event.event_type}, id={event.event_id}") - try: - # Check if this is a completion event that should update an existing saga - completion_events = { - EventType.EXECUTION_COMPLETED, - EventType.EXECUTION_FAILED, - EventType.EXECUTION_TIMEOUT, - } - if event.event_type in completion_events: - await self._handle_completion_event(event) - return - - # Check if this event should trigger a new saga - saga_triggered = False - for saga_name, saga_class in self._sagas.items(): - self.logger.debug(f"Checking if {saga_name} should be triggered by {event.event_type}") - if self._should_trigger_saga(saga_class, event): - self.logger.info(f"Event {event.event_type} triggers saga {saga_name}") - saga_triggered = True - saga_id = await self._start_saga(saga_name, event) - if not saga_id: - raise RuntimeError(f"Failed to create saga {saga_name} for event {event.event_id}") - - if not saga_triggered: - self.logger.debug(f"Event {event.event_type} did not trigger any saga") - - except Exception as e: - self.logger.error(f"Error handling event {event.event_id}: {e}", exc_info=True) - raise - - async def _handle_completion_event(self, event: DomainEvent) -> None: - """Handle execution completion events to update saga state.""" - execution_id = getattr(event, "execution_id", None) - if not execution_id: - self.logger.warning(f"Completion event {event.event_type} has no execution_id") - return - - # Find the execution saga specifically (not other saga types) - saga = await self._repo.get_saga_by_execution_and_name(execution_id, ExecutionSaga.get_name()) + async def _resolve_completion( + self, execution_id: str, state: SagaState, error_message: str | None = None + ) -> None: + """Look up the active saga for an execution and transition it to a terminal state.""" + saga = await self._repo.get_saga_by_execution_and_name(execution_id, _SAGA_NAME) if not saga: self.logger.debug(f"No execution_saga found for execution {execution_id}") return - # Only update if saga is still in a running state if saga.state not in (SagaState.RUNNING, SagaState.CREATED): self.logger.debug(f"Saga {saga.saga_id} already in terminal state {saga.state}") return - # Update saga state based on completion event type - if event.event_type == EventType.EXECUTION_COMPLETED: - self.logger.info(f"Marking saga {saga.saga_id} as COMPLETED due to execution completion") - saga.state = SagaState.COMPLETED - saga.completed_at = datetime.now(UTC) - elif event.event_type == EventType.EXECUTION_TIMEOUT: - timeout_seconds = getattr(event, "timeout_seconds", None) - self.logger.info(f"Marking saga {saga.saga_id} as TIMEOUT after {timeout_seconds}s") - saga.state = SagaState.TIMEOUT - saga.error_message = f"Execution timed out after {timeout_seconds} seconds" - saga.completed_at = datetime.now(UTC) - else: - # EXECUTION_FAILED - error_msg = getattr(event, "error_message", None) or f"Execution {event.event_type}" - self.logger.info(f"Marking saga {saga.saga_id} as FAILED: {error_msg}") - saga.state = SagaState.FAILED - saga.error_message = error_msg - saga.completed_at = datetime.now(UTC) - + self.logger.info(f"Marking saga {saga.saga_id} as {state}") + saga.state = state + saga.error_message = error_message + saga.completed_at = datetime.now(UTC) await self._save_saga(saga) - self._running_instances.pop(saga.saga_id, None) - - def _should_trigger_saga(self, saga_class: type[BaseSaga], event: DomainEvent) -> bool: - trigger_event_types = saga_class.get_trigger_events() - should_trigger = event.event_type in trigger_event_types - self.logger.debug( - f"Saga {saga_class.get_name()} triggers on {trigger_event_types}, " - f"event is {event.event_type}, should trigger: {should_trigger}" - ) - return should_trigger - - async def _start_saga(self, saga_name: str, trigger_event: DomainEvent) -> str | None: - """Start a new saga instance""" - self.logger.info(f"Starting saga {saga_name} for event {trigger_event.event_type}") - saga_class = self._sagas.get(saga_name) - if not saga_class: - raise ValueError(f"Unknown saga: {saga_name}") - - execution_id = getattr(trigger_event, "execution_id", None) - self.logger.debug(f"Extracted execution_id={execution_id} from event") - if not execution_id: - self.logger.warning(f"Could not extract execution ID from event: {trigger_event}") - return None - - existing = await self._repo.get_saga_by_execution_and_name(execution_id, saga_name) - if existing: - self.logger.info(f"Saga {saga_name} already exists for execution {execution_id}") - saga_id: str = existing.saga_id - return saga_id - - instance = Saga( + + async def _start_saga(self, trigger_event: ExecutionRequestedEvent) -> str: + """Start a new saga instance.""" + execution_id = trigger_event.execution_id + self.logger.info(f"Starting saga {_SAGA_NAME} for execution {execution_id}") + + candidate = Saga( saga_id=str(uuid4()), - saga_name=saga_name, + saga_name=_SAGA_NAME, execution_id=execution_id, state=SagaState.RUNNING, ) - await self._save_saga(instance) - self._running_instances[instance.saga_id] = instance + instance, created = await self._repo.get_or_create_saga(candidate) + if not created: + self.logger.info(f"Saga {_SAGA_NAME} already exists for execution {execution_id}") + return instance.saga_id - self.logger.info(f"Started saga {saga_name} (ID: {instance.saga_id}) for execution {execution_id}") - - saga = saga_class() - # Inject runtime dependencies explicitly (no DI via context) - try: - saga.bind_dependencies( - producer=self._producer, - alloc_repo=self._alloc_repo, - publish_commands=bool(getattr(self.config, "publish_commands", False)), - ) - except Exception: - # Back-compat: if saga doesn't support binding, it will fallback to context where needed - pass + self.logger.info(f"Started saga {_SAGA_NAME} (ID: {instance.saga_id}) for execution {execution_id}") + saga = self._create_saga_instance() context = SagaContext(instance.saga_id, execution_id) asyncio.create_task(self._execute_saga(saga, instance, context, trigger_event)) return instance.saga_id + def _create_saga_instance(self) -> ExecutionSaga: + """Create and bind an ExecutionSaga instance.""" + saga = ExecutionSaga() + saga.bind_dependencies( + producer=self._producer, + alloc_repo=self._alloc_repo, + publish_commands=self.config.publish_commands, + ) + return saga + async def _execute_saga( self, - saga: BaseSaga, + saga: ExecutionSaga, instance: Saga, context: SagaContext, trigger_event: DomainEvent, ) -> None: - """Execute saga steps""" + """Execute saga steps.""" tracer = get_tracer() try: - # Get saga steps steps = saga.get_steps() - # Execute each step for step in steps: - if not self.is_running: - break - - # Update current step instance.current_step = step.name await self._save_saga(instance) self.logger.info(f"Executing saga step: {step.name} for saga {instance.saga_id}") - # Execute step within a span with tracer.start_as_current_span( name="saga.step", kind=SpanKind.INTERNAL, @@ -322,8 +160,6 @@ async def _execute_saga( if success: instance.completed_steps.append(step.name) - - # Persist only safe, public context (no ephemeral objects) instance.context_data = context.to_public_dict() await self._save_saga(instance) @@ -331,7 +167,6 @@ async def _execute_saga( if compensation: context.add_compensation(compensation) else: - # Step failed, start compensation self.logger.error(f"Saga step {step.name} failed for saga {instance.saga_id}") if self.config.enable_compensation: @@ -341,12 +176,7 @@ async def _execute_saga( return - # All steps completed successfully - # Execution saga waits for external completion events (EXECUTION_COMPLETED/FAILED) - if instance.saga_name == ExecutionSaga.get_name(): - self.logger.info(f"Saga {instance.saga_id} steps done, waiting for execution completion event") - else: - await self._complete_saga(instance) + self.logger.info(f"Saga {instance.saga_id} steps done, waiting for execution completion event") except Exception as e: self.logger.error(f"Error executing saga {instance.saga_id}: {e}", exc_info=True) @@ -357,15 +187,13 @@ async def _execute_saga( await self._fail_saga(instance, str(e)) async def _compensate_saga(self, instance: Saga, context: SagaContext) -> None: - """Execute compensation steps""" + """Execute compensation steps.""" self.logger.info(f"Starting compensation for saga {instance.saga_id}") - # Only update state if not already cancelled if instance.state != SagaState.CANCELLED: instance.state = SagaState.COMPENSATING await self._save_saga(instance) - # Execute compensations in reverse order for compensation in reversed(context.compensations): try: self.logger.info(f"Executing compensation: {compensation.name} for saga {instance.saga_id}") @@ -380,98 +208,54 @@ async def _compensate_saga(self, instance: Saga, context: SagaContext) -> None: except Exception as e: self.logger.error(f"Error in compensation {compensation.name}: {e}", exc_info=True) - # Mark saga as failed or keep as cancelled if instance.state == SagaState.CANCELLED: - # Keep cancelled state but update compensated steps instance.updated_at = datetime.now(UTC) await self._save_saga(instance) self.logger.info(f"Saga {instance.saga_id} compensation completed after cancellation") else: - # Mark as failed for non-cancelled compensations await self._fail_saga(instance, "Saga compensated due to failure") - async def _complete_saga(self, instance: Saga) -> None: - """Mark saga as completed""" - instance.state = SagaState.COMPLETED - instance.completed_at = datetime.now(UTC) - await self._save_saga(instance) - - # Remove from running instances - self._running_instances.pop(instance.saga_id, None) - - self.logger.info(f"Saga {instance.saga_id} completed successfully") - async def _fail_saga(self, instance: Saga, error_message: str) -> None: - """Mark saga as failed""" + """Mark saga as failed.""" instance.state = SagaState.FAILED instance.error_message = error_message instance.completed_at = datetime.now(UTC) await self._save_saga(instance) - - # Remove from running instances - self._running_instances.pop(instance.saga_id, None) - self.logger.error(f"Saga {instance.saga_id} failed: {error_message}") - async def _check_timeouts(self) -> None: - """Check for saga timeouts""" - while self.is_running: - try: - # Check every 30 seconds - await asyncio.sleep(30) - - cutoff_time = datetime.now(UTC) - timedelta(seconds=self.config.timeout_seconds) - - timed_out = await self._repo.find_timed_out_sagas(cutoff_time) - - for instance in timed_out: - self.logger.warning(f"Saga {instance.saga_id} timed out") - - instance.state = SagaState.TIMEOUT - instance.error_message = f"Saga timed out after {self.config.timeout_seconds} seconds" - instance.completed_at = datetime.now(UTC) - - await self._save_saga(instance) - self._running_instances.pop(instance.saga_id, None) - - except Exception as e: - self.logger.error(f"Error checking timeouts: {e}") + async def check_timeouts(self) -> None: + """Check for timed-out sagas and mark them. Single invocation.""" + cutoff_time = datetime.now(UTC) - timedelta(seconds=self.config.timeout_seconds) + timed_out = await self._repo.find_timed_out_sagas(cutoff_time) + for instance in timed_out: + self.logger.warning(f"Saga {instance.saga_id} timed out") + instance.state = SagaState.TIMEOUT + instance.error_message = f"Saga timed out after {self.config.timeout_seconds} seconds" + instance.completed_at = datetime.now(UTC) + await self._save_saga(instance) async def _save_saga(self, instance: Saga) -> None: - """Persist saga through repository""" + """Persist saga through repository.""" instance.updated_at = datetime.now(UTC) await self._repo.upsert_saga(instance) async def get_saga_status(self, saga_id: str) -> Saga | None: - """Get saga instance status""" - # Check memory first - if saga_id in self._running_instances: - return self._running_instances[saga_id] - + """Get saga instance status.""" return await self._repo.get_saga(saga_id) async def get_execution_sagas(self, execution_id: str) -> list[Saga]: - """Get all sagas for an execution, sorted by created_at descending (newest first)""" + """Get all sagas for an execution, sorted by created_at descending (newest first).""" result = await self._repo.get_sagas_by_execution(execution_id) return result.sagas async def cancel_saga(self, saga_id: str) -> bool: - """Cancel a running saga and trigger compensation. - - Args: - saga_id: The ID of the saga to cancel - - Returns: - True if cancelled successfully, False otherwise - """ + """Cancel a running saga and trigger compensation.""" try: - # Get saga instance saga_instance = await self.get_saga_status(saga_id) if not saga_instance: self.logger.error("Saga not found", extra={"saga_id": saga_id}) return False - # Check if saga can be cancelled if saga_instance.state not in [SagaState.RUNNING, SagaState.CREATED]: self.logger.warning( "Cannot cancel saga in current state. Only RUNNING or CREATED sagas can be cancelled.", @@ -479,12 +263,10 @@ async def cancel_saga(self, saga_id: str) -> bool: ) return False - # Update state to CANCELLED saga_instance.state = SagaState.CANCELLED saga_instance.error_message = "Saga cancelled by user request" saga_instance.completed_at = datetime.now(UTC) - # Log cancellation with user context if available user_id = saga_instance.context_data.get("user_id") self.logger.info( "Saga cancellation initiated", @@ -495,52 +277,26 @@ async def cancel_saga(self, saga_id: str) -> bool: }, ) - # Save state await self._save_saga(saga_instance) - # Remove from running instances - self._running_instances.pop(saga_id, None) - - # Publish cancellation event if self._producer and self.config.store_events: await self._publish_saga_cancelled_event(saga_instance) - # Trigger compensation if saga was running and has completed steps if saga_instance.completed_steps and self.config.enable_compensation: - # Get saga class - saga_class = self._sagas.get(saga_instance.saga_name) - if saga_class: - # Create saga instance and context - saga = saga_class() - try: - saga.bind_dependencies( - producer=self._producer, - alloc_repo=self._alloc_repo, - publish_commands=bool(getattr(self.config, "publish_commands", False)), - ) - except Exception: - pass - context = SagaContext(saga_instance.saga_id, saga_instance.execution_id) - - # Restore context data - for key, value in saga_instance.context_data.items(): - context.set(key, value) - - # Get steps and build compensation list - steps = saga.get_steps() - for step in steps: - if step.name in saga_instance.completed_steps: - compensation = step.get_compensation() - if compensation: - context.add_compensation(compensation) - - # Execute compensation - await self._compensate_saga(saga_instance, context) - else: - self.logger.error( - "Saga class not found for compensation", - extra={"saga_name": saga_instance.saga_name, "saga_id": saga_id}, - ) + saga = self._create_saga_instance() + context = SagaContext(saga_instance.saga_id, saga_instance.execution_id) + + for key, value in saga_instance.context_data.items(): + context.set(key, value) + + steps = saga.get_steps() + for step in steps: + if step.name in saga_instance.completed_steps: + compensation = step.get_compensation() + if compensation: + context.add_compensation(compensation) + + await self._compensate_saga(saga_instance, context) self.logger.info("Saga cancelled successfully", extra={"saga_id": saga_id}) return True @@ -554,11 +310,7 @@ async def cancel_saga(self, saga_id: str) -> bool: return False async def _publish_saga_cancelled_event(self, saga_instance: Saga) -> None: - """Publish saga cancelled event. - - Args: - saga_instance: The cancelled saga instance - """ + """Publish saga cancelled event.""" try: cancelled_by = saga_instance.context_data.get("user_id") if saga_instance.context_data else None metadata = EventMetadata( @@ -586,43 +338,3 @@ async def _publish_saga_cancelled_event(self, saga_instance: Saga) -> None: except Exception as e: self.logger.error(f"Failed to publish saga cancellation event: {e}") - - -def create_saga_orchestrator( - saga_repository: SagaRepository, - producer: UnifiedProducer, - schema_registry_manager: SchemaRegistryManager, - settings: Settings, - event_store: EventStore, - resource_allocation_repository: ResourceAllocationRepository, - config: SagaConfig, - logger: logging.Logger, - event_metrics: EventMetrics, -) -> SagaOrchestrator: - """Factory function to create a saga orchestrator. - - Args: - saga_repository: Repository for saga persistence - producer: Kafka producer instance - schema_registry_manager: Schema registry manager for event serialization - settings: Application settings - event_store: Event store instance for event sourcing - resource_allocation_repository: Repository for resource allocations - config: Saga configuration - logger: Logger instance - event_metrics: Event metrics for tracking Kafka consumption - - Returns: - A new saga orchestrator instance - """ - return SagaOrchestrator( - config, - saga_repository=saga_repository, - producer=producer, - schema_registry_manager=schema_registry_manager, - settings=settings, - event_store=event_store, - resource_allocation_repository=resource_allocation_repository, - logger=logger, - event_metrics=event_metrics, - ) diff --git a/backend/app/services/saga/saga_service.py b/backend/app/services/saga/saga_service.py index 40297155..12cba068 100644 --- a/backend/app/services/saga/saga_service.py +++ b/backend/app/services/saga/saga_service.py @@ -167,24 +167,3 @@ async def get_saga_statistics(self, user: User, include_all: bool = False) -> di return await self.saga_repo.get_saga_statistics(saga_filter) - async def get_saga_status_from_orchestrator(self, saga_id: str, user: User) -> Saga | None: - """Get saga status from orchestrator with fallback to database.""" - self.logger.debug("Getting live saga status", extra={"saga_id": saga_id}) - - # Try orchestrator first for live status - saga = await self.orchestrator.get_saga_status(saga_id) - if saga: - # Check access - if not await self.check_execution_access(saga.execution_id, user): - self.logger.warning( - "Access denied to live saga", - extra={"user_id": user.user_id, "saga_id": saga_id, "execution_id": saga.execution_id}, - ) - raise SagaAccessDeniedError(saga_id, user.user_id) - - self.logger.debug("Retrieved live status for saga", extra={"saga_id": saga_id}) - return saga - - # Fall back to repository - self.logger.debug("No live status found for saga, checking database", extra={"saga_id": saga_id}) - return await self.get_saga_with_access_check(saga_id, user) diff --git a/backend/app/services/saga/saga_step.py b/backend/app/services/saga/saga_step.py index 87f7f4b9..81d65e1e 100644 --- a/backend/app/services/saga/saga_step.py +++ b/backend/app/services/saga/saga_step.py @@ -1,4 +1,3 @@ -import logging from abc import ABC, abstractmethod from typing import Any, Generic, TypeVar @@ -6,8 +5,6 @@ from app.domain.events.typed import DomainEvent -logger = logging.getLogger(__name__) - T = TypeVar("T", bound=DomainEvent) @@ -18,10 +15,7 @@ def __init__(self, saga_id: str, execution_id: str): self.saga_id = saga_id self.execution_id = execution_id self.data: dict[str, Any] = {} - self.events: list[DomainEvent] = [] self.compensations: list[CompensationStep] = [] - self.current_step: str | None = None - self.error: Exception | None = None def set(self, key: str, value: Any) -> None: """Set context data""" @@ -31,18 +25,10 @@ def get(self, key: str, default: Any = None) -> Any: """Get context data""" return self.data.get(key, default) - def add_event(self, event: DomainEvent) -> None: - """Add event to context""" - self.events.append(event) - def add_compensation(self, compensation: "CompensationStep") -> None: """Add compensation step""" self.compensations.append(compensation) - def set_error(self, error: Exception) -> None: - """Set error in context""" - self.error = error - def to_public_dict(self) -> dict[str, Any]: """Return a safe, persistable snapshot of context data. @@ -92,10 +78,6 @@ def get_compensation(self) -> "CompensationStep | None": """Get compensation step for this action""" pass - async def can_execute(self, context: SagaContext, event: T) -> bool: - """Check if step can be executed""" - return True - def __str__(self) -> str: return f"SagaStep({self.name})" 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 bcd517f8..f0349f12 100644 --- a/backend/tests/unit/services/saga/test_execution_saga_steps.py +++ b/backend/tests/unit/services/saga/test_execution_saga_steps.py @@ -8,8 +8,6 @@ CreatePodStep, DeletePodCompensation, ExecutionSaga, - MonitorExecutionStep, - QueueExecutionStep, ReleaseResourcesCompensation, ValidateExecutionStep, ) @@ -30,16 +28,16 @@ async def test_validate_execution_step_success_and_failures() -> None: ok = await ValidateExecutionStep().execute(ctx, _req()) assert ok is True and ctx.get("execution_id") == "e1" - # Timeout too large + # Timeout too large → raises ctx2 = SagaContext("s1", "e1") - ok2 = await ValidateExecutionStep().execute(ctx2, _req(timeout=301)) - assert ok2 is False and ctx2.error is not None + with pytest.raises(ValueError, match="Timeout exceeds maximum"): + await ValidateExecutionStep().execute(ctx2, _req(timeout=301)) - # Script too big + # Script too big → raises ctx3 = SagaContext("s1", "e1") big = "x" * (1024 * 1024 + 1) - ok3 = await ValidateExecutionStep().execute(ctx3, _req(script=big)) - assert ok3 is False and ctx3.error is not None + with pytest.raises(ValueError, match="Script size exceeds limit"): + await ValidateExecutionStep().execute(ctx3, _req(script=big)) class _FakeAllocRepo(ResourceAllocationRepository): @@ -76,37 +74,11 @@ async def test_allocate_resources_step_paths() -> None: ok = await AllocateResourcesStep(alloc_repo=_FakeAllocRepo(active=0, alloc_id="alloc-1")).execute(ctx, _req()) assert ok is True and ctx.get("resources_allocated") is True and ctx.get("allocation_id") == "alloc-1" - # Limit exceeded + # Limit exceeded → raises ctx2 = SagaContext("s2", "e2") ctx2.set("execution_id", "e2") - ok2 = await AllocateResourcesStep(alloc_repo=_FakeAllocRepo(active=100)).execute(ctx2, _req()) - assert ok2 is False - - # Missing repo - ctx3 = SagaContext("s3", "e3") - ctx3.set("execution_id", "e3") - ok3 = await AllocateResourcesStep(alloc_repo=None).execute(ctx3, _req()) - assert ok3 is False - - -@pytest.mark.asyncio -async def test_queue_and_monitor_steps() -> None: - ctx = SagaContext("s1", "e1") - ctx.set("execution_id", "e1") - assert await QueueExecutionStep().execute(ctx, _req()) is True - assert ctx.get("queued") is True - - assert await MonitorExecutionStep().execute(ctx, _req()) is True - assert ctx.get("monitoring_active") is True - - # Force exceptions to exercise except paths - class _BadCtx(SagaContext): - def set(self, key: str, value: object) -> None: - raise RuntimeError("boom") - - bad = _BadCtx("s", "e") - assert await QueueExecutionStep().execute(bad, _req()) is False - assert await MonitorExecutionStep().execute(bad, _req()) is False + with pytest.raises(ValueError, match="Resource limit exceeded"): + await AllocateResourcesStep(alloc_repo=_FakeAllocRepo(active=100)).execute(ctx2, _req()) class _FakeProducer(UnifiedProducer): @@ -122,28 +94,22 @@ async def produce(self, event_to_produce: DomainEvent, key: str | None = None, @pytest.mark.asyncio async def test_create_pod_step_publish_flag_and_compensation() -> None: + prod = _FakeProducer() + + # Skip publish path ctx = SagaContext("s1", "e1") ctx.set("execution_id", "e1") - # Skip publish path - s1 = CreatePodStep(producer=None, publish_commands=False) + s1 = CreatePodStep(producer=prod, publish_commands=False) 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") - prod = _FakeProducer() s2 = CreatePodStep(producer=prod, publish_commands=True) ok2 = await s2.execute(ctx2, _req()) assert ok2 is True and ctx2.get("pod_creation_triggered") is True and prod.events - # Missing producer -> failure - ctx3 = SagaContext("s3", "e3") - ctx3.set("execution_id", "e3") - s3 = CreatePodStep(producer=None, publish_commands=True) - ok3 = await s3.execute(ctx3, _req()) - assert ok3 is False and ctx3.error is not None - # DeletePod compensation triggers only when flagged and producer exists comp = DeletePodCompensation(producer=prod) ctx2.set("pod_creation_triggered", True) @@ -158,9 +124,6 @@ async def test_release_resources_compensation() -> None: ctx.set("allocation_id", "alloc-1") assert await comp.compensate(ctx) is True and repo.released == ["alloc-1"] - # Missing repo -> failure - comp2 = ReleaseResourcesCompensation(alloc_repo=None) - assert await comp2.compensate(ctx) is False # Missing allocation_id -> True short-circuit ctx2 = SagaContext("sX", "eX") assert await ReleaseResourcesCompensation(alloc_repo=repo).compensate(ctx2) is True @@ -168,39 +131,38 @@ async def test_release_resources_compensation() -> None: @pytest.mark.asyncio async def test_delete_pod_compensation_variants() -> None: + prod = _FakeProducer() + # Not triggered -> True early - comp_none = DeletePodCompensation(producer=None) ctx = SagaContext("s", "e") ctx.set("pod_creation_triggered", False) - assert await comp_none.compensate(ctx) is True + assert await DeletePodCompensation(producer=prod).compensate(ctx) is True - # Triggered but missing producer -> False + # Triggered -> publishes delete command ctx2 = SagaContext("s2", "e2") ctx2.set("pod_creation_triggered", True) ctx2.set("execution_id", "e2") - assert await comp_none.compensate(ctx2) is False + assert await DeletePodCompensation(producer=prod).compensate(ctx2) is True + assert len(prod.events) == 1 - # Exercise get_compensation methods return types (coverage for lines returning comps/None) + # get_compensation return types assert ValidateExecutionStep().get_compensation() is None assert isinstance(AllocateResourcesStep(_FakeAllocRepo()).get_compensation(), ReleaseResourcesCompensation) - assert isinstance(QueueExecutionStep().get_compensation(), type(DeletePodCompensation(None)).__bases__[0]) or True - assert CreatePodStep(None, publish_commands=False).get_compensation() is not None - assert MonitorExecutionStep().get_compensation() is None + assert isinstance(CreatePodStep(prod, publish_commands=False).get_compensation(), DeletePodCompensation) def test_execution_saga_bind_and_get_steps_sets_flags_and_types() -> None: - # Dummy subclasses to satisfy isinstance checks without real deps class DummyProd(UnifiedProducer): def __init__(self) -> None: - pass # Skip parent __init__ + pass class DummyAlloc(ResourceAllocationRepository): def __init__(self) -> None: - pass # Skip parent __init__ + pass s = ExecutionSaga() s.bind_dependencies(producer=DummyProd(), alloc_repo=DummyAlloc(), publish_commands=True) steps = s.get_steps() - # CreatePod step should be configured and present + assert len(steps) == 3 cps = [st for st in steps if isinstance(st, CreatePodStep)][0] assert cps.publish_commands is True diff --git a/backend/tests/unit/services/saga/test_saga_comprehensive.py b/backend/tests/unit/services/saga/test_saga_comprehensive.py index a473c2b3..d5eea475 100644 --- a/backend/tests/unit/services/saga/test_saga_comprehensive.py +++ b/backend/tests/unit/services/saga/test_saga_comprehensive.py @@ -6,11 +6,9 @@ """ import pytest -from app.domain.enums.events import EventType from app.domain.enums.saga import SagaState from app.domain.events.typed import DomainEvent, ExecutionRequestedEvent from app.domain.saga.models import Saga -from app.services.saga.execution_saga import ExecutionSaga from app.services.saga.saga_step import CompensationStep, SagaContext, SagaStep from tests.conftest import make_execution_requested_event @@ -47,11 +45,6 @@ def test_saga_context_public_filtering() -> None: assert "public" in out and "_private" not in out -def test_execution_saga_triggers_on_request() -> None: - # No orchestrator needed: this is pure metadata - assert EventType.EXECUTION_REQUESTED in ExecutionSaga.get_trigger_events() - - @pytest.mark.asyncio async def test_step_success_and_compensation_chain() -> None: ctx = SagaContext("s1", "e1") diff --git a/backend/tests/unit/services/saga/test_saga_orchestrator_unit.py b/backend/tests/unit/services/saga/test_saga_orchestrator_unit.py index f2ce43f4..a46b611a 100644 --- a/backend/tests/unit/services/saga/test_saga_orchestrator_unit.py +++ b/backend/tests/unit/services/saga/test_saga_orchestrator_unit.py @@ -1,21 +1,14 @@ import logging -from unittest.mock import MagicMock import pytest -from app.core.metrics import EventMetrics from app.db.repositories.resource_allocation_repository import ResourceAllocationRepository from app.db.repositories.saga_repository import SagaRepository -from app.domain.enums.events import EventType from app.domain.enums.saga import SagaState -from app.domain.events.typed import DomainEvent, ExecutionRequestedEvent +from app.domain.events.typed import DomainEvent from app.domain.saga.models import Saga, SagaConfig from app.events.core import UnifiedProducer -from app.events.event_store import EventStore -from app.events.schema.schema_registry import SchemaRegistryManager -from app.services.saga.base_saga import BaseSaga +from app.services.saga.execution_saga import ExecutionSaga from app.services.saga.saga_orchestrator import SagaOrchestrator -from app.services.saga.saga_step import CompensationStep, SagaContext, SagaStep -from app.settings import Settings from tests.conftest import make_execution_requested_event @@ -31,6 +24,14 @@ def __init__(self) -> None: self.saved: list[Saga] = [] self.existing: dict[tuple[str, str], Saga] = {} + async def get_or_create_saga(self, saga: Saga) -> tuple[Saga, bool]: + key = (saga.execution_id, saga.saga_name) + if key in self.existing: + return self.existing[key], False + self.existing[key] = saga + self.saved.append(saga) + return saga, True + async def get_saga_by_execution_and_name(self, execution_id: str, saga_name: str) -> Saga | None: return self.existing.get((execution_id, saga_name)) @@ -51,13 +52,6 @@ async def produce( return None -class _FakeStore(EventStore): - """Fake EventStore for testing.""" - - def __init__(self) -> None: - pass # Skip parent __init__ - - class _FakeAlloc(ResourceAllocationRepository): """Fake ResourceAllocationRepository for testing.""" @@ -65,73 +59,36 @@ def __init__(self) -> None: pass # No special attributes needed -class _StepOK(SagaStep[ExecutionRequestedEvent]): - def __init__(self) -> None: - super().__init__("ok") - - async def execute(self, context: SagaContext, event: ExecutionRequestedEvent) -> bool: - return True - - def get_compensation(self) -> CompensationStep | None: - return None - - -class _Saga(BaseSaga): - @classmethod - def get_name(cls) -> str: - return "s" - - @classmethod - def get_trigger_events(cls) -> list[EventType]: - return [EventType.EXECUTION_REQUESTED] - - def get_steps(self) -> list[SagaStep[ExecutionRequestedEvent]]: - return [_StepOK()] - - -def _orch(event_metrics: EventMetrics) -> SagaOrchestrator: +def _orch(repo: SagaRepository | None = None) -> SagaOrchestrator: return SagaOrchestrator( config=SagaConfig(name="t", enable_compensation=True, store_events=True, publish_commands=False), - saga_repository=_FakeRepo(), + saga_repository=repo or _FakeRepo(), producer=_FakeProd(), - schema_registry_manager=MagicMock(spec=SchemaRegistryManager), - settings=MagicMock(spec=Settings), - event_store=_FakeStore(), resource_allocation_repository=_FakeAlloc(), logger=_test_logger, - event_metrics=event_metrics, ) @pytest.mark.asyncio -async def test_min_success_flow(event_metrics: EventMetrics) -> None: - orch = _orch(event_metrics) - orch.register_saga(_Saga) - # Set orchestrator running state via lifecycle property - orch._lifecycle_started = True - await orch._handle_event(make_execution_requested_event(execution_id="e")) - # basic sanity; deep behavior covered by integration - assert orch.is_running is True +async def test_handle_event_triggers_saga() -> None: + fake_repo = _FakeRepo() + orch = _orch(repo=fake_repo) + await orch.handle_execution_requested(make_execution_requested_event(execution_id="e")) + assert len(fake_repo.saved) == 1 + saved = fake_repo.saved[0] + assert saved.execution_id == "e" + assert saved.saga_name == ExecutionSaga.get_name() + assert saved.state == SagaState.RUNNING @pytest.mark.asyncio -async def test_should_trigger_and_existing_short_circuit(event_metrics: EventMetrics) -> None: +async def test_existing_saga_short_circuits() -> None: fake_repo = _FakeRepo() - orch = SagaOrchestrator( - config=SagaConfig(name="t", enable_compensation=True, store_events=True, publish_commands=False), - saga_repository=fake_repo, - producer=_FakeProd(), - schema_registry_manager=MagicMock(spec=SchemaRegistryManager), - settings=MagicMock(spec=Settings), - event_store=_FakeStore(), - resource_allocation_repository=_FakeAlloc(), - logger=_test_logger, - event_metrics=event_metrics, - ) - orch.register_saga(_Saga) - assert orch._should_trigger_saga(_Saga, make_execution_requested_event(execution_id="e")) is True - # Existing short-circuit returns existing ID - s = Saga(saga_id="sX", saga_name="s", execution_id="e", state=SagaState.RUNNING) - fake_repo.existing[("e", "s")] = s - sid = await orch._start_saga("s", make_execution_requested_event(execution_id="e")) - assert sid == "sX" + saga_name = ExecutionSaga.get_name() + s = Saga(saga_id="sX", saga_name=saga_name, execution_id="e", state=SagaState.RUNNING) + fake_repo.existing[("e", saga_name)] = s + orch = _orch(repo=fake_repo) + # Should not create a duplicate — returns existing + await orch.handle_execution_requested(make_execution_requested_event(execution_id="e")) + # No new sagas saved — existing saga was returned as-is + assert fake_repo.saved == [] diff --git a/backend/tests/unit/services/saga/test_saga_step_and_base.py b/backend/tests/unit/services/saga/test_saga_step_and_base.py index d56acab6..850ab187 100644 --- a/backend/tests/unit/services/saga/test_saga_step_and_base.py +++ b/backend/tests/unit/services/saga/test_saga_step_and_base.py @@ -1,10 +1,5 @@ -import asyncio -from unittest.mock import MagicMock - import pytest -from app.domain.enums.events import EventType -from app.domain.events.typed import EventMetadata, SystemErrorEvent -from app.services.saga.base_saga import BaseSaga +from app.domain.events.typed import SystemErrorEvent from app.services.saga.saga_step import CompensationStep, SagaContext, SagaStep pytestmark = pytest.mark.unit @@ -41,46 +36,14 @@ async def compensate(self, context: SagaContext) -> bool: # noqa: ARG002 @pytest.mark.asyncio -async def test_context_adders() -> None: +async def test_context_add_compensation() -> None: ctx = SagaContext("s1", "e1") - evt = SystemErrorEvent( - error_type="test_error", - message="test", - service_name="test_service", - metadata=EventMetadata(service_name="t", service_version="1"), - ) - ctx.add_event(evt) - assert len(ctx.events) == 1 comp = _DummyComp() ctx.add_compensation(comp) assert len(ctx.compensations) == 1 -def test_base_saga_abstract_calls_cover_pass_lines() -> None: - # Abstract classmethods can still be called on the class to hit 'pass' lines - assert BaseSaga.get_name() is None - assert BaseSaga.get_trigger_events() is None - # Instance-less call to abstract instance method to hit 'pass' - assert BaseSaga.get_steps(None) is None # type: ignore[arg-type] - - # And the default bind hook returns None when called - - class Dummy(BaseSaga): - @classmethod - def get_name(cls) -> str: - return "d" - - @classmethod - def get_trigger_events(cls) -> list[EventType]: - return [] - - def get_steps(self) -> list[SagaStep[SystemErrorEvent]]: - return [] - - Dummy().bind_dependencies() - - -def test_saga_step_str_and_can_execute() -> None: +def test_saga_step_str() -> None: class S(SagaStep[SystemErrorEvent]): async def execute(self, context: SagaContext, event: SystemErrorEvent) -> bool: return True @@ -90,5 +53,3 @@ def get_compensation(self) -> CompensationStep | None: s = S("nm") assert str(s) == "SagaStep(nm)" - # can_execute default True - assert asyncio.run(s.can_execute(SagaContext("s", "e"), MagicMock(spec=SystemErrorEvent))) is True diff --git a/backend/workers/run_saga_orchestrator.py b/backend/workers/run_saga_orchestrator.py index f54ccfa1..fd1c1e97 100644 --- a/backend/workers/run_saga_orchestrator.py +++ b/backend/workers/run_saga_orchestrator.py @@ -27,10 +27,9 @@ async def run_saga_orchestrator(settings: Settings) -> None: schema_registry = await container.get(SchemaRegistryManager) await initialize_event_schemas(schema_registry) - # Services are already started by the DI container providers - orchestrator = await container.get(SagaOrchestrator) + # Triggers consumer start + timeout checker via DI + await container.get(SagaOrchestrator) - # Shutdown event - signal handlers just set this shutdown_event = asyncio.Event() loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM): @@ -39,11 +38,8 @@ async def run_saga_orchestrator(settings: Settings) -> None: logger.info("Saga orchestrator started and running") try: - # Wait for shutdown signal or service to stop - while orchestrator.is_running and not shutdown_event.is_set(): - await asyncio.sleep(1) + await shutdown_event.wait() finally: - # Container cleanup stops everything logger.info("Initiating graceful shutdown...") await container.close()