Loading src/open_exposure_gateway/application/services/edge_application_management_service.py +38 −0 Original line number Diff line number Diff line Loading @@ -285,6 +285,28 @@ _SETTLED_APP_INSTANCE_STATES = frozenset( ) def _synthetic_failure_completion(event: SRMOperationStatus) -> SRMOperationCompleted: """Recast a failed/failed_before_start event.srm.operation.status as a total-failure event.srm.operation.completed so handle_completed's existing failure handling (instance/deployment transitions, callback delivery) can run early, without waiting for SRM's own completed event to arrive.""" title = ( "SRM rejected the command before starting." if event.state == "failed_before_start" else "SRM reported the operation as failed." ) return SRMOperationCompleted( schema_version=event.schema_version, operation_id=event.operation_id, status="failed", service_order_id=event.service_order_id, instances=[], error={"type": "about:blank", "title": title, "status": 502}, correlation_id=event.correlation_id, completed_at=event.emitted_at, ) def _expected_instance_keys(operation: Operation) -> Optional[set[str]]: """The fan-out target set recorded when a multi-zone /deployments request put N commands under one operation_id. Loading Loading @@ -1278,6 +1300,22 @@ class EdgeApplicationManagementService: if self._operation_repo is None: raise RuntimeError("OperationRepository is not available") if event.state in ("failed", "failed_before_start"): # SRM always follows a failed/failed_before_start status event with # a matching event.srm.operation.completed(status=failed) for the # same operation_id. Route through handle_completed rather than # writing FAILED here directly: that keeps the instance/deployment # transitions and callback delivery in one place, and lets that # method's own terminal-status guard treat the (guaranteed) later # completed event as the safe, idempotent redelivery it is. logger.info( "operation_early_failure_from_status_event", operation_id=event.operation_id, state=event.state, ) await self.handle_completed(_synthetic_failure_completion(event)) return operation_id = UUID(event.operation_id) operation = await self._operation_repo.get_by_id(operation_id) if operation is None: Loading tests/unit/test_eam_service.py +131 −3 Original line number Diff line number Diff line from datetime import datetime, timedelta, timezone from typing import Any from typing import Any, Literal from unittest.mock import AsyncMock from uuid import UUID, uuid4 Loading Loading @@ -4534,15 +4534,22 @@ class TestHandleOperationStatus: OPERATION_ID = UUID("eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee") def _accepted_event(self, operation_id: UUID = OPERATION_ID) -> SRMOperationStatus: def _status_event( self, state: Literal["accepted", "failed_before_start", "completed", "failed", "in_progress"], operation_id: UUID = OPERATION_ID, ) -> SRMOperationStatus: return SRMOperationStatus( schema_version="1.0", operation_id=str(operation_id), state="accepted", state=state, correlation_id="corr-1", emitted_at="2026-07-04T10:00:05+00:00", ) def _accepted_event(self, operation_id: UUID = OPERATION_ID) -> SRMOperationStatus: return self._status_event("accepted", operation_id) async def _seed_operation( self, operation_repo: FakeOperationRepository, Loading Loading @@ -4629,6 +4636,127 @@ class TestHandleOperationStatus: assert await operation_repo.get_by_id(unknown_operation_id) is None @pytest.mark.parametrize("failure_state", ["failed", "failed_before_start"]) async def test_failure_state_moves_pending_deploy_operation_and_instance_to_failed( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_instance_repo: FakeAppInstanceRepository, failure_state: Literal["failed", "failed_before_start"], ) -> None: """A failed/failed_before_start status event carries no instances[] of its own -- it must still flip the pre-created app_instance row to failed, via the same total-failure path handle_completed already uses, rather than leaving it stuck instantiating.""" await self._seed_operation(operation_repo, OperationStatus.PENDING) await app_instance_repo.save( AppInstance( app_instance_id=INSTANCE_ID, operation_id=self.OPERATION_ID, app_registration_id=uuid4(), edge_cloud_zone_id=ZONE_ID, state=AppInstanceState.INSTANTIATING, ) ) await service.handle_operation_status(self._status_event(failure_state)) updated_operation = await operation_repo.get_by_id(self.OPERATION_ID) assert updated_operation is not None assert updated_operation.status == OperationStatus.FAILED updated_instance = await app_instance_repo.get_by_id(INSTANCE_ID) assert updated_instance is not None assert updated_instance.state == AppInstanceState.FAILED async def test_failed_moves_in_progress_terminate_operation_to_failed( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository ) -> None: await self._seed_operation( operation_repo, OperationStatus.IN_PROGRESS, operation_type=OperationType.TERMINATE ) await service.handle_operation_status(self._status_event("failed")) updated = await operation_repo.get_by_id(self.OPERATION_ID) assert updated is not None assert updated.status == OperationStatus.FAILED @pytest.mark.parametrize( "terminal_status", [OperationStatus.COMPLETED, OperationStatus.PARTIALLY_COMPLETED, OperationStatus.FAILED], ) async def test_redelivered_failure_state_on_terminal_operation_is_noop( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, terminal_status: OperationStatus, ) -> None: await self._seed_operation(operation_repo, terminal_status) await service.handle_operation_status(self._status_event("failed_before_start")) updated = await operation_repo.get_by_id(self.OPERATION_ID) assert updated is not None assert updated.status == terminal_status async def test_failure_state_for_unknown_operation_is_noop( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository ) -> None: unknown_operation_id = uuid4() await service.handle_operation_status( self._status_event("failed_before_start", unknown_operation_id) ) assert await operation_repo.get_by_id(unknown_operation_id) is None async def test_later_real_completed_failure_is_a_safe_redelivery_after_early_failure( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_instance_repo: FakeAppInstanceRepository, ) -> None: """SRM always follows failed_before_start with its own event.srm.operation.completed(status=failed) for the same operation_id. By the time that arrives, this operation has already been failed via the status event -- handle_completed's own terminal guard must treat it as a harmless redelivery, not raise or re-process it.""" await self._seed_operation(operation_repo, OperationStatus.PENDING) await app_instance_repo.save( AppInstance( app_instance_id=INSTANCE_ID, operation_id=self.OPERATION_ID, app_registration_id=uuid4(), edge_cloud_zone_id=ZONE_ID, state=AppInstanceState.INSTANTIATING, ) ) await service.handle_operation_status(self._status_event("failed_before_start")) real_completed_event = SRMOperationCompleted( schema_version="1.0", operation_id=str(self.OPERATION_ID), status="failed", instances=[], error={ "type": "https://etsi.org/sdg/oop/problems/invalid-command", "title": "Deploy service command is invalid.", "status": 400, }, correlation_id="corr-1", completed_at="2026-07-04T10:02:35+00:00", ) await service.handle_completed(real_completed_event) updated_operation = await operation_repo.get_by_id(self.OPERATION_ID) assert updated_operation is not None assert updated_operation.status == OperationStatus.FAILED updated_instance = await app_instance_repo.get_by_id(INSTANCE_ID) assert updated_instance is not None assert updated_instance.state == AppInstanceState.FAILED class TestReconcileStaleOperations: """The at-most-once NATS subject can drop event.srm.operation.completed, which Loading Loading
src/open_exposure_gateway/application/services/edge_application_management_service.py +38 −0 Original line number Diff line number Diff line Loading @@ -285,6 +285,28 @@ _SETTLED_APP_INSTANCE_STATES = frozenset( ) def _synthetic_failure_completion(event: SRMOperationStatus) -> SRMOperationCompleted: """Recast a failed/failed_before_start event.srm.operation.status as a total-failure event.srm.operation.completed so handle_completed's existing failure handling (instance/deployment transitions, callback delivery) can run early, without waiting for SRM's own completed event to arrive.""" title = ( "SRM rejected the command before starting." if event.state == "failed_before_start" else "SRM reported the operation as failed." ) return SRMOperationCompleted( schema_version=event.schema_version, operation_id=event.operation_id, status="failed", service_order_id=event.service_order_id, instances=[], error={"type": "about:blank", "title": title, "status": 502}, correlation_id=event.correlation_id, completed_at=event.emitted_at, ) def _expected_instance_keys(operation: Operation) -> Optional[set[str]]: """The fan-out target set recorded when a multi-zone /deployments request put N commands under one operation_id. Loading Loading @@ -1278,6 +1300,22 @@ class EdgeApplicationManagementService: if self._operation_repo is None: raise RuntimeError("OperationRepository is not available") if event.state in ("failed", "failed_before_start"): # SRM always follows a failed/failed_before_start status event with # a matching event.srm.operation.completed(status=failed) for the # same operation_id. Route through handle_completed rather than # writing FAILED here directly: that keeps the instance/deployment # transitions and callback delivery in one place, and lets that # method's own terminal-status guard treat the (guaranteed) later # completed event as the safe, idempotent redelivery it is. logger.info( "operation_early_failure_from_status_event", operation_id=event.operation_id, state=event.state, ) await self.handle_completed(_synthetic_failure_completion(event)) return operation_id = UUID(event.operation_id) operation = await self._operation_repo.get_by_id(operation_id) if operation is None: Loading
tests/unit/test_eam_service.py +131 −3 Original line number Diff line number Diff line from datetime import datetime, timedelta, timezone from typing import Any from typing import Any, Literal from unittest.mock import AsyncMock from uuid import UUID, uuid4 Loading Loading @@ -4534,15 +4534,22 @@ class TestHandleOperationStatus: OPERATION_ID = UUID("eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee") def _accepted_event(self, operation_id: UUID = OPERATION_ID) -> SRMOperationStatus: def _status_event( self, state: Literal["accepted", "failed_before_start", "completed", "failed", "in_progress"], operation_id: UUID = OPERATION_ID, ) -> SRMOperationStatus: return SRMOperationStatus( schema_version="1.0", operation_id=str(operation_id), state="accepted", state=state, correlation_id="corr-1", emitted_at="2026-07-04T10:00:05+00:00", ) def _accepted_event(self, operation_id: UUID = OPERATION_ID) -> SRMOperationStatus: return self._status_event("accepted", operation_id) async def _seed_operation( self, operation_repo: FakeOperationRepository, Loading Loading @@ -4629,6 +4636,127 @@ class TestHandleOperationStatus: assert await operation_repo.get_by_id(unknown_operation_id) is None @pytest.mark.parametrize("failure_state", ["failed", "failed_before_start"]) async def test_failure_state_moves_pending_deploy_operation_and_instance_to_failed( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_instance_repo: FakeAppInstanceRepository, failure_state: Literal["failed", "failed_before_start"], ) -> None: """A failed/failed_before_start status event carries no instances[] of its own -- it must still flip the pre-created app_instance row to failed, via the same total-failure path handle_completed already uses, rather than leaving it stuck instantiating.""" await self._seed_operation(operation_repo, OperationStatus.PENDING) await app_instance_repo.save( AppInstance( app_instance_id=INSTANCE_ID, operation_id=self.OPERATION_ID, app_registration_id=uuid4(), edge_cloud_zone_id=ZONE_ID, state=AppInstanceState.INSTANTIATING, ) ) await service.handle_operation_status(self._status_event(failure_state)) updated_operation = await operation_repo.get_by_id(self.OPERATION_ID) assert updated_operation is not None assert updated_operation.status == OperationStatus.FAILED updated_instance = await app_instance_repo.get_by_id(INSTANCE_ID) assert updated_instance is not None assert updated_instance.state == AppInstanceState.FAILED async def test_failed_moves_in_progress_terminate_operation_to_failed( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository ) -> None: await self._seed_operation( operation_repo, OperationStatus.IN_PROGRESS, operation_type=OperationType.TERMINATE ) await service.handle_operation_status(self._status_event("failed")) updated = await operation_repo.get_by_id(self.OPERATION_ID) assert updated is not None assert updated.status == OperationStatus.FAILED @pytest.mark.parametrize( "terminal_status", [OperationStatus.COMPLETED, OperationStatus.PARTIALLY_COMPLETED, OperationStatus.FAILED], ) async def test_redelivered_failure_state_on_terminal_operation_is_noop( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, terminal_status: OperationStatus, ) -> None: await self._seed_operation(operation_repo, terminal_status) await service.handle_operation_status(self._status_event("failed_before_start")) updated = await operation_repo.get_by_id(self.OPERATION_ID) assert updated is not None assert updated.status == terminal_status async def test_failure_state_for_unknown_operation_is_noop( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository ) -> None: unknown_operation_id = uuid4() await service.handle_operation_status( self._status_event("failed_before_start", unknown_operation_id) ) assert await operation_repo.get_by_id(unknown_operation_id) is None async def test_later_real_completed_failure_is_a_safe_redelivery_after_early_failure( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_instance_repo: FakeAppInstanceRepository, ) -> None: """SRM always follows failed_before_start with its own event.srm.operation.completed(status=failed) for the same operation_id. By the time that arrives, this operation has already been failed via the status event -- handle_completed's own terminal guard must treat it as a harmless redelivery, not raise or re-process it.""" await self._seed_operation(operation_repo, OperationStatus.PENDING) await app_instance_repo.save( AppInstance( app_instance_id=INSTANCE_ID, operation_id=self.OPERATION_ID, app_registration_id=uuid4(), edge_cloud_zone_id=ZONE_ID, state=AppInstanceState.INSTANTIATING, ) ) await service.handle_operation_status(self._status_event("failed_before_start")) real_completed_event = SRMOperationCompleted( schema_version="1.0", operation_id=str(self.OPERATION_ID), status="failed", instances=[], error={ "type": "https://etsi.org/sdg/oop/problems/invalid-command", "title": "Deploy service command is invalid.", "status": 400, }, correlation_id="corr-1", completed_at="2026-07-04T10:02:35+00:00", ) await service.handle_completed(real_completed_event) updated_operation = await operation_repo.get_by_id(self.OPERATION_ID) assert updated_operation is not None assert updated_operation.status == OperationStatus.FAILED updated_instance = await app_instance_repo.get_by_id(INSTANCE_ID) assert updated_instance is not None assert updated_instance.state == AppInstanceState.FAILED class TestReconcileStaleOperations: """The at-most-once NATS subject can drop event.srm.operation.completed, which Loading
