SRM Refactor: Service Specification/Application deployment

Paris Stentoumisrequested to merge
refactor/app-deployment into refactor/tf-sdk-ports

Summary

Implements the first SRM service deployment path for command.srm.service.deploy, focused on deploy_workload.

Changes

  • Adds deploy command parsing and two-stage validation.
  • Adds deploy command coordinator for accept, completion, and post-accept failure handling.
  • Creates service_order, service_instance, and deploy capability_instance rows.
  • Publishes event.srm.operation.status and event.srm.operation.completed.
  • Handles duplicate operation_id as idempotent replay:
    • non-terminal duplicates re-emit accepted
    • terminal duplicates re-emit operation.completed
  • Adds naive placement support across zones/domains/capability bindings.
  • Adds repository methods needed for deploy state updates and replay reconstruction.

Scope Notes

  • TF-SDK execution is intentionally left as a TODO because TF-SDK is still WIP.
  • Placement is naive. The first found active domain which supports that specific capability is chosen. Available capacities are ignored for now. As for the Control Path Binding the default is chosen if present else the one with the highest priority (lowest value).
  • Service specification creation has known gaps and is not addressed here.
  • JetStream ACK/work queue semantics are not fully implemented yet; current NATS handling is a stepping stone.
Edited by Paris Stentoumis

Merge request reports

src/srm/adapters/database/repos/catalog.py

+20 −0
Changes for src/srm/adapters/database/repos/catalog.py: 20 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -147,6 +147,16 @@ class SqlServiceDeploymentUnitRepository(ServiceDeploymentUnitRepository): 
        row = await self._session.scalar(stmt)

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



    async def list_by_service_specification_id(

        self,

        service_specification_id: UUID,

    ) -> list[ServiceDeploymentUnit]:

        stmt = select(ServiceDeploymentUnitRow).where(

            ServiceDeploymentUnitRow.service_specification_id == service_specification_id

        )

        rows = (await self._session.scalars(stmt)).all()

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



    async def create(self, deployment_unit: ServiceDeploymentUnit) -> ServiceDeploymentUnit:

        row = ServiceDeploymentUnitMapper.to_row(deployment_unit)

        self._session.add(row)
@@ -173,6 +183,16 @@ class SqlServiceCapabilityRequirementRepository(ServiceCapabilityRequirementRepo 
        row = await self._session.scalar(stmt)

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



    async def list_by_service_specification_id(

        self,

        service_specification_id: UUID,

    ) -> list[ServiceCapabilityRequirement]:

        stmt = select(ServiceCapabilityRequirementRow).where(

            ServiceCapabilityRequirementRow.service_specification_id == service_specification_id

        )

        rows = (await self._session.scalars(stmt)).all()

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



    async def create(

        self,

        capability_requirement: ServiceCapabilityRequirement,

src/srm/adapters/database/repos/runtime_inventory.py

+56 −0
Changes for src/srm/adapters/database/repos/runtime_inventory.py: 56 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from __future__ import annotations



from datetime import datetime

from uuid import UUID



import structlog
@@ -19,6 +20,7 @@ from srm.domain.models.runtime_inventory import ( 
    ServiceInstance,

    ServiceOrder,

)

from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState

from srm.domain.ports.database.runtime_inventory import (

    CapabilityInstanceRepository,

    ServiceInstanceRepository,
@@ -51,6 +53,11 @@ class SqlServiceOrderRepository(ServiceOrderRepository): 
        row = await self._session.scalar(stmt)

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



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

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

        row = await self._session.scalar(stmt)

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



    async def create(self, service_order: ServiceOrder) -> ServiceOrder:

        row = ServiceOrderMapper.to_row(service_order)

        self._session.add(row)
@@ -65,6 +72,26 @@ class SqlServiceOrderRepository(ServiceOrderRepository): 
            raise RuntimeError("Created service order could not be reloaded")

        return saved



    async def update_state(

        self,

        id: UUID,

        state: ServiceOrderState,

        *,

        completed_at: datetime | None = None,

        failure_detail: dict[str, object] | None = None,

    ) -> ServiceOrder:

        row = await self._session.get(ServiceOrderRow, id)

        if row is None:

            raise RuntimeError("Service order could not be reloaded")

        row.state = state

        row.completed_at = completed_at

        row.failure_detail = failure_detail

        await self._session.flush()

        saved = await self.get_by_id(id)

        if saved is None:

            raise RuntimeError("Updated service order could not be reloaded")

        return saved





class SqlServiceInstanceRepository(ServiceInstanceRepository):

    def __init__(self, session: AsyncSession) -> None:
@@ -81,6 +108,20 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): 
        row = await self._session.scalar(stmt)

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



    async def list_by_originating_service_order_id(

        self,

        originating_service_order_id: UUID,

    ) -> list[ServiceInstance]:

        stmt = (

            select(ServiceInstanceRow)

            .options(

                selectinload(ServiceInstanceRow.capability_instances),

            )

            .where(ServiceInstanceRow.originating_service_order_id == originating_service_order_id)

        )

        rows = (await self._session.scalars(stmt)).all()

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



    async def create(self, service_instance: ServiceInstance) -> ServiceInstance:

        row = ServiceInstanceMapper.to_row(service_instance)

        self._session.add(row)
@@ -95,6 +136,21 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): 
            raise RuntimeError("Created service instance could not be reloaded")

        return saved



    async def update_state(

        self,

        id: UUID,

        state: ServiceInstanceState,

    ) -> ServiceInstance:

        row = await self._session.get(ServiceInstanceRow, id)

        if row is None:

            raise RuntimeError("Service instance could not be reloaded")

        row.state = state

        await self._session.flush()

        saved = await self.get_by_id(id)

        if saved is None:

            raise RuntimeError("Updated service instance could not be reloaded")

        return saved





class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):

    def __init__(self, session: AsyncSession) -> None:

src/srm/adapters/database/repos/topology.py

+15 −0
Changes for src/srm/adapters/database/repos/topology.py: 15 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -16,6 +16,7 @@ from srm.adapters.database.mappers import ( 
)

from srm.adapters.database.sql import CapabilityRow, ControlPathBindingRow, DomainRow, ZoneRow

from srm.domain.models.topology import Capability, ControlPathBinding, Domain, Zone

from srm.domain.models.topology.enums import ZoneKind, ZoneState

from srm.domain.ports.database.topology import (

    CapabilityRepository,

    ControlPathBindingRepository,
@@ -57,6 +58,20 @@ class SqlZoneRepository(ZoneRepository): 
        row = await self._session.scalar(stmt)

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



    async def list_active_resource_zones(self) -> list[Zone]:

        stmt = (

            select(ZoneRow)

            .options(

                selectinload(ZoneRow.domains)

                .selectinload(DomainRow.capabilities)

                .selectinload(CapabilityRow.control_path_bindings)

            )

            .where(ZoneRow.kind == ZoneKind.RESOURCE, ZoneRow.state == ZoneState.ACTIVE)

            .order_by(ZoneRow.ref, ZoneRow.id)

        )

        rows = (await self._session.scalars(stmt)).all()

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



    async def create(self, zone: Zone) -> Zone:

        row = ZoneMapper.to_row(zone)

        self._session.add(row)

src/srm/api/databus/dependencies.py

0 → 100644
+53 −0
Changes for src/srm/api/databus/dependencies.py: 53 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker



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

    SqlServiceCapabilityRequirementRepository,

    SqlServiceDeploymentUnitRepository,

    SqlServiceSpecificationRepository,

)

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

    SqlCapabilityInstanceRepository,

    SqlServiceInstanceRepository,

    SqlServiceOrderRepository,

)

from srm.adapters.database.repos.topology import SqlDomainRepository, SqlZoneRepository

from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

from srm.adapters.databus.nats_publisher import NatsPublisher

from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator

from srm.application.use_cases.deploy_service import DeployServiceUseCase

from srm.domain.ports.databus.publisher import DataBusPublisher





def get_deploy_service_use_case(

    session: AsyncSession,

    connection_manager: NatsConnectionManager,

) -> DeployServiceUseCase:

    return build_deploy_service_use_case(session, NatsPublisher(connection_manager))





def build_deploy_service_use_case(

    session: AsyncSession,

    publisher: DataBusPublisher,

) -> DeployServiceUseCase:

    return DeployServiceUseCase(

        service_specifications=SqlServiceSpecificationRepository(session),

        service_deployment_units=SqlServiceDeploymentUnitRepository(session),

        service_capability_requirements=SqlServiceCapabilityRequirementRepository(session),

        service_orders=SqlServiceOrderRepository(session),

        service_instances=SqlServiceInstanceRepository(session),

        capability_instances=SqlCapabilityInstanceRepository(session),

        zones=SqlZoneRepository(session),

        domains=SqlDomainRepository(session),

        publisher=publisher,

    )





def get_deploy_service_command_coordinator(

    session_maker: async_sessionmaker[AsyncSession],

    connection_manager: NatsConnectionManager,

) -> DeployServiceCommandCoordinator:

    return DeployServiceCommandCoordinator(

        session_maker=session_maker,

        use_case_factory=build_deploy_service_use_case,

        publisher=NatsPublisher(connection_manager),

    )

src/srm/api/databus/nats_subscriber.py

+51 −18
Changes for src/srm/api/databus/nats_subscriber.py: 51 added lines, 18 removed lines.
Original line number Diff line number Diff line
@@ -3,34 +3,59 @@ from collections.abc import Awaitable, Callable 
import structlog

from nats.aio.msg import Msg

from nats.aio.subscription import Subscription

from pydantic import ValidationError

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker



from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

from srm.api.databus.schemas import InboundMessage

from srm.api.databus.dependencies import get_deploy_service_command_coordinator

from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage



logger: structlog.BoundLogger = structlog.get_logger(__name__)



COMMAND_SUBJECTS = [

    "command.srm.service.deploy",

    "command.srm.service.scale",

    "command.srm.service.terminate",

    "command.srm.network.capability.activate",

    "command.srm.network.capability.update",

    "command.srm.network.capability.deactivate",

]



MessageHandler = Callable[[AnswerableCommandV1], Awaitable[None]]





async def _noop_callback(envelope: AnswerableCommandV1) -> None:

    # TODO: non-deploy command handlers must validate in two stages (interface-contract.md §A).

    logger.info(

        "Message Received",

        operation_id=str(envelope.operation_id),

        correlation_id=envelope.correlation_id,

    )





COMMAND_HANDLERS: dict[str, MessageHandler] = {

    "command.srm.service.scale": _noop_callback,

    "command.srm.service.terminate": _noop_callback,

    "command.srm.network.capability.activate": _noop_callback,

    "command.srm.network.capability.update": _noop_callback,

    "command.srm.network.capability.deactivate": _noop_callback,

}



async def _noop_router(message: InboundMessage) -> None:

    # TODO: the real router must validate in two stages (interface-contract.md §A, ADR-0032).

    logger.info("Message Received", subject=message.subject, data=message.payload)

    return None

COMMAND_SUBJECTS = ("command.srm.service.deploy", *COMMAND_HANDLERS.keys())





async def subscribe_to_subjects(

    connection_manager: NatsConnectionManager,

    session_maker: async_sessionmaker[AsyncSession],

) -> list["NatsSubscriber"]:

    deploy_coordinator = get_deploy_service_command_coordinator(

        session_maker,

        connection_manager,

    )



    command_handlers = {

        "command.srm.service.deploy": deploy_coordinator.handle,

        **COMMAND_HANDLERS,

    }

    subscribers = [

        NatsSubscriber(connection_manager=connection_manager, subject=subject, router=_noop_router)

        for subject in COMMAND_SUBJECTS

        NatsSubscriber(

            connection_manager=connection_manager,

            subject=subject,

            callback=callback,

        )

        for subject, callback in command_handlers.items()

    ]



    for subscriber in subscribers:
@@ -44,11 +69,11 @@ class NatsSubscriber: 
        self,

        connection_manager: NatsConnectionManager,

        subject: str,

        router: Callable[[InboundMessage], Awaitable[None]],

        callback: MessageHandler,

    ) -> None:

        self._connection_manager = connection_manager

        self._subject = subject

        self._router = router

        self._callback = callback

        self._subscription: Subscription | None = None



    async def start(self) -> None:
@@ -64,6 +89,14 @@ class NatsSubscriber: 
            headers=dict(msg.headers) if msg.headers is not None else {},

        )

        try:

            await self._router(inbound_message)

            envelope = AnswerableCommandV1.model_validate_json(inbound_message.payload)

        except ValidationError:

            # Stage 1 failures are unanswerable: publish no SRM event;

            # route to JetStream/DLQ when wired.

            logger.exception("databus_stage1_unanswerable", subject=msg.subject)

            return



        try:

            await self._callback(envelope)

        except Exception:

            logger.exception("databus_handler_failed", subject=msg.subject)