feat(fm): keep partner zones in step, and release subscriptions with the federation (262f5d38) · Commits · OOP / code / federation-manager

src/federation_manager/application/agreements.py

+24 −4

Changes for src/federation_manager/application/agreements.py: 24 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,12 +3,13 @@ from dataclasses import replace
		from typing import Any
		from uuid import UUID, uuid4

		from federation_manager.application.zone_notifications import ZoneOfferNotifier
		from federation_manager.domain.errors import (
		AgreementNotFound,
		AgreementValidityInvalid,
		PartnerNotRegistered,
		)
		from federation_manager.domain.models import Agreement
		from federation_manager.domain.models import Agreement, LocalZone, PartnerOP
		from federation_manager.domain.ports import AgreementRepositoryPort, PartnerRepositoryPort


		@@ -20,17 +21,21 @@ class AgreementService:
		agreements: AgreementRepositoryPort,
		partners: PartnerRepositoryPort,
		*,
		notifier: ZoneOfferNotifier \| None = None,
		id_factory: Callable[[], UUID] = uuid4,
		) -> None:
		self._agreements = agreements
		self._partners = partners
		self._notifier = notifier
		self._new_id = id_factory

		async def create(self, fields: dict[str, Any]) -> Agreement:
		await self._require_partner(fields["partner_op_id"])
		partner = await self._require_partner(fields["partner_op_id"])
		agreement = Agreement(id=self._new_id(), **fields)
		_require_valid_period(agreement)
		before = await self._offer(partner)
		await self._agreements.add(agreement)
		await self._tell(partner, before)
		return agreement

		async def get(self, agreement_id: UUID) -> Agreement:
		@@ -42,15 +47,30 @@ class AgreementService:
		async def update(self, agreement_id: UUID, changes: dict[str, Any]) -> Agreement:
		updated = replace(await self.get(agreement_id), **changes)
		_require_valid_period(updated)
		partner = await self._require_partner(updated.partner_op_id)
		before = await self._offer(partner)
		await self._agreements.update(updated)
		await self._tell(partner, before)
		return updated

		async def list_for_partner(self, partner_id: UUID) -> list[Agreement]:
		return await self._agreements.list_for_partner(partner_id)

		async def _require_partner(self, partner_id: UUID) -> None:
		if await self._partners.find_by_id(partner_id) is None:
		async def _require_partner(self, partner_id: UUID) -> PartnerOP:
		partner = await self._partners.find_by_id(partner_id)
		if partner is None:
		raise PartnerNotRegistered(partner_id)
		return partner

		async def _offer(self, partner: PartnerOP) -> dict[UUID, LocalZone] \| None:
		if self._notifier is None:
		return None
		return await self._notifier.offered(partner.id)

		async def _tell(self, partner: PartnerOP, before: dict[UUID, LocalZone] \| None) -> None:
		# a partner already federated learns the change now, not at its next read (REQ-FED-04)
		if self._notifier is not None:
		await self._notifier.notify(partner, before, await self._notifier.offered(partner.id))


		def _require_valid_period(agreement: Agreement) -> None:

src/federation_manager/application/callbacks.py

+28 −4

Changes for src/federation_manager/application/callbacks.py: 28 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -5,12 +5,14 @@ from uuid import UUID

		from federation_manager.application.federated_handoff import publish_outcome
		from federation_manager.application.federation import TERMINATED
		from federation_manager.application.partner_zones import add_to_offer, remove_from_offer
		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 (
		CatalogueRepositoryPort,
		DataBusPublisherPort,
		FederationContextRepositoryPort,
		TransactionRepositoryPort,
		@@ -31,16 +33,38 @@ _STATUS: dict[str, str] = {
		class PartnerStatusCallbackService:
		"""onPartnerStatusEvent: the partner tells us something about a federation we created."""

		def __init__(self, contexts: FederationContextRepositoryPort) -> None:
		def __init__(
		self,
		contexts: FederationContextRepositoryPort,
		catalogue: CatalogueRepositoryPort \| None = None,
		) -> None:
		self._contexts = contexts
		self._catalogue = catalogue

		async def handle(self, partner: PartnerOP, event: PartnerStatusEvent) -> None:
		context = await outbound_context(self._contexts, partner, event.federation_context_id)

		if event.object_type == "ZONES" and self._catalogue is not None:
		if event.operation_type == "ADD" and event.add_zones:
		await add_to_offer(
		self._catalogue,
		partner.id,
		[zone.model_dump(by_alias=True, exclude_none=True) for zone in event.add_zones],
		)
		elif event.operation_type == "REMOVE" and event.remove_zones:
		await remove_from_offer(self._catalogue, partner.id, event.remove_zones)
		logger.info(
		"partner_zones_callback_applied",
		partner_op_id=str(partner.id),
		federation_context_id=event.federation_context_id,
		operation_type=event.operation_type,
		zones=len(event.add_zones or event.remove_zones or []),
		)
		return

		if event.object_type != "FEDERATION" or event.federation_status is None:
		# Zone, network-code and service-API deltas need shared_resource_catalogues
		# (REQ-FED-04), which does not exist yet. Accept the notification rather than
		# make the partner retry a delivery we will never process.
		# Network-code, service-API and zone-status changes aren't tracked. Accept the
		# notification rather than make the partner retry a delivery we will never process.
		logger.info(
		"partner_status_callback_ignored",
		partner_op_id=str(partner.id),

src/federation_manager/application/federation.py

+9 −19

Changes for src/federation_manager/application/federation.py: 9 added lines, 19 removed lines.
Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ from uuid import UUID, uuid4

		from pydantic import ValidationError

		from federation_manager.application.partner_zones import replace_offer
		from federation_manager.application.zone_offer import offered_zones
		from federation_manager.contracts.ewbi import (
		FederationDetails,
		@@ -29,7 +30,6 @@ from federation_manager.domain.ewbi import CREATE_FEDERATION_PATH
		from federation_manager.domain.models import (
		INBOUND,
		OUTBOUND,
		FederatedZone,
		FederationContext,
		LocalZone,
		PartnerOP,
		@@ -42,7 +42,6 @@ from federation_manager.domain.ports import (
		PartnerRepositoryPort,
		SrmQueryPort,
		)
		from federation_manager.domain.zones import federated_zone_id

		logger = get_logger(__name__)

		@@ -182,24 +181,12 @@ class FederationEstablishmentService:
		) -> None:
		if self._catalogue is None:
		return
		zones = []
		for offered in accepted.offered_availability_zones or []:
		external = offered.get("zoneId")
		if not external:
		continue
		zones.append(
		FederatedZone(
		zone_id=federated_zone_id(partner.id, str(external)),
		partner_op_id=partner.id,
		external_zone_id=str(external),
		geography_details=offered.get("geographyDetails"),
		geolocation=offered.get("geolocation"),
		# what the partner says its platform can do, until partner_ops.supported_apis
		# exists (federation-manager#24)
		supported_apis=tuple(accepted.platform_caps),
		)
		await replace_offer(
		self._catalogue,
		partner.id,
		accepted.offered_availability_zones or [],
		supported_apis=accepted.platform_caps,
		)
		await self._catalogue.replace_partner_zones(partner.id, zones)

		async def establish_missing(self) -> list[FederationContext]:
		established: list[FederationContext] = []
		@@ -280,6 +267,9 @@ class InboundFederationService:
		async def terminate(self, partner: PartnerOP, federation_context_id: str) -> None:
		context = await self._live_inbound(partner, federation_context_id)
		await self._contexts.set_status(context.id, TERMINATED)
		# subscriptions live under the federation (OPG.04 /{federationContextId}/zones)
		for subscription in await self._catalogue.list_subscriptions(partner.id):
		await self._catalogue.unsubscribe(partner.id, subscription.zone_id)

		async def _offered_zones(self, partner: PartnerOP) -> list[ZoneDetails]:
		agreement = await self._agreements.find_active_for_partner(partner.id)

src/federation_manager/application/partner_federations.py

+13 −1

Changes for src/federation_manager/application/partner_federations.py: 13 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@ from federation_manager.application.federation import (
		FederationEstablishmentService,
		LocalOperator,
		)
		from federation_manager.application.partner_zones import forget_offer, replace_offer
		from federation_manager.contracts.ewbi import (
		FederationDetails,
		FederationRequestData,
		@@ -72,7 +73,16 @@ class PartnerFederationService:
		# OPG.04 §2.2.1.10: a partner that no longer knows the context has terminated it.
		if response.status_code == 404:
		await self._contexts.set_status(context.id, TERMINATED)
		return _accepted(partner, response, FederationDetails)
		body = _accepted(partner, response, FederationDetails)
		if self._catalogue is not None:
		# the partner states its whole offer here, so what we hold catches up with it
		offered = FederationDetails.model_validate(body).offered_availability_zones or []
		await replace_offer(
		self._catalogue,
		partner.id,
		[zone.model_dump(by_alias=True, exclude_none=True) for zone in offered],
		)
		return body

		async def delete(self, partner_op_id: UUID, federation_context_id: str) -> None:
		partner, context = await self._outbound(partner_op_id, federation_context_id)
		@@ -81,6 +91,8 @@ class PartnerFederationService:
		)
		if response.is_success() or response.status_code in _ALREADY_GONE:
		await self._contexts.set_status(context.id, TERMINATED)
		if self._catalogue is not None:
		await forget_offer(self._catalogue, partner.id)
		_accepted(partner, response)

		async def zone_data(

src/federation_manager/application/partner_zones.py

0 → 100644

+95 −0

Changes for src/federation_manager/application/partner_zones.py: 95 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""The zones a partner offers us, kept in step with what the partner tells us.

		The partner states its whole offer in the CreateFederation answer and in GetFederationDetails,
		and changes it in between with ZONES ADD/REMOVE on our partner-status callback (REQ-FED-04).
		"""

		from collections.abc import Iterable, Mapping
		from typing import Any
		from uuid import UUID

		from federation_manager.domain.models import FederatedZone
		from federation_manager.domain.onboarding import ACCEPTED_ZONE
		from federation_manager.domain.ports import CatalogueRepositoryPort
		from federation_manager.domain.zones import federated_zone_id


		def offered_zone(
		partner_id: UUID, offered: Mapping[str, Any], supported_apis: Iterable[str]
		) -> FederatedZone \| None:
		external = offered.get("zoneId")
		if not external:
		return None
		return FederatedZone(
		zone_id=federated_zone_id(partner_id, str(external)),
		partner_op_id=partner_id,
		external_zone_id=str(external),
		geography_details=offered.get("geographyDetails"),
		geolocation=offered.get("geolocation"),
		# what the partner says its platform can do, until partner_ops.supported_apis
		# exists (federation-manager#24)
		supported_apis=tuple(supported_apis),
		)


		async def replace_offer(
		catalogue: CatalogueRepositoryPort,
		partner_id: UUID,
		offered: Iterable[Mapping[str, Any]],
		supported_apis: Iterable[str] \| None = None,
		) -> None:
		"""The partner stated its whole offer; keep exactly that."""
		if supported_apis is None:
		supported_apis = await _known_apis(catalogue, partner_id)
		apis = tuple(supported_apis)
		zones = [z for o in offered if (z := offered_zone(partner_id, o, apis)) is not None]
		await catalogue.replace_partner_zones(partner_id, zones)


		async def add_to_offer(
		catalogue: CatalogueRepositoryPort, partner_id: UUID, added: Iterable[Mapping[str, Any]]
		) -> None:
		current = await _partner_zones(catalogue, partner_id)
		apis = await _known_apis(catalogue, partner_id)
		by_id = {zone.zone_id: zone for zone in current}
		for offered in added:
		zone = offered_zone(partner_id, offered, apis)
		if zone is not None:
		by_id[zone.zone_id] = zone
		await catalogue.replace_partner_zones(partner_id, list(by_id.values()))


		async def remove_from_offer(
		catalogue: CatalogueRepositoryPort, partner_id: UUID, removed: Iterable[str]
		) -> None:
		gone = set(removed)
		current = await _partner_zones(catalogue, partner_id)
		kept = [zone for zone in current if zone.external_zone_id not in gone]
		if len(kept) != len(current):
		await catalogue.replace_partner_zones(partner_id, kept)


		async def forget_offer(catalogue: CatalogueRepositoryPort, partner_id: UUID) -> None:
		"""The federation is gone, and with it the zones and our subscriptions to them."""
		accepted = await catalogue.list_resources(partner_id, ACCEPTED_ZONE)
		for zone in accepted:
		external = str(zone.get("external_zone_id") or "")
		if external:
		await catalogue.delete_resource(
		partner_id, ACCEPTED_ZONE, str(federated_zone_id(partner_id, external))
		)
		await catalogue.replace_partner_zones(partner_id, [])


		async def _partner_zones(
		catalogue: CatalogueRepositoryPort, partner_id: UUID
		) -> list[FederatedZone]:
		return [zone for zone in await catalogue.list_zones() if zone.partner_op_id == partner_id]


		async def _known_apis(catalogue: CatalogueRepositoryPort, partner_id: UUID) -> tuple[str, ...]:
		# GetFederationDetails and the zone callback don't repeat the platform caps
		# CreateFederation gave us, so the zones we already hold keep them
		for zone in await _partner_zones(catalogue, partner_id):
		return zone.supported_apis
		return ()