orchestrata_ultimate.postman.delivery

Reference › orchestrata_ultimate › postman › delivery

The delivery manager of the broker, SPEC-002 story US-001 (REQ-002-US-001-002 to -005) with queue transport from FR-004. DeliveryManager implements IDeliveryManager from interfaces and takes a validated MessageEnvelope and a MessagePayload through to one delivery receipt per recipient. It composes three collaborators, each replaceable by passing one in: the RoutingEngine to resolve recipients, the EnvelopeHandler to validate, and the QueueManager to provide queues. No other module imports it; the Postman facade has its own simpler path.

deliver(envelope, payload, mode) follows these steps. It extracts or generates a message id. _validate_envelope returns a failed DeliveryResult if the envelope handler rejects the envelope or raises. _queue_message records a queued status and keeps the envelope and payload in a pending table. _resolve_recipients asks the routing engine and turns routing errors into a failed result with _record_failed. The result is set to delivering, and _deliver_to_all calls _deliver_to_recipient for each recipient. That method gets, or lazily creates, an in-memory queue named for the recipient (_get_or_create_recipient_queue, tolerating a creation race) and enqueues a QueueEntry with the envelope's priority. The receipt is DELIVERED only after the queue accepts the entry; any exception gives a FAILED receipt carrying the error, so a message that never arrived is not reported as delivered. The aggregate result is delivered only if every recipient succeeded, and successful messages leave the pending table.

get_status returns the stored result for a message id. cancel works only on a message still queued and marks it failed with the reason "Cancelled by sender". retry re-delivers a failed message from the stored envelope and payload and raises DeliveryError if the message is unknown, not failed, or no longer stored. A lock guards the shared tables.

At a glance

Class diagram

Dependency diagram

API

Classes

DeliveryManager

Bases: IDeliveryManager · Decorators: traces(spec='002', story='US-001', reqs=['REQ-002-US-001-002', 'REQ-002-US-001-003', 'REQ-002-US-001-004', 'REQ-002-US-… · Line 58

Manages message delivery lifecycle from sender to recipient(s).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/postman/delivery.py

# mypy: ignore-errors
"""
Delivery Manager - Core delivery orchestration for Postman message broker.

This module implements the IDeliveryManager protocol, coordinating message
delivery from sender to recipient(s) using the routing engine and envelope
validation.

Author: Mohamed
"""

import asyncio
import logging
import uuid
from datetime import UTC, datetime

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.envelope import EnvelopeHandler
from orchestrata_ultimate.postman.exceptions import (
    DeliveryError,
    QueueError,
    RecipientNotFoundError,
)
from orchestrata_ultimate.postman.interfaces import (
    IDeliveryManager,
    IMessageEnvelope,
    IQueue,
    IQueueManager,
    IRoutingEngine,
)
from orchestrata_ultimate.postman.queue_manager import QueueManager
from orchestrata_ultimate.postman.router import RoutingEngine
from orchestrata_ultimate.postman.types import (
    DeliveryMode,
    DeliveryReceipt,
    DeliveryResult,
    DeliveryStatus,
    MessageEnvelope,
    MessagePayload,
    QueueConfig,
    QueueEntry,
    ReceiptType,
)

logger = logging.getLogger(__name__)


@traces(
    spec="002",
    story="US-001",
    reqs=[
        "REQ-002-US-001-002",
        "REQ-002-US-001-003",
        "REQ-002-US-001-004",
        "REQ-002-US-001-005",
    ],
)
class DeliveryManager(IDeliveryManager):
    """
    Manages message delivery lifecycle from sender to recipient(s).

    Orchestrates the complete delivery process:
    1. Validates envelope structure and headers
    2. Resolves recipients via routing engine
    3. Delivers message to each recipient
    4. Tracks delivery status and generates receipts
    5. Handles retries and cancellations

    Thread-safe for concurrent delivery operations.
    """

    def __init__(
        self,
        routing_engine: IRoutingEngine | None = None,
        envelope_handler: IMessageEnvelope | None = None,
        queue_manager: IQueueManager | None = None,
    ) -> None:
        """
        Initialize delivery manager with required dependencies.

        Args:
            routing_engine: Engine for resolving message recipients. Defaults
                to a fresh :class:`~orchestrata_ultimate.postman.router.RoutingEngine`
                (SPEC-002 FR-003) when not supplied.
            envelope_handler: Handler for envelope validation. Defaults to a
                fresh :class:`~orchestrata_ultimate.postman.envelope.EnvelopeHandler`
                (SPEC-002 FR-001) when not supplied.
            queue_manager: Manager providing each recipient's real delivery
                queue (SPEC-002 FR-004) -- the actual transport
                ``_deliver_to_recipient`` routes messages through. Defaults to
                a fresh :class:`~orchestrata_ultimate.postman.queue_manager.QueueManager`
                when not supplied.
        """
        self._routing_engine: IRoutingEngine = routing_engine or RoutingEngine()
        self._envelope_handler: IMessageEnvelope = envelope_handler or EnvelopeHandler()
        self._queue_manager: IQueueManager = queue_manager or QueueManager()

        # Delivery tracking state. Pending holds the original (envelope, payload)
        # so a failed delivery can be retried with the real payload (not a
        # placeholder). Cleared only on successful delivery.
        self._deliveries: dict[str, DeliveryResult] = {}
        self._pending: dict[str, tuple[MessageEnvelope, MessagePayload]] = {}

        # Thread safety
        self._lock = asyncio.Lock()

    @traces(
        spec="002",
        story="US-001",
        reqs=["REQ-002-US-001-002", "REQ-002-US-001-003", "REQ-002-US-001-005"],
    )
    async def deliver(
        self,
        envelope: MessageEnvelope,
        payload: MessagePayload,
        mode: DeliveryMode = DeliveryMode.AT_LEAST_ONCE,
    ) -> DeliveryResult:
        """
        Deliver message to recipients according to envelope routing.

        Process:
        1. Validate envelope structure
        2. Resolve recipients from TO/CC/BCC headers
        3. Attempt delivery to each recipient
        4. Generate delivery receipts
        5. Track delivery status

        Args:
            envelope: Message envelope with routing headers
            payload: Message payload to deliver
            mode: Delivery guarantee mode (default: AT_LEAST_ONCE)

        Returns:
            DeliveryResult with success status, receipts, and any errors

        Raises:
            DeliveryError: If critical delivery error occurs
        """
        message_id = self._extract_message_id(envelope)
        timestamp = datetime.now(UTC)

        # Step 1: Validate envelope
        validation_result = self._validate_envelope(envelope, message_id, timestamp)
        if validation_result is not None:
            return validation_result

        # Step 2: Queue message and update status
        await self._queue_message(message_id, envelope, payload, timestamp)

        # Step 3: Resolve recipients
        recipients_result = await self._resolve_recipients(message_id, envelope)
        if isinstance(recipients_result, DeliveryResult):
            return recipients_result
        recipients = recipients_result

        # Step 4: Update status to DELIVERING
        async with self._lock:
            delivering_result = DeliveryResult(
                message_id=message_id,
                timestamp=datetime.now(UTC),
                success=False,
                status=DeliveryStatus.DELIVERING,
                receipts=[],
            )
            self._deliveries[message_id] = delivering_result

        # Step 5: Deliver to each recipient and generate final result
        final_result = await self._deliver_to_all(
            message_id, recipients, envelope, payload
        )

        # Step 6: Update tracking. Keep the pending (envelope, payload) when the
        # delivery failed so retry() can re-deliver the real payload; clear it
        # only on success.
        async with self._lock:
            self._deliveries[message_id] = final_result
            if final_result.success:
                self._pending.pop(message_id, None)

        return final_result

    @traces(spec="002", story="US-001", reqs=["REQ-002-US-001-005"])
    def _extract_message_id(self, envelope: MessageEnvelope) -> str:
        """Extract or generate message ID from envelope.

        Uses ``getattr`` rather than direct attribute access: real
        ``MessageEnvelope`` instances never carry a ``headers`` dict (it is
        not a declared field), but some duck-typed callers still supply one.
        """
        headers = getattr(envelope, "headers", None)
        if headers:
            return headers.get("MESSAGE_ID", str(uuid.uuid4()))
        elif hasattr(envelope, "message_id") and envelope.message_id:
            return envelope.message_id
        return str(uuid.uuid4())

    @traces(spec="002", story="FR-001", reqs=["REQ-002-FR1-014"])
    def _validate_envelope(
        self,
        envelope: MessageEnvelope,
        message_id: str,
        timestamp: datetime,
    ) -> DeliveryResult | None:
        """Validate envelope structure. Returns DeliveryResult on failure, None on success."""
        try:
            is_valid, validation_error = self._envelope_handler.validate(envelope)
            if not is_valid:
                return DeliveryResult(
                    message_id=message_id,
                    timestamp=timestamp,
                    success=False,
                    status=DeliveryStatus.FAILED,
                    receipts=[],
                    error=f"Envelope validation failed: {validation_error}",
                )
        except Exception as e:
            return DeliveryResult(
                message_id=message_id,
                timestamp=timestamp,
                success=False,
                status=DeliveryStatus.FAILED,
                receipts=[],
                error=f"Validation error: {str(e)}",
            )
        return None

    @traces(spec="002", story="US-001", reqs=["REQ-002-US-001-005"])
    async def _queue_message(
        self,
        message_id: str,
        envelope: MessageEnvelope,
        payload: MessagePayload,
        timestamp: datetime,
    ) -> None:
        """Queue message and record initial QUEUED status."""
        async with self._lock:
            self._pending[message_id] = (envelope, payload)
            self._deliveries[message_id] = DeliveryResult(
                message_id=message_id,
                timestamp=timestamp,
                success=False,
                status=DeliveryStatus.QUEUED,
                receipts=[],
            )

    @traces(
        spec="002",
        story="US-001/FR-003",
        reqs=["REQ-002-US-001-002", "REQ-002-FR3-010"],
    )
    async def _resolve_recipients(
        self,
        message_id: str,
        envelope: MessageEnvelope,
    ) -> list[str] | DeliveryResult:
        """Resolve recipients via routing engine.

        Returns list of recipients on success, or DeliveryResult on failure.
        """
        try:
            recipients = await self._routing_engine.resolve(envelope)
            if not recipients:
                return await self._record_failed(message_id, "No recipients found")
            return recipients
        except RecipientNotFoundError as e:
            return await self._record_failed(
                message_id, f"Recipient resolution failed: {str(e)}"
            )
        except Exception as e:
            return await self._record_failed(message_id, f"Routing error: {str(e)}")

    @traces(spec="002", story="US-001", reqs=["REQ-002-US-001-005"])
    async def _record_failed(self, message_id: str, error_msg: str) -> DeliveryResult:
        """Record a FAILED result and remove from pending."""
        async with self._lock:
            failed_result = DeliveryResult(
                message_id=message_id,
                timestamp=datetime.now(UTC),
                success=False,
                status=DeliveryStatus.FAILED,
                receipts=[],
                error=error_msg,
            )
            self._deliveries[message_id] = failed_result
            # Keep the pending (envelope, payload) so retry() can re-deliver.
        return failed_result

    @traces(spec="002", story="US-001", reqs=["REQ-002-US-001-002"])
    async def _deliver_to_all(
        self,
        message_id: str,
        recipients: list[str],
        envelope: MessageEnvelope,
        payload: MessagePayload,
    ) -> DeliveryResult:
        """Deliver to all recipients and return final aggregated result."""
        receipts: list[DeliveryReceipt] = []
        all_success = True

        for recipient in recipients:
            receipt = await self._deliver_to_recipient(
                message_id=message_id,
                recipient=recipient,
                envelope=envelope,
                payload=payload,
            )
            receipts.append(receipt)
            if receipt.status != DeliveryStatus.DELIVERED:
                all_success = False

        final_status = (
            DeliveryStatus.DELIVERED if all_success else DeliveryStatus.FAILED
        )
        return DeliveryResult(
            message_id=message_id,
            timestamp=datetime.now(UTC),
            success=all_success,
            status=final_status,
            receipts=receipts,
            error=None if all_success else "Some deliveries failed",
        )

    @traces(
        spec="002",
        story="FR-004",
        reqs=["REQ-002-FR4-001"],
    )
    async def _get_or_create_recipient_queue(self, recipient: str) -> IQueue:
        """Get (or lazily create) the recipient's real delivery queue.

        This is the actual transport ``_deliver_to_recipient`` routes messages
        through (SPEC-002 FR-004): an in-memory queue keyed by recipient,
        created on first use via the composed ``IQueueManager``. A concurrent
        first delivery to the same recipient can race with queue creation;
        losing that race just falls back to the queue the winner created.

        Args:
            recipient: Recipient agent ID the queue belongs to.

        Returns:
            The recipient's real ``IQueue`` instance.
        """
        queue = await self._queue_manager.get_queue(recipient)
        if queue is not None:
            return queue

        try:
            return await self._queue_manager.create_queue(
                recipient, QueueConfig(name=recipient, persistence_enabled=False)
            )
        except QueueError:
            # Lost a race with a concurrent delivery to the same recipient.
            existing = await self._queue_manager.get_queue(recipient)
            if existing is None:
                raise
            return existing

    @traces(
        spec="002",
        story="FR-004/FR-007",
        reqs=["REQ-002-FR4-006", "REQ-002-FR7-002", "REQ-002-FR7-003"],
    )
    async def _deliver_to_recipient(
        self,
        message_id: str,
        recipient: str,
        envelope: MessageEnvelope,
        payload: MessagePayload,
    ) -> DeliveryReceipt:
        """
        Deliver message to a single recipient.

        Actually enqueues the message onto the recipient's real delivery
        queue (SPEC-002 FR-004, via the composed ``IQueueManager``) -- the
        transport a real subscriber dequeues from. A receipt is only ever
        DELIVERED once the queue confirms the entry was accepted; any
        transport failure (queue full, queue unavailable, ...) is reported
        FAILED so callers never see a false DELIVERED for a message that
        never arrived.

        Args:
            message_id: Unique message identifier
            recipient: Recipient address (e.g., "U:Cd", "U:PM")
            envelope: Message envelope
            payload: Message payload

        Returns:
            DeliveryReceipt with delivery status and metadata
        """
        try:
            queue = await self._get_or_create_recipient_queue(recipient)
            entry = QueueEntry(message_id=message_id, priority=envelope.priority)
            await queue.enqueue(entry)

            # Generate successful receipt -- only once the entry actually
            # landed in the recipient's real queue.
            return DeliveryReceipt(
                receipt_id=str(uuid.uuid4()),
                message_id=message_id,
                recipient=recipient,
                timestamp=datetime.now(UTC),
                status=DeliveryStatus.DELIVERED,
                receipt_type=ReceiptType.DELIVERY,
            )

        except Exception as e:
            # Real transport failure (queue full, queue unreachable, ...) --
            # report FAILED, never a false DELIVERED.
            logger.warning(f"Delivery to recipient {recipient} failed: {e}")
            return DeliveryReceipt(
                receipt_id=str(uuid.uuid4()),
                message_id=message_id,
                recipient=recipient,
                timestamp=datetime.now(UTC),
                status=DeliveryStatus.FAILED,
                receipt_type=ReceiptType.DELIVERY,
                error=str(e),
            )

    @traces(spec="002", story="US-001", reqs=["REQ-002-US-001-005"])
    async def get_status(self, message_id: str) -> DeliveryResult | None:
        """
        Get current delivery status for a message.

        Args:
            message_id: Unique message identifier

        Returns:
            DeliveryResult if message found, None otherwise
        """
        async with self._lock:
            return self._deliveries.get(message_id)

    async def cancel(self, message_id: str) -> bool:
        """
        Cancel pending message delivery.

        Only cancels messages in QUEUED status. Messages already
        in DELIVERING or DELIVERED status cannot be cancelled.

        Args:
            message_id: Unique message identifier

        Returns:
            True if cancelled successfully, False otherwise
        """
        async with self._lock:
            delivery = self._deliveries.get(message_id)
            if not delivery:
                return False

            # Can only cancel queued messages
            if delivery.status != DeliveryStatus.QUEUED:
                return False

            # Update status to FAILED with cancellation note
            cancelled_result = DeliveryResult(
                message_id=message_id,
                timestamp=datetime.now(UTC),
                success=False,
                status=DeliveryStatus.FAILED,
                receipts=[],
                error="Cancelled by sender",
            )
            self._deliveries[message_id] = cancelled_result
            self._pending.pop(message_id, None)

            return True

    @traces(spec="002", story="US-001", reqs=["REQ-002-US-001-004"])
    async def retry(self, message_id: str) -> DeliveryResult:
        """
        Retry failed message delivery.

        Re-attempts delivery for failed messages using the original
        envelope and payload (if still available).

        Args:
            message_id: Unique message identifier

        Returns:
            DeliveryResult with retry outcome

        Raises:
            DeliveryError: If message not found or not retryable
        """
        async with self._lock:
            delivery = self._deliveries.get(message_id)
            pending = self._pending.get(message_id)

        if not delivery:
            raise DeliveryError(f"Message {message_id} not found")

        if delivery.status != DeliveryStatus.FAILED:
            raise DeliveryError(
                f"Message {message_id} is not in failed status (current: {delivery.status})"
            )

        if not pending:
            raise DeliveryError(
                f"Original envelope/payload for message {message_id} no longer available"
            )

        # Retry delivery with the original, persisted envelope AND payload.
        envelope, payload = pending
        return await self.deliver(envelope, payload)