feat: expire sessions at the of their duration (0f28b1dd) · Commits · OOP / code / open-exposure-gateway

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

+18 −1

Changes for src/open_exposure_gateway/adapters/database/repos/qod_sessions.py: 18 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		from datetime import datetime
		from typing import Any
		from uuid import UUID

		from sqlalchemy import or_, select, type_coerce
		from sqlalchemy import literal_column, or_, select, type_coerce
		from sqlalchemy.dialects.postgresql import JSONB
		from sqlalchemy.ext.asyncio import AsyncSession

		@@ -51,6 +52,22 @@ class SqlQodSessionRepository(QodSessionRepository):
		row = await self._session.scalar(stmt)
		return QodSessionMapper.to_domain(row) if row is not None else None

		async def list_available_due(self, now: datetime) -> list[QodSession]:
		ends_at = QodSessionRow.started_at + QodSessionRow.duration_seconds * literal_column(
		"INTERVAL '1 second'"
		)
		stmt = (
		select(QodSessionRow)
		.where(
		QodSessionRow.state == QodSessionState.AVAILABLE,
		QodSessionRow.started_at.is_not(None),
		ends_at <= now,
		)
		.order_by(QodSessionRow.started_at)
		)
		rows = await self._session.scalars(stmt)
		return [QodSessionMapper.to_domain(row) for row in rows]

		async def save(self, qod_session: QodSession) -> QodSession:
		merged = await self._session.merge(QodSessionMapper.to_row(qod_session))
		await self._session.flush()

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+76 −8

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 76 added lines, 8 removed lines.
Original line number	Diff line number	Diff line
		@@ -521,7 +521,11 @@ class QualityOnDemandService:
		return

		qod_session = await self._qod_session_repo.get_by_operation_id(operation_id)
		if qod_session is None:
		if (
		qod_session is None
		or qod_session.state in _TERMINAL_STATES
		or qod_session.status_info == "DURATION_EXPIRED"
		):
		return

		qos_status = (event.metadata or {}).get("qos_status")
		@@ -613,8 +617,37 @@ class QualityOnDemandService:
		x_correlator: Optional[str] = None,
		) -> None:
		qod_session = await self._get_owned_session(session_id, app_provider_id)
		assert self._operation_repo is not None and self._qod_session_repo is not None
		assert self._qod_session_repo is not None

		correlation_id, requested_at = await self._request_deactivation(
		qod_session, tenant_id, app_provider_id, x_correlator
		)

		await self._qod_session_repo.save(
		qod_session.model_copy(update={"state": QodSessionState.DELETION_REQUESTED})
		)

		if qod_session.state == QodSessionState.AVAILABLE:
		await self._deliver_qos_status_changed(
		operation_id=qod_session.operation_id,
		correlation_id=correlation_id,
		session_id=qod_session.session_id,
		qos_status="UNAVAILABLE",
		status_info="DELETE_REQUESTED",
		occurred_at=requested_at,
		)

		# DELETE_REQUESTED is the session's last notification: drop the token.
		await self._forget_callback_credential(qod_session.operation_id)

		async def _request_deactivation(
		self,
		qod_session: QodSession,
		tenant_id: str,
		app_provider_id: str,
		x_correlator: Optional[str],
		) -> tuple[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=qod_session.session_id,
		@@ -633,7 +666,7 @@ class QualityOnDemandService:
		operation_type=OperationType.NETWORK_CAPABILITY_DEACTIVATE,
		status=OperationStatus.PENDING,
		subject=Subject.TASK_DEACTIVATE,
		metadata={"session_id": session_id},
		metadata={"session_id": str(qod_session.session_id)},
		)
		)

		@@ -642,23 +675,58 @@ class QualityOnDemandService:
		command,
		"Failed to publish QoD session deactivation",
		)
		return correlation_id, requested_at

		async def list_due_session_ids(self, now: datetime) -> list[UUID]:
		if self._qod_session_repo is None:
		raise RuntimeError("QodSession repository is not available")
		return [s.session_id for s in await self._qod_session_repo.list_available_due(now)]

		async def expire_session(self, session_id: UUID, now: datetime) -> bool:
		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)
		if (
		qod_session is None
		or qod_session.state != QodSessionState.AVAILABLE
		or qod_session.started_at is None
		or qod_session.started_at + timedelta(seconds=qod_session.duration_seconds) > now
		):
		return False

		activate_operation = await self._operation_repo.get_by_id(qod_session.operation_id)
		if activate_operation is None:
		logger.warning("qod_session_expiry_without_operation", session_id=str(session_id))
		return False

		correlation_id, _ = await self._request_deactivation(
		qod_session,
		activate_operation.tenant_id,
		activate_operation.app_provider_id,
		x_correlator=None,
		)

		await self._qod_session_repo.save(
		qod_session.model_copy(update={"state": QodSessionState.DELETION_REQUESTED})
		qod_session.model_copy(
		update={
		"state": QodSessionState.UNAVAILABLE,
		**_transition_fields("UNAVAILABLE", "DURATION_EXPIRED", now),
		}
		)
		)

		if qod_session.state == QodSessionState.AVAILABLE:
		await self._deliver_qos_status_changed(
		operation_id=qod_session.operation_id,
		correlation_id=correlation_id,
		session_id=qod_session.session_id,
		qos_status="UNAVAILABLE",
		status_info="DELETE_REQUESTED",
		occurred_at=requested_at,
		status_info="DURATION_EXPIRED",
		occurred_at=now.isoformat(),
		)

		# DELETE_REQUESTED is the session's last notification: drop the token.
		await self._forget_callback_credential(qod_session.operation_id)
		return True

		async def _forget_callback_credential(self, operation_id: UUID) -> None:
		if self._callback_registration_repo is None or self._callback_credential_repo is None:

src/open_exposure_gateway/main.py

+48 −1

Changes for src/open_exposure_gateway/main.py: 48 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -74,6 +74,7 @@ from open_exposure_gateway.domain.edge_application_management import (
		from open_exposure_gateway.domain.quality_on_demand import SRMOperationStatus
		from open_exposure_gateway.domain.quality_on_demand import Subject as QodSubject
		from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort
		from open_exposure_gateway.ports.databus_port import DataBusPort
		from open_exposure_gateway.ports.qod_callback_port import QodCallbackDeliveryPort
		from open_exposure_gateway.ports.srm_port import SRMClientPort

		@@ -138,6 +139,8 @@ async def _run_reconciliation_pass(
		srm_client: SRMClientPort,
		callback_delivery_port: CallbackDeliveryPort,
		deadline: timedelta,
		publisher: DataBusPort,
		qod_callback_client: QodCallbackDeliveryPort,
		) -> None:
		# One transaction per operation so a single failure does not block the rest.
		logger = structlog.get_logger()
		@@ -153,6 +156,8 @@ async def _run_reconciliation_pass(
		await session.rollback()
		logger.exception("callback_credential_purge_failed")

		await _expire_due_qod_sessions(session_maker, srm_client, publisher, qod_callback_client)

		async with session_maker() as session:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		stale = await service.list_stale_operations(deadline)
		@@ -173,19 +178,57 @@ async def _run_reconciliation_pass(
		logger.exception("reconciliation_force_fail_failed", operation_id=str(operation_id))


		async def _expire_due_qod_sessions(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		publisher: DataBusPort,
		qod_callback_client: QodCallbackDeliveryPort,
		) -> None:
		logger = structlog.get_logger()
		now = datetime.now(timezone.utc)
		try:
		async with session_maker() as session:
		service = _build_qod_service(session, srm_client, qod_callback_client, publisher)
		session_ids = await service.list_due_session_ids(now)
		except Exception:
		logger.exception("qod_session_expiry_listing_failed")
		return

		expired = 0
		for session_id in session_ids:
		async with session_maker() as session:
		try:
		service = _build_qod_service(session, srm_client, qod_callback_client, publisher)
		if await service.expire_session(session_id, now):
		expired += 1
		await session.commit()
		except Exception:
		await session.rollback()
		logger.exception("qod_session_expiry_failed", session_id=str(session_id))
		if expired:
		logger.info("qod_sessions_expired", count=expired)


		async def _reconciliation_loop(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		callback_delivery_port: CallbackDeliveryPort,
		interval: float,
		deadline: timedelta,
		publisher: DataBusPort,
		qod_callback_client: QodCallbackDeliveryPort,
		) -> None:
		logger = structlog.get_logger()
		while True:
		try:
		await asyncio.sleep(interval)
		await _run_reconciliation_pass(
		session_maker, srm_client, callback_delivery_port, deadline
		session_maker,
		srm_client,
		callback_delivery_port,
		deadline,
		publisher,
		qod_callback_client,
		)
		except asyncio.CancelledError:
		raise
		@@ -197,9 +240,11 @@ def _build_qod_service(
		session: AsyncSession,
		srm_client: SRMClientPort,
		qod_callback_client: QodCallbackDeliveryPort,
		publisher: Optional[DataBusPort] = None,
		) -> QualityOnDemandService:
		return QualityOnDemandService(
		srm_client=srm_client,
		publisher=publisher,
		operation_repo=SqlOperationRepository(session),
		qod_session_repo=SqlQodSessionRepository(session),
		callback_registration_repo=SqlCallbackRegistrationRepository(session),
		@@ -421,6 +466,8 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		HttpCallbackClient(),
		interval=reconciliation_settings.interval_seconds,
		deadline=timedelta(seconds=reconciliation_settings.operation_deadline_seconds),
		publisher=publisher,
		qod_callback_client=qod_callback_client,
		)
		)
		logger.info(

src/open_exposure_gateway/ports/database/qod_sessions.py

+6 −0

Changes for src/open_exposure_gateway/ports/database/qod_sessions.py: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""QoD session repository ports."""

		from abc import ABC, abstractmethod
		from datetime import datetime
		from typing import Any
		from uuid import UUID

		@@ -22,6 +23,11 @@ class QodSessionRepository(ABC):
		) -> QodSession \| None:
		pass

		@abstractmethod
		async def list_available_due(self, now: datetime) -> list[QodSession]:
		"""AVAILABLE sessions whose started_at + duration_seconds <= now."""
		pass

		@abstractmethod
		async def save(self, qod_session: QodSession) -> QodSession:
		pass

tests/integration/test_postgres.py

+29 −1

Changes for tests/integration/test_postgres.py: 29 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -462,7 +462,12 @@ async def test_reconciliation_pass_purges_expired_callback_credentials(
		await session.commit()

		await _run_reconciliation_pass(
		session_maker, AsyncMock(), AsyncMock(), deadline=timedelta(days=365)
		session_maker,
		AsyncMock(),
		AsyncMock(),
		deadline=timedelta(days=365),
		publisher=AsyncMock(),
		qod_callback_client=AsyncMock(),
		)

		async with session_maker() as session:
		@@ -601,6 +606,29 @@ async def test_qod_session_repo_find_active_by_device_scopes_by_identifier_state
		assert await repo.find_active_by_device({"phoneNumber": "+123456789"}, "tenant-a") is None


		async def test_qod_session_repo_lists_available_sessions_past_their_duration(
		db_session: AsyncSession,
		) -> None:
		operation = await SqlOperationRepository(db_session).save(_operation())
		repo = SqlQodSessionRepository(db_session)
		now = datetime.now(timezone.utc)

		async def seed(state: QodSessionState, started_ago: timedelta \| None) -> QodSession:
		qod_session = _qod_session(operation_id=operation.operation_id)
		qod_session.state = state
		qod_session.duration_seconds = 60
		qod_session.started_at = now - started_ago if started_ago is not None else None
		return await repo.save(qod_session)

		due = await seed(QodSessionState.AVAILABLE, timedelta(seconds=60))
		await seed(QodSessionState.AVAILABLE, timedelta(seconds=59))
		await seed(QodSessionState.AVAILABLE, None)
		await seed(QodSessionState.UNAVAILABLE, timedelta(hours=1))
		await seed(QodSessionState.DELETION_REQUESTED, timedelta(hours=1))

		assert [s.session_id for s in await repo.list_available_due(now)] == [due.session_id]


		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)