feat: add terminal operation status handling and improve idempotency in completion callbacks (2abc89dc) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+16 −0

Original line number	Diff line number	Diff line
		@@ -84,6 +84,14 @@ _APP_INSTANCE_COMPLETION_STATE_MAP: dict[str, AppInstanceState] = {
		"failed": AppInstanceState.FAILED,
		}

		_TERMINAL_OPERATION_STATUSES = frozenset(
		{
		OperationStatus.COMPLETED,
		OperationStatus.PARTIALLY_COMPLETED,
		OperationStatus.FAILED,
		}
		)


		def _is_final(app_instance: AppInstance) -> bool:
		"""Whether no completion may move this instance any further."""
		@@ -507,6 +515,14 @@ class EdgeApplicationManagementService:
		)
		return

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

		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]
		result: Optional[dict[str, Any]] = None
		if status != OperationStatus.FAILED:

tests/unit/test_eam_service.py

+42 −0

Original line number	Diff line number	Diff line
		@@ -1707,6 +1707,48 @@ class TestHandleCompleted:
		assert final.state == AppInstanceState.TERMINATED
		assert len(callback_delivery_port.delivered) == 1

		async def test_redelivered_deploy_completion_delivers_one_callback(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_registration_repo: FakeAppRegistrationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_delivery_port: FakeCallbackDeliveryPort,
		callback_delivery_repo: FakeCallbackDeliveryRepository,
		) -> None:
		"""JetStream delivers at least once, so the same completion can arrive
		repeatedly; §L.2 requires OEG to apply it idempotently by operation_id.
		`ready` is not a final instance state, so only the operation-level guard
		stops the customer's webhook firing once per redelivery."""
		await self._seed_pending_operation(operation_repo)
		await self._seed_app_instance_with_registration(
		app_registration_repo, app_instance_repo, INSTANCE_ID
		)
		await self._seed_active_callback_registration(callback_registration_repo)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(INSTANCE_ID), zone_id=str(ZONE_ID), status="completed"
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)
		await service.handle_completed(event)
		await service.handle_completed(event)

		final = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert final is not None
		assert final.state == AppInstanceState.READY
		assert len(callback_delivery_port.delivered) == 1
		assert len(callback_delivery_repo.rows) == 1

		async def test_stale_total_failure_does_not_overwrite_terminated_instance(
		self,
		service: EdgeApplicationManagementService,

Admin message