feat: reconcile operations stranded by a lost completion event (3f4cedea) · Commits · OOP / code / open-exposure-gateway

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

+21 −1

Original line number	Diff line number	Diff line
		from collections.abc import Collection
		from datetime import datetime
		from uuid import UUID

		from sqlalchemy import select
		@@ -7,7 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
		from open_exposure_gateway.adapters.database.mappers import OperationMapper
		from open_exposure_gateway.adapters.database.sql import OperationRow
		from open_exposure_gateway.adapters.errors import DuplicateOperationError
		from open_exposure_gateway.domain.models import Operation
		from open_exposure_gateway.domain.models import Operation, OperationStatus, OperationType
		from open_exposure_gateway.ports.database.operations import OperationRepository

		_UNIQUE_VIOLATION = "23505"
		@@ -44,3 +46,21 @@ class SqlOperationRepository(OperationRepository):
		if saved is None:
		raise RuntimeError("Saved operation could not be reloaded")
		return saved

		async def list_stale(
		self,
		statuses: Collection[OperationStatus],
		operation_types: Collection[OperationType],
		older_than: datetime,
		) -> list[Operation]:
		stmt = (
		select(OperationRow)
		.where(
		OperationRow.status.in_(statuses),
		OperationRow.operation_type.in_(operation_types),
		OperationRow.created_at < older_than,
		)
		.order_by(OperationRow.created_at)
		)
		rows = await self._session.scalars(stmt)
		return [OperationMapper.to_domain(row) for row in rows]

src/open_exposure_gateway/application/services/edge_application_management_service.py

+108 −1

Original line number	Diff line number	Diff line
		from collections import defaultdict
		from datetime import datetime, timezone
		from datetime import datetime, timedelta, timezone
		from typing import Any, Optional
		from uuid import UUID, uuid4

		@@ -124,6 +124,36 @@ _TERMINAL_OPERATION_STATUSES = frozenset(
		}
		)

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

		# Operation types finalised by handle_completed() off event.srm.operation.completed,
		# and therefore the ones the reconciliation sweep may force-fail. NETWORK_CAPABILITY*
		# are QoD's and are driven by QualityOnDemandService's own handler -- never swept here.
		_RECONCILABLE_OPERATION_TYPES = frozenset(
		{
		OperationType.DEPLOY,
		OperationType.TERMINATE,
		}
		)

		# Recorded on operation.error (and surfaced to the app provider through the normal
		# failure callback) when the sweep force-fails an operation that never received a
		# completion event. The SRM-side workload may still exist and need manual cleanup.
		_RECONCILE_TIMEOUT_ERROR: dict[str, Any] = {
		"type": "https://etsi.org/sdg/oop/problems/operation-reconcile-timeout",
		"title": "Operation Reconcile Timeout",
		"detail": (
		"No completion event was received within the reconciliation deadline. "
		"The operation was force-failed to unblock the edge cloud zone; the "
		"underlying SRM resources may still exist and require manual cleanup."
		),
		}

		# Vendored EAM event type. A subscription registered with this type gets the
		# onAppDeploymentStatusChange contract (one delivery, array of CloudEvents);
		# anything else falls through to the per-instance onAppInstanceStatusChange path.
		@@ -142,6 +172,25 @@ def _deployment_is_final(app_deployment: AppDeployment) -> bool:
		return app_deployment.state == AppDeploymentState.TERMINATED


		def _build_reconcile_timeout_event(operation: Operation) -> SRMOperationCompleted:
		"""A synthetic terminal-failure event for an operation that timed out.

		The reconciliation sweep feeds this back through handle_completed() so a
		stranded operation fails through the exact path a real SRM failure takes --
		operation row, app_instances fan-out, app_deployments rollup and provider
		callback all move identically, with nothing duplicated here.
		"""
		return SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(operation.operation_id),
		status="failed",
		instances=[],
		error=_RECONCILE_TIMEOUT_ERROR,
		correlation_id=operation.correlation_id,
		completed_at=datetime.now(timezone.utc).isoformat(),
		)


		def _log_stale_completion(app_instance: AppInstance, operation_id: UUID) -> None:
		logger.info(
		"stale_completion_ignored_for_final_app_instance",
		@@ -998,6 +1047,64 @@ class EdgeApplicationManagementService:
		if updated_instances:
		await self._deliver_callbacks(operation_id, event.completed_at, updated_instances)

		async def list_stale_operations(
		self, deadline: timedelta, now: Optional[datetime] = None
		) -> list[Operation]:
		"""EAM operations still non-terminal `deadline` after they were created.

		These are the candidates for force-failing. handle_completed() is the only
		thing that moves an operation out of PENDING/IN_PROGRESS, and it runs off
		an at-most-once NATS subject, so a dropped completion event strands the
		operation -- and with it the zone (exists_in_zone keeps returning true) and
		any re-DELETE (the guard rejects TERMINATING).
		"""
		if self._operation_repo is None:
		raise RuntimeError("OperationRepository is not available")
		now = now or datetime.now(timezone.utc)
		return await self._operation_repo.list_stale(
		statuses=_NON_TERMINAL_OPERATION_STATUSES,
		operation_types=_RECONCILABLE_OPERATION_TYPES,
		older_than=now - deadline,
		)

		async def force_fail_stale_operation(self, operation_id: UUID) -> bool:
		"""Force one stranded operation to FAILED, unblocking its zone.

		Routes a synthetic terminal-failure event through handle_completed() so the
		operation row, its app_instances, the app_deployments rollup and the
		provider failure callback all move exactly as a real SRM failure would --
		nothing here duplicates that logic. Returns False (a logged no-op) if the
		operation has vanished or already reached a terminal state, which makes
		this safe to run from every replica and safe if a real event lands first.
		"""
		if self._operation_repo is None:
		raise RuntimeError("OperationRepository is not available")
		operation = await self._operation_repo.get_by_id(operation_id)
		if operation is None:
		return False
		if operation.status in _TERMINAL_OPERATION_STATUSES:
		logger.info(
		"reconcile_skipped_already_terminal_operation",
		operation_id=str(operation_id),
		status=operation.status.value,
		)
		return False

		logger.warning(
		"operation_reconcile_timeout",
		operation_id=str(operation_id),
		operation_type=operation.operation_type.value,
		status=operation.status.value,
		created_at=operation.created_at.isoformat() if operation.created_at else None,
		detail=(
		"no completion event within the reconciliation deadline; force-failing "
		"to unblock the zone -- SRM may hold an orphaned resource that needs "
		"manual cleanup"
		),
		)
		await self.handle_completed(_build_reconcile_timeout_event(operation))
		return True

		async def _deliver_callbacks(
		self,
		operation_id: UUID,

src/open_exposure_gateway/core/config.py

+19 −0

Original line number	Diff line number	Diff line
		@@ -69,6 +69,24 @@ class LocationRetrievalSettings(BaseModel):
		return self.service_specification_id == DEFAULT_LOCATION_RETRIEVAL_SERVICE_SPECIFICATION_ID


		class ReconciliationSettings(BaseModel):
		"""Sweep that force-fails EAM operations stranded by a lost completion event.

		event.srm.operation.completed rides an at-most-once NATS subject, so a single
		dropped event leaves an operation in PENDING/IN_PROGRESS forever -- which 409s
		the zone (exists_in_zone) and blocks a re-DELETE. This sweep is the recovery
		until the subject becomes a durable JetStream consumer.
		"""

		enabled: bool = True
		# How often the sweep runs.
		interval_seconds: int = 60
		# An operation still non-terminal this long after creation is force-failed.
		# Keep it comfortably longer than the slowest real SRM deploy/terminate so
		# only genuine stalls are hit.
		operation_deadline_seconds: int = 1800


		class Settings(BaseSettings):
		model_config = SettingsConfigDict(
		env_file=".env",
		@@ -91,6 +109,7 @@ class Settings(BaseSettings):
		callback_settings: CallbackSettings = CallbackSettings()
		qod_settings: QodSettings = QodSettings()
		location_retrieval_settings: LocationRetrievalSettings = LocationRetrievalSettings()
		reconciliation_settings: ReconciliationSettings = ReconciliationSettings()


		@lru_cache

src/open_exposure_gateway/main.py

+93 −11

Original line number	Diff line number	Diff line
		import asyncio
		from collections.abc import AsyncGenerator, Awaitable, Callable
		from contextlib import asynccontextmanager
		from contextlib import asynccontextmanager, suppress
		from datetime import timedelta
		from typing import Optional

		import structlog
		@@ -65,15 +67,12 @@ from open_exposure_gateway.ports.qod_callback_port import QodCallbackDeliveryPor
		from open_exposure_gateway.ports.srm_port import SRMClientPort


		def _build_operation_completed_handler(
		session_maker: async_sessionmaker[AsyncSession],
		def _build_eam_service(
		session: AsyncSession,
		srm_client: SRMClientPort,
		callback_delivery_port: CallbackDeliveryPort,
		) -> Callable[[SRMOperationCompleted], Awaitable[None]]:
		async def handle(event: SRMOperationCompleted) -> None:
		async with session_maker() as session:
		try:
		service = EdgeApplicationManagementService(
		) -> EdgeApplicationManagementService:
		return EdgeApplicationManagementService(
		srm_client=srm_client,
		app_registration_repo=SqlAppRegistrationRepository(session),
		operation_repo=SqlOperationRepository(session),
		@@ -83,6 +82,17 @@ def _build_operation_completed_handler(
		callback_delivery_port=callback_delivery_port,
		callback_delivery_repo=SqlCallbackDeliveryRepository(session),
		)


		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:
		async with session_maker() as session:
		try:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		await service.handle_completed(event)
		await session.commit()
		except Exception:
		@@ -92,6 +102,55 @@ def _build_operation_completed_handler(
		return handle


		async def _run_reconciliation_pass(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		callback_delivery_port: CallbackDeliveryPort,
		deadline: timedelta,
		) -> None:
		"""One sweep: list operations stranded past `deadline`, then force-fail each in
		its own transaction so one poison operation can't block the rest."""
		logger = structlog.get_logger()
		async with session_maker() as session:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		stale = await service.list_stale_operations(deadline)

		operation_ids = [operation.operation_id for operation in stale]
		if not operation_ids:
		return

		logger.info("reconciliation_found_stale_operations", count=len(operation_ids))
		for operation_id in operation_ids:
		async with session_maker() as session:
		try:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		await service.force_fail_stale_operation(operation_id)
		await session.commit()
		except Exception:
		await session.rollback()
		logger.exception("reconciliation_force_fail_failed", operation_id=str(operation_id))


		async def _reconciliation_loop(
		session_maker: async_sessionmaker[AsyncSession],
		srm_client: SRMClientPort,
		callback_delivery_port: CallbackDeliveryPort,
		interval: float,
		deadline: timedelta,
		) -> None:
		logger = structlog.get_logger()
		while True:
		try:
		await asyncio.sleep(interval)
		await _run_reconciliation_pass(
		session_maker, srm_client, callback_delivery_port, deadline
		)
		except asyncio.CancelledError:
		raise
		except Exception:
		logger.exception("reconciliation_pass_failed")


		def _build_qod_service(
		session: AsyncSession,
		srm_client: SRMClientPort,
		@@ -272,6 +331,24 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		await operation_status_consumer.start()
		logger.info("NATS consumer started", subject=QodSubject.OPERATION_STATUS)

		reconciliation_settings = settings.reconciliation_settings
		reconciliation_task: Optional[asyncio.Task[None]] = None
		if reconciliation_settings.enabled:
		reconciliation_task = asyncio.create_task(
		_reconciliation_loop(
		session_maker,
		srm_client,
		HttpCallbackClient(),
		interval=reconciliation_settings.interval_seconds,
		deadline=timedelta(seconds=reconciliation_settings.operation_deadline_seconds),
		)
		)
		logger.info(
		"Reconciliation loop started",
		interval_seconds=reconciliation_settings.interval_seconds,
		operation_deadline_seconds=reconciliation_settings.operation_deadline_seconds,
		)

		app.state.srm_client = srm_client
		app.state.publisher = publisher
		app.state.db_engine = db_engine
		@@ -281,6 +358,11 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		yield

		logger.info("Shutting down application")
		if reconciliation_task is not None:
		reconciliation_task.cancel()
		with suppress(asyncio.CancelledError):
		await reconciliation_task
		logger.info("Reconciliation loop stopped")
		await publisher.close()
		logger.info("NATS publisher closed")
		await db_engine.dispose()

src/open_exposure_gateway/ports/database/operations.py

+13 −1

Original line number	Diff line number	Diff line
		"""Operation repository ports."""

		from abc import ABC, abstractmethod
		from collections.abc import Collection
		from datetime import datetime
		from uuid import UUID

		from open_exposure_gateway.domain.models import Operation
		from open_exposure_gateway.domain.models import Operation, OperationStatus, OperationType


		class OperationRepository(ABC):
		@@ -20,3 +22,13 @@ class OperationRepository(ABC):
		@abstractmethod
		async def save(self, operation: Operation) -> Operation:
		pass

		@abstractmethod
		async def list_stale(
		self,
		statuses: Collection[OperationStatus],
		operation_types: Collection[OperationType],
		older_than: datetime,
		) -> list[Operation]:
		"""Operations in one of `statuses` and `operation_types`, created before
		`older_than`, oldest first. Backs the reconciliation sweep."""