fix(eam): let a failed app instance be deleted (f0210b60) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+66 −6

Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 66 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -267,6 +267,22 @@ def _synthetic_failure_completion(event: SRMOperationStatus) -> SRMOperationComp
		)


		_PREVIOUSLY_FAILED_APP_INSTANCE_IDS_KEY = "previously_failed_app_instance_ids"


		def _previously_failed_instance_ids(operation: Operation) -> frozenset[str]:
		raw = operation.metadata.get(_PREVIOUSLY_FAILED_APP_INSTANCE_IDS_KEY) or []
		return frozenset(str(raw_id) for raw_id in raw)


		def _log_failed_instance_treated_as_terminated(operation_id: UUID, app_instance_id: UUID) -> None:
		logger.info(
		"failed_instance_terminate_failure_treated_as_terminated",
		operation_id=str(operation_id),
		app_instance_id=str(app_instance_id),
		)


		def _expected_instance_keys(operation: Operation) -> Optional[set[str]]:
		"""Zone ids (deploy) or app_instance ids (terminate) a multi-zone operation waits for."""
		raw = (
		@@ -876,6 +892,11 @@ class EdgeApplicationManagementService:
		"app_instance_ids": [
		str(instance.app_instance_id) for instance in live_instances
		],
		_PREVIOUSLY_FAILED_APP_INSTANCE_IDS_KEY: [
		str(instance.app_instance_id)
		for instance in live_instances
		if instance.state == AppInstanceState.FAILED
		],
		},
		)
		)
		@@ -993,6 +1014,10 @@ class EdgeApplicationManagementService:
		requested_at=requested_at,
		)

		metadata: dict[str, Any] = {"app_instance_id": str(app_instance_id)}
		if app_instance.state == AppInstanceState.FAILED:
		metadata[_PREVIOUSLY_FAILED_APP_INSTANCE_IDS_KEY] = [str(app_instance_id)]

		await self._operation_repo.save(
		Operation(
		operation_id=operation_id,
		@@ -1003,7 +1028,7 @@ class EdgeApplicationManagementService:
		status=OperationStatus.PENDING,
		subject=Subject.TASK_TERMINATE,
		app_registration_id=app_instance.app_registration_id,
		metadata={"app_instance_id": str(app_instance_id)},
		metadata=metadata,
		)
		)
		await self._app_instance_repo.save(
		@@ -1083,6 +1108,10 @@ class EdgeApplicationManagementService:
		await self._operation_repo.save(updated)

		is_terminate = operation.operation_type == OperationType.TERMINATE
		previously_failed_ids = (
		_previously_failed_instance_ids(operation) if is_terminate else frozenset()
		)
		treated_as_terminated = False
		updated_instances: list[AppInstance] = []

		if event.instances:
		@@ -1097,6 +1126,14 @@ class EdgeApplicationManagementService:
		continue
		if is_terminate and instance.status == "completed":
		state = AppInstanceState.TERMINATED
		elif (
		is_terminate
		and instance.status == "failed"
		and str(app_instance_id) in previously_failed_ids
		):
		state = AppInstanceState.TERMINATED
		treated_as_terminated = True
		_log_failed_instance_treated_as_terminated(operation_id, app_instance_id)
		else:
		state = _APP_INSTANCE_COMPLETION_STATE_MAP[instance.status]
		if not _instance_transition_allowed(app_instance.state, state):
		@@ -1132,19 +1169,29 @@ class EdgeApplicationManagementService:
		else:
		failed_instances = await self._app_instance_repo.list_by_operation_id(operation_id)
		for app_instance in failed_instances:
		target_state = (
		AppInstanceState.TERMINATED
		if str(app_instance.app_instance_id) in previously_failed_ids
		else AppInstanceState.FAILED
		)
		# A reconcile force-fail must not fail siblings that already came up READY.
		if not _instance_transition_allowed(
		app_instance.state, AppInstanceState.FAILED
		) or (not is_terminate and app_instance.state == AppInstanceState.READY):
		if not _instance_transition_allowed(app_instance.state, target_state) or (
		not is_terminate and app_instance.state == AppInstanceState.READY
		):
		_log_rejected_transition(
		app_instance.state.value,
		AppInstanceState.FAILED.value,
		target_state.value,
		operation_id,
		app_instance_id=str(app_instance.app_instance_id),
		)
		continue
		if target_state == AppInstanceState.TERMINATED:
		treated_as_terminated = True
		_log_failed_instance_treated_as_terminated(
		operation_id, app_instance.app_instance_id
		)
		saved = await self._app_instance_repo.save(
		app_instance.model_copy(update={"state": AppInstanceState.FAILED})
		app_instance.model_copy(update={"state": target_state})
		)
		updated_instances.append(saved)

		@@ -1165,6 +1212,19 @@ class EdgeApplicationManagementService:
		new_state = (
		AppDeploymentState.PARTIAL if partial_delivery else deployment_state_map[status]
		)
		if (
		treated_as_terminated
		and not partial_delivery
		and new_state != AppDeploymentState.TERMINATED
		):
		deployment_instances = await self._app_instance_repo.list_by_app_deployment_id(
		app_deployment.app_deployment_id
		)
		if all(
		instance.state == AppInstanceState.TERMINATED
		for instance in deployment_instances
		):
		new_state = AppDeploymentState.TERMINATED
		if _deployment_transition_allowed(app_deployment.state, new_state):
		await self._app_deployment_repo.save(
		app_deployment.model_copy(update={"state": new_state})

tests/unit/test_eam_service.py

+376 −4

Changes for tests/unit/test_eam_service.py: 376 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -2262,6 +2262,96 @@ class TestDeleteAppInstance:
		assert instance is not None
		assert instance.state == AppInstanceState.READY

		async def _mark_failed(self, app_instance_repo: FakeAppInstanceRepository) -> None:
		instance = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert instance is not None
		await app_instance_repo.save(instance.model_copy(update={"state": AppInstanceState.FAILED}))

		async def test_records_failed_instance_as_previously_failed(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._mark_failed(app_instance_repo)
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)
		(operation,) = _terminate_operations(operation_repo)
		assert operation.metadata["previously_failed_app_instance_ids"] == [str(INSTANCE_ID)]

		async def test_does_not_record_ready_instance_as_previously_failed(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		) -> None:
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)
		(operation,) = _terminate_operations(operation_repo)
		assert "previously_failed_app_instance_ids" not in operation.metadata

		async def test_failed_instance_with_failed_terminate_is_gone_on_retry(
		self,
		service: EdgeApplicationManagementService,
		publisher: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._mark_failed(app_instance_repo)
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)
		publisher.publish.assert_called_once()
		(operation,) = _terminate_operations(operation_repo)

		await service.handle_completed(
		SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(operation.operation_id),
		status="failed",
		instances=[],
		error={"type": "about:blank", "title": "Instance Not Federated", "status": 404},
		correlation_id=operation.correlation_id,
		completed_at="2026-07-04T10:02:35+00:00",
		)
		)

		instance = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert instance is not None
		assert instance.state == AppInstanceState.TERMINATED
		with pytest.raises(NotFoundException):
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		async def test_ready_instance_with_failed_terminate_stays_failed(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)
		(operation,) = _terminate_operations(operation_repo)

		await service.handle_completed(
		SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(operation.operation_id),
		status="failed",
		instances=[],
		error={"type": "about:blank", "title": "Termination Failed", "status": 503},
		correlation_id=operation.correlation_id,
		completed_at="2026-07-04T10:02:35+00:00",
		)
		)

		instance = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert instance is not None
		assert instance.state == AppInstanceState.FAILED


		class TestDeleteAppDeployment:
		ZONE_A = UUID("aaaaaaaa-1111-4000-8000-000000000001")
		@@ -2354,6 +2444,33 @@ class TestDeleteAppDeployment:
		str(self.INSTANCE_A),
		str(self.INSTANCE_B),
		}
		assert operation.metadata["previously_failed_app_instance_ids"] == []

		async def test_records_only_failed_live_instances_as_previously_failed(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_deployment(
		app_deployment_repo,
		app_instance_repo,
		deployment_state=AppDeploymentState.PARTIAL,
		instance_states={
		self.INSTANCE_A: AppInstanceState.FAILED,
		self.INSTANCE_B: AppInstanceState.READY,
		},
		)
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)
		(operation,) = _terminate_operations(operation_repo)
		assert set(operation.metadata["app_instance_ids"]) == {
		str(self.INSTANCE_A),
		str(self.INSTANCE_B),
		}
		assert operation.metadata["previously_failed_app_instance_ids"] == [str(self.INSTANCE_A)]

		async def test_moves_deployment_and_every_instance_to_terminating(
		self,
		@@ -4411,6 +4528,87 @@ class TestHandleCompleted:
		assert updated is not None
		assert updated.state == AppInstanceState.FAILED

		async def _seed_terminating_previously_failed_instance(
		self, operation_repo: FakeOperationRepository, app_instance_repo: FakeAppInstanceRepository
		) -> None:
		await self._seed_pending_operation(
		operation_repo,
		operation_type=OperationType.TERMINATE,
		metadata={
		"app_instance_id": str(INSTANCE_ID),
		"previously_failed_app_instance_ids": [str(INSTANCE_ID)],
		},
		)
		await app_instance_repo.save(
		AppInstance(
		app_instance_id=INSTANCE_ID,
		operation_id=uuid4(),
		app_registration_id=uuid4(),
		edge_cloud_zone_id=ZONE_ID,
		state=AppInstanceState.TERMINATING,
		)
		)

		async def test_terminate_failure_for_previously_failed_instance_marks_terminated(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_terminating_previously_failed_instance(operation_repo, app_instance_repo)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="failed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(INSTANCE_ID),
		zone_id=str(ZONE_ID),
		status="failed",
		error={"type": "about:blank", "title": "Not Found", "status": 404},
		)
		],
		error={"type": "about:blank", "title": "Not Found", "status": 404},
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		updated = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert updated is not None
		assert updated.state == AppInstanceState.TERMINATED
		operation = await operation_repo.get_by_id(self.OPERATION_ID)
		assert operation is not None
		assert operation.status == OperationStatus.FAILED
		assert operation.error == {"type": "about:blank", "title": "Not Found", "status": 404}

		async def test_terminate_total_failure_for_previously_failed_instance_marks_terminated(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_terminating_previously_failed_instance(operation_repo, app_instance_repo)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="failed",
		instances=[],
		error={"type": "about:blank", "title": "Termination Failed", "status": 404},
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		updated = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert updated is not None
		assert updated.state == AppInstanceState.TERMINATED
		operation = await operation_repo.get_by_id(self.OPERATION_ID)
		assert operation is not None
		assert operation.status == OperationStatus.FAILED

		async def test_terminate_total_failure_fallback_resolves_instance_via_stashed_id(
		self,
		service: EdgeApplicationManagementService,
		@@ -4985,9 +5183,16 @@ class TestDeploymentTerminateCompletion:
		*,
		terminate_status: OperationStatus = OperationStatus.PENDING,
		app_registration_repo: FakeAppRegistrationRepository \| None = None,
		previously_failed: list[UUID] \| None = None,
		) -> UUID:
		app_registration_id = uuid4()
		app_id = uuid4()
		metadata: dict[str, Any] = {
		"app_deployment_id": str(DEPLOYMENT_ID),
		"app_instance_ids": [str(i) for i in instance_ids],
		}
		if previously_failed is not None:
		metadata["previously_failed_app_instance_ids"] = [str(i) for i in previously_failed]
		if app_registration_repo is not None:
		await app_registration_repo.save(
		AppRegistration(
		@@ -5010,10 +5215,7 @@ class TestDeploymentTerminateCompletion:
		status=terminate_status,
		subject=Subject.TASK_TERMINATE,
		app_registration_id=app_registration_id,
		metadata={
		"app_deployment_id": str(DEPLOYMENT_ID),
		"app_instance_ids": [str(i) for i in instance_ids],
		},
		metadata=metadata,
		)
		)
		await app_deployment_repo.save(
		@@ -5386,6 +5588,176 @@ class TestDeploymentTerminateCompletion:
		assert instance is not None
		assert instance.state == AppInstanceState.FAILED

		async def test_mixed_teardown_terminates_previously_failed_and_fails_the_ready_one(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A, self.INSTANCE_B],
		previously_failed=[self.INSTANCE_A],
		)
		error = {"type": "about:blank", "title": "boom", "status": 500}
		event = self._completed_event(
		"failed",
		[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(ZONE_ID),
		status="failed",
		error=error,
		),
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_B),
		zone_id=str(ZONE_ID),
		status="failed",
		error=error,
		),
		],
		)

		await service.handle_completed(event)

		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.TERMINATED
		assert instance_b is not None and instance_b.state == AppInstanceState.FAILED
		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.FAILED

		async def test_total_failure_mixed_teardown_only_terminates_previously_failed(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A, self.INSTANCE_B],
		previously_failed=[self.INSTANCE_A],
		)
		event = self._completed_event(
		"failed",
		[],
		error={"title": "Termination Failed", "status": 503, "detail": "backend down"},
		)

		await service.handle_completed(event)

		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.TERMINATED
		assert instance_b is not None and instance_b.state == AppInstanceState.FAILED
		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.FAILED

		async def test_failed_teardown_of_only_previously_failed_instance_terminates_deployment(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_registration_repo: FakeAppRegistrationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_delivery_port: FakeCallbackDeliveryPort,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A],
		app_registration_repo=app_registration_repo,
		previously_failed=[self.INSTANCE_A],
		)
		await callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=self.DEPLOY_OP_ID,
		tenant_id="tenant-1",
		api_family="edge-application-management",
		sink="https://client.example.com/callback",
		event_types=[
		"org.camaraproject.edge-application-management.v0.app-deployment-status-change"
		],
		is_active=True,
		)
		)
		event = self._completed_event(
		"failed",
		[],
		error={"title": "Termination Failed", "status": 404, "detail": "not federated"},
		)

		await service.handle_completed(event)

		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.TERMINATED
		instance = await app_instance_repo.get_by_id(self.INSTANCE_A)
		assert instance is not None and instance.state == AppInstanceState.TERMINATED
		operation = await operation_repo.get_by_id(self.TERMINATE_OP_ID)
		assert operation is not None
		assert operation.status == OperationStatus.FAILED
		assert operation.error is not None and operation.error["status"] == 404
		assert len(callback_delivery_port.delivered) == 1
		_, payload = callback_delivery_port.delivered[0]
		assert isinstance(payload, list)
		assert [ce.data.status for ce in payload] == ["unknown"]

		async def test_previously_failed_reported_after_sibling_completed_terminates_deployment(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A, self.INSTANCE_B],
		previously_failed=[self.INSTANCE_A],
		)
		await service.handle_completed(
		self._completed_event(
		"completed",
		[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_B),
		zone_id=str(ZONE_ID),
		status="completed",
		)
		],
		)
		)
		await service.handle_completed(
		self._completed_event(
		"failed",
		[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(ZONE_ID),
		status="failed",
		error={"type": "about:blank", "title": "boom", "status": 404},
		)
		],
		)
		)

		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.TERMINATED


		class TestHandleOperationStatus:
		"""event.srm.operation.status(state=accepted) narrows the "no idea what's