Commit 548b8076 authored by Paris Stentoumis's avatar Paris Stentoumis
Browse files

feat: log databus operation and correlation ids through publish and subscribe 

Emit structured receive, publish, accept, and complete events so deploy and network-capability flows can be traced by operation_id.

Co-authored-by: default avatarCursor <cursoragent@cursor.com>
parent 5883b516

src/srm/adapters/databus/nats_publisher.py

+16 −0
Changes for src/srm/adapters/databus/nats_publisher.py: 16 added lines, 0 removed lines.
Original line number Diff line number Diff line 
import json

from typing import Any



import structlog



from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

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



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





def _optional_str(value: Any) -> str | None:

    if value is None:

        return None

    return str(value)





class NatsPublisher(DataBusPublisher):

    def __init__(self, connection_manager: NatsConnectionManager) -> None:
@@ -18,3 +28,9 @@ class NatsPublisher(DataBusPublisher): 


        body = json.dumps(payload).encode("utf-8")

        await self._connection_manager.client.publish(subject, body, headers=headers)

        logger.info(

            "databus_message_published",

            subject=subject,

            operation_id=_optional_str(payload.get("operation_id")),

            correlation_id=_optional_str(payload.get("correlation_id")),

        )

src/srm/api/databus/nats_subscriber.py

+13 −2
Changes for src/srm/api/databus/nats_subscriber.py: 13 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -28,7 +28,7 @@ EventHandler = Callable[[AnswerableEventV1], 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",

        "databus_command_unhandled",

        operation_id=str(envelope.operation_id),

        correlation_id=envelope.correlation_id,

    )
@@ -137,10 +137,21 @@ class NatsSubscriber: 
            logger.exception("databus_stage1_unanswerable", subject=msg.subject)

            return



        logger.info(

            "databus_command_received",

            subject=msg.subject,

            operation_id=str(envelope.operation_id),

            correlation_id=envelope.correlation_id,

        )

        try:

            await self._callback(envelope)

        except Exception:

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

            logger.exception(

                "databus_handler_failed",

                subject=msg.subject,

                operation_id=str(envelope.operation_id),

                correlation_id=envelope.correlation_id,

            )





class NatsEventSubscriber:

src/srm/application/use_cases/deploy_service.py

+5 −3
Changes for src/srm/application/use_cases/deploy_service.py: 5 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -193,9 +193,10 @@ class DeployServiceUseCase: 
                )

            )



        structlog.get_logger(__name__).info(

        logger.info(

            "service_deploy_accepted",

            operation_id=str(command.operation_id),

            correlation_id=command.correlation_id,

            service_specification_id=str(command.service_specification_id),

            target_count=len(command.targets),

        )
@@ -356,9 +357,10 @@ class DeployServiceUseCase: 
            completed_at=completed_at,

        )



        structlog.get_logger(__name__).info(

            "service_deploy_use_case_invoked",

        logger.info(

            "service_deploy_completed",

            operation_id=str(command.operation_id),

            correlation_id=command.correlation_id,

            service_specification_id=str(command.service_specification_id),

            target_count=len(command.targets),

        )

src/srm/application/use_cases/network_capability_activate.py

+20 −1
Changes for src/srm/application/use_cases/network_capability_activate.py: 20 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -6,6 +6,8 @@ from enum import StrEnum 
from typing import Any, Literal

from uuid import UUID



import structlog



from srm.application.services.capability_placement import (

    CapabilityPlacement,

    PlacementPin,
@@ -59,6 +61,8 @@ from srm.domain.ports.databus.events import ( 
from srm.domain.ports.databus.publisher import DataBusPublisher

from srm.domain.ports.transformation_functions.registry import ControlPathRegistry



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





@dataclass(frozen=True, slots=True)

class NetworkCapabilityPayloadCommand:
@@ -261,6 +265,13 @@ class NetworkCapabilityActivateUseCase: 
            )

        )



        logger.info(

            "network_capability_activate_accepted",

            operation_id=str(command.operation_id),

            correlation_id=command.correlation_id,

            service_instance_id=str(command.service_instance_id),

            capability_type=command.network_capability.capability_type,

        )

        return NetworkCapabilityActivateResult(

            accepted_order_id=service_order.id,

            requirement=requirement,
@@ -321,7 +332,7 @@ class NetworkCapabilityActivateUseCase: 
            completed_at=completed_at,

        )



        return SrmOperationCompletedV1(

        completed_event = SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            status=OperationCompletedStatus.COMPLETED,
@@ -336,6 +347,14 @@ class NetworkCapabilityActivateUseCase: 
            correlation_id=command.correlation_id,

            completed_at=completed_at,

        )

        logger.info(

            "network_capability_activate_completed",

            operation_id=str(command.operation_id),

            correlation_id=command.correlation_id,

            service_instance_id=str(command.service_instance_id),

            capability_type=command.network_capability.capability_type,

        )

        return completed_event



    async def fail_after_accept(

        self,

tests/api/databus/test_nats_subscriber.py

+46 −2
Changes for tests/api/databus/test_nats_subscriber.py: 46 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -168,6 +168,46 @@ async def test_handle_message_defaults_headers_to_empty_dict_when_none( 
    assert envelope.correlation_id == "corr-1"





async def test_handle_message_logs_command_received_when_envelope_is_valid(

    connection_manager: MagicMock,

) -> None:

    subscriber, callback = make_subscriber(connection_manager)

    msg = FakeMsg(

        subject="command.srm.service.deploy",

        data=(

            b'{"schema_version": "1.0",'

            b' "operation_id": "11111111-2222-3333-4444-555555555555",'

            b' "correlation_id": "corr-1",'

            b' "requested_at": "2026-06-19T12:00:00Z",'

            b' "app_provider_id": "VideoAppsCo",'

            b' "source": "nbi_camara"}'

        ),

    )



    with structlog.testing.capture_logs() as logs:

        await subscriber._handle_message(msg)  # type: ignore[arg-type]



    callback.assert_awaited_once()

    assert [

        (

            entry["event"],

            entry["subject"],

            entry["operation_id"],

            entry["correlation_id"],

            entry["log_level"],

        )

        for entry in logs

    ] == [

        (

            "databus_command_received",

            "command.srm.service.deploy",

            "11111111-2222-3333-4444-555555555555",

            "corr-1",

            "info",

        )

    ]





async def test_handle_message_passes_malformed_payload_through_unparsed(

    connection_manager: MagicMock,

) -> None:
@@ -215,9 +255,13 @@ async def test_handle_message_logs_dropped_command_when_callback_fails( 
        await subscriber._handle_message(msg)  # type: ignore[arg-type]



    callback.assert_awaited_once()

    assert [(entry["event"], entry["subject"], entry["log_level"]) for entry in logs] == [

        ("databus_handler_failed", "command.srm.service.deploy", "error")

    assert [(entry["event"], entry.get("subject"), entry["log_level"]) for entry in logs] == [

        ("databus_command_received", "command.srm.service.deploy", "info"),

        ("databus_handler_failed", "command.srm.service.deploy", "error"),

    ]

    failed = logs[1]

    assert failed["operation_id"] == "11111111-2222-3333-4444-555555555555"

    assert failed["correlation_id"] == "corr-1"





EXPECTED_COMMAND_SUBJECTS = [