Commit 76b109b6 authored by Sergio Gimenez's avatar Sergio Gimenez
Browse files

fix(federation): report and remove partner-hosted instances correctly 

Two bugs on the federated path, both seen on the three-operator test env.

A failed partner deploy left the OEG instance stuck in "instantiating" for 30
minutes, blocking the same app in that zone with a 409. When FM reported the
failure, SRM marked the hand-off target failed but dropped FM's error, so the
completion's instance entry had status failed and no error. The SRM contract
(Sec. C.2) requires one, OEG refused the event as invalid, and only its
reconciler freed the zone. The failed target now keeps FM's error (or a
generic problem when none came) and the instance entry carries it; failed
local instances on that path get the generic problem too. The two replay paths
attached an error only when the whole operation failed, so a failed instance
in a partly completed order also went without one; they now follow the
instance's own status.

Deleting an instance a partner hosts failed before reaching FM: the federated
branch of TerminateServiceUseCase.accept wrote target_service_instance_id =
the instance id, a foreign key to a service_instance row SRM never creates
for partner-hosted instances (ADR-0053, 2026-10-08 amendment). Postgres
refused the insert, RemoveApp was never sent and the release stayed on the
partner. The order now leaves it empty, as a federated deploy order already
does; the hand-off target names the instance. The unit test asserted the old
value because its mock has no foreign keys; a new integration test runs
accept() against Postgres.

The terminate hand-off now records the zone from FM's internal instance
read, so its completion satisfies the zone_id contract and OEG accepts it.
An instance unknown to both SRM and FM fails before starting.

Co-Authored-By: default avatarClaude Opus 5.5 <noreply@anthropic.com>
parent 69479613

src/srm/application/use_cases/deploy_service.py

+6 −1
Changes for src/srm/application/use_cases/deploy_service.py: 6 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -307,7 +307,12 @@ class DeployServiceUseCase: 
                    "service_instance_id": str(target["app_instance_id"]),

                    "zone_id": str(target["zone_id"]),

                    "status": "completed" if federated_status == "completed" else "failed",

                    **({"error": problem} if status == OperationCompletedStatus.FAILED else {}),

                    # a failed instance carries an error even when the order completed partly

                    **(

                        {"error": target.get("error") or problem}

                        if federated_status != "completed"

                        else {}

                    ),

                }

            )

src/srm/application/use_cases/federated_deploy_outcome.py

+23 −3
Changes for src/srm/application/use_cases/federated_deploy_outcome.py: 23 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -20,6 +20,12 @@ logger: structlog.BoundLogger = structlog.get_logger(__name__) 
_TERMINAL = {ServiceOrderState.COMPLETED, ServiceOrderState.FAILED, ServiceOrderState.CANCELLED}



_HANDOFF_TARGETS_KEY = "targets"

# SRM interface contract C.2: a failed instance carries an error; this when nobody said why

_UNEXPLAINED_FAILURE: dict[str, Any] = {

    "type": "about:blank",

    "title": "The instance failed.",

    "status": 500,

}





def build_federated_handoff(federated_targets: list[tuple[UUID, UUID]]) -> dict[str, Any]:
@@ -37,11 +43,15 @@ def build_federated_handoff(federated_targets: list[tuple[UUID, UUID]]) -> dict[ 
    }





def build_terminate_federated_handoff(app_instance_id: UUID) -> dict[str, Any]:

def build_terminate_federated_handoff(

    app_instance_id: UUID, zone_id: UUID | None = None

) -> dict[str, Any]:

    return {

        _HANDOFF_TARGETS_KEY: [

            {

                "app_instance_id": str(app_instance_id),

                # contract C.2: the completion names the instance's zone

                **({"zone_id": str(zone_id)} if zone_id is not None else {}),

                "status": "pending",

            }

        ]
@@ -59,6 +69,8 @@ def _handoff_instance_entry(target: dict[str, Any]) -> dict[str, Any]: 
    }

    if target.get("zone_id") is not None:

        entry["zone_id"] = str(target["zone_id"])

    if entry["status"] == "failed":

        entry["error"] = target.get("error") or _UNEXPLAINED_FAILURE

    return entry
@@ -94,6 +106,9 @@ class FederatedDeployOutcomeUseCase: 
            ):

                continue

            target["status"] = target_status

            if target_status == "failed":

                # kept with the target: the completion event needs it per instance

                target["error"] = outcome.error or _UNEXPLAINED_FAILURE

            matched = True

            if outcome.service_instance_id is not None:

                break
@@ -116,9 +131,14 @@ class FederatedDeployOutcomeUseCase: 
            {

                "service_instance_id": str(instance.id),

                "zone_id": str(instance.zone_id),

                "status": "completed"

                "status": "completed",

            }

            if instance.state == ServiceInstanceState.ACTIVE

                else "failed",

            else {

                "service_instance_id": str(instance.id),

                "zone_id": str(instance.zone_id),

                "status": "failed",

                "error": _UNEXPLAINED_FAILURE,

            }

            for instance in local_instances

        ]

src/srm/application/use_cases/terminate_service.py

+28 −3
Changes for src/srm/application/use_cases/terminate_service.py: 28 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -9,6 +9,7 @@ import structlog 
from anyio import to_thread

from fastapi.encoders import jsonable_encoder



from srm.adapters.errors import DownstreamServiceException, NotFoundException

from srm.adapters.federation_manager.fm_client import FederationManager

from srm.application.use_cases.federated_deploy_outcome import (

    build_terminate_federated_handoff,
@@ -139,17 +140,23 @@ class TerminateServiceUseCase: 


        service_instance = await self._service_instances.get_by_id(command.service_instance_id)

        if service_instance is None:

            zone_id = await self._partner_hosted_zone(command.service_instance_id)

            if zone_id is None:

                await self._publish_failed_before_start(command, service_order_id=None)

                return TerminateServiceResult()

            service_order = await self._service_orders.create(

                ServiceOrder(

                    operation_id=command.operation_id,

                    correlation_id=command.correlation_id,

                    order_type=ServiceOrderType.TERMINATE_SERVICE,

                    service_specification_id=command.service_specification_id,

                    target_service_instance_id=command.service_instance_id,

                    # SRM keeps no row for a partner-hosted instance, so this can't point at

                    # one; the instance is the federated hand-off's target, as on deploy

                    target_service_instance_id=None,

                    state=ServiceOrderState.ACCEPTED,

                    payload_snapshot=command.payload_snapshot,

                    federated_handoff=build_terminate_federated_handoff(

                        command.service_instance_id

                        command.service_instance_id, zone_id

                    ),

                    app_provider_id=command.app_provider_id,

                    federation_partner_ref=command.federation_partner_ref,
@@ -206,6 +213,19 @@ class TerminateServiceUseCase: 
            target_zone_id=service_instance.zone_id,

        )



    async def _partner_hosted_zone(self, service_instance_id: UUID) -> UUID | None:

        """The zone of an instance a partner hosts; None when FM doesn't know it either."""

        try:

            instance = await self._fm_client.get_service_instance(service_instance_id)

        except (NotFoundException, DownstreamServiceException) as error:

            logger.info(

                "service_terminate_instance_unknown",

                service_instance_id=str(service_instance_id),

                reason=type(error).__name__,

            )

            return None

        return instance.zone_id



    async def hand_off_to_partner(

        self,

        command: TerminateServiceCommand,
@@ -568,7 +588,12 @@ class TerminateServiceUseCase: 
                entry: dict[str, Any] = {

                    "service_instance_id": str(target["app_instance_id"]),

                    "status": "completed" if federated_status == "completed" else "failed",

                    **({"error": problem} if status == OperationCompletedStatus.FAILED else {}),

                    # a failed instance carries an error even when the order completed partly

                    **(

                        {"error": target.get("error") or problem}

                        if federated_status != "completed"

                        else {}

                    ),

                }

                if target.get("zone_id") is not None:

                    entry["zone_id"] = str(target["zone_id"])

tests/integration/test_federated_terminate.py

0 → 100644
+50 −0
Changes for tests/integration/test_federated_terminate.py: 50 added lines, 0 removed lines.
Original line number Diff line number Diff line 
"""Terminating a partner-hosted instance against the real schema, foreign keys included."""



from __future__ import annotations



from types import SimpleNamespace

from unittest.mock import AsyncMock, MagicMock

from uuid import uuid4



from sqlalchemy.ext.asyncio import AsyncSession



from srm.adapters.database.repos.runtime_inventory import (

    SqlServiceInstanceRepository,

    SqlServiceOrderRepository,

)

from srm.application.use_cases.terminate_service import TerminateServiceUseCase

from tests.unit.test_terminate_service_use_case import _command





async def test_terminating_a_partner_hosted_instance_stores_its_order(

    db_session: AsyncSession,

) -> None:

    # SRM keeps no service_instance row for an instance a partner hosts; FM knows its zone

    instance_id = uuid4()

    zone_id = uuid4()

    fm_client = AsyncMock()

    fm_client.get_service_instance.return_value = SimpleNamespace(zone_id=zone_id)

    orders = SqlServiceOrderRepository(db_session)

    use_case = TerminateServiceUseCase(

        service_orders=orders,

        service_instances=SqlServiceInstanceRepository(db_session),

        capability_instances=AsyncMock(),

        capabilities=AsyncMock(),

        domains=AsyncMock(),

        zones=AsyncMock(),

        publisher=AsyncMock(),

        fm_client=fm_client,

        control_paths=MagicMock(),

    )

    operation_id = uuid4()



    result = await use_case.accept(_command(operation_id, service_instance_id=instance_id))



    assert result.is_federated is True

    stored = await orders.get_by_operation_id(operation_id)

    assert stored is not None

    assert stored.federated_handoff == {

        "targets": [

            {"app_instance_id": str(instance_id), "zone_id": str(zone_id), "status": "pending"}

        ]

    }

tests/unit/test_federated_deploy_outcome.py

+16 −0
Changes for tests/unit/test_federated_deploy_outcome.py: 16 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -104,6 +104,22 @@ async def test_a_failed_partner_deploy_fails_the_order() -> None: 
    assert event is not None and event.status == OperationCompletedStatus.FAILED

    assert event.error == {"type": "x", "title": "partner failed"}

    assert orders.update_state.await_args.args[1] == ServiceOrderState.FAILED

    # contract C.2: OEG drops a completion whose failed instance has no error

    [entry] = event.instances

    assert entry["status"] == "failed"

    assert entry["error"] == {"type": "x", "title": "partner failed"}





async def test_a_failed_target_without_a_reason_still_carries_an_error() -> None:

    instance = uuid4()

    use_case, _, _ = await _setup(_order(federated_targets=[_pending(instance)]))

    outcome = _outcome("failed", instance).model_copy(update={"error": None})



    event = await use_case.execute(outcome)



    assert event is not None

    [entry] = event.instances

    assert entry["error"]["status"] == 500





async def test_an_outcome_for_a_finished_or_unknown_order_is_ignored() -> None: