feat(fm): remove a federated instance at the partner when SRM hands over a terminate (ADR-0053) (!27) · Merge requests · OOP / code / federation-manager

src/federation_manager/adapters/database/transaction_repo.py

+17 −1

Changes for src/federation_manager/adapters/database/transaction_repo.py: 17 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@ from sqlalchemy import Select, insert, select, update
		from sqlalchemy.ext.asyncio import AsyncSession

		from federation_manager.adapters.database.tables import federation_transactions as transactions
		from federation_manager.domain.models import FederationTransaction
		from federation_manager.domain.models import OUTBOUND, FederationTransaction


		class PostgresTransactionRepo:
		@@ -72,6 +72,22 @@ class PostgresTransactionRepo:
		.limit(1)
		)

		async def find_outbound_by_service_instance(
		self, api_type: str, service_instance_id: UUID
		) -> FederationTransaction \| None:
		# SRM hands us its own instance id; the partner's appInstIdentifier is theirs to pick
		return await self._one(
		select(transactions)
		.where(
		transactions.c.direction == OUTBOUND,
		transactions.c.api_type == api_type,
		transactions.c.request_summary["service_instance_id"].astext
		== str(service_instance_id),
		)
		.order_by(transactions.c.started_at.desc())
		.limit(1)
		)

		async def _one(self, stmt: Select[tuple[object, ...]]) -> FederationTransaction \| None:
		row = (await self._session.execute(stmt)).one_or_none()
		if row is None:

src/federation_manager/adapters/databus/nats_adapter.py

+38 −1

Changes for src/federation_manager/adapters/databus/nats_adapter.py: 38 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@ from nats.aio.client import Client
		from nats.aio.msg import Msg
		from nats.js import JetStreamContext
		from nats.js.api import AckPolicy, ConsumerConfig, RetentionPolicy, StreamConfig
		from nats.js.errors import NotFoundError

		from federation_manager.contracts.srm import (
		EVENT_STREAM,
		@@ -97,10 +98,45 @@ class NatsEventConsumer:

		async def subscribe(
		self, subject: str, handler: Callable[[dict[str, Any]], Awaitable[None]]
		) -> None:
		async def by_payload(_subject: str, payload: dict[str, Any]) -> None:
		await handler(payload)

		await self._subscribe(subject, by_payload)

		async def subscribe_commands(
		self, handlers: dict[str, Callable[[dict[str, Any]], Awaitable[None]]]
		) -> None:
		# One durable for every command FM takes (ADR-0053's fm-task-worker on command.fm.>),
		# dispatched by subject. A second subscribe() on the same durable would silently bind
		# to its old filter and share its deliver subject with the first handler.
		js = self._require_js()
		try:
		info = await js.consumer_info(TASK_STREAM, TASK_CONSUMER)
		except NotFoundError:
		pass
		else:
		if info.config.filter_subject != TASK_FILTER:
		# an older FM filtered it on command.fm.service.deploy alone; the unacked
		# commands stay in the work-queue stream for the recreated consumer
		await js.delete_consumer(TASK_STREAM, TASK_CONSUMER)

		async def dispatch(subject: str, payload: dict[str, Any]) -> None:
		handler = handlers.get(subject)
		if handler is None:
		# acked: no FM replica has a handler, so redelivery can't help
		logger.warning("command_without_handler", subject=subject)
		return
		await handler(payload)

		await self._subscribe(TASK_FILTER, dispatch)

		async def _subscribe(
		self, subject: str, handler: Callable[[str, dict[str, Any]], Awaitable[None]]
		) -> None:
		async def on_message(message: Msg) -> None:
		try:
		await handler(json.loads(message.data))
		await handler(message.subject, json.loads(message.data))
		except Exception as error:
		# left unacked so JetStream redelivers up to max_deliver
		logger.error(
		@@ -142,6 +178,7 @@ class NatsEventConsumer:


		TASK_CONSUMER = "fm-task-worker"
		TASK_FILTER = "command.fm.>"


		def _stream_for(subject: str) -> str:

src/federation_manager/application/federated_handoff.py

+145 −3

Changes for src/federation_manager/application/federated_handoff.py: 145 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -2,23 +2,39 @@ import re
		from collections.abc import Callable
		from datetime import datetime, timezone
		from typing import Any
		from uuid import UUID
		from uuid import UUID, uuid4

		from federation_manager.application.outbound_deploy import OutboundDeploymentService, SrmHandOff
		from federation_manager.application.outbound_deploy import (
		INSTALL_APP,
		OutboundDeploymentService,
		SrmHandOff,
		uuid_or_none,
		)
		from federation_manager.contracts.srm import (
		SUBJECT_FM_COMPLETED,
		FmOperationCompletedV1,
		SrmServiceDeployV1,
		SrmServiceTerminateV1,
		)
		from federation_manager.core.logging import get_logger
		from federation_manager.domain.errors import FederationError
		from federation_manager.domain.errors import (
		FederationError,
		PartnerNotRegistered,
		PartnerRejectedRequest,
		)
		from federation_manager.domain.ewbi import remove_app_path
		from federation_manager.domain.models import OUTBOUND, FederationTransaction
		from federation_manager.domain.ports import (
		CatalogueRepositoryPort,
		DataBusPublisherPort,
		EwbiClientPort,
		FederationContextRepositoryPort,
		PartnerRepositoryPort,
		TransactionRepositoryPort,
		)

		REMOVE_APP = "remove-app"

		logger = get_logger(__name__)


		@@ -91,6 +107,132 @@ class FederatedDeployHandler:
		)


		# command.fm.service.terminate: SRM's instance row says a partner hosts it (ADR-0053)
		class FederatedTerminateHandler:
		def __init__(
		self,
		partners: PartnerRepositoryPort,
		transactions: TransactionRepositoryPort,
		ewbi_client: EwbiClientPort,
		publisher: DataBusPublisherPort,
		*,
		clock: Callable[[], datetime] = _utcnow,
		id_factory: Callable[[], UUID] = uuid4,
		) -> None:
		self._partners = partners
		self._transactions = transactions
		self._ewbi = ewbi_client
		self._publisher = publisher
		self._clock = clock
		self._new_id = id_factory

		async def handle(self, payload: dict[str, Any]) -> None:
		command = SrmServiceTerminateV1.model_validate(payload)
		done = await self._transactions.find_by_operation_id(command.operation_id)
		if done is not None:
		# redelivered: the RemoveApp already went out, but the outcome may not have
		# (a failed publish is what leaves the command unacked), so say it again
		await publish_outcome(self._publisher, _removed(command, done))
		return
		try:
		outcome = await self._terminate(command)
		except FederationError as error:
		logger.warning(
		"federated_terminate_refused",
		operation_id=str(command.operation_id),
		error=type(error).__name__,
		)
		outcome = FmOperationCompletedV1(
		operation_id=command.operation_id,
		correlation_id=command.correlation_id,
		operation="terminate",
		status="failed",
		partner_op_id=None,
		federation_transaction_id=None,
		service_instance_id=command.service_instance_id,
		error=_problem(error),
		completed_at=self._clock(),
		)
		await publish_outcome(self._publisher, outcome)

		async def _terminate(self, command: SrmServiceTerminateV1) -> FmOperationCompletedV1:
		install = await self._transactions.find_outbound_by_service_instance(
		INSTALL_APP, command.service_instance_id
		)
		if install is None or install.external_resource_id is None:
		raise InstanceNotFederated(command.service_instance_id)
		partner = await self._partners.find_by_id(install.partner_op_id)
		if partner is None:
		raise PartnerNotRegistered(install.partner_op_id)
		summary = install.request_summary
		response = await self._ewbi.delete(
		partner,
		remove_app_path(
		str(summary["federation_context_id"]),
		str(summary["app_id"]),
		install.external_resource_id,
		str(summary["partner_zone_id"]),
		),
		)
		# OPG.04 gives RemoveApp no callback: the partner's 200 is the most we can observe.
		# A 404 means the partner no longer has the instance, which is what was asked for.
		if response.status_code == 200:
		basis = "partner_accepted"
		elif response.status_code == 404:
		basis = "partner_not_found"
		else:
		raise PartnerRejectedRequest(partner.id, response.status_code, response.body)

		completed_at = self._clock()
		transaction = FederationTransaction(
		id=self._new_id(),
		partner_op_id=partner.id,
		direction=OUTBOUND,
		api_type=REMOVE_APP,
		status="completed",
		federation_context_row_id=install.federation_context_row_id,
		external_resource_id=install.external_resource_id,
		operation_id=command.operation_id,
		correlation_id=uuid_or_none(command.correlation_id),
		request_summary={
		"federation_context_id": summary["federation_context_id"],
		"app_id": summary["app_id"],
		"partner_zone_id": summary["partner_zone_id"],
		"service_instance_id": str(command.service_instance_id),
		"correlation_id": command.correlation_id,
		"install_transaction_id": str(install.id),
		},
		response_summary={"status_code": response.status_code, "completion_basis": basis},
		started_at=completed_at,
		completed_at=completed_at,
		)
		await self._transactions.add(transaction)
		return _removed(command, transaction)


		def _removed(
		command: SrmServiceTerminateV1, removal: FederationTransaction
		) -> FmOperationCompletedV1:
		basis = (removal.response_summary or {}).get("completion_basis", "partner_accepted")
		return FmOperationCompletedV1(
		operation_id=command.operation_id,
		correlation_id=command.correlation_id,
		operation="terminate",
		status="completed",
		partner_op_id=removal.partner_op_id,
		federation_transaction_id=removal.id,
		service_instance_id=command.service_instance_id,
		external_ref=removal.external_resource_id,
		metadata={"completion_basis": basis},
		completed_at=removal.completed_at or removal.started_at,
		)


		class InstanceNotFederated(FederationError):
		def __init__(self, service_instance_id: UUID) -> None:
		super().__init__(f"no partner hosts service instance {service_instance_id}")


		class ZoneNotFederatedHere(FederationError):
		def __init__(self, zone_id: UUID \| None) -> None:
		super().__init__(f"no partner offers zone {zone_id}")

src/federation_manager/application/outbound_deploy.py

+2 −2

Changes for src/federation_manager/application/outbound_deploy.py: 2 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -176,7 +176,7 @@ class OutboundDeploymentService:
		idempotency_key=idempotency_key,
		external_resource_id=installed.app_inst_identifier,
		operation_id=hand_off.operation_id if hand_off else None,
		correlation_id=_uuid_or_none(hand_off.correlation_id) if hand_off else None,
		correlation_id=uuid_or_none(hand_off.correlation_id) if hand_off else None,
		request_summary={
		"federation_context_id": context.federation_context_id,
		"app_id": request.app_id,
		@@ -337,7 +337,7 @@ def _component_spec(unit: dict[str, Any], image: UUID) -> ComponentSpec:
		)


		def _uuid_or_none(value: str) -> UUID \| None:
		def uuid_or_none(value: str) -> UUID \| None:
		try:
		return UUID(value)
		except ValueError:

src/federation_manager/contracts/srm.py

+1 −0

Changes for src/federation_manager/contracts/srm.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -17,6 +17,7 @@ SUBJECT_CAPABILITY_ACTIVATE = "command.srm.network.capability.activate"
		SUBJECT_CAPABILITY_DEACTIVATE = "command.srm.network.capability.deactivate"
		# SRM's federated hand-off to FM and FM's answer (ADR-0053)
		SUBJECT_FM_DEPLOY = "command.fm.service.deploy"
		SUBJECT_FM_TERMINATE = "command.fm.service.terminate"
		SUBJECT_FM_COMPLETED = "event.fm.operation.completed"

feat(fm): remove a federated instance at the partner when SRM hands over a terminate (ADR-0053)

Merge request reports