fix: retry lookup when SRM reply arrives before the commit (0c3fc1c1) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/main.py

+22 −0

Changes for src/open_exposure_gateway/main.py: 22 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -4,6 +4,7 @@ from collections.abc import AsyncGenerator, Awaitable, Callable
		from contextlib import asynccontextmanager, suppress
		from datetime import datetime, timedelta, timezone
		from typing import Optional
		from uuid import UUID

		import structlog
		from fastapi import FastAPI, Request, Response
		@@ -130,12 +131,30 @@ def _build_eam_service(
		)


		_OPERATION_LOOKUP_RETRY_DELAYS: tuple[float, ...] = (0.05, 0.1, 0.2, 0.4)


		async def _wait_for_operation(
		session_maker: async_sessionmaker[AsyncSession], operation_id: str
		) -> None:
		try:
		parsed_id = UUID(operation_id)
		except ValueError:
		return
		for delay in _OPERATION_LOOKUP_RETRY_DELAYS:
		async with session_maker() as session:
		if await SqlOperationRepository(session).get_by_id(parsed_id) is not None:
		return
		await asyncio.sleep(delay)


		def _build_operation_completed_handler(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		callback_delivery_port: CallbackDeliveryPort,
		) -> Callable[[SRMOperationCompleted], Awaitable[None]]:
		async def handle(event: SRMOperationCompleted) -> None:
		await _wait_for_operation(session_maker, event.operation_id)
		async with session_maker() as session:
		try:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		@@ -154,6 +173,7 @@ def _build_eam_operation_status_handler(
		callback_delivery_port: CallbackDeliveryPort,
		) -> Callable[[SRMOperationStatus], Awaitable[None]]:
		async def handle(event: SRMOperationStatus) -> None:
		await _wait_for_operation(session_maker, event.operation_id)
		async with session_maker() as session:
		try:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		@@ -356,6 +376,7 @@ def _build_qod_operation_completed_handler(
		"""

		async def handle(event: SRMOperationCompleted) -> None:
		await _wait_for_operation(session_maker, event.operation_id)
		async with session_maker() as session:
		try:
		qod_service = _build_qod_service(session, srm_client, qod_callback_client)
		@@ -386,6 +407,7 @@ def _build_qod_operation_status_handler(
		"""

		async def handle(event: SRMOperationStatus) -> None:
		await _wait_for_operation(session_maker, event.operation_id)
		async with session_maker() as session:
		try:
		qod_service = _build_qod_service(session, srm_client, qod_callback_client)

tests/integration/test_postgres.py

+66 −1

Changes for tests/integration/test_postgres.py: 66 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -29,6 +29,10 @@ from open_exposure_gateway.adapters.errors import (
		DuplicateAppRegistrationError,
		DuplicateOperationError,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMCompletedInstance,
		SRMOperationCompleted,
		)
		from open_exposure_gateway.domain.models import (
		AppInstance,
		AppInstanceState,
		@@ -44,7 +48,10 @@ from open_exposure_gateway.domain.models import (
		QodSession,
		QodSessionState,
		)
		from open_exposure_gateway.main import _run_reconciliation_pass
		from open_exposure_gateway.main import (
		_build_qod_operation_completed_handler,
		_run_reconciliation_pass,
		)


		def _app_registration() -> AppRegistration:
		@@ -517,6 +524,64 @@ async def test_reconciliation_pass_purges_expired_callback_credentials(
		assert kept.encrypted_token is not None


		async def test_completion_arriving_before_its_operation_commits_is_still_applied(
		db_engine: AsyncEngine,
		) -> None:
		session_maker = async_sessionmaker(db_engine, expire_on_commit=False)
		async with session_maker() as session:
		activate_operation = await SqlOperationRepository(session).save(_operation())
		qod_session = await SqlQodSessionRepository(session).save(
		_qod_session(activate_operation.operation_id).model_copy(
		update={"state": QodSessionState.DELETION_REQUESTED}
		)
		)
		await session.commit()

		deactivate_operation = _operation().model_copy(
		update={
		"operation_type": OperationType.NETWORK_CAPABILITY_DEACTIVATE,
		"subject": "command.srm.network_capability.deactivate",
		"metadata": {"session_id": str(qod_session.session_id)},
		}
		)
		handler = _build_qod_operation_completed_handler(
		session_maker, AsyncMock(), AsyncMock(), AsyncMock()
		)
		handling = asyncio.ensure_future(
		handler(
		SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(deactivate_operation.operation_id),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(qod_session.session_id),
		zone_id=str(uuid4()),
		status="completed",
		)
		],
		correlation_id=deactivate_operation.correlation_id,
		completed_at=datetime.now(timezone.utc).isoformat(),
		)
		)
		)
		await asyncio.sleep(0.1)
		async with session_maker() as session:
		await SqlOperationRepository(session).save(deactivate_operation)
		await session.commit()
		await handling

		async with session_maker() as session:
		operation = await SqlOperationRepository(session).get_by_id(
		deactivate_operation.operation_id
		)
		reloaded = await SqlQodSessionRepository(session).get_by_id(qod_session.session_id)
		assert operation is not None
		assert operation.status == OperationStatus.COMPLETED
		assert reloaded is not None
		assert reloaded.state == QodSessionState.DELETED


		async def test_callback_delivery_repo_persists_and_lists(db_session: AsyncSession) -> None:
		operation = await SqlOperationRepository(db_session).save(_operation())
		callback_registration = await SqlCallbackRegistrationRepository(db_session).save(