feat: consume SRM's accepted event to move EAM operations to IN_PROGRESS state (690f73fb) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+28 −0

Original line number	Diff line number	Diff line
		@@ -64,6 +64,7 @@ from open_exposure_gateway.domain.models import (
		OperationType,
		PackageType,
		)
		from open_exposure_gateway.domain.quality_on_demand import SRMOperationStatus
		from open_exposure_gateway.ports.callback_delivery_port import (
		CallbackDeliveryPort,
		CallbackEvent,
		@@ -1273,6 +1274,33 @@ class EdgeApplicationManagementService:
		if updated_instances:
		await self._deliver_callbacks(operation_id, event.completed_at, updated_instances)

		async def handle_operation_status(self, event: SRMOperationStatus) -> None:
		if self._operation_repo is None:
		raise RuntimeError("OperationRepository is not available")

		operation_id = UUID(event.operation_id)
		operation = await self._operation_repo.get_by_id(operation_id)
		if operation is None:
		logger.warning(
		"operation_status_for_unknown_operation", operation_id=event.operation_id
		)
		return

		if operation.status in _TERMINAL_OPERATION_STATUSES:
		logger.info(
		"redelivered_status_ignored_for_terminal_operation",
		operation_id=event.operation_id,
		status=operation.status.value,
		)
		return

		if event.state != "accepted" or operation.status != OperationStatus.PENDING:
		return

		await self._operation_repo.save(
		operation.model_copy(update={"status": OperationStatus.IN_PROGRESS})
		)

		async def _reported_or_settled_keys(
		self,
		operation: Operation,

src/open_exposure_gateway/domain/edge_application_management.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -11,6 +11,7 @@ class Subject(StrEnum):
		TASK_DEPLOY = "command.srm.service.deploy"
		TASK_TERMINATE = "command.srm.service.terminate"
		OPERATION_COMPLETED = "event.srm.operation.completed"
		OPERATION_STATUS = "event.srm.operation.status"


		class AppRepo(BaseModel):

src/open_exposure_gateway/main.py

+28 −0

Original line number	Diff line number	Diff line
		@@ -102,6 +102,24 @@ def _build_operation_completed_handler(
		return handle


		def _build_eam_operation_status_handler(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		callback_delivery_port: CallbackDeliveryPort,
		) -> Callable[[SRMOperationStatus], Awaitable[None]]:
		async def handle(event: SRMOperationStatus) -> None:
		async with session_maker() as session:
		try:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		await service.handle_operation_status(event)
		await session.commit()
		except Exception:
		await session.rollback()
		raise

		return handle


		async def _run_reconciliation_pass(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		@@ -313,6 +331,16 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		)
		await eam_operation_consumer.start()

		eam_operation_status_consumer = NatsOperationStatusConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_STATUS,
		handler=_build_eam_operation_status_handler(
		session_maker, srm_client, HttpCallbackClient()
		),
		)
		await eam_operation_status_consumer.start()
		logger.info("NATS consumer started", subject=Subject.OPERATION_STATUS)

		operation_completed_consumer = NatsOperationConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_COMPLETED,

tests/unit/test_eam_service.py

+105 −0

Original line number	Diff line number	Diff line
		@@ -64,6 +64,7 @@ from open_exposure_gateway.domain.models import (
		OperationType,
		PackageType,
		)
		from open_exposure_gateway.domain.quality_on_demand import SRMOperationStatus
		from tests.unit.fakes import (
		FakeAppDeploymentRepository,
		FakeAppInstanceRepository,
		@@ -4525,6 +4526,110 @@ class TestDeploymentTerminateCompletion:
		assert instance.state == AppInstanceState.FAILED


		class TestHandleOperationStatus:
		"""event.srm.operation.status(state=accepted) narrows the "no idea what's
		happening" window between an operation going PENDING and SRM's terminal
		event.srm.operation.completed arriving: it moves PENDING -> IN_PROGRESS
		only, and never touches AppInstance/AppDeployment state."""

		OPERATION_ID = UUID("eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee")

		def _accepted_event(self, operation_id: UUID = OPERATION_ID) -> SRMOperationStatus:
		return SRMOperationStatus(
		schema_version="1.0",
		operation_id=str(operation_id),
		state="accepted",
		correlation_id="corr-1",
		emitted_at="2026-07-04T10:00:05+00:00",
		)

		async def _seed_operation(
		self,
		operation_repo: FakeOperationRepository,
		status: OperationStatus,
		operation_type: OperationType = OperationType.DEPLOY,
		) -> None:
		subject = (
		Subject.TASK_TERMINATE
		if operation_type == OperationType.TERMINATE
		else Subject.TASK_DEPLOY
		)
		await operation_repo.save(
		Operation(
		operation_id=self.OPERATION_ID,
		correlation_id="corr-1",
		tenant_id="tenant-1",
		app_provider_id="provider-1",
		operation_type=operation_type,
		status=status,
		subject=subject,
		metadata={},
		)
		)

		async def test_accepted_moves_pending_deploy_operation_to_in_progress(
		self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository
		) -> None:
		await self._seed_operation(operation_repo, OperationStatus.PENDING)

		await service.handle_operation_status(self._accepted_event())

		updated = await operation_repo.get_by_id(self.OPERATION_ID)
		assert updated is not None
		assert updated.status == OperationStatus.IN_PROGRESS

		async def test_accepted_moves_pending_terminate_operation_to_in_progress(
		self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository
		) -> None:
		await self._seed_operation(
		operation_repo, OperationStatus.PENDING, operation_type=OperationType.TERMINATE
		)

		await service.handle_operation_status(self._accepted_event())

		updated = await operation_repo.get_by_id(self.OPERATION_ID)
		assert updated is not None
		assert updated.status == OperationStatus.IN_PROGRESS

		@pytest.mark.parametrize(
		"terminal_status",
		[OperationStatus.COMPLETED, OperationStatus.PARTIALLY_COMPLETED, OperationStatus.FAILED],
		)
		async def test_redelivered_accepted_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._accepted_event())

		updated = await operation_repo.get_by_id(self.OPERATION_ID)
		assert updated is not None
		assert updated.status == terminal_status

		async def test_redelivered_accepted_on_in_progress_operation_is_noop(
		self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository
		) -> None:
		await self._seed_operation(operation_repo, OperationStatus.IN_PROGRESS)

		await service.handle_operation_status(self._accepted_event())

		updated = await operation_repo.get_by_id(self.OPERATION_ID)
		assert updated is not None
		assert updated.status == OperationStatus.IN_PROGRESS

		async def test_accepted_for_unknown_operation_is_noop(
		self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository
		) -> None:
		unknown_operation_id = uuid4()

		await service.handle_operation_status(self._accepted_event(unknown_operation_id))

		assert await operation_repo.get_by_id(unknown_operation_id) is None


		class TestReconcileStaleOperations:
		"""The at-most-once NATS subject can drop event.srm.operation.completed, which
		strands an operation in PENDING/IN_PROGRESS forever. list_stale_operations()