feat: handle service terminate with per-capability persistence (10881d69) · Commits · OOP / code / service-resource-manager

src/srm/application/command_handlers/terminate_service.py

+31 −1

Changes for src/srm/application/command_handlers/terminate_service.py: 31 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -109,10 +109,40 @@ class TerminateServiceCommandCoordinator:
		command: TerminateServiceCommand,
		accepted: TerminateServiceResult,
		) -> SrmOperationCompletedV1 \| None:
		async with self._session_maker() as session:
		use_case = self._use_case_factory(session, self._publisher)
		plan = await use_case.prepare_fulfill(command, accepted)

		if plan is None:
		return None

		terminated_at = datetime.now(UTC)
		for target in plan.capabilities:
		# 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)

		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		await use_case.persist_capability_terminated(
		target.capability_instance_id,
		terminated_at,
		)
		await session.commit()
		except Exception:
		await session.rollback()
		raise

		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		completed_event = await use_case.complete(command, accepted)
		completed_event = await use_case.persist_completed(
		command,
		accepted,
		terminated_at=terminated_at,
		)
		await session.commit()
		return completed_event
		except Exception:

src/srm/application/use_cases/terminate_service.py

+80 −21

Changes for src/srm/application/use_cases/terminate_service.py: 80 added lines, 21 removed lines.
Original line number	Diff line number	Diff line
		@@ -7,7 +7,7 @@ from uuid import UUID

		import structlog

		from srm.domain.models.runtime_inventory import ServiceInstance, ServiceOrder
		from srm.domain.models.runtime_inventory import ServiceOrder
		from srm.domain.models.runtime_inventory.enums import (
		CapabilityInstanceState,
		ServiceInstanceState,
		@@ -56,6 +56,21 @@ class TerminateServiceResult:
		completed_event: SrmOperationCompletedV1 \| None = None


		@dataclass(frozen=True, slots=True)
		class TerminateCapabilityTarget:
		capability_instance_id: UUID
		external_id: str \| None
		control_path_ref_snapshot: str \| None


		@dataclass(frozen=True, slots=True)
		class TerminateFulfillmentPlan:
		service_instance_id: UUID
		service_order_id: UUID
		zone_id: UUID
		capabilities: tuple[TerminateCapabilityTarget, ...]


		class TerminateServiceUseCase:
		def __init__(
		self,
		@@ -78,6 +93,15 @@ class TerminateServiceUseCase:
		if service_instance is None:
		await self._publish_failed_before_start(command, service_order_id=None)
		return TerminateServiceResult()
		if (
		service_instance.state == ServiceInstanceState.TERMINATED
		or service_instance.state == ServiceInstanceState.TERMINATING
		):
		await self._publish_failed_before_start(command, service_order_id=None)
		return TerminateServiceResult()
		if service_instance.app_provider_id != command.app_provider_id:
		await self._publish_failed_before_start(command, service_order_id=None)
		return TerminateServiceResult()

		service_order = await self._service_orders.create(
		ServiceOrder(
		@@ -106,10 +130,60 @@ class TerminateServiceUseCase:
		target_zone_id=service_instance.zone_id,
		)

		async def complete(
		async def prepare_fulfill(
		self,
		command: TerminateServiceCommand,
		accepted: TerminateServiceResult,
		) -> TerminateFulfillmentPlan \| None:
		if accepted.accepted_order_id is None:
		return None

		service_order = await self._service_orders.get_by_id(accepted.accepted_order_id)
		if service_order is None:
		raise RuntimeError("Accepted terminate service order could not be reloaded")

		service_instance = await self._service_instances.get_by_id(command.service_instance_id)
		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(
		TerminateCapabilityTarget(
		capability_instance_id=capability.id,
		external_id=capability.external_id,
		control_path_ref_snapshot=capability.control_path_ref_snapshot,
		)
		for capability in service_instance.capabilities
		),
		)

		@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 persist_capability_terminated(
		self,
		capability_instance_id: UUID,
		terminated_at: datetime,
		) -> None:
		await self._capability_instances.update_state(
		capability_instance_id,
		CapabilityInstanceState.TERMINATED,
		terminated_at=terminated_at,
		)

		async def persist_completed(
		self,
		command: TerminateServiceCommand,
		accepted: TerminateServiceResult,
		*,
		terminated_at: datetime \| None = None,
		) -> SrmOperationCompletedV1 \| None:
		if accepted.accepted_order_id is None:
		return None
		@@ -122,17 +196,16 @@ class TerminateServiceUseCase:
		if service_instance is None:
		raise RuntimeError("Accepted terminate service instance could not be reloaded")

		terminated_at = datetime.now(UTC)
		await self._terminate_child_capabilities(service_instance, terminated_at)
		completed_at = terminated_at or datetime.now(UTC)
		await self._service_instances.update_state(
		service_instance.id,
		ServiceInstanceState.TERMINATED,
		terminated_at=terminated_at,
		terminated_at=completed_at,
		)
		await self._service_orders.update_state(
		service_order.id,
		ServiceOrderState.COMPLETED,
		completed_at=terminated_at,
		completed_at=completed_at,
		)

		structlog.get_logger(__name__).info(
		@@ -153,7 +226,7 @@ class TerminateServiceUseCase:
		}
		],
		correlation_id=command.correlation_id,
		completed_at=terminated_at,
		completed_at=completed_at,
		)

		async def fail_after_accept(
		@@ -275,20 +348,6 @@ class TerminateServiceUseCase:
		completed_at=service_order.completed_at or datetime.now(UTC),
		)

		async def _terminate_child_capabilities(
		self,
		service_instance: ServiceInstance,
		terminated_at: datetime,
		) -> None:
		for capability in service_instance.capabilities:
		# TODO: TF-SDK CapabilityExecutionAdapter.deactivate_capability/terminate(...)
		# call using capability.external_id and capability.control_path_ref_snapshot.
		await self._capability_instances.update_state(
		capability.id,
		CapabilityInstanceState.TERMINATED,
		terminated_at=terminated_at,
		)

		async def _publish_failed_before_start(
		self,
		command: TerminateServiceCommand,

tests/application/test_terminate_service_command_coordinator.py

+187 −20

Changes for tests/application/test_terminate_service_command_coordinator.py: 187 added lines, 20 removed lines.
Original line number	Diff line number	Diff line
		@@ -2,14 +2,18 @@ from __future__ import annotations

		from dataclasses import dataclass
		from datetime import datetime
		from unittest.mock import AsyncMock, MagicMock
		from unittest.mock import AsyncMock, MagicMock, patch
		from uuid import UUID

		from srm.api.databus.schemas import AnswerableCommandV1
		from srm.application.command_handlers.terminate_service import (
		TerminateServiceCommandCoordinator,
		)
		from srm.application.use_cases.terminate_service import TerminateServiceResult
		from srm.application.use_cases.terminate_service import (
		TerminateCapabilityTarget,
		TerminateFulfillmentPlan,
		TerminateServiceResult,
		)
		from srm.domain.ports.databus.events import (
		OperationCompletedStatus,
		SrmOperationCompletedV1,
		@@ -52,6 +56,27 @@ def _valid_envelope(*, terminate: dict[str, object] \| None = None) -> Answerable
		)


		def _fulfillment_plan(
		accepted_order_id: UUID,
		*,
		capability_count: int = 1,
		) -> TerminateFulfillmentPlan:
		capabilities = tuple(
		TerminateCapabilityTarget(
		capability_instance_id=UUID(f"bbbbbbbb-0000-4000-8000-{i:012d}"),
		external_id=f"ext-{i}",
		control_path_ref_snapshot=f"cp-ref-{i}",
		)
		for i in range(1, capability_count + 1)
		)
		return TerminateFulfillmentPlan(
		service_instance_id=UUID("9a3f1c22-0000-4000-8000-000000000002"),
		service_order_id=accepted_order_id,
		zone_id=UUID("642f6105-7015-4af1-a4d1-e1ecb8437abc"),
		capabilities=capabilities,
		)


		async def test_handle_publishes_failed_before_start_for_stage2_schema_failure() -> None:
		session_maker = FakeSessionMaker()
		use_case_factory = MagicMock()
		@@ -93,33 +118,83 @@ async def test_handle_commits_accept_before_publishing_accepted_then_completes()
		correlation_id="corr-appinst-001",
		completed_at=datetime.fromisoformat("2026-06-19T12:01:00+00:00"),
		)
		plan = _fulfillment_plan(accepted_order_id, capability_count=2)
		accept_use_case = AsyncMock()
		accept_use_case.accept.return_value = TerminateServiceResult(
		accepted_order_id=accepted_order_id
		)
		complete_use_case = AsyncMock()
		complete_use_case.complete.return_value = completed_event
		use_case_factory = MagicMock(side_effect=[accept_use_case, complete_use_case])
		prepare_use_case = AsyncMock()
		prepare_use_case.prepare_fulfill.return_value = plan
		persist_cap_use_case_1 = AsyncMock()
		persist_cap_use_case_2 = AsyncMock()
		persist_completed_use_case = AsyncMock()
		persist_completed_use_case.persist_completed.return_value = completed_event
		use_case_factory = MagicMock(
		side_effect=[
		accept_use_case,
		prepare_use_case,
		persist_cap_use_case_1,
		persist_cap_use_case_2,
		persist_completed_use_case,
		]
		)
		publisher = AsyncMock()
		coordinator = TerminateServiceCommandCoordinator(
		session_maker=session_maker, # type: ignore[arg-type]
		use_case_factory=use_case_factory,
		publisher=publisher,
		)
		call_order: list[str] = []

		async def track_prepare(args: object, *kwargs: object) -> TerminateFulfillmentPlan:
		call_order.append("prepare")
		return plan

		async def track_fulfill_one(args: object, *kwargs: object) -> None:
		call_order.append("fulfill_one")

		async def track_persist_cap(args: object, *kwargs: object) -> None:
		call_order.append("persist_cap")

		async def track_persist_completed(args: object, *kwargs: object) -> SrmOperationCompletedV1:
		call_order.append("persist_completed")
		return completed_event

		prepare_use_case.prepare_fulfill.side_effect = track_prepare
		persist_cap_use_case_1.persist_capability_terminated.side_effect = track_persist_cap
		persist_cap_use_case_2.persist_capability_terminated.side_effect = track_persist_cap
		persist_completed_use_case.persist_completed.side_effect = track_persist_completed

		with patch(
		"srm.application.command_handlers.terminate_service.TerminateServiceUseCase.fulfill_one",
		side_effect=track_fulfill_one,
		) as fulfill_mock:
		await coordinator.handle(_valid_envelope())

		assert len(session_maker.contexts) == 2
		# accept + prepare + 2 capability persists + final persist
		assert len(session_maker.contexts) == 5
		session_maker.contexts[0].session.commit.assert_awaited_once()
		session_maker.contexts[0].session.rollback.assert_not_awaited()
		session_maker.contexts[1].session.commit.assert_awaited_once()
		session_maker.contexts[1].session.rollback.assert_not_awaited()
		session_maker.contexts[2].session.commit.assert_awaited_once()
		session_maker.contexts[3].session.commit.assert_awaited_once()
		session_maker.contexts[4].session.commit.assert_awaited_once()
		assert fulfill_mock.await_count == 2
		assert fulfill_mock.await_args_list[0].args == (plan.capabilities[0],)
		assert fulfill_mock.await_args_list[1].args == (plan.capabilities[1],)
		persist_cap_use_case_1.persist_capability_terminated.assert_awaited_once()
		persist_cap_use_case_2.persist_capability_terminated.assert_awaited_once()
		persist_completed_use_case.persist_completed.assert_awaited_once()
		assert call_order == [
		"prepare",
		"fulfill_one",
		"persist_cap",
		"fulfill_one",
		"persist_cap",
		"persist_completed",
		]
		assert publisher.publish.await_count == 2
		subject, payload = publisher.publish.await_args_list[0].args
		assert subject == "event.srm.operation.status"
		assert payload["state"] == "accepted"
		assert payload["service_order_id"] == str(accepted_order_id)
		assert payload["service_instance_id"] == "9a3f1c22-0000-4000-8000-000000000002"
		subject, payload = publisher.publish.await_args_list[1].args
		assert subject == "event.srm.operation.completed"
		assert payload["status"] == "completed"
		@@ -166,7 +241,7 @@ async def test_handle_republishes_terminal_event_for_terminal_duplicate() -> Non
		assert payload["service_order_id"] == str(duplicate_order_id)


		async def test_handle_publishes_failed_event_when_complete_raises() -> None:
		async def test_handle_keeps_prior_capability_commits_when_fulfill_one_fails_later() -> None:
		session_maker = FakeSessionMaker()
		accepted_order_id = UUID("aaaaaaaa-0000-4000-8000-000000000001")
		failed_event = SrmOperationCompletedV1(
		@@ -178,15 +253,26 @@ async def test_handle_publishes_failed_event_when_complete_raises() -> None:
		correlation_id="corr-appinst-001",
		completed_at=datetime.fromisoformat("2026-06-19T12:01:00+00:00"),
		)
		plan = _fulfillment_plan(accepted_order_id, capability_count=3)
		accept_use_case = AsyncMock()
		accept_use_case.accept.return_value = TerminateServiceResult(
		accepted_order_id=accepted_order_id
		)
		complete_use_case = AsyncMock()
		complete_use_case.complete.side_effect = RuntimeError("adapter timeout")
		prepare_use_case = AsyncMock()
		prepare_use_case.prepare_fulfill.return_value = plan
		persist_cap_use_case_1 = AsyncMock()
		persist_cap_use_case_2 = AsyncMock()
		fail_use_case = AsyncMock()
		fail_use_case.fail_after_accept.return_value = failed_event
		use_case_factory = MagicMock(side_effect=[accept_use_case, complete_use_case, fail_use_case])
		use_case_factory = MagicMock(
		side_effect=[
		accept_use_case,
		prepare_use_case,
		persist_cap_use_case_1,
		persist_cap_use_case_2,
		fail_use_case,
		]
		)
		publisher = AsyncMock()
		coordinator = TerminateServiceCommandCoordinator(
		session_maker=session_maker, # type: ignore[arg-type]
		@@ -194,10 +280,79 @@ async def test_handle_publishes_failed_event_when_complete_raises() -> None:
		publisher=publisher,
		)

		async def fulfill_two_then_fail(target: TerminateCapabilityTarget) -> None:
		if target.capability_instance_id == plan.capabilities[2].capability_instance_id:
		raise RuntimeError("adapter timeout")

		with patch(
		"srm.application.command_handlers.terminate_service.TerminateServiceUseCase.fulfill_one",
		side_effect=fulfill_two_then_fail,
		):
		await coordinator.handle(_valid_envelope())

		session_maker.contexts[1].session.rollback.assert_awaited_once()
		# accept + prepare + 2 capability commits + fail_after_accept
		assert len(session_maker.contexts) == 5
		session_maker.contexts[2].session.commit.assert_awaited_once()
		session_maker.contexts[3].session.commit.assert_awaited_once()
		session_maker.contexts[2].session.rollback.assert_not_awaited()
		session_maker.contexts[3].session.rollback.assert_not_awaited()
		persist_cap_use_case_1.persist_capability_terminated.assert_awaited_once()
		persist_cap_use_case_2.persist_capability_terminated.assert_awaited_once()
		fail_use_case.fail_after_accept.assert_awaited_once()
		session_maker.contexts[4].session.commit.assert_awaited_once()
		assert publisher.publish.await_count == 2
		subject, payload = publisher.publish.await_args_list[1].args
		assert subject == "event.srm.operation.completed"
		assert payload["status"] == "failed"


		async def test_handle_publishes_failed_event_when_persist_completed_raises() -> None:
		session_maker = FakeSessionMaker()
		accepted_order_id = UUID("aaaaaaaa-0000-4000-8000-000000000001")
		failed_event = SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=UUID("11111111-2222-3333-4444-555555555555"),
		status=OperationCompletedStatus.FAILED,
		service_order_id=accepted_order_id,
		error={"type": "about:blank", "title": "failed", "status": 500},
		correlation_id="corr-appinst-001",
		completed_at=datetime.fromisoformat("2026-06-19T12:01:00+00:00"),
		)
		plan = _fulfillment_plan(accepted_order_id)
		accept_use_case = AsyncMock()
		accept_use_case.accept.return_value = TerminateServiceResult(
		accepted_order_id=accepted_order_id
		)
		prepare_use_case = AsyncMock()
		prepare_use_case.prepare_fulfill.return_value = plan
		persist_cap_use_case = AsyncMock()
		persist_completed_use_case = AsyncMock()
		persist_completed_use_case.persist_completed.side_effect = RuntimeError("db write failed")
		fail_use_case = AsyncMock()
		fail_use_case.fail_after_accept.return_value = failed_event
		use_case_factory = MagicMock(
		side_effect=[
		accept_use_case,
		prepare_use_case,
		persist_cap_use_case,
		persist_completed_use_case,
		fail_use_case,
		]
		)
		publisher = AsyncMock()
		coordinator = TerminateServiceCommandCoordinator(
		session_maker=session_maker, # type: ignore[arg-type]
		use_case_factory=use_case_factory,
		publisher=publisher,
		)

		await coordinator.handle(_valid_envelope())

		# accept + prepare + cap persist + persist_completed (rollback) + fail
		assert len(session_maker.contexts) == 5
		session_maker.contexts[2].session.commit.assert_awaited_once()
		session_maker.contexts[3].session.rollback.assert_awaited_once()
		session_maker.contexts[4].session.commit.assert_awaited_once()
		fail_use_case.fail_after_accept.assert_awaited_once()
		assert publisher.publish.await_count == 2
		subject, payload = publisher.publish.await_args_list[1].args
		@@ -208,16 +363,28 @@ async def test_handle_publishes_failed_event_when_complete_raises() -> None:
		async def test_handle_publishes_fallback_failed_event_when_fail_after_accept_raises() -> None:
		session_maker = FakeSessionMaker()
		accepted_order_id = UUID("aaaaaaaa-0000-4000-8000-000000000001")
		plan = _fulfillment_plan(accepted_order_id)
		accept_use_case = AsyncMock()
		accept_use_case.accept.return_value = TerminateServiceResult(
		accepted_order_id=accepted_order_id,
		target_zone_id=UUID("642f6105-7015-4af1-a4d1-e1ecb8437abc"),
		)
		complete_use_case = AsyncMock()
		complete_use_case.complete.side_effect = RuntimeError("adapter timeout")
		prepare_use_case = AsyncMock()
		prepare_use_case.prepare_fulfill.return_value = plan
		persist_cap_use_case = AsyncMock()
		persist_completed_use_case = AsyncMock()
		persist_completed_use_case.persist_completed.side_effect = RuntimeError("adapter timeout")
		fail_use_case = AsyncMock()
		fail_use_case.fail_after_accept.side_effect = RuntimeError("database down")
		use_case_factory = MagicMock(side_effect=[accept_use_case, complete_use_case, fail_use_case])
		use_case_factory = MagicMock(
		side_effect=[
		accept_use_case,
		prepare_use_case,
		persist_cap_use_case,
		persist_completed_use_case,
		fail_use_case,
		]
		)
		publisher = AsyncMock()
		coordinator = TerminateServiceCommandCoordinator(
		session_maker=session_maker, # type: ignore[arg-type]
		@@ -227,7 +394,7 @@ async def test_handle_publishes_fallback_failed_event_when_fail_after_accept_rai

		await coordinator.handle(_valid_envelope())

		session_maker.contexts[2].session.rollback.assert_awaited_once()
		session_maker.contexts[4].session.rollback.assert_awaited_once()
		assert publisher.publish.await_count == 2
		subject, payload = publisher.publish.await_args_list[1].args
		assert subject == "event.srm.operation.completed"

tests/unit/test_terminate_service_use_case.py

+88 −10

File changed.

Preview size limit exceeded, changes collapsed.