feat(fm): take SRM's federated deploys and report the outcome (ADR-0053) (1b9b7595) · Commits · OOP / code / federation-manager

src/federation_manager/adapters/databus/nats_adapter.py

+29 −12

Changes for src/federation_manager/adapters/databus/nats_adapter.py: 29 added lines, 12 removed lines.
Original line number	Diff line number	Diff line
		@@ -41,7 +41,7 @@ class NatsCommandPublisher:
		js = self._require_js()
		config = StreamConfig(
		name=TASK_STREAM,
		subjects=["command.srm.>"],
		subjects=["command.srm.>", "command.fm.>"],
		retention=RetentionPolicy.WORK_QUEUE,
		max_age=TASK_STREAM_MAX_AGE_SECONDS,
		)
		@@ -82,16 +82,18 @@ class NatsEventConsumer:

		async def ensure_event_stream(self) -> None:
		js = self._require_js()
		try:
		await js.stream_info(EVENT_STREAM)
		except Exception:
		await js.add_stream(
		StreamConfig(
		config = StreamConfig(
		name=EVENT_STREAM,
		subjects=["event.srm.>"],
		subjects=["event.srm.>", "event.fm.>"],
		retention=RetentionPolicy.LIMITS,
		)
		)
		try:
		await js.stream_info(EVENT_STREAM)
		except Exception:
		await js.add_stream(config)
		else:
		# a stream from an older FM lacks event.fm.>, and FM's own publishes would fail
		await js.update_stream(config)

		async def subscribe(
		self, subject: str, handler: Callable[[dict[str, Any]], Awaitable[None]]
		@@ -110,23 +112,38 @@ class NatsEventConsumer:
		return
		await message.ack()

		name = f"{self._durable}-{subject.replace('.', '-')}"
		name = self._consumer_name(subject)
		# queue group so replicas of this FM share one durable instead of fighting over it
		await self._require_js().subscribe(
		subject,
		queue=name,
		durable=name,
		stream=EVENT_STREAM,
		stream=_stream_for(subject),
		cb=on_message,
		manual_ack=True,
		config=ConsumerConfig(max_deliver=3, ack_policy=AckPolicy.EXPLICIT),
		)

		def _consumer_name(self, subject: str) -> str:
		# A work-queue stream allows one consumer per subject, so commands share one durable
		# (ADR-0053's fm-task-worker); events keep a per-FM durable each.
		if subject.startswith("command."):
		return TASK_CONSUMER
		return f"{self._durable}-{subject.replace('.', '-')}"

		async def delete_durable(self, subject: str) -> None:
		name = f"{self._durable}-{subject.replace('.', '-')}"
		await self._require_js().delete_consumer(EVENT_STREAM, name)
		name = self._consumer_name(subject)
		await self._require_js().delete_consumer(_stream_for(subject), name)

		def _require_js(self) -> JetStreamContext:
		if self._js is None:
		raise RuntimeError("NATS event consumer is not connected")
		return self._js


		TASK_CONSUMER = "fm-task-worker"


		def _stream_for(subject: str) -> str:
		# commands live in the work-queue stream, events in the event stream
		return TASK_STREAM if subject.startswith("command.") else EVENT_STREAM

src/federation_manager/application/callbacks.py

+35 −1

Changes for src/federation_manager/application/callbacks.py: 35 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		from collections.abc import Callable
		from datetime import datetime, timezone
		from typing import Literal
		from uuid import UUID

		from federation_manager.application.federated_handoff import publish_outcome
		from federation_manager.application.federation import TERMINATED
		from federation_manager.contracts.ewbi import InstanceStatusCallback, PartnerStatusEvent
		from federation_manager.contracts.srm import FmOperationCompletedV1
		from federation_manager.core.logging import get_logger
		from federation_manager.domain.errors import FederationContextUnknown, InstanceUnknown
		from federation_manager.domain.models import OUTBOUND, FederationContext, PartnerOP
		from federation_manager.domain.ports import (
		DataBusPublisherPort,
		FederationContextRepositoryPort,
		TransactionRepositoryPort,
		)
		@@ -76,9 +81,11 @@ class InstanceStatusCallbackService:
		self,
		contexts: FederationContextRepositoryPort,
		transactions: TransactionRepositoryPort,
		publisher: DataBusPublisherPort \| None = None,
		*,
		clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
		) -> None:
		self._publisher = publisher
		self._contexts = contexts
		self._transactions = transactions
		self._clock = clock
		@@ -92,6 +99,7 @@ class InstanceStatusCallbackService:
		raise InstanceUnknown(event.app_instance_id)

		state = event.app_instance_info.app_instance_state
		status: Literal["completed", "failed"]
		if state == "READY":
		status = "completed"
		elif state == "FAILED":
		@@ -104,15 +112,41 @@ class InstanceStatusCallbackService:
		state=state,
		)
		return
		completed_at = self._clock()
		await self._transactions.record_outcome(
		transaction.id,
		status=status,
		completed_at=self._clock(),
		completed_at=completed_at,
		response_summary={
		"app_instance_state": state,
		"message": event.app_instance_info.message,
		},
		)
		if transaction.operation_id is not None and self._publisher is not None:
		# SRM handed this deploy to us and is waiting for the outcome (ADR-0053)
		summary = transaction.request_summary
		instance = summary.get("service_instance_id")
		await publish_outcome(
		self._publisher,
		FmOperationCompletedV1(
		operation_id=transaction.operation_id,
		correlation_id=str(summary.get("correlation_id", transaction.operation_id)),
		operation="deploy",
		status=status,
		partner_op_id=partner.id,
		federation_transaction_id=transaction.id,
		service_instance_id=UUID(str(instance)) if instance else None,
		external_ref=event.app_instance_id,
		metadata={"completion_basis": "partner_confirmed"},
		error=None
		if status == "completed"
		else {
		"type": "urn:oop:fm:error:partner-instance-failed",
		"title": event.app_instance_info.message or "The partner reported FAILED.",
		},
		completed_at=completed_at,
		),
		)
		logger.info(
		"instance_status_callback_applied",
		partner_op_id=str(partner.id),

src/federation_manager/application/federated_handoff.py

0 → 100644

+110 −0

Changes for src/federation_manager/application/federated_handoff.py: 110 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import re
		from collections.abc import Callable
		from datetime import datetime, timezone
		from typing import Any
		from uuid import UUID

		from federation_manager.application.outbound_deploy import OutboundDeploymentService, SrmHandOff
		from federation_manager.contracts.srm import (
		SUBJECT_FM_COMPLETED,
		FmOperationCompletedV1,
		SrmServiceDeployV1,
		)
		from federation_manager.core.logging import get_logger
		from federation_manager.domain.errors import FederationError
		from federation_manager.domain.ports import (
		CatalogueRepositoryPort,
		DataBusPublisherPort,
		FederationContextRepositoryPort,
		TransactionRepositoryPort,
		)

		logger = get_logger(__name__)


		def _utcnow() -> datetime:
		return datetime.now(timezone.utc)


		# command.fm.service.deploy: SRM found the target zone at a partner
		class FederatedDeployHandler:
		def __init__(
		self,
		catalogue: CatalogueRepositoryPort,
		contexts: FederationContextRepositoryPort,
		transactions: TransactionRepositoryPort,
		deployments: OutboundDeploymentService,
		publisher: DataBusPublisherPort,
		*,
		clock: Callable[[], datetime] = _utcnow,
		) -> None:
		self._catalogue = catalogue
		self._contexts = contexts
		self._transactions = transactions
		self._deployments = deployments
		self._publisher = publisher
		self._clock = clock

		async def handle(self, payload: dict[str, Any]) -> None:
		command = SrmServiceDeployV1.model_validate(payload)
		if await self._transactions.find_by_operation_id(command.operation_id) is not None:
		return # redelivered: the InstallApp for this operation already went out
		for target in command.targets:
		try:
		await self._deploy(command, target.app_instance_id, target.zone_id)
		except FederationError as error:
		logger.warning(
		"federated_deploy_refused",
		operation_id=str(command.operation_id),
		error=type(error).__name__,
		)
		await publish_outcome(
		self._publisher,
		FmOperationCompletedV1(
		operation_id=command.operation_id,
		correlation_id=command.correlation_id,
		operation="deploy",
		status="failed",
		partner_op_id=None,
		federation_transaction_id=None,
		service_instance_id=target.app_instance_id,
		error=_problem(error),
		completed_at=self._clock(),
		),
		)

		async def _deploy(
		self, command: SrmServiceDeployV1, service_instance_id: UUID, zone_id: UUID \| None
		) -> None:
		zone = await self._catalogue.find_zone(zone_id) if zone_id else None
		if zone is None:
		raise ZoneNotFederatedHere(zone_id)
		context = await self._contexts.find_active_outbound(zone.partner_op_id)
		if context is None:
		raise NoFederationWithPartner(zone.partner_op_id)
		await self._deployments.deploy(
		zone.partner_op_id,
		context.federation_context_id,
		command.service_specification_id,
		zone.zone_id,
		SrmHandOff(command.operation_id, command.correlation_id, service_instance_id),
		)


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


		class NoFederationWithPartner(FederationError):
		def __init__(self, partner_id: UUID) -> None:
		super().__init__(f"no active federation with partner {partner_id}")


		async def publish_outcome(publisher: DataBusPublisherPort, outcome: FmOperationCompletedV1) -> None:
		await publisher.publish(SUBJECT_FM_COMPLETED, outcome.model_dump(mode="json"))


		def _problem(error: FederationError) -> dict[str, Any]:
		kind = re.sub(r"(?<!^)(?=[A-Z])", "-", type(error).__name__).lower()
		return {"type": f"urn:oop:fm:error:{kind}", "title": str(error)}

src/federation_manager/application/outbound_deploy.py

+26 −0

Changes for src/federation_manager/application/outbound_deploy.py: 26 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -70,6 +70,14 @@ def _utcnow() -> datetime:
		return datetime.now(timezone.utc)


		@dataclass(frozen=True)
		class SrmHandOff:
		# set when SRM handed the deploy to us, so the outcome can be reported back to it
		operation_id: UUID
		correlation_id: str
		service_instance_id: UUID


		@dataclass(frozen=True)
		class OutboundInstall:
		app_id: str
		@@ -108,6 +116,7 @@ class OutboundDeploymentService:
		federation_context_id: str,
		service_specification_id: UUID,
		zone_id: UUID,
		hand_off: SrmHandOff \| None = None,
		) -> OutboundInstall:
		partner = await self._partners.find_by_id(partner_op_id)
		if partner is None:
		@@ -166,6 +175,8 @@ class OutboundDeploymentService:
		federation_context_row_id=context.id,
		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,
		request_summary={
		"federation_context_id": context.federation_context_id,
		"app_id": request.app_id,
		@@ -173,6 +184,14 @@ class OutboundDeploymentService:
		"zone_id": str(zone_id),
		"partner_zone_id": request.zone_info.zone_id,
		"service_specification_id": str(service_specification_id),
		**(
		{
		"service_instance_id": str(hand_off.service_instance_id),
		"correlation_id": hand_off.correlation_id,
		}
		if hand_off
		else {}
		),
		},
		started_at=self._clock(),
		)
		@@ -316,3 +335,10 @@ def _component_spec(unit: dict[str, Any], image: UUID) -> ComponentSpec:
		]
		or None,
		)


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

src/federation_manager/contracts/srm.py

+18 −0

Changes for src/federation_manager/contracts/srm.py: 18 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -15,6 +15,9 @@ SUBJECT_DEPLOY = "command.srm.service.deploy"
		SUBJECT_TERMINATE = "command.srm.service.terminate"
		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_COMPLETED = "event.fm.operation.completed"


		class CommandEnvelopeV1(BaseModel):
		@@ -153,3 +156,18 @@ class NetworkCapabilityDeactivatePayloadV1(BaseModel):
		class SrmNetworkCapabilityDeactivateV1(CommandEnvelopeV1):
		network_capability: NetworkCapabilityDeactivatePayloadV1
		service_specification_id: UUID \| None = None


		class FmOperationCompletedV1(BaseModel):
		schema_version: str = "1.0"
		operation_id: UUID
		correlation_id: str
		operation: Literal["deploy", "terminate"]
		status: Literal["completed", "failed"]
		partner_op_id: UUID \| None
		federation_transaction_id: UUID \| None
		service_instance_id: UUID \| None = None
		external_ref: str \| None = None
		metadata: dict[str, Any] = Field(default_factory=dict)
		error: dict[str, Any] \| None = None
		completed_at: datetime