fix: guard app instance/deployment state moves with an explicit transition... (b612f9bf) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+148 −29

Original line number	Diff line number	Diff line
		@@ -155,6 +155,20 @@ _RECONCILE_TIMEOUT_ERROR: dict[str, Any] = {
		),
		}

		# Recorded on the deploy operation's error when a DELETE /deployments arrives
		# before that deploy completes. Closes the operation so its late completion is
		# ignored by handle_completed's terminal-operation guard; the teardown that
		# follows drives the deployment to TERMINATED as usual.
		_DEPLOY_SUPERSEDED_BY_DELETE_ERROR: dict[str, Any] = {
		"type": "https://etsi.org/sdg/oop/problems/deploy-superseded-by-delete",
		"title": "Deploy Superseded By Delete",
		"detail": (
		"A DELETE /deployments request was received before this deployment "
		"operation completed. The operation was closed so the teardown could "
		"proceed without a stale completion reviving terminated instances."
		),
		}

		# 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.
		@@ -163,14 +177,80 @@ _DEPLOYMENT_STATUS_CHANGE_EVENT_TYPE = (
		)


		def _is_final(app_instance: AppInstance) -> bool:
		"""Whether no completion may move this instance any further."""
		return app_instance.state == AppInstanceState.TERMINATED
		# The only state moves handle_completed() is allowed to make off a completion
		# event. Anything not listed here -- notably every move OUT of FAILED/TERMINATED,
		# and TERMINATING -> READY -- is a late, duplicate or out-of-order event racing a
		# row that has already reported its outcome, and is dropped (logged, not applied)
		# rather than allowed to revive or rewrite it. These transitions are an OEG
		# design decision: the CAMARA EAM spec models instance status as a flat enum with
		# no lifecycle rules and no deployment-level status at all (see ADR for rationale).
		_ALLOWED_APP_INSTANCE_TRANSITIONS: dict[AppInstanceState, frozenset[AppInstanceState]] = {
		AppInstanceState.INSTANTIATING: frozenset(
		{AppInstanceState.READY, AppInstanceState.FAILED, AppInstanceState.TERMINATED}
		),
		AppInstanceState.READY: frozenset({AppInstanceState.FAILED, AppInstanceState.TERMINATED}),
		AppInstanceState.TERMINATING: frozenset({AppInstanceState.TERMINATED, AppInstanceState.FAILED}),
		# FAILED is a resting state, not a dead end: terminating a broken instance
		# must still be able to reach TERMINATED. Every other move out of FAILED --
		# notably back to READY off a stale deploy completion -- is rejected.
		AppInstanceState.FAILED: frozenset({AppInstanceState.TERMINATED}),
		AppInstanceState.TERMINATED: frozenset(),
		}

		_ALLOWED_APP_DEPLOYMENT_TRANSITIONS: dict[AppDeploymentState, frozenset[AppDeploymentState]] = {
		AppDeploymentState.INSTANTIATING: frozenset(
		{
		AppDeploymentState.READY,
		AppDeploymentState.PARTIAL,
		AppDeploymentState.FAILED,
		AppDeploymentState.TERMINATED,
		}
		),
		AppDeploymentState.READY: frozenset(
		{AppDeploymentState.PARTIAL, AppDeploymentState.FAILED, AppDeploymentState.TERMINATED}
		),
		AppDeploymentState.PARTIAL: frozenset(
		{AppDeploymentState.READY, AppDeploymentState.FAILED, AppDeploymentState.TERMINATED}
		),
		AppDeploymentState.TERMINATING: frozenset(
		{AppDeploymentState.TERMINATED, AppDeploymentState.PARTIAL, AppDeploymentState.FAILED}
		),
		# As with instances: a failed deployment can still be torn down, so a
		# terminate completion may drive FAILED -> TERMINATED (or PARTIAL on a
		# partial teardown). It may never roll back toward READY.
		AppDeploymentState.FAILED: frozenset(
		{AppDeploymentState.TERMINATED, AppDeploymentState.PARTIAL}
		),
		AppDeploymentState.TERMINATED: frozenset(),
		}


		def _instance_transition_allowed(current: AppInstanceState, target: AppInstanceState) -> bool:
		"""Whether a completion event may move an app_instance from `current` to `target`.

		A no-op (current == target) is always allowed so a debounced/redelivered
		completion that re-asserts the same outcome is harmless.
		"""
		if current == target:
		return True
		return target in _ALLOWED_APP_INSTANCE_TRANSITIONS.get(current, frozenset())


		def _deployment_is_final(app_deployment: AppDeployment) -> bool:
		"""Whether no completion may move this deployment rollup any further."""
		return app_deployment.state == AppDeploymentState.TERMINATED
		def _deployment_transition_allowed(current: AppDeploymentState, target: AppDeploymentState) -> bool:
		"""Whether a completion event may move an app_deployment rollup from `current` to `target`."""
		if current == target:
		return True
		return target in _ALLOWED_APP_DEPLOYMENT_TRANSITIONS.get(current, frozenset())


		def _log_rejected_transition(current: str, target: str, operation_id: UUID, **ids: str) -> None:
		logger.info(
		"rejected_state_transition",
		from_state=current,
		to_state=target,
		operation_id=str(operation_id),
		**ids,
		)


		def _build_reconcile_timeout_event(operation: Operation) -> SRMOperationCompleted:
		@@ -192,15 +272,6 @@ def _build_reconcile_timeout_event(operation: Operation) -> SRMOperationComplete
		)


		def _log_stale_completion(app_instance: AppInstance, operation_id: UUID) -> None:
		logger.info(
		"stale_completion_ignored_for_final_app_instance",
		app_instance_id=str(app_instance.app_instance_id),
		operation_id=str(operation_id),
		state=app_instance.state.value,
		)


		# 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.
		@@ -783,6 +854,31 @@ class EdgeApplicationManagementService:
		):
		raise NotFoundException(message=f"App deployment {app_deployment_id} not found")

		# A deploy still in flight for this deployment would otherwise finalise
		# after our teardown and roll the rollup / its instances back toward
		# READY. CAMARA defines no 409 for DELETE /deployments, so we stay 202
		# and instead close the deploy operation now: its late completion is then
		# dropped by the terminal-operation guard at the top of handle_completed.
		deploy_operation = await self._operation_repo.get_by_id(app_deployment.operation_id)
		if (
		deploy_operation is not None
		and deploy_operation.status not in _TERMINAL_OPERATION_STATUSES
		):
		await self._operation_repo.save(
		deploy_operation.model_copy(
		update={
		"status": OperationStatus.FAILED,
		"error": _DEPLOY_SUPERSEDED_BY_DELETE_ERROR,
		"completed_at": datetime.now(timezone.utc),
		}
		)
		)
		logger.info(
		"deploy_operation_superseded_by_delete",
		operation_id=str(deploy_operation.operation_id),
		app_deployment_id=str(app_deployment_id),
		)

		app_instances = await self._app_instance_repo.list_by_app_deployment_id(app_deployment_id)
		# Instances already gone (individually DELETEd via /appinstances, or a
		# prior failure) get no fresh terminate command -- SRM has nothing left
		@@ -1070,13 +1166,21 @@ class EdgeApplicationManagementService:
		app_instance_id=instance.service_instance_id,
		)
		continue
		if _is_final(app_instance):
		_log_stale_completion(app_instance, operation_id)
		continue
		if is_terminate and instance.status == "completed":
		state = AppInstanceState.TERMINATED
		else:
		state = _APP_INSTANCE_COMPLETION_STATE_MAP[instance.status]
		# A late/duplicate/out-of-order completion must not revive or
		# rewrite a row that has already reported its outcome (FAILED,
		# TERMINATED) or is mid-teardown (TERMINATING -> READY).
		if not _instance_transition_allowed(app_instance.state, state):
		_log_rejected_transition(
		app_instance.state.value,
		state.value,
		operation_id,
		app_instance_id=str(app_instance.app_instance_id),
		)
		continue
		saved = await self._app_instance_repo.save(
		app_instance.model_copy(update={"state": state})
		)
		@@ -1110,13 +1214,19 @@ 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:
		# Skip rows a prior partial completion already settled: a deploy
		# Skip rows already settled (FAILED/TERMINATED), plus the one
		# case the transition table would otherwise permit: 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)
		if not _instance_transition_allowed(
		app_instance.state, AppInstanceState.FAILED
		) or (not is_terminate and app_instance.state == AppInstanceState.READY):
		_log_rejected_transition(
		app_instance.state.value,
		AppInstanceState.FAILED.value,
		operation_id,
		app_instance_id=str(app_instance.app_instance_id),
		)
		continue
		saved = await self._app_instance_repo.save(
		app_instance.model_copy(update={"state": AppInstanceState.FAILED})
		@@ -1129,8 +1239,9 @@ class EdgeApplicationManagementService:
		# operation, so it carries the deployment id in metadata instead. A
		# single-zone /appinstances operation resolves neither -- soft no-op,
		# same optional-dependency pattern as the callback repos below. The
		# _is_final guard mirrors the instances above so a stale redelivery
		# can't revive a terminated deployment.
		# transition guard mirrors the instances above so a stale redelivery
		# can't revive a terminated/failed deployment or roll a mid-teardown
		# rollup back toward READY.
		if self._app_deployment_repo is not None:
		if is_terminate:
		raw_deployment_id = operation.metadata.get("app_deployment_id")
		@@ -1143,13 +1254,21 @@ class EdgeApplicationManagementService:
		else:
		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):
		if app_deployment is not None:
		new_state = (
		AppDeploymentState.PARTIAL if partial_delivery else deployment_state_map[status]
		)
		if _deployment_transition_allowed(app_deployment.state, new_state):
		await self._app_deployment_repo.save(
		app_deployment.model_copy(update={"state": new_state})
		)
		else:
		_log_rejected_transition(
		app_deployment.state.value,
		new_state.value,
		operation_id,
		app_deployment_id=str(app_deployment.app_deployment_id),
		)

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

tests/unit/test_eam_service.py

+185 −0

Original line number	Diff line number	Diff line
		@@ -1987,6 +1987,75 @@ class TestDeleteAppDeployment:
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)

		async def test_supersedes_an_in_flight_deploy_operation(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		"""A DELETE arriving while the deploy is still in progress closes that
		deploy operation, so its late completion is dropped by handle_completed's
		terminal-operation guard instead of reviving the torn-down instances."""
		await operation_repo.save(
		Operation(
		operation_id=self.DEPLOY_OPERATION_ID,
		correlation_id="corr-deploy",
		tenant_id="tenant-1",
		app_provider_id="VideoAppsCo",
		operation_type=OperationType.DEPLOY,
		status=OperationStatus.IN_PROGRESS,
		subject=Subject.TASK_DEPLOY,
		app_registration_id=APP_REGISTRATION_ID,
		metadata={},
		)
		)
		await self._seed_deployment(
		app_deployment_repo,
		app_instance_repo,
		deployment_state=AppDeploymentState.INSTANTIATING,
		instance_states={
		self.INSTANCE_A: AppInstanceState.INSTANTIATING,
		self.INSTANCE_B: AppInstanceState.INSTANTIATING,
		},
		)

		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID,
		tenant_id="tenant-1",
		app_provider_id="VideoAppsCo",
		x_correlator="corr-1",
		)

		deploy_op = await operation_repo.get_by_id(self.DEPLOY_OPERATION_ID)
		assert deploy_op is not None
		assert deploy_op.status == OperationStatus.FAILED
		assert deploy_op.completed_at is not None
		assert deploy_op.error is not None
		assert deploy_op.error["type"].endswith("deploy-superseded-by-delete")

		# the deploy's now-stale completion arrives late: dropped, instances
		# stay torn down rather than snapping back to READY.
		await service.handle_completed(
		SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.DEPLOY_OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(self.ZONE_A),
		status="completed",
		)
		],
		correlation_id="corr-deploy",
		completed_at="2026-07-04T10:02:35+00:00",
		)
		)
		instance_a = await app_instance_repo.get_by_id(self.INSTANCE_A)
		assert instance_a is not None
		assert instance_a.state == AppInstanceState.TERMINATING

		async def test_already_terminal_instances_are_skipped(
		self,
		service: EdgeApplicationManagementService,
		@@ -3233,6 +3302,122 @@ class TestHandleCompleted:
		assert updated is not None
		assert updated.state == AppInstanceState.TERMINATED

		async def test_stale_deploy_completion_does_not_revive_failed_instance(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		"""A duplicate/out-of-order DEPLOY completion landing on an instance that
		has already failed must not roll it back to READY -- only a terminate
		may move it further (to TERMINATED)."""
		await self._seed_pending_operation(operation_repo, OperationType.DEPLOY)
		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.FAILED,
		)
		)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(INSTANCE_ID), zone_id=str(ZONE_ID), status="completed"
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

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

		async def test_late_deploy_completion_does_not_revive_terminating_instance(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		"""DELETE moved the instance to TERMINATING while its deploy was still in
		flight; that deploy's completion must not flip it back to READY."""
		await self._seed_pending_operation(operation_repo, OperationType.DEPLOY)
		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.TERMINATING,
		)
		)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(INSTANCE_ID), zone_id=str(ZONE_ID), status="completed"
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		unchanged = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert unchanged is not None
		assert unchanged.state == AppInstanceState.TERMINATING

		async def test_stale_deploy_completion_does_not_revive_failed_deployment(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		"""The deployment rollup carries the same guard: a stale deploy
		completion must not roll a FAILED deployment back to READY."""
		deployment_id = uuid4()
		await self._seed_pending_operation(operation_repo, OperationType.DEPLOY)
		await self._seed_instantiating_app_instance(app_instance_repo, INSTANCE_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=[ZONE_ID],
		state=AppDeploymentState.FAILED,
		)
		)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(INSTANCE_ID), zone_id=str(ZONE_ID), status="completed"
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		unchanged = await app_deployment_repo.get_by_id(deployment_id)
		assert unchanged is not None
		assert unchanged.state == AppDeploymentState.FAILED

		async def test_unknown_app_instance_id_is_skipped_not_raised(
		self,
		service: EdgeApplicationManagementService,