fix: defer deactivate when DELETE arrives before activate completes (4fa700b0) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/database/repos/qod_sessions.py

+8 −2

Changes for src/open_exposure_gateway/adapters/database/repos/qod_sessions.py: 8 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -19,13 +19,19 @@ class SqlQodSessionRepository(QodSessionRepository):
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def get_by_id(self, session_id: UUID) -> QodSession \| None:
		async def get_by_id(self, session_id: UUID, for_update: bool = False) -> QodSession \| None:
		stmt = select(QodSessionRow).where(QodSessionRow.session_id == session_id)
		if for_update:
		stmt = stmt.with_for_update().execution_options(populate_existing=True)
		row = await self._session.scalar(stmt)
		return QodSessionMapper.to_domain(row) if row is not None else None

		async def get_by_operation_id(self, operation_id: UUID) -> QodSession \| None:
		async def get_by_operation_id(
		self, operation_id: UUID, for_update: bool = False
		) -> QodSession \| None:
		stmt = select(QodSessionRow).where(QodSessionRow.operation_id == operation_id)
		if for_update:
		stmt = stmt.with_for_update().execution_options(populate_existing=True)
		row = await self._session.scalar(stmt)
		return QodSessionMapper.to_domain(row) if row is not None else None

src/open_exposure_gateway/adapters/database/repos/traffic_influences.py

+10 −2

Changes for src/open_exposure_gateway/adapters/database/repos/traffic_influences.py: 10 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -14,15 +14,23 @@ class SqlTrafficInfluenceRepository(TrafficInfluenceRepository):
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def get_by_id(self, traffic_influence_id: UUID) -> TrafficInfluence \| None:
		async def get_by_id(
		self, traffic_influence_id: UUID, for_update: bool = False
		) -> TrafficInfluence \| None:
		stmt = select(TrafficInfluenceRow).where(
		TrafficInfluenceRow.traffic_influence_id == traffic_influence_id
		)
		if for_update:
		stmt = stmt.with_for_update().execution_options(populate_existing=True)
		row = await self._session.scalar(stmt)
		return TrafficInfluenceMapper.to_domain(row) if row is not None else None

		async def get_by_operation_id(self, operation_id: UUID) -> TrafficInfluence \| None:
		async def get_by_operation_id(
		self, operation_id: UUID, for_update: bool = False
		) -> TrafficInfluence \| None:
		stmt = select(TrafficInfluenceRow).where(TrafficInfluenceRow.operation_id == operation_id)
		if for_update:
		stmt = stmt.with_for_update().execution_options(populate_existing=True)
		row = await self._session.scalar(stmt)
		return TrafficInfluenceMapper.to_domain(row) if row is not None else None

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+63 −11

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 63 added lines, 11 removed lines.
Original line number	Diff line number	Diff line
		@@ -40,6 +40,7 @@ from open_exposure_gateway.core.exceptions import (
		UnprocessableEntityException,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMCompletedInstance,
		SRMOperationCompleted,
		completion_error,
		)
		@@ -93,6 +94,8 @@ _TERMINAL_STATES = frozenset(
		{QodSessionState.DELETION_REQUESTED, QodSessionState.DELETED, QodSessionState.ERROR}
		)

		_IN_FLIGHT_OPERATION_STATUSES = frozenset({OperationStatus.PENDING, OperationStatus.IN_PROGRESS})

		_QOS_STATUS_CHANGED_EVENT_TYPE = "org.camaraproject.quality-on-demand.v1.qos-status-changed"


		@@ -451,13 +454,23 @@ class QualityOnDemandService:
		await self._handle_deactivate_completed(operation, status)
		return

		qod_session = await self._qod_session_repo.get_by_operation_id(operation_id)
		qod_session = await self._qod_session_repo.get_by_operation_id(
		operation_id, for_update=True
		)
		if qod_session is None:
		logger.warning(
		"qod_session_completed_for_unknown_operation", operation_id=event.operation_id
		)
		return

		instance = event.instances[0] if event.instances else None
		if (
		qod_session.state == QodSessionState.DELETION_REQUESTED
		and operation.status in _IN_FLIGHT_OPERATION_STATUSES
		):
		await self._finish_deferred_delete(qod_session, operation, instance)
		return

		# A DELETE issued before the activate completed is valid (ADR-0034), so this
		# completion can arrive after teardown already started. The requested delete
		# wins: without this the row would go back to AVAILABLE, the repeat DELETE would
		@@ -471,7 +484,6 @@ class QualityOnDemandService:
		)
		return

		instance = event.instances[0] if event.instances else None
		update: dict[str, object]
		qos_status: Literal["AVAILABLE", "UNAVAILABLE"]
		status_info: Literal["NETWORK_TERMINATED"] \| None
		@@ -501,6 +513,34 @@ class QualityOnDemandService:
		occurred_at=event.completed_at,
		)

		async def _finish_deferred_delete(
		self,
		qod_session: QodSession,
		activate_operation: Operation,
		instance: Optional[SRMCompletedInstance],
		) -> None:
		assert self._qod_session_repo is not None
		if instance is None or instance.status != "completed":
		logger.info(
		"deferred_qod_delete_finished_locally",
		session_id=str(qod_session.session_id),
		)
		await self._qod_session_repo.save(
		qod_session.model_copy(update={"state": QodSessionState.DELETED})
		)
		return

		qod_session = await self._qod_session_repo.save(
		qod_session.model_copy(update={"external_ref": instance.external_ref})
		)
		logger.info("deferred_qod_deactivate_requested", session_id=str(qod_session.session_id))
		await self._request_deactivation(
		qod_session,
		activate_operation.tenant_id,
		activate_operation.app_provider_id,
		x_correlator=None,
		)

		async def _handle_deactivate_completed(
		self, operation: Operation, status: OperationStatus
		) -> None:
		@@ -512,7 +552,7 @@ class QualityOnDemandService:
		)
		return

		qod_session = await self._qod_session_repo.get_by_id(UUID(session_id_raw))
		qod_session = await self._qod_session_repo.get_by_id(UUID(session_id_raw), for_update=True)
		if qod_session is None:
		logger.warning("deactivate_completed_for_unknown_session", session_id=session_id_raw)
		return
		@@ -527,7 +567,9 @@ class QualityOnDemandService:
		raise RuntimeError("Operation/QodSession repositories are not available")

		operation_id = operation.operation_id
		qod_session = await self._qod_session_repo.get_by_operation_id(operation_id)
		qod_session = await self._qod_session_repo.get_by_operation_id(
		operation_id, for_update=True
		)
		if (
		qod_session is None
		or qod_session.state in _TERMINAL_STATES
		@@ -581,18 +623,22 @@ class QualityOnDemandService:
		occurred_at=event.emitted_at,
		)

		async def _get_owned_session(self, session_id: str, app_provider_id: str) -> QodSession:
		async def _get_owned_session(
		self, session_id: str, app_provider_id: str, for_update: bool = False
		) -> tuple[QodSession, Operation]:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		qod_session = await self._qod_session_repo.get_by_id(self._parse_session_id(session_id))
		qod_session = await self._qod_session_repo.get_by_id(
		self._parse_session_id(session_id), for_update=for_update
		)
		if qod_session is None or qod_session.state in _TERMINAL_STATES:
		raise NotFoundException(message=f"Session {session_id} not found")
		# 404 rather than 403 so other callers can't probe which session IDs exist.
		activate_operation = await self._operation_repo.get_by_id(qod_session.operation_id)
		if activate_operation is None or activate_operation.app_provider_id != app_provider_id:
		raise NotFoundException(message=f"Session {session_id} not found")
		return qod_session
		return qod_session, activate_operation

		async def get_session(
		self,
		@@ -600,7 +646,7 @@ class QualityOnDemandService:
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> SessionInfo:
		qod_session = await self._get_owned_session(session_id, app_provider_id)
		qod_session, _ = await self._get_owned_session(session_id, app_provider_id)

		target = session_target_from_row(qod_session) or await self._legacy_session_target(
		session_id, x_correlator
		@@ -630,9 +676,15 @@ class QualityOnDemandService:
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> None:
		qod_session = await self._get_owned_session(session_id, app_provider_id)
		qod_session, activate_operation = await self._get_owned_session(
		session_id, app_provider_id, for_update=True
		)
		assert self._qod_session_repo is not None

		if activate_operation.status in _IN_FLIGHT_OPERATION_STATUSES:
		logger.info("qod_session_deactivate_deferred", session_id=session_id)
		_, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		else:
		correlation_id, requested_at = await self._request_deactivation(
		qod_session, tenant_id, app_provider_id, x_correlator
		)
		@@ -700,7 +752,7 @@ class QualityOnDemandService:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		qod_session = await self._qod_session_repo.get_by_id(session_id)
		qod_session = await self._qod_session_repo.get_by_id(session_id, for_update=True)
		if (
		qod_session is None
		or qod_session.state != QodSessionState.AVAILABLE
		@@ -755,7 +807,7 @@ class QualityOnDemandService:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		qod_session = await self._qod_session_repo.get_by_id(session_id)
		qod_session = await self._qod_session_repo.get_by_id(session_id, for_update=True)
		if (
		qod_session is None
		or qod_session.state != QodSessionState.UNAVAILABLE

src/open_exposure_gateway/application/services/traffic_influence_service.py

+81 −16

Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 81 added lines, 16 removed lines.
Original line number	Diff line number	Diff line
		@@ -39,6 +39,7 @@ from open_exposure_gateway.core.exceptions import (
		UnprocessableEntityException,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMCompletedInstance,
		SRMOperationCompleted,
		completion_error,
		)
		@@ -99,6 +100,8 @@ _TEARDOWN_STATES = frozenset(
		}
		)

		_IN_FLIGHT_OPERATION_STATUSES = frozenset({OperationStatus.PENDING, OperationStatus.IN_PROGRESS})


		def _reported_selected_app_instance_ids(operation: Optional[Operation]) -> list[UUID] \| None:
		if operation is None or not operation.result:
		@@ -623,13 +626,23 @@ class TrafficInfluenceService:
		succeeded = instance is not None and instance.status == "completed"

		if operation.operation_type == OperationType.TRAFFIC_INFLUENCE:
		traffic_influence = await self._traffic_influence_repo.get_by_operation_id(operation_id)
		traffic_influence = await self._traffic_influence_repo.get_by_operation_id(
		operation_id, for_update=True
		)
		if traffic_influence is None:
		logger.warning(
		"traffic_influence_completed_for_unknown_operation",
		operation_id=event.operation_id,
		)
		return
		if (
		traffic_influence.state == RecordTrafficInfluenceState.DELETION_IN_PROGRESS
		and operation.status in _IN_FLIGHT_OPERATION_STATUSES
		):
		await self._finish_deferred_delete(
		traffic_influence, operation, instance if succeeded else None
		)
		return
		if traffic_influence.state in _TEARDOWN_STATES:
		logger.info(
		"activate_completion_ignored_for_terminal_traffic_influence",
		@@ -661,7 +674,9 @@ class TrafficInfluenceService:
		operation_id=event.operation_id,
		)
		return
		traffic_influence = await self._traffic_influence_repo.get_by_id(UUID(raw_id))
		traffic_influence = await self._traffic_influence_repo.get_by_id(
		UUID(raw_id), for_update=True
		)
		if traffic_influence is None:
		return
		deactivate_update = {
		@@ -690,7 +705,9 @@ class TrafficInfluenceService:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		operation_id = operation.operation_id
		traffic_influence = await self._traffic_influence_repo.get_by_operation_id(operation_id)
		traffic_influence = await self._traffic_influence_repo.get_by_operation_id(
		operation_id, for_update=True
		)
		if traffic_influence is None:
		return

		@@ -753,13 +770,13 @@ class TrafficInfluenceService:
		return build_traffic_influence_response(traffic_influence, request_metadata)

		async def _get_owned_traffic_influence(
		self, traffic_influence_id: str, app_provider_id: str
		self, traffic_influence_id: str, app_provider_id: str, for_update: bool = False
		) -> TrafficInfluenceRecord:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		traffic_influence = await self._traffic_influence_repo.get_by_id(
		self._parse_traffic_influence_id(traffic_influence_id)
		self._parse_traffic_influence_id(traffic_influence_id), for_update=for_update
		)
		if traffic_influence is None:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")
		@@ -862,7 +879,7 @@ class TrafficInfluenceService:
		x_correlator: Optional[str] = None,
		) -> None:
		traffic_influence = await self._get_owned_traffic_influence(
		traffic_influence_id, app_provider_id
		traffic_influence_id, app_provider_id, for_update=True
		)
		if traffic_influence.state in _TEARDOWN_STATES:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")
		@@ -873,6 +890,40 @@ class TrafficInfluenceService:
		await self._delete_never_activated_traffic_influence(traffic_influence)
		return

		if (
		create_operation is not None
		and create_operation.status in _IN_FLIGHT_OPERATION_STATUSES
		):
		logger.info(
		"traffic_influence_deactivate_deferred",
		traffic_influence_id=traffic_influence_id,
		)
		requested_at = datetime.now(timezone.utc).isoformat()
		else:
		requested_at = await self._request_deactivation(
		traffic_influence, tenant_id, app_provider_id, x_correlator
		)

		was_active = traffic_influence.state == RecordTrafficInfluenceState.ACTIVE
		saved = await self._traffic_influence_repo.save(
		traffic_influence.model_copy(
		update={"state": RecordTrafficInfluenceState.DELETION_IN_PROGRESS}
		)
		)
		# Same condition QoD's delete_session uses to skip notifying (via qod_session.state
		# == AVAILABLE): don't tell subscribers something's "going away" if it was never
		# reported as up in the first place (still ORDERED/CREATED when DELETE arrived).
		if was_active:
		await self._deliver_traffic_influence_changed(saved.operation_id, saved, requested_at)

		async def _request_deactivation(
		self,
		traffic_influence: TrafficInfluenceRecord,
		tenant_id: str,
		app_provider_id: str,
		x_correlator: Optional[str],
		) -> str:
		assert self._operation_repo is not None
		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		command = build_deactivate_command(
		service_instance_id=traffic_influence.traffic_influence_id,
		@@ -891,7 +942,7 @@ class TrafficInfluenceService:
		operation_type=OperationType.TRAFFIC_INFLUENCE_DEACTIVATE,
		status=OperationStatus.PENDING,
		subject=Subject.TASK_DEACTIVATE,
		metadata={"traffic_influence_id": str(traffic_influence_id)},
		metadata={"traffic_influence_id": str(traffic_influence.traffic_influence_id)},
		)
		)

		@@ -900,18 +951,32 @@ class TrafficInfluenceService:
		command,
		"Failed to publish traffic influence deactivation",
		)
		return requested_at

		was_active = traffic_influence.state == RecordTrafficInfluenceState.ACTIVE
		saved = await self._traffic_influence_repo.save(
		traffic_influence.model_copy(
		update={"state": RecordTrafficInfluenceState.DELETION_IN_PROGRESS}
		async def _finish_deferred_delete(
		self,
		traffic_influence: TrafficInfluenceRecord,
		create_operation: Operation,
		instance: Optional[SRMCompletedInstance],
		) -> None:
		assert self._traffic_influence_repo is not None
		if instance is None:
		await self._delete_never_activated_traffic_influence(traffic_influence)
		return

		traffic_influence = await self._traffic_influence_repo.save(
		traffic_influence.model_copy(update={"external_ref": instance.external_ref})
		)
		logger.info(
		"deferred_traffic_influence_deactivate_requested",
		traffic_influence_id=str(traffic_influence.traffic_influence_id),
		)
		await self._request_deactivation(
		traffic_influence,
		create_operation.tenant_id,
		create_operation.app_provider_id,
		x_correlator=None,
		)
		# Same condition QoD's delete_session uses to skip notifying (via qod_session.state
		# == AVAILABLE): don't tell subscribers something's "going away" if it was never
		# reported as up in the first place (still ORDERED/CREATED when DELETE arrived).
		if was_active:
		await self._deliver_traffic_influence_changed(saved.operation_id, saved, requested_at)

		async def _delete_never_activated_traffic_influence(
		self, traffic_influence: TrafficInfluenceRecord

src/open_exposure_gateway/main.py

+10 −3

Changes for src/open_exposure_gateway/main.py: 10 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -355,9 +355,11 @@ def _build_traffic_influence_service(
		session: AsyncSession,
		srm_client: SRMClientPort,
		traffic_influence_callback_client: TrafficInfluenceCallbackDeliveryPort,
		publisher: Optional[DataBusPort] = None,
		) -> TrafficInfluenceService:
		return TrafficInfluenceService(
		srm_client=srm_client,
		publisher=publisher,
		operation_repo=SqlOperationRepository(session),
		traffic_influence_repo=SqlTrafficInfluenceRepository(session),
		callback_registration_repo=SqlCallbackRegistrationRepository(session),
		@@ -376,6 +378,7 @@ def _build_shared_operation_completed_handler(
		srm_client: SRMClientPort,
		qod_callback_client: QodCallbackDeliveryPort,
		traffic_influence_callback_client: TrafficInfluenceCallbackDeliveryPort,
		publisher: DataBusPort,
		) -> Callable[[SRMOperationCompleted], Awaitable[None]]:
		"""Routes ``event.srm.operation.completed`` to QoD or TI by ``operation_type``.

		@@ -399,13 +402,13 @@ def _build_shared_operation_completed_handler(
		match operation.operation_type:
		case OperationType.QOD_SESSION \| OperationType.QOD_SESSION_DEACTIVATE:
		await _build_qod_service(
		session, srm_client, qod_callback_client
		session, srm_client, qod_callback_client, publisher
		).handle_completed(event, operation)
		case (
		OperationType.TRAFFIC_INFLUENCE \| OperationType.TRAFFIC_INFLUENCE_DEACTIVATE
		):
		await _build_traffic_influence_service(
		session, srm_client, traffic_influence_callback_client
		session, srm_client, traffic_influence_callback_client, publisher
		).handle_completed(event, operation)
		case _:
		return
		@@ -641,7 +644,11 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		client=publisher.client,
		subject=Subject.OPERATION_COMPLETED,
		handler=_build_shared_operation_completed_handler(
		session_maker, srm_client, qod_callback_client, traffic_influence_callback_client
		session_maker,
		srm_client,
		qod_callback_client,
		traffic_influence_callback_client,
		publisher,
		),
		)
		await operation_completed_consumer.start()