fix: hold a deployment operatoin open until every fan-out instance reports (ddd2ec5b) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+123 −10

Original line number	Diff line number	Diff line
		@@ -46,6 +46,7 @@ from open_exposure_gateway.core.exceptions import (
		NotImplementedException,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMCompletedInstance,
		SRMOperationCompleted,
		Subject,
		)
		@@ -200,6 +201,53 @@ def _log_stale_completion(app_instance: AppInstance, operation_id: UUID) -> None
		)


		# An app_instance in one of these states has already reported its outcome for
		# the fan-out; a debounced completion that omits it is still "complete enough"
		# for that instance.
		_SETTLED_APP_INSTANCE_STATES = frozenset(
		{
		AppInstanceState.READY,
		AppInstanceState.FAILED,
		AppInstanceState.TERMINATED,
		}
		)


		def _expected_instance_keys(operation: Operation) -> Optional[set[str]]:
		"""The fan-out target set recorded when a multi-zone /deployments request
		put N commands under one operation_id.

		POST /deployments stashes the target zone ids (metadata.edge_cloud_zones);
		DELETE /deployments stashes the target app_instance ids
		(metadata.app_instance_ids). A single-zone /appinstances operation stashes
		neither -- returns None, and handle_completed keeps trusting event.status.
		"""
		raw = (
		operation.metadata.get("app_instance_ids")
		if operation.operation_type == OperationType.TERMINATE
		else operation.metadata.get("edge_cloud_zones")
		)
		if not raw:
		return None
		return {str(key) for key in raw}


		def _completed_instance_key(operation: Operation, instance: SRMCompletedInstance) -> str:
		"""The half of a completion instance that lines up with
		_expected_instance_keys: the zone for a deploy, the service_instance_id for
		a terminate."""
		if operation.operation_type == OperationType.TERMINATE:
		return instance.service_instance_id
		return instance.zone_id


		def _persisted_instance_key(operation: Operation, app_instance: AppInstance) -> str:
		"""Same key, taken from a stored app_instance row."""
		if operation.operation_type == OperationType.TERMINATE:
		return str(app_instance.app_instance_id)
		return str(app_instance.edge_cloud_zone_id)


		class EdgeApplicationManagementService:
		def __init__(
		self,
		@@ -935,6 +983,31 @@ class EdgeApplicationManagementService:
		return

		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]

		# Multi-zone POST/DELETE /deployments fan N commands out under one
		# operation_id; SRM debounces the N terminal events into a single
		# completion. If that completion doesn't account for every targeted
		# instance (a lost fan-out event, an early debounce flush), the
		# operation is NOT finished -- leave it IN_PROGRESS and roll the
		# aggregate to PARTIAL so a later completion, or the reconcile sweep,
		# settles it, rather than marking the deployment ready/terminated with
		# an instance stranded at instantiating/terminating.
		expected_keys = _expected_instance_keys(operation)
		partial_delivery = False
		if expected_keys is not None and event.status != "failed":
		reported_keys = await self._reported_or_settled_keys(operation, expected_keys, event)
		missing_keys = expected_keys - reported_keys
		if missing_keys:
		partial_delivery = True
		status = OperationStatus.IN_PROGRESS
		logger.warning(
		"operation_completion_missing_expected_instances",
		operation_id=event.operation_id,
		operation_type=operation.operation_type.value,
		missing=sorted(missing_keys),
		reported=sorted(reported_keys),
		)

		result: Optional[dict[str, Any]] = None
		if status != OperationStatus.FAILED:
		result = {
		@@ -948,14 +1021,14 @@ class EdgeApplicationManagementService:
		]
		}

		updated = operation.model_copy(
		update={
		operation_update: dict[str, Any] = {
		"status": status,
		"result": result,
		"error": event.error,
		"completed_at": datetime.fromisoformat(event.completed_at),
		}
		)
		if not partial_delivery:
		operation_update["completed_at"] = datetime.fromisoformat(event.completed_at)
		updated = operation.model_copy(update=operation_update)
		await self._operation_repo.save(updated)

		is_terminate = operation.operation_type == OperationType.TERMINATE
		@@ -1011,7 +1084,12 @@ class EdgeApplicationManagementService:
		# lookup happens to return.
		failed_instances = await self._app_instance_repo.list_by_operation_id(operation_id)
		for app_instance in failed_instances:
		if _is_final(app_instance):
		# Skip rows a prior partial completion already settled: a deploy
		# straggler being force-failed by the reconcile sweep must not
		# drag a sibling that already came up READY down with it.
		if _is_final(app_instance) or (
		not is_terminate and app_instance.state == AppInstanceState.READY
		):
		_log_stale_completion(app_instance, operation_id)
		continue
		saved = await self._app_instance_repo.save(
		@@ -1040,13 +1118,48 @@ class EdgeApplicationManagementService:
		app_deployment = await self._app_deployment_repo.get_by_operation_id(operation_id)
		deployment_state_map = _APP_DEPLOYMENT_COMPLETION_STATE_MAP
		if app_deployment is not None and not _deployment_is_final(app_deployment):
		new_state = (
		AppDeploymentState.PARTIAL if partial_delivery else deployment_state_map[status]
		)
		await self._app_deployment_repo.save(
		app_deployment.model_copy(update={"state": deployment_state_map[status]})
		app_deployment.model_copy(update={"state": new_state})
		)

		if updated_instances:
		await self._deliver_callbacks(operation_id, event.completed_at, updated_instances)

		async def _reported_or_settled_keys(
		self,
		operation: Operation,
		expected_keys: set[str],
		event: SRMOperationCompleted,
		) -> set[str]:
		"""Every expected fan-out instance this completion accounts for: the
		ones named in event.instances[], plus any that an earlier partial
		completion already drove to a terminal state -- so a later event
		carrying only the stragglers can still close the operation.
		"""
		if self._app_instance_repo is None:
		raise RuntimeError("AppInstanceRepository is not available")

		keys = {_completed_instance_key(operation, instance) for instance in event.instances}

		if operation.operation_type == OperationType.TERMINATE:
		# Terminate never re-homes the app_instance rows onto its own
		# operation_id, so resolve each stashed id directly.
		for raw_id in expected_keys:
		app_instance = await self._app_instance_repo.get_by_id(UUID(raw_id))
		if app_instance is not None and app_instance.state in _SETTLED_APP_INSTANCE_STATES:
		keys.add(raw_id)
		else:
		for app_instance in await self._app_instance_repo.list_by_operation_id(
		operation.operation_id
		):
		if app_instance.state in _SETTLED_APP_INSTANCE_STATES:
		keys.add(_persisted_instance_key(operation, app_instance))

		return keys

		async def list_stale_operations(
		self, deadline: timedelta, now: Optional[datetime] = None
		) -> list[Operation]:

tests/unit/test_eam_service.py

+230 −5

Original line number	Diff line number	Diff line
		@@ -2117,12 +2117,14 @@ class TestHandleCompleted:
		operation_repo: FakeOperationRepository,
		operation_type: OperationType = OperationType.DEPLOY,
		terminate_target_instance_id: UUID \| None = None,
		metadata: dict[str, Any] \| None = None,
		) -> None:
		subject = (
		Subject.TASK_TERMINATE
		if operation_type == OperationType.TERMINATE
		else Subject.TASK_DEPLOY
		)
		if metadata is None:
		metadata = (
		{"app_instance_id": str(terminate_target_instance_id)}
		if terminate_target_instance_id is not None
		@@ -2555,6 +2557,188 @@ class TestHandleCompleted:
		assert unchanged is not None
		assert unchanged.state == AppInstanceState.TERMINATED

		ZONE_A = UUID("aaaaaaaa-0000-4000-8000-0000000000a1")
		ZONE_B = UUID("aaaaaaaa-0000-4000-8000-0000000000b2")
		INSTANCE_A = UUID("bbbbbbbb-0000-4000-8000-0000000000a1")
		INSTANCE_B = UUID("bbbbbbbb-0000-4000-8000-0000000000b2")

		async def _seed_multi_zone_deploy(
		self,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> UUID:
		"""A multi-zone POST /deployments: one DEPLOY operation carrying
		metadata.edge_cloud_zones, two pre-created instances (one per zone), an
		instantiating aggregate row."""
		deployment_id = uuid4()
		await self._seed_pending_operation(
		operation_repo,
		metadata={"edge_cloud_zones": [str(self.ZONE_A), str(self.ZONE_B)]},
		)
		pairs = ((self.INSTANCE_A, self.ZONE_A), (self.INSTANCE_B, self.ZONE_B))
		for instance_id, zone_id in pairs:
		await app_instance_repo.save(
		AppInstance(
		app_instance_id=instance_id,
		operation_id=self.OPERATION_ID,
		app_registration_id=uuid4(),
		edge_cloud_zone_id=zone_id,
		state=AppInstanceState.INSTANTIATING,
		app_deployment_id=deployment_id,
		)
		)
		await app_deployment_repo.save(
		AppDeployment(
		app_deployment_id=deployment_id,
		operation_id=self.OPERATION_ID,
		app_registration_id=uuid4(),
		app_deployment_name="video_analytics_eu",
		edge_cloud_zones=[self.ZONE_A, self.ZONE_B],
		state=AppDeploymentState.INSTANTIATING,
		)
		)
		return deployment_id

		async def test_completion_missing_a_zone_keeps_operation_open_and_partial(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		"""A debounced completion that only accounts for zone A must not roll
		the operation or the aggregate to a terminal state while zone B is
		still stranded at instantiating."""
		deployment_id = await self._seed_multi_zone_deploy(
		operation_repo, app_instance_repo, app_deployment_repo
		)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(self.ZONE_A),
		status="completed",
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		operation = await operation_repo.get_by_id(self.OPERATION_ID)
		assert operation is not None
		assert operation.status == OperationStatus.IN_PROGRESS
		assert operation.completed_at is None

		deployment = await app_deployment_repo.get_by_id(deployment_id)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.PARTIAL

		instance_a = await app_instance_repo.get_by_id(self.INSTANCE_A)
		instance_b = await app_instance_repo.get_by_id(self.INSTANCE_B)
		assert instance_a is not None and instance_a.state == AppInstanceState.READY
		assert instance_b is not None and instance_b.state == AppInstanceState.INSTANTIATING

		async def test_completion_covering_every_zone_still_finalises(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		"""The guard only holds the operation open when a zone is missing -- a
		completion that names both zones finalises exactly as before."""
		deployment_id = await self._seed_multi_zone_deploy(
		operation_repo, app_instance_repo, app_deployment_repo
		)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(self.ZONE_A),
		status="completed",
		),
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_B),
		zone_id=str(self.ZONE_B),
		status="completed",
		),
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		operation = await operation_repo.get_by_id(self.OPERATION_ID)
		assert operation is not None
		assert operation.status == OperationStatus.COMPLETED
		deployment = await app_deployment_repo.get_by_id(deployment_id)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.READY

		async def test_follow_up_completion_for_stragglers_closes_the_operation(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		"""After a partial completion leaves the operation open, a second event
		that only carries the previously-missing zone must finalise it -- the
		already-READY zone A counts as reported."""
		deployment_id = await self._seed_multi_zone_deploy(
		operation_repo, app_instance_repo, app_deployment_repo
		)
		first = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(self.ZONE_A),
		status="completed",
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)
		second = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_B),
		zone_id=str(self.ZONE_B),
		status="completed",
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:03:35+00:00",
		)

		await service.handle_completed(first)
		await service.handle_completed(second)

		operation = await operation_repo.get_by_id(self.OPERATION_ID)
		assert operation is not None
		assert operation.status == OperationStatus.COMPLETED
		deployment = await app_deployment_repo.get_by_id(deployment_id)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.READY
		instance_b = await app_instance_repo.get_by_id(self.INSTANCE_B)
		assert instance_b is not None and instance_b.state == AppInstanceState.READY

		async def test_rolls_up_app_deployment_to_ready_on_full_completion(
		self,
		service: EdgeApplicationManagementService,
		@@ -3879,6 +4063,47 @@ class TestDeploymentTerminateCompletion:
		assert a is not None and a.state == AppInstanceState.TERMINATED
		assert b is not None and b.state == AppInstanceState.FAILED

		async def test_completion_missing_an_instance_keeps_teardown_open(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		"""A debounced teardown completion that only names instance A must not
		roll the operation terminal or the aggregate to TERMINATED while
		instance B is still stranded at terminating."""
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A, self.INSTANCE_B],
		)
		event = self._completed_event(
		"completed",
		[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(ZONE_ID),
		status="completed",
		)
		],
		)

		await service.handle_completed(event)

		operation = await operation_repo.get_by_id(self.TERMINATE_OP_ID)
		assert operation is not None
		assert operation.status == OperationStatus.IN_PROGRESS
		assert operation.completed_at is None
		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.PARTIAL
		a = await app_instance_repo.get_by_id(self.INSTANCE_A)
		b = await app_instance_repo.get_by_id(self.INSTANCE_B)
		assert a is not None and a.state == AppInstanceState.TERMINATED
		assert b is not None and b.state == AppInstanceState.TERMINATING

		async def test_total_failure_fans_out_over_every_stashed_instance(
		self,
		service: EdgeApplicationManagementService,