fix: prevent duplicate notifications on redelivery of QoD events (c9444810) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+9 −0

Original line number	Diff line number	Diff line
		@@ -295,8 +295,13 @@ class QualityOnDemandService:
		else:
		update = {"state": QodSessionState.UNAVAILABLE}
		qos_status, status_info = "UNAVAILABLE", "NETWORK_TERMINATED"

		prior_state = qod_session.state
		await self._qod_session_repo.save(qod_session.model_copy(update=update))

		if prior_state == update["state"]:
		return

		await self._deliver_qos_status_changed(
		operation_id=operation_id,
		session_id=qod_session.session_id,
		@@ -351,8 +356,12 @@ class QualityOnDemandService:
		status = (
		QodSessionState.AVAILABLE if qos_status == "AVAILABLE" else QodSessionState.UNAVAILABLE
		)
		prior_state = qod_session.state
		await self._qod_session_repo.save(qod_session.model_copy(update={"state": status}))

		if prior_state == status:
		return

		await self._deliver_qos_status_changed(
		operation_id=operation_id,
		session_id=qod_session.session_id,

tests/integration/test_postgres.py

+98 −0

Original line number	Diff line number	Diff line
		@@ -17,6 +17,7 @@ from open_exposure_gateway.adapters.database.repos.callback_registrations import
		SqlCallbackRegistrationRepository,
		)
		from open_exposure_gateway.adapters.database.repos.operations import SqlOperationRepository
		from open_exposure_gateway.adapters.database.repos.qod_sessions import SqlQodSessionRepository
		from open_exposure_gateway.adapters.errors import (
		DuplicateAppRegistrationError,
		DuplicateOperationError,
		@@ -32,6 +33,8 @@ from open_exposure_gateway.domain.models import (
		OperationStatus,
		OperationType,
		PackageType,
		QodSession,
		QodSessionState,
		)


		@@ -85,6 +88,17 @@ def _callback_registration(operation_id: UUID) -> CallbackRegistration:
		)


		def _qod_session(operation_id: UUID) -> QodSession:
		return QodSession(
		session_id=uuid4(),
		operation_id=operation_id,
		service_specification_id=uuid4(),
		qos_profile="QOS_E",
		duration_seconds=3600,
		state=QodSessionState.REQUESTED,
		)


		def _callback_delivery(callback_registration_id: UUID, operation_id: UUID) -> CallbackDelivery:
		return CallbackDelivery(
		id=uuid4(),
		@@ -339,3 +353,87 @@ async def test_callback_delivery_repo_persists_and_lists(db_session: AsyncSessio
		assert len(deliveries) == 1
		assert deliveries[0].id == saved.id
		assert deliveries[0].state == "delivered"


		async def test_qod_session_repo_persists_and_loads(db_session: AsyncSession) -> None:
		operation = await SqlOperationRepository(db_session).save(_operation())
		repo = SqlQodSessionRepository(db_session)
		qod_session = _qod_session(operation_id=operation.operation_id)

		saved = await repo.save(qod_session)

		assert saved.session_id == qod_session.session_id
		assert saved.created_at is not None
		assert saved.updated_at is not None

		by_id = await repo.get_by_id(saved.session_id)
		by_operation_id = await repo.get_by_operation_id(operation.operation_id)

		assert by_id is not None
		assert by_operation_id is not None
		assert by_id == by_operation_id
		assert by_id.service_specification_id == qod_session.service_specification_id
		assert by_id.qos_profile == qod_session.qos_profile
		assert by_id.duration_seconds == qod_session.duration_seconds
		assert by_id.state == QodSessionState.REQUESTED
		assert by_id.external_ref is None
		assert by_id.device_ports is None
		assert by_id.application_server_ports is None


		async def test_qod_session_repo_persists_optional_fields(db_session: AsyncSession) -> None:
		operation = await SqlOperationRepository(db_session).save(_operation())
		repo = SqlQodSessionRepository(db_session)
		qod_session = _qod_session(operation_id=operation.operation_id)
		qod_session.external_ref = "external-ref-1"
		qod_session.device_ports = {"ports": [1000, 2000]}
		qod_session.application_server_ports = {"ports": [3000]}

		saved = await repo.save(qod_session)
		reloaded = await repo.get_by_id(saved.session_id)

		assert reloaded is not None
		assert reloaded.external_ref == "external-ref-1"
		assert reloaded.device_ports == {"ports": [1000, 2000]}
		assert reloaded.application_server_ports == {"ports": [3000]}


		async def test_qod_session_repo_get_by_id_returns_none_when_missing(
		db_session: AsyncSession,
		) -> None:
		repo = SqlQodSessionRepository(db_session)

		assert await repo.get_by_id(uuid4()) is None


		async def test_qod_session_repo_get_by_operation_id_returns_none_when_missing(
		db_session: AsyncSession,
		) -> None:
		repo = SqlQodSessionRepository(db_session)

		assert await repo.get_by_operation_id(uuid4()) is None


		async def test_qod_session_repo_updates_updated_at_on_second_save(db_engine: AsyncEngine) -> None:
		session_factory = async_sessionmaker(bind=db_engine, expire_on_commit=False)

		async with session_factory() as session:
		operation = await SqlOperationRepository(session).save(_operation())
		saved = await SqlQodSessionRepository(session).save(
		_qod_session(operation_id=operation.operation_id)
		)
		await session.commit()

		await asyncio.sleep(0.01)

		async with session_factory() as session:
		repo = SqlQodSessionRepository(session)
		saved.state = QodSessionState.AVAILABLE
		updated = await repo.save(saved)
		await session.commit()

		assert updated.created_at == saved.created_at
		assert updated.created_at is not None
		assert updated.updated_at is not None
		assert updated.updated_at > updated.created_at
		assert updated.state == QodSessionState.AVAILABLE

tests/unit/test_qod_service.py

+92 −0

Original line number	Diff line number	Diff line
		@@ -631,6 +631,59 @@ class TestHandleCompleted:
		assert cloud_event.data.qosStatus == "UNAVAILABLE"
		assert cloud_event.data.statusInfo == "NETWORK_TERMINATED"

		async def test_redelivery_of_same_event_does_not_duplicate_notification(
		self,
		operation_repo: FakeOperationRepository,
		qod_session_repo: FakeQodSessionRepository,
		) -> None:
		"""The databus can redeliver event.srm.operation.completed at least
		once (e.g. handler crashes after acting but before acking). Replaying
		the exact same event must not re-notify a session that is already
		UNAVAILABLE -- the CAMARA QoD spec forbids sending a qos-status-changed
		event when qosStatus was already UNAVAILABLE."""
		callback_registration_repo = FakeCallbackRegistrationRepository()
		callback_delivery_repo = FakeCallbackDeliveryRepository()
		callback_delivery_port = FakeQodCallbackDeliveryPort()
		service = QualityOnDemandService(
		srm_client=AsyncMock(),
		operation_repo=operation_repo,
		qod_session_repo=qod_session_repo,
		callback_registration_repo=callback_registration_repo,
		callback_delivery_repo=callback_delivery_repo,
		callback_delivery_port=callback_delivery_port,
		)
		operation, qod_session = await self._seed(operation_repo, qod_session_repo)
		await callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=operation.operation_id,
		tenant_id="tenant-1",
		api_family="quality-on-demand",
		sink="https://client.example.com/cb",
		event_types=["org.camaraproject.qod.v0.qos-status-changed"],
		)
		)
		event = SRMOperationCompleted.model_validate(
		completion_payload(
		str(operation.operation_id),
		status="failed",
		instances=[],
		error={"status": 500, "code": "INTERNAL", "message": "backend rejected"},
		)
		)

		await service.handle_completed(event)
		await service.handle_completed(event) # simulated redelivery of the same message

		updated_session = await qod_session_repo.get_by_id(qod_session.session_id)
		assert updated_session is not None
		assert updated_session.state == QodSessionState.UNAVAILABLE

		assert len(callback_delivery_port.delivered) == 1
		deliveries = list(callback_delivery_repo.rows.values())
		assert len(deliveries) == 1
		assert deliveries[0].attempt == 1

		async def test_failed_delivery_still_persists_session_update(
		self,
		operation_repo: FakeOperationRepository,
		@@ -850,6 +903,45 @@ class TestHandleStatusChanged:
		second_delivery = next(d for d in deliveries if d.attempt == 2)
		assert second_delivery.state == "delivered"

		async def test_redelivery_of_same_event_does_not_duplicate_notification(
		self,
		service: QualityOnDemandService,
		operation_repo: FakeOperationRepository,
		qod_session_repo: FakeQodSessionRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_delivery_repo: FakeCallbackDeliveryRepository,
		callback_delivery_port: FakeQodCallbackDeliveryPort,
		) -> None:
		"""event.srm.operation.status can be redelivered at least once (ADR-0036
		notes redelivery/clock-skew can cause duplicate processing). Replaying
		the same UNAVAILABLE status a second time must not send a second
		qos-status-changed notification -- the session is already UNAVAILABLE."""
		operation, qod_session = await self._seed_available(operation_repo, qod_session_repo)
		await callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=operation.operation_id,
		tenant_id="tenant-1",
		api_family="quality-on-demand",
		sink="https://client.example.com/cb",
		event_types=["org.camaraproject.qod.v0.qos-status-changed"],
		)
		)
		event = SRMOperationStatus.model_validate(
		operation_status_payload(str(operation.operation_id))
		)

		await service.handle_status_changed(event)
		await service.handle_status_changed(event) # simulated redelivery of the same message

		updated_session = await qod_session_repo.get_by_id(qod_session.session_id)
		assert updated_session is not None
		assert updated_session.state == QodSessionState.UNAVAILABLE

		assert len(callback_delivery_port.delivered) == 1
		deliveries = list(callback_delivery_repo.rows.values())
		assert len(deliveries) == 1

		async def test_ignores_status_for_a_different_domains_operation(
		self,
		service: QualityOnDemandService,

Admin message