fix(eam): show refused attempts and hide confirmed deletions

Sergio Gimenezrequested to merge
fix/instance-list-lifecycle into develop

Instance list, refused partner deployment:

sequenceDiagram
  participant AP as App Provider
  participant OEG
  participant SRM
  participant FM
  AP->>OEG: POST /appinstances (Paris zone)
  OEG-->>AP: 202
  OEG->>SRM: command.srm.service.deploy
  SRM->>FM: command.fm.service.deploy
  FM-->>SRM: partner refused
  SRM-->>OEG: event.srm.operation.completed (failed + error)
  OEG->>OEG: save failed attempt, release zone
  AP->>OEG: GET /appinstances
  OEG->>SRM: GET /internal/service-instances
  SRM-->>OEG: nothing for the refused attempt
  Note over OEG: before: attempt missing from the list<br/>now: listed as failed for its App Provider
  OEG-->>AP: SRM rows + failed attempt, minus confirmed deletions

App list (second commit):

flowchart LR
  G["GET /apps"] -- "GET /internal/catalog/service-specifications" --> S["SRM specs<br/>(retired specs still listed)"] --> Q{"OEG registration<br/>DELETED?"}
  Q -- "yes" --> X["left out<br/>(same as GET /apps/{appId}, 404)"]
  Q -- "no, or never registered here" --> L["listed"]
flowchart LR
  I["fix/instance-list-lifecycle<br/>OEG !43 (this MR)"] -- "targets now" --> A["feat/agreements-relay<br/>OEG !42"] -- "merges first" --> D["develop"]
  I -. "retarget after !42 merges" .-> D

A partner deployment refused before fulfillment has no runtime instance for SRM to return. Even after OEG consumes the failure, GET /appinstances omitted the accepted attempt. A confirmed delete could also reappear as unknown from an old runtime transaction.

Supplement SRM results with OEG's saved failed attempts, scoped by the original operation's App Provider. Apply app, instance and region filters, retain the requested name, and return SRM's runtime result once when both sources have an instance. Suppress a runtime row when OEG has confirmed that instance as terminated. Historical rows that missed earlier terminal events are not rewritten.

A second commit fixes GET /apps listing an app after DELETE /apps/{appId} answered 204. SRM only retires the specification of an app that ever had instances, and its list still returns it. The list now skips specifications whose registration here is marked deleted, which matches GET /apps/{appId} (404 for the same app). Specifications OEG never registered are listed as before.

The accompanying architecture amendment explicitly changes the previous SRM-only read contract. Please review that decision with this MR. SRM remains the runtime source, and failed attempts receive no invented endpoints or cluster references.

Validation: 1234 unit tests passed, 1 skipped (2 new for the app list); 40 integration tests passed, including provider isolation against Postgres; mypy, ruff and import-linter passed. The portal UI showed two refused attempts as failed, accepted the retry with 202, and removed a successfully deleted Paris instance.

Target feat/agreements-relay to exclude its parent commit from the initial diff; retarget to develop after that merges. The stale-409 fix also requires SRM to emit valid failed completion entries with errors.

Edited by Sergio Gimenez

Merge request reports

src/open_exposure_gateway/adapters/database/repos/app_instances.py

+13 −1
Changes for src/open_exposure_gateway/adapters/database/repos/app_instances.py: 13 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -4,7 +4,7 @@ from sqlalchemy import select 
from sqlalchemy.ext.asyncio import AsyncSession



from open_exposure_gateway.adapters.database.mappers import AppInstanceMapper

from open_exposure_gateway.adapters.database.sql import AppInstanceRow

from open_exposure_gateway.adapters.database.sql import AppInstanceRow, OperationRow

from open_exposure_gateway.domain.models import AppInstance, AppInstanceState

from open_exposure_gateway.ports.database.instances import AppInstanceRepository
@@ -20,6 +20,18 @@ class SqlAppInstanceRepository(AppInstanceRepository): 
        row = await self._session.scalar(stmt)

        return AppInstanceMapper.to_domain(row) if row is not None else None



    async def list_failed_by_provider(self, app_provider_id: str) -> list[AppInstance]:

        stmt = (

            select(AppInstanceRow)

            .join(OperationRow, AppInstanceRow.operation_id == OperationRow.operation_id)

            .where(

                OperationRow.app_provider_id == app_provider_id,

                AppInstanceRow.state == AppInstanceState.FAILED,

            )

        )

        rows = await self._session.scalars(stmt)

        return [AppInstanceMapper.to_domain(row) for row in rows]



    async def get_by_operation_id(self, operation_id: UUID) -> AppInstance | None:

        stmt = select(AppInstanceRow).where(AppInstanceRow.operation_id == operation_id)

        row = await self._session.scalar(stmt)

src/open_exposure_gateway/application/services/edge_application_management_service.py

+61 −0
Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 61 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -369,6 +369,7 @@ class EdgeApplicationManagementService: 
        catalogs = await self.srm_client.get_apps(

            app_provider_id=app_provider_id, x_correlator=x_correlator

        )

        catalogs = [catalog for catalog in catalogs if not await self._deleted_here(catalog)]

        try:

            return [build_app_manifest(catalog) for catalog in catalogs]

        except (ValueError, TypeError, IndexError) as exc:
@@ -967,6 +968,10 @@ class EdgeApplicationManagementService: 
        result: list[AppInstanceInfo] = []

        registration_cache: dict[UUID, Optional[AppRegistration]] = {}

        for instance in instances:

            if self._app_instance_repo is not None:

                saved = await self._app_instance_repo.get_by_id(UUID(instance.service_instance_id))

                if saved is not None and saved.state == AppInstanceState.TERMINATED:

                    continue

            instance_registration: Optional[AppRegistration]

            if filter_registration is not None:

                instance_registration = filter_registration
@@ -1010,6 +1015,47 @@ class EdgeApplicationManagementService: 
                    _camara_app_provider(instance_registration),

                )

            )

        # A refusal before fulfillment has no SRM instance. OEG owns the accepted

        # attempt and its terminal event state, so keep that failure visible too.

        if self._app_instance_repo is not None and self._operation_repo is not None:

            failed = await self._app_instance_repo.list_failed_by_provider(app_provider_id)

            seen = {row.appInstanceId for row in result}

            region_zones: set[UUID] | None = None

            if failed and region is not None:

                zones = await self.srm_client.get_zones(

                    region=region, status=None, x_correlator=x_correlator

                )

                region_zones = {UUID(zone.id) for zone in zones}

            for saved in failed:

                if saved.app_instance_id in seen:

                    continue

                if app_instance_id is not None and saved.app_instance_id != app_instance_id:

                    continue

                if app_id is not None and saved.app_registration_id != resolved_app_id:

                    continue

                if region_zones is not None and saved.edge_cloud_zone_id not in region_zones:

                    continue

                if self._app_registration_repo is None:

                    raise RuntimeError("AppRegistrationRepository is not available")

                registration = await self._app_registration_repo.get_by_id(

                    saved.app_registration_id

                )

                operation = await self._operation_repo.get_by_id(saved.operation_id)

                if registration is None or operation is None:

                    continue

                name = operation.metadata.get("name") or operation.metadata.get(

                    "app_deployment_name"

                )

                result.append(

                    AppInstanceInfo(

                        appInstanceId=saved.app_instance_id,

                        name=str(name or saved.app_instance_id),

                        appId=registration.app_id,

                        appProvider=_camara_app_provider(registration, app_provider_id),

                        status=AppInstanceStatus.FAILED,

                        edgeCloudZoneId=saved.edge_cloud_zone_id,

                    )

                )

        return result



    async def delete_app_instance(
@@ -1575,6 +1621,21 @@ class EdgeApplicationManagementService: 
            return None

        return app_registration, catalog



    async def _deleted_here(self, catalog: SRMCatalogPayload) -> bool:

        """

        SRM only retires the specification of an app that ever had instances, and still

        lists it. The app was deleted once its registration here was, which is also why

        GET /apps/{appId} answers 404 for it.

        """

        if self._app_registration_repo is None:

            return False

        try:

            registration_id = UUID(catalog.service_specification.id)

        except ValueError:

            return False

        registration = await self._app_registration_repo.get_by_id(registration_id)

        return registration is not None and registration.status == AppRegistrationStatus.DELETED



    async def _resolve_app_id(self, app_registration_id: UUID) -> Optional[UUID]:

        if self._app_registration_repo is None:

            return None

src/open_exposure_gateway/ports/database/instances.py

+4 −0
Changes for src/open_exposure_gateway/ports/database/instances.py: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -11,6 +11,10 @@ class AppInstanceRepository(ABC): 
    async def get_by_id(self, app_instance_id: UUID) -> AppInstance | None:

        pass



    @abstractmethod

    async def list_failed_by_provider(self, app_provider_id: str) -> list[AppInstance]:

        pass



    @abstractmethod

    async def get_by_operation_id(self, operation_id: UUID) -> AppInstance | None:

        pass

tests/integration/test_postgres.py

+27 −0
Changes for tests/integration/test_postgres.py: 27 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -731,3 +731,30 @@ async def test_qod_session_repo_updates_updated_at_on_second_save(db_engine: Asy 
    assert updated.updated_at is not None

    assert updated.updated_at > updated.created_at

    assert updated.state == QodSessionState.AVAILABLE





async def test_failed_instance_listing_scopes_by_operation_provider(

    db_session: AsyncSession,

) -> None:

    registration = await SqlAppRegistrationRepository(db_session).save(_app_registration())

    operations = SqlOperationRepository(db_session)

    instances = SqlAppInstanceRepository(db_session)

    expected = []

    for provider, state in [

        ("provider-a", AppInstanceState.FAILED),

        ("provider-b", AppInstanceState.FAILED),

        ("provider-a", AppInstanceState.TERMINATED),

        ("provider-a", AppInstanceState.READY),

    ]:

        operation = _operation(app_registration_id=registration.app_registration_id)

        operation.app_provider_id = provider

        await operations.save(operation)

        instance = _app_instance(operation.operation_id, registration.app_registration_id)

        instance.state = state

        await instances.save(instance)

        if provider == "provider-a" and state == AppInstanceState.FAILED:

            expected.append(instance.app_instance_id)

    assert [

        i.app_instance_id for i in await instances.list_failed_by_provider("provider-a")

    ] == expected

    assert await instances.list_failed_by_provider("nobody") == []

tests/unit/fakes.py

+13 −1
Changes for tests/unit/fakes.py: 13 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -608,8 +608,20 @@ class FakeOperationRepository(OperationRepository): 




class FakeAppInstanceRepository(AppInstanceRepository):

    def __init__(self) -> None:

    def __init__(self, operations: FakeOperationRepository | None = None) -> None:

        self.rows: dict[UUID, AppInstance] = {}

        self.operations = operations



    async def list_failed_by_provider(self, app_provider_id: str) -> list[AppInstance]:

        if self.operations is None:

            return []

        return [

            row.model_copy(deep=True)

            for row in self.rows.values()

            if row.state == AppInstanceState.FAILED

            and (operation := self.operations.rows.get(row.operation_id)) is not None

            and operation.app_provider_id == app_provider_id

        ]



    async def get_by_id(self, app_instance_id: UUID) -> AppInstance | None:

        found = self.rows.get(app_instance_id)