feat: terminate service tf-sdk wiring (4ac0a895) · Commits · OOP / code / service-resource-manager

src/srm/api/databus/dependencies.py

+21 −14

Changes for src/srm/api/databus/dependencies.py: 21 added lines, 14 removed lines.
Original line number	Diff line number	Diff line
		@@ -10,9 +10,11 @@ from srm.adapters.database.repos.runtime_inventory import (
		SqlServiceInstanceRepository,
		SqlServiceOrderRepository,
		)
		from srm.adapters.database.repos.topology import SqlZoneRepository
		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.adapters.databus.nats_publisher import NatsPublisher
		from srm.adapters.database.repos.topology import (
		SqlCapabilityRepository,
		SqlDomainRepository,
		SqlZoneRepository,
		)
		from srm.adapters.federation_manager.fm_client import FederationManager
		from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator
		from srm.application.command_handlers.federated_deploy_outcome import (
		@@ -116,37 +118,42 @@ def get_network_capability_activate_command_coordinator(
		)


		def get_terminate_service_use_case(
		session: AsyncSession,
		connection_manager: NatsConnectionManager,
		fm_client: FederationManager,
		) -> TerminateServiceUseCase:
		return build_terminate_service_use_case(session, NatsPublisher(connection_manager), fm_client)


		def build_terminate_service_use_case(
		session: AsyncSession,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		control_paths: ControlPathRegistry,
		) -> TerminateServiceUseCase:
		return TerminateServiceUseCase(
		service_orders=SqlServiceOrderRepository(session),
		service_instances=SqlServiceInstanceRepository(session),
		capability_instances=SqlCapabilityInstanceRepository(session),
		capabilities=SqlCapabilityRepository(session),
		domains=SqlDomainRepository(session),
		zones=SqlZoneRepository(session),
		publisher=publisher,
		fm_client=fm_client,
		control_paths=control_paths,
		)


		def get_terminate_service_command_coordinator(
		session_maker: async_sessionmaker[AsyncSession],
		connection_manager: NatsConnectionManager,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		control_paths: ControlPathRegistry,
		) -> TerminateServiceCommandCoordinator:
		def use_case_factory(
		session: AsyncSession,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		) -> TerminateServiceUseCase:
		return build_terminate_service_use_case(session, publisher, fm_client, control_paths)

		return TerminateServiceCommandCoordinator(
		session_maker=session_maker,
		use_case_factory=build_terminate_service_use_case,
		publisher=NatsPublisher(connection_manager),
		use_case_factory=use_case_factory,
		publisher=publisher,
		fm_client=fm_client,
		)

src/srm/api/databus/nats_subscriber.py

+3 −2

Changes for src/srm/api/databus/nats_subscriber.py: 3 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -63,8 +63,9 @@ async def subscribe_to_subjects(
		)
		terminate_coordinator = get_terminate_service_command_coordinator(
		session_maker,
		connection_manager,
		fm_client,
		publisher=publisher,
		fm_client=fm_client,
		control_paths=control_paths,
		)
		network_activate_coordinator = get_network_capability_activate_command_coordinator(
		session_maker,

src/srm/application/command_handlers/terminate_service.py

+7 −5

Changes for src/srm/application/command_handlers/terminate_service.py: 7 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -138,12 +138,14 @@ class TerminateServiceCommandCoordinator:
		# Outside any session — TF-SDK side effect.
		# On raise: no new DB write; prior committed caps stay TERMINATED;
		# handle()'s except → fail_after_accept.
		await TerminateServiceUseCase.fulfill_one(target)
		await use_case.fulfill_one(target)

		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		await use_case.persist_capability_terminated(
		persist_use_case = self._use_case_factory(
		session, self._publisher, self._fm_client
		)
		await persist_use_case.persist_capability_terminated(
		target.capability_instance_id,
		terminated_at,
		)
		@@ -154,8 +156,8 @@ class TerminateServiceCommandCoordinator:

		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		completed_event = await use_case.persist_completed(
		persist_use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		completed_event = await persist_use_case.persist_completed(
		command,
		accepted,
		terminated_at=terminated_at,

src/srm/application/use_cases/terminate_service.py

+189 −19

Changes for src/srm/application/use_cases/terminate_service.py: 189 added lines, 19 removed lines.
Original line number	Diff line number	Diff line
		@@ -6,6 +6,7 @@ from typing import Any, Literal
		from uuid import UUID

		import structlog
		from anyio import to_thread
		from fastapi.encoders import jsonable_encoder

		from srm.adapters.federation_manager.fm_client import FederationManager
		@@ -13,6 +14,7 @@ from srm.application.use_cases.federated_deploy_outcome import (
		build_terminate_federated_handoff,
		federated_handoff_targets,
		)
		from srm.domain.models.canonical_parameters.parameters import Parameters
		from srm.domain.models.runtime_inventory import ServiceOrder
		from srm.domain.models.runtime_inventory.enums import (
		CapabilityInstanceState,
		@@ -20,11 +22,17 @@ from srm.domain.models.runtime_inventory.enums import (
		ServiceOrderState,
		ServiceOrderType,
		)
		from srm.domain.models.topology import CapabilityKind
		from srm.domain.ports.database.runtime_inventory import (
		CapabilityInstanceRepository,
		ServiceInstanceRepository,
		ServiceOrderRepository,
		)
		from srm.domain.ports.database.topology import (
		CapabilityRepository,
		DomainRepository,
		ZoneRepository,
		)
		from srm.domain.ports.databus.events import (
		OperationCompletedStatus,
		OperationStatusState,
		@@ -32,6 +40,25 @@ from srm.domain.ports.databus.events import (
		SrmOperationStatusV1,
		)
		from srm.domain.ports.databus.publisher import DataBusPublisher
		from srm.domain.ports.transformation_functions import (
		CapabilityLifecycleRequest,
		CapabilityRuntimeStateRequest,
		ControlPathRegistry,
		DomainExecutionFailed,
		ResolvedCapabilityTarget,
		ResourceNotFound,
		TransformationFunctionExecutionPort,
		)

		logger: structlog.BoundLogger = structlog.get_logger(__name__)

		_TERMINATE_ACTION = "terminate"
		_ALREADY_GONE_STATES = frozenset(
		{
		CapabilityInstanceState.UNKNOWN,
		CapabilityInstanceState.TERMINATED,
		}
		)


		@dataclass(frozen=True, slots=True)
		@@ -66,8 +93,12 @@ class TerminateServiceResult:
		@dataclass(frozen=True, slots=True)
		class TerminateCapabilityTarget:
		capability_instance_id: UUID
		external_id: str \| None
		control_path_ref_snapshot: str \| None
		external_id: str
		capability_kind: CapabilityKind
		resolved_target: ResolvedCapabilityTarget
		parameters: Parameters
		operation_id: str
		tenant_id: str


		@dataclass(frozen=True, slots=True)
		@@ -84,14 +115,22 @@ class TerminateServiceUseCase:
		service_orders: ServiceOrderRepository,
		service_instances: ServiceInstanceRepository,
		capability_instances: CapabilityInstanceRepository,
		capabilities: CapabilityRepository,
		domains: DomainRepository,
		zones: ZoneRepository,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		control_paths: ControlPathRegistry,
		) -> None:
		self._service_orders = service_orders
		self._service_instances = service_instances
		self._capability_instances = capability_instances
		self._capabilities = capabilities
		self._domains = domains
		self._zones = zones
		self._publisher = publisher
		self._fm_client = fm_client
		self._control_paths = control_paths

		async def accept(self, command: TerminateServiceCommand) -> TerminateServiceResult:
		existing_order = await self._service_orders.get_by_operation_id(command.operation_id)
		@@ -116,7 +155,7 @@ class TerminateServiceUseCase:
		federation_partner_ref=command.federation_partner_ref,
		)
		)
		structlog.get_logger(__name__).info(
		logger.info(
		"service_terminate_accepted_federated",
		operation_id=str(command.operation_id),
		service_instance_id=str(command.service_instance_id),
		@@ -151,8 +190,12 @@ class TerminateServiceUseCase:
		federation_partner_ref=command.federation_partner_ref,
		)
		)
		await self._service_instances.update_state(
		service_instance.id,
		ServiceInstanceState.TERMINATING,
		)

		structlog.get_logger(__name__).info(
		logger.info(
		"service_terminate_accepted",
		operation_id=str(command.operation_id),
		service_instance_id=str(command.service_instance_id),
		@@ -177,7 +220,7 @@ class TerminateServiceUseCase:

		await self._fm_client.request_terminate_service(jsonable_encoder(command.payload_snapshot))
		await self._service_orders.update_state(service_order.id, ServiceOrderState.IN_PROGRESS)
		structlog.get_logger(__name__).info(
		logger.info(
		"service_terminate_handed_to_partner",
		operation_id=str(command.operation_id),
		service_instance_id=str(command.service_instance_id),
		@@ -200,26 +243,148 @@ class TerminateServiceUseCase:
		if service_instance is None:
		raise RuntimeError("Accepted terminate service instance could not be reloaded")

		return TerminateFulfillmentPlan(
		service_instance_id=service_instance.id,
		service_order_id=service_order.id,
		zone_id=service_instance.zone_id,
		capabilities=tuple(
		targets: list[TerminateCapabilityTarget] = []
		for capability in service_instance.capabilities:
		if capability.state is not CapabilityInstanceState.ACTIVE:
		continue
		if not capability.external_id or not capability.control_path_ref_snapshot:
		logger.info(
		"terminate_skip_capability_missing_refs",
		operation_id=str(command.operation_id),
		capability_instance_id=str(capability.id),
		external_id=capability.external_id,
		control_path_ref_snapshot=capability.control_path_ref_snapshot,
		)
		continue

		topology_capability = await self._capabilities.get_by_id(capability.capability_id)
		if topology_capability is None:
		raise RuntimeError(
		f"Capability {capability.capability_id} could not be reloaded for terminate"
		)
		domain = await self._domains.get_by_id(topology_capability.domain_id)
		if domain is None:
		raise RuntimeError(
		f"Domain {topology_capability.domain_id} could not be reloaded for terminate"
		)
		zone = await self._zones.get_by_id(domain.zone_id)
		if zone is None:
		raise RuntimeError(f"Zone {domain.zone_id} could not be reloaded for terminate")

		targets.append(
		TerminateCapabilityTarget(
		capability_instance_id=capability.id,
		external_id=capability.external_id,
		control_path_ref_snapshot=capability.control_path_ref_snapshot,
		capability_kind=topology_capability.kind,
		resolved_target=ResolvedCapabilityTarget(
		zone_ref=zone.ref,
		zone_kind=zone.kind,
		domain_ref=domain.ref,
		domain_kind=domain.kind,
		capability_ref=topology_capability.ref,
		capability_kind=topology_capability.kind,
		control_path_ref=capability.control_path_ref_snapshot,
		action=_TERMINATE_ACTION,
		),
		parameters=capability.parameters_snapshot,
		operation_id=str(command.operation_id),
		tenant_id=command.app_provider_id,
		)
		)
		for capability in service_instance.capabilities

		return TerminateFulfillmentPlan(
		service_instance_id=service_instance.id,
		service_order_id=service_order.id,
		zone_id=service_instance.zone_id,
		capabilities=tuple(targets),
		)

		async def fulfill_one(self, target: TerminateCapabilityTarget) -> None:
		adapter = self._control_paths.get_adapter(target.resolved_target.control_path_ref)
		request = CapabilityLifecycleRequest(
		operation_id=target.operation_id,
		tenant_id=target.tenant_id,
		external_id=target.external_id,
		capability_kind=target.capability_kind,
		target=target.resolved_target,
		parameters=target.parameters,
		)
		logger.info(
		"terminate_adapter_deactivate",
		operation_id=target.operation_id,
		control_path_ref=target.resolved_target.control_path_ref,
		adapter=type(adapter).__name__,
		external_id=target.external_id,
		capability_kind=target.capability_kind,
		capability_instance_id=str(target.capability_instance_id),
		)
		try:
		result = await to_thread.run_sync(adapter.deactivate_capability, request)
		except ResourceNotFound:
		logger.info(
		"terminate_adapter_already_gone",
		operation_id=target.operation_id,
		control_path_ref=target.resolved_target.control_path_ref,
		external_id=target.external_id,
		reason="resource_not_found",
		)
		return
		except DomainExecutionFailed as exc:
		if await self._is_already_gone(adapter, target):
		logger.info(
		"terminate_adapter_already_gone",
		operation_id=target.operation_id,
		control_path_ref=target.resolved_target.control_path_ref,
		external_id=target.external_id,
		reason="runtime_state",
		deactivate_error=exc.message,
		)
		return
		raise

		logger.info(
		"terminate_adapter_deactivate_result",
		operation_id=target.operation_id,
		control_path_ref=target.resolved_target.control_path_ref,
		adapter=type(adapter).__name__,
		external_id=result.external_id or target.external_id,
		state=result.state,
		details=result.details,
		)
		if result.state is not CapabilityInstanceState.TERMINATED:
		raise DomainExecutionFailed(
		error_code="domain_execution_failed",
		message=(
		"capability deactivate did not become terminated on "
		f"{target.resolved_target.control_path_ref}: {result.state}"
		),
		details=result.details,
		)

		@staticmethod
		async def fulfill_one(target: TerminateCapabilityTarget) -> None:
		# TODO: TF-SDK CapabilityExecutionAdapter.deactivate_capability/terminate(...)
		# using target.external_id and target.control_path_ref_snapshot.
		# Must not touch repositories — runs outside any DB session.
		_ = target
		async def _is_already_gone(
		self,
		adapter: TransformationFunctionExecutionPort,
		target: TerminateCapabilityTarget,
		) -> bool:
		try:
		runtime = await to_thread.run_sync(
		adapter.get_capability_runtime_state,
		CapabilityRuntimeStateRequest(
		external_id=target.external_id,
		capability_kind=target.capability_kind,
		target=target.resolved_target,
		),
		)
		except Exception:
		logger.warning(
		"terminate_runtime_state_probe_failed",
		operation_id=target.operation_id,
		control_path_ref=target.resolved_target.control_path_ref,
		external_id=target.external_id,
		exc_info=True,
		)
		return False
		return runtime.state in _ALREADY_GONE_STATES

		async def persist_capability_terminated(
		self,
		@@ -262,7 +427,7 @@ class TerminateServiceUseCase:
		completed_at=completed_at,
		)

		structlog.get_logger(__name__).info(
		logger.info(
		"service_terminate_completed",
		operation_id=str(command.operation_id),
		service_instance_id=str(command.service_instance_id),
		@@ -303,6 +468,11 @@ class TerminateServiceUseCase:
		"status": 500,
		"detail": str(error),
		}
		if not accepted.is_federated:
		await self._service_instances.update_state(
		command.service_instance_id,
		ServiceInstanceState.FAILED,
		)
		await self._service_orders.update_state(
		service_order.id,
		ServiceOrderState.FAILED,

tests/api/databus/test_dependencies.py

+20 −9

Changes for tests/api/databus/test_dependencies.py: 20 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,7 +3,7 @@ from unittest.mock import MagicMock, patch
		from srm.api.databus.dependencies import (
		build_deploy_service_use_case,
		build_network_capability_activate_use_case,
		get_terminate_service_use_case,
		build_terminate_service_use_case,
		)


		@@ -49,30 +49,41 @@ def test_build_deploy_service_use_case_wires_required_repositories() -> None:
		assert result is use_case.return_value


		def test_get_terminate_service_use_case_wires_required_repositories() -> None:
		def test_build_terminate_service_use_case_wires_required_repositories() -> None:
		session = MagicMock()
		connection_manager = MagicMock()
		publisher = MagicMock()
		fm_client = MagicMock()
		control_paths = MagicMock()

		with (
		patch("srm.api.databus.dependencies.SqlServiceOrderRepository") as orders,
		patch("srm.api.databus.dependencies.SqlServiceInstanceRepository") as instances,
		patch("srm.api.databus.dependencies.SqlCapabilityInstanceRepository") as capabilities,
		patch("srm.api.databus.dependencies.NatsPublisher") as publisher,
		patch(
		"srm.api.databus.dependencies.SqlCapabilityInstanceRepository"
		) as capability_instances,
		patch("srm.api.databus.dependencies.SqlCapabilityRepository") as capabilities,
		patch("srm.api.databus.dependencies.SqlDomainRepository") as domains,
		patch("srm.api.databus.dependencies.SqlZoneRepository") as zones,
		patch("srm.api.databus.dependencies.TerminateServiceUseCase") as use_case,
		):
		result = get_terminate_service_use_case(session, connection_manager, fm_client)
		result = build_terminate_service_use_case(session, publisher, fm_client, control_paths)

		orders.assert_called_once_with(session)
		instances.assert_called_once_with(session)
		capability_instances.assert_called_once_with(session)
		capabilities.assert_called_once_with(session)
		publisher.assert_called_once_with(connection_manager)
		domains.assert_called_once_with(session)
		zones.assert_called_once_with(session)
		use_case.assert_called_once_with(
		service_orders=orders.return_value,
		service_instances=instances.return_value,
		capability_instances=capabilities.return_value,
		publisher=publisher.return_value,
		capability_instances=capability_instances.return_value,
		capabilities=capabilities.return_value,
		domains=domains.return_value,
		zones=zones.return_value,
		publisher=publisher,
		fm_client=fm_client,
		control_paths=control_paths,
		)
		assert result is use_case.return_value