feat: delete NETWORK_TERMINATED sessions 360s after the event (65284f73) · Commits · OOP / code / open-exposure-gateway

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

+14 −0

Changes for src/open_exposure_gateway/adapters/database/repos/qod_sessions.py: 14 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -68,6 +68,20 @@ class SqlQodSessionRepository(QodSessionRepository):
		rows = await self._session.scalars(stmt)
		return [QodSessionMapper.to_domain(row) for row in rows]

		async def list_network_terminated(self, ended_before: datetime) -> list[QodSession]:
		stmt = (
		select(QodSessionRow)
		.where(
		QodSessionRow.state == QodSessionState.UNAVAILABLE,
		QodSessionRow.status_info == "NETWORK_TERMINATED",
		QodSessionRow.ended_at.is_not(None),
		QodSessionRow.ended_at <= ended_before,
		)
		.order_by(QodSessionRow.ended_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

+43 −0

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 43 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -80,6 +80,8 @@ _QOS_STATUS_INFO_MAP: dict[str, Literal["DURATION_EXPIRED", "NETWORK_TERMINATED"
		# Lets the final DURATION_EXPIRED / NETWORK_TERMINATED notification still carry the token.
		SINK_CREDENTIAL_GRACE = timedelta(hours=1)

		NETWORK_TERMINATED_DELETE_DELAY = timedelta(seconds=360)

		_TERMINAL_STATES = frozenset(
		{QodSessionState.DELETION_REQUESTED, QodSessionState.DELETED, QodSessionState.ERROR}
		)
		@@ -728,6 +730,47 @@ class QualityOnDemandService:
		await self._forget_callback_credential(qod_session.operation_id)
		return True

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

		async def delete_terminated_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.UNAVAILABLE
		or qod_session.status_info != "NETWORK_TERMINATED"
		or qod_session.ended_at is None
		or qod_session.ended_at + NETWORK_TERMINATED_DELETE_DELAY > 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_cleanup_without_operation", session_id=str(session_id))
		return False

		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})
		)

		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:
		return

src/open_exposure_gateway/main.py

+32 −0

Changes for src/open_exposure_gateway/main.py: 32 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -157,6 +157,7 @@ async def _run_reconciliation_pass(
		logger.exception("callback_credential_purge_failed")

		await _expire_due_qod_sessions(session_maker, srm_client, publisher, qod_callback_client)
		await _delete_terminated_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)
		@@ -209,6 +210,37 @@ async def _expire_due_qod_sessions(
		logger.info("qod_sessions_expired", count=expired)


		async def _delete_terminated_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_terminated_session_ids(now)
		except Exception:
		logger.exception("qod_session_cleanup_listing_failed")
		return

		deleted = 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.delete_terminated_session(session_id, now):
		deleted += 1
		await session.commit()
		except Exception:
		await session.rollback()
		logger.exception("qod_session_cleanup_failed", session_id=str(session_id))
		if deleted:
		logger.info("qod_terminated_sessions_deleted", count=deleted)


		async def _reconciliation_loop(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,

src/open_exposure_gateway/ports/database/qod_sessions.py

+5 −0

Changes for src/open_exposure_gateway/ports/database/qod_sessions.py: 5 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -28,6 +28,11 @@ class QodSessionRepository(ABC):
		"""AVAILABLE sessions whose started_at + duration_seconds <= now."""
		pass

		@abstractmethod
		async def list_network_terminated(self, ended_before: datetime) -> list[QodSession]:
		"""UNAVAILABLE / NETWORK_TERMINATED sessions whose ended_at <= ended_before."""
		pass

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

tests/integration/test_postgres.py

+29 −0

Changes for tests/integration/test_postgres.py: 29 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import asyncio
		from datetime import datetime, timedelta, timezone
		from typing import Literal
		from unittest.mock import AsyncMock
		from uuid import UUID, uuid4

		@@ -629,6 +630,34 @@ async def test_qod_session_repo_lists_available_sessions_past_their_duration(
		assert [s.session_id for s in await repo.list_available_due(now)] == [due.session_id]


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

		async def seed(
		state: QodSessionState,
		status_info: Literal["DURATION_EXPIRED", "NETWORK_TERMINATED"] \| None,
		ended_at: datetime \| None,
		) -> QodSession:
		qod_session = _qod_session(operation_id=operation.operation_id)
		qod_session.state = state
		qod_session.status_info = status_info
		qod_session.ended_at = ended_at
		return await repo.save(qod_session)

		old = timedelta(seconds=1)
		due = await seed(QodSessionState.UNAVAILABLE, "NETWORK_TERMINATED", cutoff)
		await seed(QodSessionState.UNAVAILABLE, "NETWORK_TERMINATED", cutoff + old)
		await seed(QodSessionState.UNAVAILABLE, "NETWORK_TERMINATED", None)
		await seed(QodSessionState.UNAVAILABLE, "DURATION_EXPIRED", cutoff - old)
		await seed(QodSessionState.DELETION_REQUESTED, "NETWORK_TERMINATED", cutoff - old)

		assert [s.session_id for s in await repo.list_network_terminated(cutoff)] == [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)